merge: continuous concurrent gate speedup (algo #6 semantic revision, design v3)

.env.example

@@ -47,7 +47,7 @@ LLM_TTFT_TIMEOUT=30
 LLM_INTER_TOKEN_TIMEOUT=15
 LLM_RETRY_MAX_DELAY=30.0
 # 正整数秒，禁止 0（0 会被拒绝启动）；训练场景建议 >= 单次训练时长
-REDIS_CACHE_TTL=86400
+REDIS_CACHE_TTL=604800
 
 # 建树批量并行：全局 VLM/LLM 在途调用上限（Spec-2 工程配置）
 TREE_BUILD_API_CONCURRENCY=16

app/harness/checkpoint.py

@@ -58,7 +58,6 @@ _DECISION_KEYS = (
     "gate_delta_min",
     "gate_lambda_dir",
     "gate_e_rollback",
-    "gate_block",
     "gate_n_max",
     "gate_p_low",
     "gate_p_high",

app/harness/config.py

@@ -70,14 +70,13 @@ class RunConfig:
         gate_delta_min: 最小点估计效应量下限（承接旧 margin 语义）。
         gate_lambda_dir: Wald 方向拒绝的对数似然比阈值（必须为负）。
         gate_e_rollback: 试用期对称回滚门（回滚 e 值门槛）。
-        gate_block: 块序贯验证的块大小（=推理并发度，块内跑满）。
         gate_n_max: 单次 gate 消耗的题数上限。
         gate_p_low: 信息量阶梯 p-hat 保留区间下界（剔除必错零信息题）。
         gate_p_high: 信息量阶梯 p-hat 保留区间上界（剔除必对零信息题）。
         gate_probe_quota: 冷启动探针集比例（全错题中插尾的比例）。
         gate_gamma_decay: 逐题正确率估计 p-hat 的 EMA 衰减系数。
         gate_cooldown_steps: 回滚后该题型跳过进化的冷却 step 数。
-        gate_guard_err: gate 内跨块累计 INFRA 错误率护栏。
+        gate_guard_err: gate 内累计 INFRA 错误率护栏。
         skill_update_mode: skill 进化模式，"patch"（局部 edit）/ "rewrite"（整篇重写）。
         appendix_consolidate_threshold: appendix note 条数达此值触发 LLM consolidation。
         run_id: diagnose/evolve 模式要分析的运行 ID，默认空字符串。
@@ -125,7 +124,6 @@ class RunConfig:
     gate_delta_min: float
     gate_lambda_dir: float
     gate_e_rollback: float
-    gate_block: int
     gate_n_max: int
     gate_p_low: float
     gate_p_high: float
@@ -361,7 +359,7 @@ def _validate_gate_thresholds(config: RunConfig) -> None:
 
 
 def _validate_gate_ladder(config: RunConfig) -> None:
-    """校验 CE-Gate 信息量阶梯与块序贯参数。
+    """校验 CE-Gate 信息量阶梯参数。
 
     参数:
         config: 待校验的配置实例。
@@ -369,11 +367,8 @@ def _validate_gate_ladder(config: RunConfig) -> None:
     异常:
         ValueError: 任一阶梯参数不合法。
     """
-    if config.gate_block <= 0 or config.gate_n_max < config.gate_block:
-        raise ValueError(
-            f"需 0 < gate_block <= gate_n_max，"
-            f"实际: block={config.gate_block}, n_max={config.gate_n_max}"
-        )
+    if config.gate_n_max <= 0:
+        raise ValueError(f"需 gate_n_max > 0，实际: n_max={config.gate_n_max}")
     if not (0 <= config.gate_p_low < config.gate_p_high <= 1):
         raise ValueError(
             f"需 0 <= gate_p_low < gate_p_high <= 1，"

app/harness/inference.py

@@ -409,7 +409,11 @@ async def _run_single_question(
     返回:
         预测结果字典（含 video_id, question_id, prediction, answer 等）。
     """
+    # run_id 必须显式入 record:HarnessLog.insert 缺省用**实例** run_id 填充,
+    # 连续并发 gate 共享单一 gate_log(实例 run_id 为 step 级)时,各臂行必须
+    # 落自己的臂 run_id,否则 validate 回读 _load_run_rows(臂 run_id) 为空。
     record: dict[str, Any] = {
+        "run_id": run_id,
         "video_id": qa.video_id,
         "question_id": qa.question_id,
         "task_type": qa.task_type,

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,12 +341,16 @@ 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 不上。
 
+    返回:
+        无。
+
     关键实现:
         逐行 insert（非 insert_many），保证每行独立事务。
     """
@@ -354,6 +358,12 @@ def write_gate_evidence(
 
     with HarnessLog(db_path, run_id) as log:
         log.create_table("gate_evidence", _GATE_EVIDENCE_COLS)
+        # 幂等迁移(对齐 question_gen/run_store 先例):块序贯时代的旧表只有
+        # block_idx 列,CREATE TABLE IF NOT EXISTS 不补列,直接插 ladder_rank
+        # 会 OperationalError——为旧 workspace 复用补列,新表恒为 no-op。
+        cols = {r["name"] for r in log.query("PRAGMA table_info(gate_evidence)")}
+        if "ladder_rank" not in cols:
+            log.execute("ALTER TABLE gate_evidence ADD COLUMN ladder_rank INTEGER")
         for row in rows:
             log.insert("gate_evidence", {"epoch": epoch, "step": step, **row})
 

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,95 @@ 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 模式中的全部特殊字符(`\\`、`%`、`_`)为字面匹配。
+
+    参数:
+        text: 待作为 LIKE 前缀字面使用的原始字符串。
+
+    返回:
+        可安全拼入 `LIKE ? ESCAPE '\\'` 模式的转义串。
+
+    关键实现细节:
+        反斜杠必须最先转义,否则会二次转义后续替换产生的转义符。
+    """
+    return text.replace("\\", "\\\\").replace("%", r"\%").replace("_", r"\_")
+
+
+def _clear_step_rows(db_path: str, *, baseline_run_id: str, epoch: int, step: int) -> None:
+    """清空一个 step 的全部旧行（rollout + gate 派生），保证崩溃重跑幂等。
+
+    修复前序潜伏 bug：旧实现只清 rollout run_id，gate 派生 run_id
+    （`{step_run_id}_gate_%`）从不清理，重跑会累积重复 predictions（HarnessLog
+    无主键去重），_load_run_rows 的 dict 覆盖使结果依赖 SELECT 顺序。
+    gate_evidence / quadrant_pair 以 (run_id, epoch, step) 过滤删除；
+    表不存在（首个 step）时跳过。step_report 为按文件名覆盖写的 JSON，天然幂等。
+
+    参数:
+        db_path: harness.db 路径。
+        baseline_run_id: 基线 run（gate_evidence/quadrant_pair 的 run_id 维度）。
+        epoch: 轮次（1-based）。
+        step: epoch 内 step 序号（0-based）。
+
+    返回:
+        无。
+
+    关键实现细节:
+        predictions/traces 的 gate 行按 LIKE 前缀删除，`\\`/`%`/`_` 三个 LIKE
+        特殊字符全部显式转义（ESCAPE）钉死字面匹配，避免 `..._s1` 误匹配
+        `..._s10` 类前缀陷阱，也防 run_id 含 `%`/`\\` 时通配误删他 run 行。
+    """
+    from app.harness.inference import PREDICTIONS_SCHEMA, TRACES_SCHEMA
+    from app.harness.log import HarnessLog
+
+    step_run_id = f"{baseline_run_id}_e{epoch}_s{step}"
+    escaped = _escape_sql_like(step_run_id)
+    with HarnessLog(db_path, step_run_id, register_run=False) as log:
+        log.create_table("predictions", PREDICTIONS_SCHEMA)
+        log.create_table("traces", TRACES_SCHEMA)
+        for table in ("predictions", "traces"):
+            log.execute(f"DELETE FROM {table} WHERE run_id=?", (step_run_id,))
+            log.execute(
+                f"DELETE FROM {table} WHERE run_id LIKE ? ESCAPE '\\'",
+                (escaped + r"\_gate\_%",),
+            )
+        for table in ("gate_evidence", "quadrant_pair"):
+            exists = log.query(
+                "SELECT name FROM sqlite_master WHERE type='table' AND name=?", (table,)
+            )
+            if exists:
+                log.execute(
+                    f"DELETE FROM {table} WHERE run_id=? AND epoch=? AND step=?",
+                    (baseline_run_id, epoch, step),
+                )
+
+
 # ---------------------------------------------------------------------------
 # Runner 主类
 # ---------------------------------------------------------------------------
@@ -1096,17 +1193,16 @@ class Runner:
         """单 step：rollout → correctness 增量 → 诊断 → 累加 system/tool → 按类 gate。"""
         run_id = f"{pools.baseline_run_id}_e{epoch}_s{step}"
 
-        from app.harness.inference import PREDICTIONS_SCHEMA, TRACES_SCHEMA
         from app.harness.log import HarnessLog
 
-        # 幂等：重跑同一 step 前先清旧行，避免断点续跑重复累计双计。
-        # 先 CREATE TABLE IF NOT EXISTS（fresh workspace 首跑时表尚未由 run_inference 建），
-        # register_run=False 避免只读清理污染 _runs 运行状态。
-        with HarnessLog(str(self._paths.db_path), run_id, register_run=False) as log:
-            log.create_table("predictions", PREDICTIONS_SCHEMA)
-            log.create_table("traces", TRACES_SCHEMA)
-            log.execute("DELETE FROM predictions WHERE run_id=?", (run_id,))
-            log.execute("DELETE FROM traces WHERE run_id=?", (run_id,))
+        # 幂等：重跑同一 step 前清 rollout + 全部 gate 派生旧行（修复潜伏 bug：
+        # 旧实现只清 rollout，gate 行崩溃重跑会累积重复）。
+        _clear_step_rows(
+            str(self._paths.db_path),
+            baseline_run_id=pools.baseline_run_id,
+            epoch=epoch,
+            step=step,
+        )
 
         await self._rollout_batch(batch, run_id)
 
@@ -1137,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(
@@ -1149,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,
@@ -1175,22 +1348,44 @@ 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, []),
             )
-            # 进化未产出真实改动
+
+        # 首异常先取消其余进化任务并排水再向上传播(与 validate_skills_concurrent
+        # 同款语义):避免失败后残留 in-flight LLM 任务与 pending task 警告。
+        tasks = [asyncio.ensure_future(_evolve_one(t)) for t in active_types]
+        try:
+            evolved = await asyncio.gather(*tasks)
+        except BaseException:
+            for task in tasks:
+                task.cancel()
+            await asyncio.gather(*tasks, return_exceptions=True)
+            raise
+        records = dict(zip(active_types, evolved, 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
             ):
@@ -1208,11 +1403,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,
@@ -1251,88 +1542,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
     # -----------------------------------------------------------------------
@@ -2413,10 +2622,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],
@@ -2424,8 +2646,9 @@ class Runner:
             run_id: str,
             skills_dir: Path,
         ) -> InferenceResult:
+            if run_id not in recorded:
+                recorded.add(run_id)
                 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,
@@ -2433,7 +2656,7 @@ class Runner:
                 prompt_builder=self._make_prompt_builder(
                     skills_dir=skills_dir, prompts_dir=self._paths.prompts_dir
                 ),
-                    log=log,
+                log=gate_log,
                 run_id=run_id,
                 concurrency=self._config.concurrency,
                 max_steps=self._config.max_steps,

config/default.yaml

@@ -42,7 +42,6 @@ harness:
   gate_delta_min: 0.02
   gate_lambda_dir: -0.642
   gate_e_rollback: 10.0
-  gate_block: 8
   gate_n_max: 40
   gate_p_low: 0.05
   gate_p_high: 0.95

config/question_gen_180_补.yaml

@@ -39,7 +39,6 @@ harness:
   gate_delta_min: 0.02
   gate_lambda_dir: -0.642
   gate_e_rollback: 10.0
-  gate_block: 8
   gate_n_max: 40
   gate_p_low: 0.05
   gate_p_high: 0.95

config/question_gen_360.yaml

@@ -42,7 +42,6 @@ harness:
   gate_delta_min: 0.02
   gate_lambda_dir: -0.642
   gate_e_rollback: 10.0
-  gate_block: 8
   gate_n_max: 40
   gate_p_low: 0.05
   gate_p_high: 0.95

config/train_action_recognition.yaml

@@ -22,7 +22,6 @@ harness:
   gate_delta_min: 0.02
   gate_lambda_dir: -0.642
   gate_e_rollback: 10.0
-  gate_block: 8
   gate_n_max: 40
   gate_p_low: 0.05
   gate_p_high: 0.95

config/train_ar30.yaml

@@ -23,7 +23,6 @@ harness:
   gate_delta_min: 0.02
   gate_lambda_dir: -0.642
   gate_e_rollback: 10.0
-  gate_block: 8
   gate_n_max: 40
   gate_p_low: 0.05
   gate_p_high: 0.95

config/train_videomme.yaml

@@ -10,8 +10,8 @@ harness:
   workspace_dir: "workspaces/train-videomme"
   store_dir: store
   mode: train
-  run_id: train_videomme_v1
-  concurrency: 24
+  run_id: train_videomme_v2
+  concurrency: 32
   max_steps: 40
   skill_mode: auto
   n_samples: 0
@@ -26,7 +26,6 @@ harness:
   gate_delta_min: 0.02
   gate_lambda_dir: -0.642
   gate_e_rollback: 10.0
-  gate_block: 8
   gate_n_max: 40
   gate_p_low: 0.05
   gate_p_high: 0.95
@@ -51,8 +50,10 @@ harness:
   # 可训练性预检（WP3）：val 单元 < eval_min_per_class 或 非test单元 < trainable_min_units 的题型剔除
   eval_min_per_class: 2
   trainable_min_units: 8
-  # mini-batch
-  batch_size: 10
+  # mini-batch —— 对齐 TRM4 正式实验 batch=40（sh --batch-size 40 覆盖 yaml 15 的最终生效值）：
+  # 8 可训题型 × 每型约 5 题/step，保住题型级诊断信号；同时 steps/epoch 180/40≈5，
+  # 进化/gate 验证轮数比 batch=10 少 4 倍。
+  batch_size: 40
   min_class_per_batch: 2
   batch_correct_ratio: 0.5
   momentum_samples: 20

tests/integration/test_checkpoint_pair.py

@@ -125,7 +125,6 @@ class _FakeConfig:
     gate_delta_min: float = 0.02
     gate_lambda_dir: float = -3.0
     gate_e_rollback: float = 10.0
-    gate_block: int = 4
     gate_n_max: int = 40
     gate_p_low: float = 0.1
     gate_p_high: float = 0.9

tests/integration/test_v3_contract_e2e.py

@@ -338,7 +338,7 @@ class TestInferenceUnitAggregationEndToEnd:
 
     def _assert_all_persisted(self, log: HarnessLog, questions: list[GeneratedQuestion]) -> None:
         """逐题溯源保留：含被剔除的孤儿题在内，每题仍逐题落 predictions。"""
-        rows = log.query("SELECT * FROM predictions WHERE run_id = ?", ("test-run",))
+        rows = log.query("SELECT * FROM predictions WHERE run_id = ?", ("run-v3-contract",))
         persisted = {r["question_id"] for r in rows}
         assert "orphan_o" in persisted, "孤儿题未逐题落库（逐题溯源被破坏）"
         assert persisted == {q.question_id for q in questions}, "逐题落库题数与输入不符"

tests/unit/test_gate_batch_parallel.py

@@ -0,0 +1,300 @@
+"""_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"])
+
+
+# ---------------------------------------------------------------------------
+# 共享 gate_log 的 run_id 契约(真 SQLite,Codex 质量审 C1):
+# HarnessLog.insert 缺省用实例 run_id 填充;record 自带 run_id 必须覆盖它,
+# 否则连续并发 gate 下所有臂的 predictions 会落成 step 级 run_id,
+# validate 按臂 run_id 回读为空 → gate 静默废掉。
+# ---------------------------------------------------------------------------
+
+
+def test_harness_log_insert_record_run_id_overrides_instance(tmp_path: Path) -> None:
+    """record 自带 run_id 覆盖实例 run_id;缺省时回落实例 run_id(锁死 enriched.update 语义)。"""
+    from app.harness.inference import PREDICTIONS_SCHEMA
+    from app.harness.log import HarnessLog
+
+    with HarnessLog(str(tmp_path / "harness.db"), "gate_e1_s0") as log:
+        log.create_table("predictions", PREDICTIONS_SCHEMA)
+        log.insert(
+            "predictions",
+            {"run_id": "run_e1_s0_gate_a_base_u0", "question_id": "q1", "prediction": "A"},
+        )
+        log.insert("predictions", {"question_id": "q2", "prediction": "B"})
+        rows = log.query("SELECT question_id, run_id FROM predictions ORDER BY question_id")
+    assert [(r["question_id"], r["run_id"]) for r in rows] == [
+        ("q1", "run_e1_s0_gate_a_base_u0"),
+        ("q2", "gate_e1_s0"),
+    ]
+
+
+def test_inference_prediction_row_carries_arm_run_id(tmp_path: Path) -> None:
+    """经共享 gate_log 落库的 prediction 行 run_id 必须是臂 run_id 而非实例 run_id。
+
+    prompt_builder 抛错走异常路径即落库,无需真实 LLM;
+    该路径与成功路径共用同一 record 初始 dict,契约一致。
+    """
+    from app.harness.inference import PREDICTIONS_SCHEMA, _run_single_question
+    from app.harness.log import HarnessLog
+
+    def _broken_prompt_builder(qa: GeneratedQuestion) -> tuple[str, str]:
+        raise RuntimeError("测试注入:跳过真实推理")
+
+    async def _noop_dispatch(tool_name: str, args: dict, *, context: dict) -> str:
+        raise NotImplementedError
+
+    with HarnessLog(str(tmp_path / "harness.db"), "gate_e1_s0") as gate_log:
+        gate_log.create_table("predictions", PREDICTIONS_SCHEMA)
+        asyncio.run(
+            _run_single_question(
+                _question("q-arm", _TYPE_A),
+                llm=object(),  # prompt_builder 先抛错,不会触达
+                tool_dispatch_fn=_noop_dispatch,
+                prompt_builder=_broken_prompt_builder,
+                log=gate_log,
+                max_steps=3,
+                plugins=[],
+                run_id="run_e1_s0_gate_action-reasoning_cand_u0",
+            )
+        )
+        rows = gate_log.query("SELECT run_id, stop_reason FROM predictions")
+    assert len(rows) == 1
+    assert rows[0]["run_id"] == "run_e1_s0_gate_action-reasoning_cand_u0"
+    assert rows[0]["stop_reason"] == "error"

tests/unit/test_gate_concurrent.py

@@ -0,0 +1,254 @@
+"""连续并发 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
+
+
+class _FakeInferenceResult:
+    """推理结果桩:只承载编排器消费的 run_id 与 total 两个字段。"""
+
+    def __init__(self, run_id: str, total: int) -> None:
+        self.run_id = run_id
+        self.total = total
+
+
+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+臂 决定 (对错, 延迟秒),制造乱序到达。"""
+
+    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 _FakeInferenceResult(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()
+
+    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 _FakeInferenceResult(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,
+        )
+
+
+@pytest.mark.asyncio
+async def test_partial_materialize_failure_cleans_up(tmp_path, monkeypatch) -> None:
+    """第 2 个题型物化失败:OSError 传播,且第 1 个已物化目录被清理不泄漏。"""
+    spec_a = _mk_spec("Action Reasoning", "action-reasoning", 1)
+    spec_b = _mk_spec("Counting Problem", "counting-problem", 1)
+    made: list[Path] = []
+
+    def _mat(workspace_dir, base_skills_version, target_file, content):
+        if made:  # 第 2 次调用:模拟磁盘错误
+            raise OSError("第 2 个题型物化失败(模拟)")
+        d = tmp_path / "cand_a"
+        d.mkdir()
+        made.append(d)
+        return d
+
+    monkeypatch.setattr("app.harness.validate.materialize_candidate_skill", _mat)
+
+    async def _never_called(questions, *, run_id, skills_dir):
+        raise AssertionError("物化失败后不应发起任何推理")
+
+    with pytest.raises(OSError):
+        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=_never_called,
+            log=_FakeLog(),
+            concurrency=8,
+        )
+    assert len(made) == 1
+    assert not made[0].exists()
+
+
+@pytest.mark.asyncio
+async def test_guard_raise_cancels_remaining_tasks(tmp_path, monkeypatch) -> None:
+    """护栏 raise 后其余在飞任务被取消收束:整体在超时内返回,不悬挂。
+
+    A 型 12 单元推理全 INFRA(stop_reason="error"),分母 ≥10 后错误率 1.0
+    超护栏 0.01 → RuntimeError;B 型推理挂在永不 set 的 Event 上,若无
+    取消收束,validate 将悬挂,wait_for 超时即为回归。
+    """
+    spec_a = _mk_spec("Action Reasoning", "action-reasoning", 12)
+    spec_b = _mk_spec("Counting Problem", "counting-problem", 2)
+    log = _FakeLog()
+    hang = asyncio.Event()  # 永不 set:B 型推理只能靠取消收束
+
+    async def _run(questions, *, run_id, skills_dir):
+        if "counting-problem" in run_id:
+            await hang.wait()
+        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 _FakeInferenceResult(run_id, len(questions))
+
+    monkeypatch.setattr(
+        "app.harness.validate.materialize_candidate_skill",
+        lambda *a, **k: tmp_path / "cand",
+    )
+    with pytest.raises(RuntimeError):
+        await asyncio.wait_for(
+            validate_skills_concurrent(
+                workspace_dir=tmp_path,
+                base_skills_version="v1",
+                specs=[spec_a, spec_b],
+                gate_params=_PARAMS,
+                gate_guard_err=0.01,
+                baseline_cache=BaselineCache(tmp_path / "bc.json"),
+                prompts_version="v1",
+                run_inference=_run,
+                log=log,
+                concurrency=8,
+            ),
+            timeout=5,
+        )

tests/unit/test_gate_prefix.py

@@ -0,0 +1,193 @@
+"""连续并发 gate 的前缀消费纯逻辑测试。"""
+
+from __future__ import annotations
+
+from app.harness.validate import GateSpec, _advance_prefix, _GateRun
+from core.evolution import GateParams
+from core.types import GeneratedQuestion, QuestionUnit
+
+
+def _mk_question(qid: str, task_type: str = "Action Reasoning") -> GeneratedQuestion:
+    """构造最小可用的 single 题（字段以 core.types 真实定义为准，缺省值从简）。"""
+    return GeneratedQuestion(
+        question_id=qid,
+        video_id="v1",
+        task_type=task_type,
+        question=f"q-{qid}",
+        options=("A. x", "B. y"),
+        answer="A",
+        source_nodes=(),
+        difficulty="easy",
+    )
+
+
+def _mk_unit(qid: str, task_type: str = "Action Reasoning") -> QuestionUnit:
+    """由单条题目构造 single 单元（unit_id 回填为 question_id）。"""
+    return QuestionUnit.from_single(_mk_question(qid, task_type))
+
+
+def _mk_run(n_units: int) -> _GateRun:
+    """构造含 n_units 个 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_units)),
+        gate_run_prefix="r_e1_s0_gate_action-reasoning",
+    )
+    return _GateRun.from_spec(spec)
+
+
+_PARAMS = GateParams(
+    e_confirm=20.0,
+    e_provisional=3.0,
+    w_net_min=2,
+    delta_min=0.02,
+    lambda_dir=-0.642,
+    e_rollback=10.0,
+)
+
+
+def test_prefix_blocks_on_unresolved_head() -> None:
+    """阶梯头部单元未配齐时,即使尾部全部配齐也一个都不消费。"""
+    run = _mk_run(4)
+    for i in (1, 2, 3):  # 尾部三个先到
+        run.slots[i].base = False
+        run.slots[i].cand_per_q = {f"q{i}": True}
+    _advance_prefix(run, _PARAMS)
+    assert run.n_used == 0 and run.w == 0 and run.verdict is None
+
+
+def test_prefix_consumes_in_ladder_order_after_head_arrives() -> None:
+    """头部补齐后一次性顺序消费到最长已配齐前缀。"""
+    run = _mk_run(4)
+    for i in (0, 1, 2):
+        run.slots[i].base = False
+        run.slots[i].cand_per_q = {f"q{i}": True}
+    _advance_prefix(run, _PARAMS)
+    assert run.n_used == 3 and run.w == 3 and run.l == 0
+    assert [r["ladder_rank"] for r in run.evidence_rows] == [0, 1, 2]
+
+
+def test_freeze_on_terminal_verdict_stops_consumption() -> None:
+    """过线即冻结,后续已配齐单元不再消费。
+
+    数值:W 连胜 L=0 时 E=(2^(W+1)-1)/(W+1),W=6→18.14<20,W=7→31.875≥20,
+    故 7 连胜恰好 confirmed 过线(Codex 复核)。
+    """
+    run = _mk_run(12)
+    for i in range(12):
+        run.slots[i].base = False
+        run.slots[i].cand_per_q = {f"q{i}": True}
+    _advance_prefix(run, _PARAMS)
+    assert run.frozen and run.verdict is not None
+    assert run.verdict.decision == "accept_confirmed"
+    assert run.n_used == 7  # 第 7 个净胜恰好过线,早停不吃满
+
+
+def test_tail_infra_reaches_terminal_not_continue() -> None:
+    """尾部全 INFRA:剔除后须重判(n_remaining 归 0 → 题尽第四出口),
+    verdict 不得停留在 continue(Codex plan 审 C1 回归锁)。"""
+    run = _mk_run(4)
+    run.slots[0].base = False
+    run.slots[0].cand_per_q = {"q0": True}
+    run.slots[1].base = True
+    run.slots[1].cand_per_q = {"q1": True}
+    for i in (2, 3):
+        run.slots[i].base_infra = True
+        run.slots[i].cand_per_q = {f"q{i}": True}
+    _advance_prefix(run, _PARAMS)
+    assert run.verdict is not None and run.verdict.decision != "continue"
+    assert run.frozen
+
+
+def test_infra_unit_skipped_not_counted() -> None:
+    """INFRA 单元(任一臂)剔除:不入 (W,L)、计入 n_excluded、前缀继续推进。
+
+    注意 futility 出口在小 n_remaining 下很敏感,用 6 单元(首个 INFRA、其余 5 个
+    W 翻转)保证消费全程不提前触发 futility:题尽走 accept_provisional 终态。
+    """
+    run = _mk_run(6)
+    run.slots[0].base_infra = True
+    run.slots[0].cand_per_q = {"q0": True}
+    for i in range(1, 6):
+        run.slots[i].base = False
+        run.slots[i].cand_per_q = {f"q{i}": True}
+    _advance_prefix(run, _PARAMS)
+    assert run.n_excluded == 1 and run.n_used == 5 and run.w == 5 and run.l == 0
+    assert run.verdict is not None and run.verdict.decision == "accept_provisional"
+
+
+def test_all_infra_leaves_verdict_none() -> None:
+    """全部单元被剔除:verdict 保持 None、frozen 保持 False——终态由调度编排层
+    (Task 3 的 validate_skills_concurrent)检测 verdict None 并 raise,本函数不管。"""
+    run = _mk_run(3)
+    for i in range(3):
+        run.slots[i].base_infra = True
+        run.slots[i].cand_per_q = {f"q{i}": True}
+    _advance_prefix(run, _PARAMS)
+    assert run.verdict is None and not run.frozen and run.n_excluded == 3 and run.prefix_ptr == 3
+
+
+def test_repeated_calls_are_idempotent() -> None:
+    """部分前缀消费后重复调用不改变状态;补齐剩余单元后再调用正常推进。"""
+    run = _mk_run(4)
+    for i in (0, 1):
+        run.slots[i].base = False
+        run.slots[i].cand_per_q = {f"q{i}": True}
+    _advance_prefix(run, _PARAMS)
+    snapshot = (run.n_used, run.w, run.l, run.prefix_ptr, len(run.evidence_rows))
+    _advance_prefix(run, _PARAMS)
+    _advance_prefix(run, _PARAMS)
+    assert (run.n_used, run.w, run.l, run.prefix_ptr, len(run.evidence_rows)) == snapshot
+    for i in (2, 3):
+        run.slots[i].base = False
+        run.slots[i].cand_per_q = {f"q{i}": True}
+    _advance_prefix(run, _PARAMS)
+    assert run.n_used > snapshot[0] and run.prefix_ptr > snapshot[3]
+
+
+def test_frozen_run_call_is_noop() -> None:
+    """frozen 后再调用是 no-op:不再消费已配齐单元、证据不再追加。"""
+    run = _mk_run(12)
+    for i in range(12):
+        run.slots[i].base = False
+        run.slots[i].cand_per_q = {f"q{i}": True}
+    _advance_prefix(run, _PARAMS)
+    assert run.frozen and run.n_used == 7
+    before = (
+        run.w,
+        run.l,
+        run.n_used,
+        run.prefix_ptr,
+        len(run.evidence_rows),
+        run.verdict,
+        run.frozen,
+    )
+    _advance_prefix(run, _PARAMS)
+    after = (
+        run.w,
+        run.l,
+        run.n_used,
+        run.prefix_ptr,
+        len(run.evidence_rows),
+        run.verdict,
+        run.frozen,
+    )
+    assert after == before
+
+
+def test_ties_hit_futility_early_and_freeze() -> None:
+    """全打平(无翻转对)时 futility 出口尽早触发并冻结——早停语义(数值:W=L=0
+    时乐观 E = E(n_remaining, 0),n 小易 <e_provisional=3)。"""
+    run = _mk_run(3)
+    for i in range(3):
+        run.slots[i].base = True
+        run.slots[i].cand_per_q = {f"q{i}": True}
+    _advance_prefix(run, _PARAMS)
+    assert run.frozen
+    # 精确锁定 futility 出口:首个消费后 W=L=0,n_remaining=2,
+    # 乐观 E=E(2,0)=(2^3-1)/3=2.33<3 → 立即 reject_futility(Codex 复核)
+    assert run.verdict is not None and run.verdict.decision == "reject_futility"
+    assert run.n_used == 1 and run.w == 0 and run.l == 0

tests/unit/test_gate_unit_arm.py

@@ -0,0 +1,257 @@
+"""单元臂执行任务测试:缓存命中/新鲜跑/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)
+
+
+@pytest.mark.asyncio
+async def test_acquire_cancellation_restores_capacity() -> None:
+    """acquire 半持有时被取消:已拿 permit 自动回滚,容量完全恢复(Codex 质量审 2)。"""
+    slots = _QuestionSlots(2)
+    await slots.acquire(1)  # 预占 1 槽,使 acquire(2) 卡在第二槽
+    task = asyncio.ensure_future(slots.acquire(2))
+    for _ in range(5):
+        await asyncio.sleep(0)  # 让 task 拿到第 1 个 permit 并阻塞在第 2 个
+    task.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await task
+    slots.release(1)  # 归还预占
+    # 半持有的 permit 若泄漏,此处 acquire(2) 将永久阻塞 → wait_for 超时暴露泄漏
+    await asyncio.wait_for(slots.acquire(2), timeout=1.0)
+    slots.release(2)
+
+
+@pytest.mark.asyncio
+async def test_frozen_during_inference_discards_result(tmp_path) -> None:
+    """推理 await 期间被冻结:in-flight 结果整体丢弃(不写 slot/计数器/缓存)。
+
+    设计语义:τ(冻结时刻)之后到达的结果不计入,滞后 INFRA 也不得触发护栏。
+    """
+    run, cache = _mk_gate_run(1, tmp_path)
+    log = _FakeLog()
+    gate_open = asyncio.Event()
+
+    async def _slow_run(questions, *, run_id: str, skills_dir: Path):
+        await gate_open.wait()
+        return await _fake_run_inference(log, correct=True)(
+            questions, run_id=run_id, skills_dir=skills_dir
+        )
+
+    task = asyncio.ensure_future(
+        _run_unit_arm(
+            run,
+            0,
+            "base",
+            _QuestionSlots(4),
+            _slow_run,
+            log,
+            cache,
+            "v1",
+            tmp_path,
+            tmp_path,
+            _PARAMS,
+            0.10,
+        )
+    )
+    for _ in range(5):
+        await asyncio.sleep(0)  # 让 task 进入推理等待
+    run.frozen = True
+    gate_open.set()
+    await task
+    assert run.slots[0].base is None
+    assert run.infra_denom == 0 and run.errors == 0
+    assert cache.get("Action Reasoning", run.s_hash, "v1", "q0") is None

tests/unit/test_gate_unit_scope.py

@@ -1,8 +1,9 @@
-"""tests/unit/test_gate_block_unit.py — gate 块实际执行路径按 unit 跑。
+"""tests/unit/test_gate_unit_scope.py — gate 真实执行路径按 unit 口径跑。
 
-针对 app/harness/validate.py::validate_skill_local（真实 gate 执行路径），
-断言混格阶梯下 gate 块按 unit 口径运行：baseline_cache 键含 unit_id、
-n_used 按 unit 累加、pair_block 折叠 AR pair、逐题 predictions 仍溯源。
+迁移自块序贯版 test_gate_block_unit.py（载体 validate_skill_local，Task 6 删除）：
+针对 app/harness/validate.py::validate_skills_concurrent（连续并发 gate 真实路径），
+断言混格阶梯下 gate 按 unit 口径运行：baseline_cache 键含 unit_id、n_used 按
+unit 累加、pair_block 折叠 AR pair、逐题 predictions 仍溯源。
 核心算法保真 #5（信息阶梯 e-process 口径从 question_id 迁至 unit_id）。
 """
 
@@ -15,7 +16,7 @@ import pytest
 from app.harness.gate_ladder import BaselineCache, skill_hash
 from app.harness.inference import PREDICTIONS_SCHEMA, InferenceResult
 from app.harness.log import HarnessLog
-from app.harness.validate import _ladder_units, validate_skill_local
+from app.harness.validate import GateSpec, _ladder_units, validate_skills_concurrent
 from core.evolution import GateParams
 from core.types import GeneratedQuestion
 
@@ -136,6 +137,35 @@ def _make_mock_run_inference(
     return mock_fn, call_log
 
 
+def _mk_spec(ladder: list[GeneratedQuestion]) -> GateSpec:
+    """由混格阶梯题序构造单题型 GateSpec（units 经 _ladder_units 聚合）。"""
+    return GateSpec(
+        task_type="temporal",
+        target_file="temporal.md",
+        candidate_content="improved skill",
+        base_skill_content="baseline skill content",
+        units=tuple(_ladder_units(ladder)),
+        gate_run_prefix="step1_gate_test",
+    )
+
+
+async def _run_gate(workspace: Path, spec: GateSpec, mock_fn, log: HarnessLog, cache, params):
+    """跑单 spec 的 validate_skills_concurrent 并返回该题型的 outcome。"""
+    outcomes = await validate_skills_concurrent(
+        workspace_dir=workspace,
+        base_skills_version="v1",
+        specs=[spec],
+        gate_params=params,
+        gate_guard_err=0.5,
+        baseline_cache=cache,
+        prompts_version="p1",
+        run_inference=mock_fn,
+        log=log,
+        concurrency=8,
+    )
+    return outcomes[spec.task_type]
+
+
 class TestLadderUnits:
     """_ladder_units：阶梯题序聚合为单元并保持信息阶梯序。"""
 
@@ -177,7 +207,7 @@ class TestLadderUnits:
 
 @pytest.mark.asyncio
 async def test_gate_n_used_counts_units_not_questions(tmp_path: Path) -> None:
-    """混格阶梯（1 pair + 2 single）→ n_used=3 单元，非 4 题。"""
+    """混格阶梯（1 pair + 2 single）→ n_used=3 单元，非 4 题（迁移自块序贯版）。"""
     workspace = _setup_workspace(tmp_path)
     log = _make_log(workspace)
     cache = BaselineCache(workspace / "baseline_cache.json")
@@ -186,7 +216,7 @@ async def test_gate_n_used_counts_units_not_questions(tmp_path: Path) -> None:
     # 基线全错、候选全对 → 3 单元齐翻 W=3
     baseline = {"p1_o": False, "p1_m": False, "s0": False, "s1": False}
     candidate = {"p1_o": True, "p1_m": True, "s0": True, "s1": True}
-    mock_fn, call_log = _make_mock_run_inference(log, baseline, candidate)
+    mock_fn, _ = _make_mock_run_inference(log, baseline, candidate)
 
     accept_params = GateParams(
         e_confirm=15.0,
@@ -197,30 +227,14 @@ async def test_gate_n_used_counts_units_not_questions(tmp_path: Path) -> None:
         e_rollback=10.0,
     )
     try:
-        outcome = await validate_skill_local(
-            workspace_dir=workspace,
-            base_skills_version="v1",
-            task_type="temporal",
-            target_file="temporal.md",
-            candidate_content="improved skill",
-            base_skill_content="baseline skill content",
-            ladder_items=ladder,
-            gate_params=accept_params,
-            gate_block=10,
-            gate_n_max=20,
-            gate_guard_err=0.5,
-            baseline_cache=cache,
-            prompts_version="p1",
-            run_inference=mock_fn,
-            log=log,
-            gate_run_prefix="step1_gate_test",
-        )
+        outcome = await _run_gate(workspace, _mk_spec(ladder), mock_fn, log, cache, accept_params)
         # n_used 按 unit 计（3），W 按 unit 计（3）
         assert outcome.n_used == 3
         assert outcome.w == 3
         assert outcome.l == 0
-        # 证据行按 unit 口径（3 行）
+        # 证据行按 unit 口径（3 行），ladder_rank 沿阶梯序连续
         assert len(outcome.evidence_rows) == 3
+        assert [r["ladder_rank"] for r in outcome.evidence_rows] == [0, 1, 2]
         # baseline_cache 键含 unit_id：pair 用 pair_id、single 用 question_id
         s_hash = skill_hash("baseline skill content")
         assert cache.get("temporal", s_hash, "p1", "p1") is False
@@ -235,84 +249,62 @@ async def test_gate_n_used_counts_units_not_questions(tmp_path: Path) -> None:
 
 @pytest.mark.asyncio
 async def test_gate_pair_partial_flip_not_counted(tmp_path: Path) -> None:
-    """AR pair 候选仅单向翻(T,F)→单元仍错，W 不被单题污染。"""
+    """AR pair 候选仅单向翻(T,F)→单元仍错，W 不被单题污染（迁移自块序贯版）。
+
+    前缀逐单元判定下 2 单元小阶梯会在首单元 futility 早停，观测不到 pair 语义；
+    补 2 个 single 拉长阶梯：4 单元中 3 个 single 翻转 → W=3（pair 不计入），
+    candidate_acc = 3/4。
+    """
     workspace = _setup_workspace(tmp_path)
     log = _make_log(workspace)
     cache = BaselineCache(workspace / "baseline_cache.json")
-    ladder = [*_pair("p1"), _single("s0")]
+    ladder = [*_pair("p1"), _single("s0"), _single("s1"), _single("s2")]
 
-    baseline = {"p1_o": False, "p1_m": False, "s0": False}
-    # pair 只翻一半（p1_o 对、p1_m 错）→ 单元 AND 仍错；s0 翻对
-    candidate = {"p1_o": True, "p1_m": False, "s0": True}
+    baseline = {"p1_o": False, "p1_m": False, "s0": False, "s1": False, "s2": False}
+    # pair 只翻一半（p1_o 对、p1_m 错）→ 单元 AND 仍错；singles 全翻对
+    candidate = {"p1_o": True, "p1_m": False, "s0": True, "s1": True, "s2": True}
     mock_fn, _ = _make_mock_run_inference(log, baseline, candidate)
 
     try:
-        outcome = await validate_skill_local(
-            workspace_dir=workspace,
-            base_skills_version="v1",
-            task_type="temporal",
-            target_file="temporal.md",
-            candidate_content="improved skill",
-            base_skill_content="baseline skill content",
-            ladder_items=ladder,
-            gate_params=_DEFAULT_GATE_PARAMS,
-            gate_block=10,
-            gate_n_max=20,
-            gate_guard_err=0.5,
-            baseline_cache=cache,
-            prompts_version="p1",
-            run_inference=mock_fn,
-            log=log,
-            gate_run_prefix="step1_gate_test",
+        outcome = await _run_gate(
+            workspace, _mk_spec(ladder), mock_fn, log, cache, _DEFAULT_GATE_PARAMS
         )
-        # 只有 s0 单元翻转，pair 单元不计 W（保真 #5：不被 P/Q 单题污染）
-        assert outcome.w == 1
+        # 只有 single 单元翻转，pair 单元不计 W（保真 #5：不被 P/Q 单题污染）
+        assert outcome.w == 3
         assert outcome.l == 0
-        assert outcome.n_used == 2
-        # candidate_acc 分母按 unit（2 单元，1 对）→ 0.5
-        assert outcome.candidate_acc == 0.5
+        assert outcome.n_used == 4
+        # candidate_acc 分母按 unit（4 单元，1 对）→ 3/4
+        assert outcome.candidate_acc == 0.75
     finally:
         log.close()
 
 
 @pytest.mark.asyncio
 async def test_gate_baseline_cache_hit_by_unit(tmp_path: Path) -> None:
-    """基线缓存按 unit_id 预填充 → 基线侧全命中不发起推理。"""
+    """基线缓存按 unit_id 预填充 → 基线侧全命中不发起推理（迁移自块序贯版）。
+
+    阶梯补长到 4 单元避免首单元 futility 早停，覆盖 pair 与 single 两种 unit 键。
+    """
     workspace = _setup_workspace(tmp_path)
     log = _make_log(workspace)
     cache = BaselineCache(workspace / "baseline_cache.json")
-    ladder = [*_pair("p1"), _single("s0")]
+    ladder = [*_pair("p1"), _single("s0"), _single("s1"), _single("s2")]
 
     s_hash = skill_hash("baseline skill content")
     # 按 unit_id 预填充（pair→pair_id，single→question_id），全错
-    cache.put("temporal", s_hash, "p1", "p1", False)
-    cache.put("temporal", s_hash, "p1", "s0", False)
+    for unit_id in ("p1", "s0", "s1", "s2"):
+        cache.put("temporal", s_hash, "p1", unit_id, False)
 
-    baseline = {"p1_o": False, "p1_m": False, "s0": False}
-    candidate = {"p1_o": True, "p1_m": True, "s0": True}
+    baseline = {"p1_o": False, "p1_m": False, "s0": False, "s1": False, "s2": False}
+    candidate = {"p1_o": True, "p1_m": True, "s0": True, "s1": True, "s2": True}
     mock_fn, call_log = _make_mock_run_inference(log, baseline, candidate)
 
     try:
-        outcome = await validate_skill_local(
-            workspace_dir=workspace,
-            base_skills_version="v1",
-            task_type="temporal",
-            target_file="temporal.md",
-            candidate_content="improved skill",
-            base_skill_content="baseline skill content",
-            ladder_items=ladder,
-            gate_params=_DEFAULT_GATE_PARAMS,
-            gate_block=10,
-            gate_n_max=20,
-            gate_guard_err=0.5,
-            baseline_cache=cache,
-            prompts_version="p1",
-            run_inference=mock_fn,
-            log=log,
-            gate_run_prefix="step1_gate_test",
+        outcome = await _run_gate(
+            workspace, _mk_spec(ladder), mock_fn, log, cache, _DEFAULT_GATE_PARAMS
         )
         base_calls = [c for c in call_log if c["run_id"].endswith("_base")]
         assert base_calls == [], "unit 键全命中不应发起基线推理"
-        assert outcome.n_used == 2
+        assert outcome.n_used == 4
     finally:
         log.close()

tests/unit/test_harness_checkpoint.py

@@ -171,7 +171,6 @@ class _FakeConfig:
     gate_delta_min: float = 0.02
     gate_lambda_dir: float = -3.0
     gate_e_rollback: float = 10.0
-    gate_block: int = 4
     gate_n_max: int = 40
     gate_p_low: float = 0.1
     gate_p_high: float = 0.9
@@ -306,7 +305,6 @@ class TestFingerprintStructuralVsDecision:
             "gate_delta_min",
             "gate_lambda_dir",
             "gate_e_rollback",
-            "gate_block",
             "gate_n_max",
             "gate_p_low",
             "gate_p_high",

tests/unit/test_harness_config.py

@@ -50,7 +50,6 @@ def _valid_kwargs() -> dict:
         "gate_delta_min": 0.02,
         "gate_lambda_dir": -0.642,
         "gate_e_rollback": 10.0,
-        "gate_block": 8,
         "gate_n_max": 40,
         "gate_p_low": 0.05,
         "gate_p_high": 0.95,
@@ -378,16 +377,16 @@ class TestGateValidation:
         with pytest.raises(ValueError, match="gate_lambda_dir"):
             _validate(cfg)
 
-    def test_block_exceeds_n_max_rejected(self) -> None:
-        """gate_block > gate_n_max 应抛出 ValueError。"""
-        cfg = _make_config(gate_block=50, gate_n_max=40)
-        with pytest.raises(ValueError, match="gate_block"):
+    def test_n_max_zero_rejected(self) -> None:
+        """gate_n_max <= 0 应抛出 ValueError（迁移自块序贯版 gate_block 校验）。"""
+        cfg = _make_config(gate_n_max=0)
+        with pytest.raises(ValueError, match="gate_n_max"):
             _validate(cfg)
 
-    def test_block_zero_rejected(self) -> None:
-        """gate_block <= 0 应抛出 ValueError。"""
-        cfg = _make_config(gate_block=0)
-        with pytest.raises(ValueError, match="gate_block"):
+    def test_n_max_negative_rejected(self) -> None:
+        """gate_n_max 为负也应报错。"""
+        cfg = _make_config(gate_n_max=-1)
+        with pytest.raises(ValueError, match="gate_n_max"):
             _validate(cfg)
 
     def test_p_low_exceeds_p_high_rejected(self) -> None:

tests/unit/test_harness_inference.py

@@ -117,8 +117,9 @@ def harness_log(tmp_path: Any, request: Any) -> HarnessLog:
     """创建临时 HarnessLog 实例。
 
     使用 test 节点名称的 hash 作为 db 文件名，避免冲突。
-    run_id 固定为 "test-run"，实际 run_inference 中传入的 run_id
-    由 HarnessLog.insert 自动覆盖为 HarnessLog 构造时的值。
+    实例 run_id 固定为 "test-run"；predictions 行的 run_id 由 inference
+    record 显式携带（run_inference 传入值），不回落实例 run_id——
+    连续并发 gate 共享单一 HarnessLog 的契约。
     """
     db_name = f"harness_{id(request)}.db"
     db_path = str(tmp_path / db_name)
@@ -528,8 +529,9 @@ class TestPredictionAlwaysWritten:
         assert result.correct == 0
         assert result.stop_reason_counts.get("error") == 1
 
-        # 验证 DB 中的记录（HarnessLog.insert 使用构造时的 run_id）
-        rows = harness_log.query("SELECT * FROM predictions WHERE run_id = ?", ("test-run",))
+        # 验证 DB 中的记录（record 显式携带 run_inference 的 run_id，
+        # 不再回落 HarnessLog 实例 run_id——连续并发 gate 共享 log 的契约）
+        rows = harness_log.query("SELECT * FROM predictions WHERE run_id = ?", ("run-error",))
         assert len(rows) == 1
         assert rows[0]["stop_reason"] == "error"
         assert rows[0]["prediction"] is None
@@ -553,8 +555,8 @@ class TestPredictionAlwaysWritten:
         )
 
         assert result.total == 1
-        # HarnessLog.insert 使用构造时的 run_id
-        rows = harness_log.query("SELECT * FROM predictions WHERE run_id = ?", ("test-run",))
+        # record 显式携带 run_inference 的 run_id（共享 log 契约）
+        rows = harness_log.query("SELECT * FROM predictions WHERE run_id = ?", ("run-parse-err",))
         assert len(rows) == 1
         assert rows[0]["prediction"] is None
 
@@ -612,7 +614,7 @@ class TestNonScalarPrediction:
         )
 
         assert result.total == 1
-        rows = harness_log.query("SELECT * FROM predictions WHERE run_id = ?", ("test-run",))
+        rows = harness_log.query("SELECT * FROM predictions WHERE run_id = ?", ("run-nonscalar",))
         assert len(rows) == 1
         # prediction 被 JSON 序列化为字符串，不再是 Python list
         assert rows[0]["prediction"] == '["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,
@@ -383,3 +383,39 @@ def test_write_epoch_report(tmp_path: Path) -> None:
     assert data["system_tool_action"] == "updated"
     assert data["momentum_updated_task_types"] == ["temporal", "causal"]
     assert data["best_val_acc"] == pytest.approx(0.88)
+
+
+def test_write_gate_evidence_migrates_legacy_block_idx_table(tmp_path) -> None:
+    """旧块序贯表(含 block_idx 无 ladder_rank)复用:幂等补列后写入成功(终审 C1 回归锁)。"""
+    import sqlite3
+
+    db = tmp_path / "harness.db"
+    conn = sqlite3.connect(db)
+    conn.execute(
+        "CREATE TABLE gate_evidence (run_id TEXT, timestamp TEXT, epoch INTEGER,"
+        " step INTEGER, question_id TEXT, task_type TEXT, block_idx INTEGER,"
+        " baseline_correct INTEGER, candidate_correct INTEGER, e_value REAL,"
+        " stop_reason TEXT)"
+    )
+    conn.commit()
+    conn.close()
+
+    write_gate_evidence(
+        str(db),
+        run_id="r1",
+        epoch=1,
+        step=0,
+        rows=[
+            {
+                "question_id": "q1",
+                "task_type": "Action Reasoning",
+                "ladder_rank": 0,
+                "baseline_correct": False,
+                "candidate_correct": True,
+                "e_value": 1.5,
+                "stop_reason": "",
+            }
+        ],
+    )
+    got = read_gate_evidence(str(db), run_id="r1")
+    assert len(got) == 1 and got[0]["ladder_rank"] == 0

tests/unit/test_harness_pools.py

@@ -327,7 +327,6 @@ class TestBuildOrLoadPoolsFrozen:
             gate_delta_min=0.02,
             gate_lambda_dir=-0.642,
             gate_e_rollback=10.0,
-            gate_block=8,
             gate_n_max=40,
             gate_p_low=0.05,
             gate_p_high=0.95,
@@ -881,7 +880,6 @@ class TestRunHoldoutEvalConfig:
             gate_delta_min=0.02,
             gate_lambda_dir=-0.642,
             gate_e_rollback=10.0,
-            gate_block=8,
             gate_n_max=40,
             gate_p_low=0.05,
             gate_p_high=0.95,
@@ -932,7 +930,6 @@ class TestRunHoldoutEvalConfig:
             gate_delta_min=0.02,
             gate_lambda_dir=-0.642,
             gate_e_rollback=10.0,
-            gate_block=8,
             gate_n_max=40,
             gate_p_low=0.05,
             gate_p_high=0.95,

tests/unit/test_harness_runner.py

@@ -840,7 +840,6 @@ class TestRunnerFactoryInjection:
             "gate_delta_min": 0.02,
             "gate_lambda_dir": -0.642,
             "gate_e_rollback": 10.0,
-            "gate_block": 8,
             "gate_n_max": 40,
             "gate_p_low": 0.05,
             "gate_p_high": 0.95,

tests/unit/test_harness_validate.py

@@ -1,7 +1,9 @@
 """tests/unit/test_harness_validate.py — app/harness/validate.py 的单元测试。
 
-覆盖：数据类型字段、materialize 物化与清理、async validate_skill_local
-（accept/reject/prefix 校验/INFRA 护栏/缓存命中/最后一块终态）。
+覆盖：数据类型字段、materialize 物化与清理、async validate_skills_concurrent
+（accept/reject/prefix 校验/INFRA 护栏/缓存命中/题尽终态）。async 用例迁移自
+块序贯版（validate_skill_local，Task 6 删除）：载体换连续并发 gate，语义断言
+保留；前缀逐单元判定使早停点比旧块判定更早（见各用例 docstring 的数值推导）。
 """
 
 from __future__ import annotations
@@ -14,10 +16,12 @@ from app.harness.gate_ladder import BaselineCache, skill_hash
 from app.harness.inference import PREDICTIONS_SCHEMA, InferenceResult
 from app.harness.log import HarnessLog
 from app.harness.validate import (
+    GateSpec,
     Probation,
     ValidationOutcome,
+    _ladder_units,
     materialize_candidate_skill,
-    validate_skill_local,
+    validate_skills_concurrent,
 )
 from core.evolution import GateParams, RejectedEdit
 from core.types import GeneratedQuestion
@@ -150,7 +154,7 @@ def _make_mock_run_inference(
 
 
 def _make_all_infra_mock(log: HarnessLog, stop_reason: str):
-    """构建基线全 INFRA 的 mock：每 record 写指定 INFRA stop_reason（error/parse_error）。
+    """构建全 INFRA 的 mock：每 record 写指定 INFRA stop_reason（error/parse_error）。
 
     与真实推理一致——per-record DB stop_reason 与汇总 stop_reason_counts 同源；护栏
     分子按 unit 从 DB 读（_infra_question_ids_from_db），故须真实落 DB。total 返回
@@ -199,6 +203,48 @@ def _make_all_infra_mock(log: HarnessLog, stop_reason: str):
     return mock_fn, call_log
 
 
+def _mk_spec(
+    questions: list[GeneratedQuestion],
+    *,
+    candidate_content: str = "candidate skill",
+    gate_run_prefix: str = "step1_gate_test",
+) -> GateSpec:
+    """由阶梯题序构造单题型 GateSpec（units 经 _ladder_units 聚合为阶梯序单元）。"""
+    return GateSpec(
+        task_type="temporal",
+        target_file="temporal.md",
+        candidate_content=candidate_content,
+        base_skill_content="baseline skill content",
+        units=tuple(_ladder_units(questions)),
+        gate_run_prefix=gate_run_prefix,
+    )
+
+
+async def _run_single_spec(
+    workspace: Path,
+    spec: GateSpec,
+    mock_fn,
+    log: HarnessLog,
+    cache: BaselineCache,
+    params: GateParams,
+    gate_guard_err: float = 0.5,
+) -> ValidationOutcome:
+    """跑单 spec 的 validate_skills_concurrent 并返回该题型的 outcome。"""
+    outcomes = await validate_skills_concurrent(
+        workspace_dir=workspace,
+        base_skills_version="v1",
+        specs=[spec],
+        gate_params=params,
+        gate_guard_err=gate_guard_err,
+        baseline_cache=cache,
+        prompts_version="p1",
+        run_inference=mock_fn,
+        log=log,
+        concurrency=8,
+    )
+    return outcomes[spec.task_type]
+
+
 def test_infra_stop_reasons_single_source() -> None:
     """app 侧 INFRA_STOP_REASONS 复用 core 常量（同一对象），杜绝未来漂移（M-2）。"""
     from app.harness import validate
@@ -331,13 +377,17 @@ class TestMaterializeCandidateSkill:
 
 
 # ===========================================================================
-# async 验证测试
+# async 验证测试（迁移自块序贯版 validate_skill_local）
 # ===========================================================================
 
 
 @pytest.mark.asyncio
-async def test_validate_skill_local_accept(tmp_path: Path) -> None:
-    """候选全对、基线全错 → 高 e 值 → accept_confirmed。"""
+async def test_validate_concurrent_accept(tmp_path: Path) -> None:
+    """候选全对、基线全错 → 高 e 值 → accept_confirmed（迁移自块序贯版）。
+
+    6 单元连胜：E=(2^(W+1)-1)/(W+1)，前 5 单元 E<15 且不触方向/futility，
+    第 6 单元 E=18.14 ≥ e_confirm=15 → 与旧块判定同点收敛（W=6, n_used=6）。
+    """
     workspace = _setup_workspace(tmp_path)
     log = _make_log(workspace)
     questions = _make_questions(6)
@@ -359,23 +409,13 @@ async def test_validate_skill_local_accept(tmp_path: Path) -> None:
     )
 
     try:
-        outcome = await validate_skill_local(
-            workspace_dir=workspace,
-            base_skills_version="v1",
-            task_type="temporal",
-            target_file="temporal.md",
-            candidate_content="improved skill",
-            base_skill_content="baseline skill content",
-            ladder_items=questions,
-            gate_params=accept_params,
-            gate_block=6,
-            gate_n_max=20,
-            gate_guard_err=0.5,
-            baseline_cache=cache,
-            prompts_version="p1",
-            run_inference=mock_fn,
-            log=log,
-            gate_run_prefix="step1_gate_test",
+        outcome = await _run_single_spec(
+            workspace,
+            _mk_spec(questions, candidate_content="improved skill"),
+            mock_fn,
+            log,
+            cache,
+            accept_params,
         )
 
         assert outcome.accepted is True
@@ -387,6 +427,8 @@ async def test_validate_skill_local_accept(tmp_path: Path) -> None:
         assert outcome.candidate_acc == 1.0
         assert outcome.baseline_acc == 0.0
         assert len(outcome.evidence_rows) == 6
+        # 阶梯序前缀消费：ladder_rank 连续（替代旧块边界断言）
+        assert [r["ladder_rank"] for r in outcome.evidence_rows] == list(range(6))
         # 终态证据行携带 stop_reason
         assert outcome.evidence_rows[-1]["stop_reason"] == "confirmed"
         # 候选临时目录应被清理
@@ -398,50 +440,45 @@ async def test_validate_skill_local_accept(tmp_path: Path) -> None:
 
 
 @pytest.mark.asyncio
-async def test_validate_skill_local_reject(tmp_path: Path) -> None:
-    """候选全错、基线全对 → L 高 → 方向拒绝。"""
+async def test_validate_concurrent_reject_directional(tmp_path: Path) -> None:
+    """候选全错、基线全对 → L 高 → 方向拒绝（迁移自块序贯版）。
+
+    前缀逐单元判定下早停点前移：15 单元阶梯保证 L=1..3 时 futility 不先触发
+    （E(w+n_rem, l) ≥ 3），L=4 时 Wald=4·ln0.6=-2.04 ≤ lambda_dir=-2.0 →
+    directional 早停于第 4 单元（旧块版一次性判整块故 L=6）。
+    """
     workspace = _setup_workspace(tmp_path)
     log = _make_log(workspace)
-    questions = _make_questions(6)
+    questions = _make_questions(15)
     cache = BaselineCache(workspace / "baseline_cache.json")
 
-    # 基线全对，候选全错 → W=0, L=6 → 方向拒绝
-    baseline_correct = {f"q{i}": True for i in range(6)}
-    candidate_correct = {f"q{i}": False for i in range(6)}
+    baseline_correct = {f"q{i}": True for i in range(15)}
+    candidate_correct = {f"q{i}": False for i in range(15)}
     mock_fn, _ = _make_mock_run_inference(log, baseline_correct, candidate_correct)
 
     try:
-        outcome = await validate_skill_local(
-            workspace_dir=workspace,
-            base_skills_version="v1",
-            task_type="temporal",
-            target_file="temporal.md",
-            candidate_content="bad skill",
-            base_skill_content="baseline skill content",
-            ladder_items=questions,
-            gate_params=_DEFAULT_GATE_PARAMS,
-            gate_block=6,
-            gate_n_max=20,
-            gate_guard_err=0.5,
-            baseline_cache=cache,
-            prompts_version="p1",
-            run_inference=mock_fn,
-            log=log,
-            gate_run_prefix="step1_gate_test",
+        outcome = await _run_single_spec(
+            workspace,
+            _mk_spec(questions, candidate_content="bad skill"),
+            mock_fn,
+            log,
+            cache,
+            _DEFAULT_GATE_PARAMS,
         )
 
         assert outcome.accepted is False
         assert outcome.action == "reject"
         assert outcome.stop_reason == "directional"
         assert outcome.w == 0
-        assert outcome.l == 6
+        assert outcome.l == 4
+        assert outcome.n_used == 4
     finally:
         log.close()
 
 
 @pytest.mark.asyncio
 async def test_gate_prefix_must_contain_gate(tmp_path: Path) -> None:
-    """gate_run_prefix 不含 '_gate_' 时抛 ValueError。"""
+    """gate_run_prefix 不含 '_gate_' 时抛 ValueError（迁移自块序贯版）。"""
     workspace = _setup_workspace(tmp_path)
     log = _make_log(workspace)
     questions = _make_questions(4)
@@ -452,23 +489,13 @@ async def test_gate_prefix_must_contain_gate(tmp_path: Path) -> None:
 
     try:
         with pytest.raises(ValueError, match="_gate_"):
-            await validate_skill_local(
-                workspace_dir=workspace,
-                base_skills_version="v1",
-                task_type="temporal",
-                target_file="temporal.md",
-                candidate_content="content",
-                base_skill_content="baseline",
-                ladder_items=questions,
-                gate_params=_DEFAULT_GATE_PARAMS,
-                gate_block=4,
-                gate_n_max=20,
-                gate_guard_err=0.5,
-                baseline_cache=cache,
-                prompts_version="p1",
-                run_inference=noop_fn,
-                log=log,
-                gate_run_prefix="step1_no_marker",
+            await _run_single_spec(
+                workspace,
+                _mk_spec(questions, gate_run_prefix="step1_no_marker"),
+                noop_fn,
+                log,
+                cache,
+                _DEFAULT_GATE_PARAMS,
             )
     finally:
         log.close()
@@ -476,34 +503,26 @@ async def test_gate_prefix_must_contain_gate(tmp_path: Path) -> None:
 
 @pytest.mark.asyncio
 async def test_infra_guard_threshold(tmp_path: Path) -> None:
-    """推理错误率超阈值时抛 RuntimeError（护栏分子/分母 unit 同粒度）。"""
+    """推理错误率超阈值时抛 RuntimeError（迁移自块序贯版，分子/分母 unit 同粒度）。
+
+    12 个 single 双臂全 INFRA error：errors 按单元去重逐单元 +1，分母逐臂 +1，
+    分母 ≥10 后错误率 >0.5 → 护栏熔断。
+    """
     workspace = _setup_workspace(tmp_path)
     log = _make_log(workspace)
-    # 需要 >=10 unit 分母才触发护栏：12 个 single，基线全 INFRA error。
-    # 首块全 INFRA → valid_chunk 空 → errors=12/denom=12=1.0>0.5 触发护栏。
     questions = _make_questions(12)
     cache = BaselineCache(workspace / "baseline_cache.json")
     mock_fn, _ = _make_all_infra_mock(log, "error")
 
     try:
         with pytest.raises(RuntimeError, match="错误率过高"):
-            await validate_skill_local(
-                workspace_dir=workspace,
-                base_skills_version="v1",
-                task_type="temporal",
-                target_file="temporal.md",
-                candidate_content="content",
-                base_skill_content="baseline skill content",
-                ladder_items=questions,
-                gate_params=_DEFAULT_GATE_PARAMS,
-                gate_block=12,
-                gate_n_max=20,
-                gate_guard_err=0.5,
-                baseline_cache=cache,
-                prompts_version="p1",
-                run_inference=mock_fn,
-                log=log,
-                gate_run_prefix="step1_gate_test",
+            await _run_single_spec(
+                workspace,
+                _mk_spec(questions),
+                mock_fn,
+                log,
+                cache,
+                _DEFAULT_GATE_PARAMS,
             )
     finally:
         log.close()
@@ -511,7 +530,10 @@ async def test_infra_guard_threshold(tmp_path: Path) -> None:
 
 @pytest.mark.asyncio
 async def test_baseline_cache_hit(tmp_path: Path) -> None:
-    """基线缓存全命中时不发起基线侧推理。"""
+    """基线缓存全命中时不发起基线侧推理（迁移自块序贯版）。
+
+    连续并发 gate 下候选侧逐单元发臂：4 单元 → 4 次 cand 调用（旧块版整块 1 次）。
+    """
     workspace = _setup_workspace(tmp_path)
     log = _make_log(workspace)
     questions = _make_questions(4)
@@ -528,30 +550,20 @@ async def test_baseline_cache_hit(tmp_path: Path) -> None:
     mock_fn, call_log = _make_mock_run_inference(log, baseline_correct, candidate_correct)
 
     try:
-        outcome = await validate_skill_local(
-            workspace_dir=workspace,
-            base_skills_version="v1",
-            task_type="temporal",
-            target_file="temporal.md",
-            candidate_content="improved skill",
-            base_skill_content="baseline skill content",
-            ladder_items=questions,
-            gate_params=_DEFAULT_GATE_PARAMS,
-            gate_block=4,
-            gate_n_max=20,
-            gate_guard_err=0.5,
-            baseline_cache=cache,
-            prompts_version="p1",
-            run_inference=mock_fn,
-            log=log,
-            gate_run_prefix="step1_gate_test",
+        outcome = await _run_single_spec(
+            workspace,
+            _mk_spec(questions, candidate_content="improved skill"),
+            mock_fn,
+            log,
+            cache,
+            _DEFAULT_GATE_PARAMS,
         )
 
         # 只有候选侧调用了 run_inference（_cand），基线侧全命中不调用
         base_calls = [c for c in call_log if c["run_id"].endswith("_base")]
         cand_calls = [c for c in call_log if c["run_id"].endswith("_cand")]
         assert len(base_calls) == 0, "基线缓存全命中不应发起推理"
-        assert len(cand_calls) == 1
+        assert len(cand_calls) == 4
         assert outcome.accepted is True
     finally:
         log.close()
@@ -559,21 +571,21 @@ async def test_baseline_cache_hit(tmp_path: Path) -> None:
 
 @pytest.mark.asyncio
 async def test_baseline_infra_error_not_cached(tmp_path: Path) -> None:
-    """基线臂 INFRA error 的 unit 不写入 BaselineCache（不永久污染），且从有效单元排除。"""
-    from app.harness.gate_ladder import skill_hash
-    from app.harness.question_units import build_units
-    from app.harness.validate import _resolve_baseline_block
+    """基线臂 INFRA error 的 unit 不写入 BaselineCache（不永久污染），且从配对剔除。
 
+    迁移自块序贯版 _resolve_baseline_block 直测：改经 validate_skills_concurrent
+    端到端验证同一契约——INFRA 单元不落缓存、不入配对；干净单元正常缓存并消费。
+    """
     workspace = _setup_workspace(tmp_path)
     log = _make_log(workspace)
-    questions = _make_questions(2)  # q0 干净, q1 INFRA error
-    units = build_units(questions)
+    questions = _make_questions(2)  # q0 基线 INFRA error, q1 干净
     cache = BaselineCache(workspace / "baseline_cache.json")
     s_hash = skill_hash("baseline skill content")
 
     async def mock_fn(qs, *, run_id, skills_dir):
+        is_base = run_id.endswith("_base")
         for q in qs:
-            is_err = q.question_id == "q1"
+            is_err = is_base and q.question_id == "q0"
             log.insert(
                 "predictions",
                 {
@@ -594,54 +606,49 @@ async def test_baseline_infra_error_not_cached(tmp_path: Path) -> None:
             )
         return InferenceResult(
             run_id=run_id,
-            accuracy=0.5,
-            total=2,
-            correct=1,
+            accuracy=0.0,
+            total=len(qs),
+            correct=0,
             per_task_type={},
             steps_mean=1.0,
             token_usage={"prompt_tokens": 10, "completion_tokens": 10},
-            stop_reason_counts={"completed": 1, "error": 1},
+            stop_reason_counts={},
         )
 
     try:
-        b_units, valid_units, _errors_inc, _denom_inc = await _resolve_baseline_block(
-            units=units,
-            task_type="temporal",
-            s_hash=s_hash,
-            prompts_version="p1",
-            baseline_cache=cache,
-            base_skills_dir=workspace / "skills" / "v1",
-            run_inference=mock_fn,
-            log=log,
-            run_id="step1_gate_b0_base",
+        outcome = await _run_single_spec(
+            workspace,
+            _mk_spec(questions),
+            mock_fn,
+            log,
+            cache,
+            _DEFAULT_GATE_PARAMS,
+            gate_guard_err=0.9,  # 分母 <10 不触发错误率护栏
         )
-        # q1 是 INFRA：不写缓存、不入 b_units、不在有效单元里
-        assert cache.get("temporal", s_hash, "p1", "q1") is None
-        assert "q1" not in b_units
-        assert all(u.unit_id != "q1" for u in valid_units)
-        # q0 干净：正常缓存并入 b_units/valid_units
-        assert cache.get("temporal", s_hash, "p1", "q0") is True
-        assert b_units["q0"] is True
-        assert any(u.unit_id == "q0" for u in valid_units)
+        # q0 是 INFRA：不写缓存、不入配对观测
+        assert cache.get("temporal", s_hash, "p1", "q0") is None
+        assert "q0" not in outcome.improvements + outcome.regressions
+        # q1 干净：正常缓存并被消费（唯一有效单元）
+        assert cache.get("temporal", s_hash, "p1", "q1") is True
+        assert outcome.n_used == 1
     finally:
         log.close()
 
 
 @pytest.mark.asyncio
-async def test_infra_guard_counts_units_not_records(tmp_path: Path) -> None:
-    """护栏分子按 unit 计：AR pair 两 record 全 INFRA 只计 1 个 INFRA unit（而非 2）。
+async def test_infra_errors_counted_per_unit_not_per_record(tmp_path: Path) -> None:
+    """护栏分子按 unit 去重：AR pair 两 record、双臂全 INFRA 只计 1 个 error。
 
-    回归 I-3：分子此前用 stop_reason_counts 逐 record 计数，分母 denom_inc=r.total
-    是 unit 粒度；AR pair（一 unit 两 record）致分子被放大、误触发 gate_guard_err。
-    分子改为"含 INFRA record 的 unit 数"后与分母同粒度（核心算法保真 #5/#6）。
+    迁移自块序贯版 _resolve_baseline_block 直测（回归 I-3）：分子若逐 record /
+    逐臂计数会被放大（一 unit 两 record × 两臂 = 4），与 unit 粒度分母失配致
+    gate_guard_err 误触发。新载体 _run_unit_arm + _register_arm_arrival 按
+    slot.excluded() 去重（核心算法保真 #5/#6）。
     """
-    from app.harness.gate_ladder import skill_hash
     from app.harness.question_units import build_units
-    from app.harness.validate import _resolve_baseline_block
+    from app.harness.validate import _GateRun, _QuestionSlots, _run_unit_arm
 
     workspace = _setup_workspace(tmp_path)
     log = _make_log(workspace)
-    # 一个 AR pair（两成员共享 pair_id）→ build_units 折叠为 1 个 pair unit
     common = {
         "video_id": "vp",
         "task_type": "temporal",
@@ -660,7 +667,17 @@ async def test_infra_guard_counts_units_not_records(tmp_path: Path) -> None:
     units = build_units(pair)
     assert len(units) == 1  # 前置：pair 折叠为 1 个 unit
     cache = BaselineCache(workspace / "baseline_cache.json")
-    s_hash = skill_hash("baseline skill content")
+    run = _GateRun.from_spec(
+        GateSpec(
+            task_type="temporal",
+            target_file="temporal.md",
+            candidate_content="cand",
+            base_skill_content="baseline skill content",
+            units=tuple(units),
+            gate_run_prefix="step1_gate_test",
+        )
+    )
+    s_hash = run.s_hash
 
     async def mock_fn(qs, *, run_id, skills_dir):
         # 两 record 皆 INFRA error
@@ -683,7 +700,7 @@ async def test_infra_guard_counts_units_not_records(tmp_path: Path) -> None:
                     "steps_json": "[]",
                 },
             )
-        # total 为 unit 粒度（1 个 pair unit）；stop_reason_counts 为 record 粒度（2）
+        # total 为 unit 粒度（1 个 pair unit）；record 粒度为 2
         return InferenceResult(
             run_id=run_id,
             accuracy=0.0,
@@ -695,94 +712,57 @@ async def test_infra_guard_counts_units_not_records(tmp_path: Path) -> None:
             stop_reason_counts={"error": 2},
         )
 
+    slots = _QuestionSlots(4)
     try:
-        _b_units, valid_units, errors_inc, denom_inc = await _resolve_baseline_block(
-            units=units,
-            task_type="temporal",
-            s_hash=s_hash,
-            prompts_version="p1",
-            baseline_cache=cache,
-            base_skills_dir=workspace / "skills" / "v1",
-            run_inference=mock_fn,
-            log=log,
-            run_id="step1_gate_b0_base",
+        for arm in ("base", "cand"):
+            await _run_unit_arm(
+                run,
+                0,
+                arm,
+                slots,
+                mock_fn,
+                log,
+                cache,
+                "p1",
+                workspace / "skills" / "v1",
+                workspace / "skills" / "v1",
+                _DEFAULT_GATE_PARAMS,
+                0.9,
             )
-        # 分子按 unit 计：1 个 INFRA unit（不是 2 条 record）；分母同粒度 = r.total = 1
-        assert errors_inc == 1
-        assert denom_inc == 1
-        # 整对 INFRA → 从有效单元剔除
-        assert valid_units == []
+        # 分子按 unit 去重：双臂 × 两 record 只计 1 个 error；分母按臂 total 累计 = 2
+        assert run.errors == 1
+        assert run.infra_denom == 2
+        assert run.slots[0].base_infra and run.slots[0].cand_infra
+        # INFRA 单元不写缓存
+        assert cache.get("temporal", s_hash, "p1", "p1") is None
     finally:
         log.close()
 
 
 @pytest.mark.asyncio
 async def test_all_infra_ladder_raises_clear_error(tmp_path: Path) -> None:
-    """整个阶梯所有 unit 都被判为 INFRA 排除 → 明确 RuntimeError（非误导性空阶梯断言）。"""
+    """整个阶梯所有 unit 都被判为 INFRA 排除 → 明确 RuntimeError（迁移自块序贯版）。
+
+    连续并发 gate 下双臂独立发射，候选臂不再依赖基线侧结果（旧版"全 INFRA 块
+    不空跑候选"的断言随块编排一并删除）。
+    """
     workspace = _setup_workspace(tmp_path)
     log = _make_log(workspace)
     questions = _make_questions(4)
     cache = BaselineCache(workspace / "baseline_cache.json")
-
-    candidate_calls: list[str] = []
-
-    async def mock_fn(qs, *, run_id, skills_dir):
-        if run_id.endswith("_cand"):
-            candidate_calls.append(run_id)
-        # 基线臂逐题全部 INFRA error（候选臂在修复后不应被空跑）
-        for q in qs:
-            log.insert(
-                "predictions",
-                {
-                    "run_id": run_id,
-                    "video_id": "v0",
-                    "question_id": q.question_id,
-                    "task_type": "temporal",
-                    "prediction": "",
-                    "answer": "A",
-                    "evidence": "",
-                    "reasoning": "",
-                    "steps_used": 1,
-                    "prompt_tokens": 10,
-                    "completion_tokens": 10,
-                    "stop_reason": "error",
-                    "steps_json": "[]",
-                },
-            )
-        total = len(qs)
-        return InferenceResult(
-            run_id=run_id,
-            accuracy=0.0,
-            total=total,
-            correct=0,
-            per_task_type={},
-            steps_mean=1.0,
-            token_usage={"prompt_tokens": 10, "completion_tokens": 10},
-            stop_reason_counts={"error": total},
-        )
+    mock_fn, _ = _make_all_infra_mock(log, "error")
 
     try:
         with pytest.raises(RuntimeError, match="INFRA"):
-            await validate_skill_local(
-                workspace_dir=workspace,
-                base_skills_version="v1",
-                task_type="temporal",
-                target_file="temporal.md",
-                candidate_content="content",
-                base_skill_content="baseline skill content",
-                ladder_items=questions,
-                gate_params=_DEFAULT_GATE_PARAMS,
-                gate_block=4,
-                gate_n_max=20,
-                gate_guard_err=0.9,  # 高阈值：4 题 <10 分母不触发错误率护栏
-                baseline_cache=cache,
-                prompts_version="p1",
-                run_inference=mock_fn,
-                log=log,
-                gate_run_prefix="step1_gate_test",
+            await _run_single_spec(
+                workspace,
+                _mk_spec(questions),
+                mock_fn,
+                log,
+                cache,
+                _DEFAULT_GATE_PARAMS,
+                gate_guard_err=0.9,  # 4 单元分母 <10 不触发错误率护栏 → 逼出全排除分支
             )
-        # 全 INFRA 块不应触发候选空跑
-        assert candidate_calls == []
     finally:
         log.close()
 
@@ -792,42 +772,33 @@ async def test_parse_error_counts_toward_guard(tmp_path: Path) -> None:
     """stop_reason=parse_error 也计入护栏错误率（与 INFRA 判定口径一致）→ 超阈值熔断。"""
     workspace = _setup_workspace(tmp_path)
     log = _make_log(workspace)
-    # 12 个 single，基线全 parse_error（per-record 落 DB，护栏按 unit 从 DB 读）。
-    # 首块全 INFRA → errors=12/denom=12=1.0>0.5 → parse_error 亦触发护栏。
     questions = _make_questions(12)
     cache = BaselineCache(workspace / "baseline_cache.json")
     mock_fn, _ = _make_all_infra_mock(log, "parse_error")
 
     try:
         with pytest.raises(RuntimeError, match="错误率过高"):
-            await validate_skill_local(
-                workspace_dir=workspace,
-                base_skills_version="v1",
-                task_type="temporal",
-                target_file="temporal.md",
-                candidate_content="content",
-                base_skill_content="baseline skill content",
-                ladder_items=questions,
-                gate_params=_DEFAULT_GATE_PARAMS,
-                gate_block=12,
-                gate_n_max=20,
-                gate_guard_err=0.5,
-                baseline_cache=cache,
-                prompts_version="p1",
-                run_inference=mock_fn,
-                log=log,
-                gate_run_prefix="step1_gate_test",
+            await _run_single_spec(
+                workspace,
+                _mk_spec(questions),
+                mock_fn,
+                log,
+                cache,
+                _DEFAULT_GATE_PARAMS,
             )
     finally:
         log.close()
 
 
 @pytest.mark.asyncio
-async def test_last_block_terminal(tmp_path: Path) -> None:
-    """单块 + n_remaining=0 → 终态判定（provisional 或 inertia），非 continue。"""
+async def test_ladder_exhaustion_terminal(tmp_path: Path) -> None:
+    """题尽（n_remaining=0）→ 终态判定（provisional 或 inertia），非 continue。
+
+    迁移自块序贯版"最后一块终态"：块边界不存在了，等价语义是阶梯耗尽时
+    第四出口兜底，终态行携带 stop_reason。
+    """
     workspace = _setup_workspace(tmp_path)
     log = _make_log(workspace)
-    # 4 题，gate_block=4 → 一块走完，n_remaining=0
     questions = _make_questions(4)
     cache = BaselineCache(workspace / "baseline_cache.json")
 
@@ -837,23 +808,13 @@ async def test_last_block_terminal(tmp_path: Path) -> None:
     mock_fn, _ = _make_mock_run_inference(log, baseline_correct, candidate_correct)
 
     try:
-        outcome = await validate_skill_local(
-            workspace_dir=workspace,
-            base_skills_version="v1",
-            task_type="temporal",
-            target_file="temporal.md",
-            candidate_content="candidate skill",
-            base_skill_content="baseline skill content",
-            ladder_items=questions,
-            gate_params=_DEFAULT_GATE_PARAMS,
-            gate_block=4,
-            gate_n_max=4,
-            gate_guard_err=0.5,
-            baseline_cache=cache,
-            prompts_version="p1",
-            run_inference=mock_fn,
-            log=log,
-            gate_run_prefix="step1_gate_test",
+        outcome = await _run_single_spec(
+            workspace,
+            _mk_spec(questions),
+            mock_fn,
+            log,
+            cache,
+            _DEFAULT_GATE_PARAMS,
         )
 
         # n_remaining=0 → 不可能是 continue
@@ -865,6 +826,8 @@ async def test_last_block_terminal(tmp_path: Path) -> None:
             "futility",
         )
         assert outcome.n_used == 4
+        # 阶梯序前缀消费：ladder_rank 连续
+        assert [r["ladder_rank"] for r in outcome.evidence_rows] == list(range(4))
         # 终态行标记 stop_reason
         assert outcome.evidence_rows[-1]["stop_reason"] != ""
     finally:

tests/unit/test_inference_pair_aggregate.py

@@ -332,8 +332,8 @@ class TestRunInferencePairEndToEnd:
         assert result.total == 1
         assert result.correct == 1
 
-        # 逐题溯源：predictions 表两条 record 都在
-        rows = harness_log.query("SELECT * FROM predictions WHERE run_id = ?", ("test-run",))
+        # 逐题溯源：predictions 表两条 record 都在（record 显式携带传入的 run_id）
+        rows = harness_log.query("SELECT * FROM predictions WHERE run_id = ?", ("run-pair-e2e",))
         qids = {r["question_id"] for r in rows}
         assert qids == {"po", "pm"}
 
@@ -360,6 +360,6 @@ class TestRunInferencePairEndToEnd:
         )
 
         assert result.total == 1  # single 存活，孤儿剔除
-        # 逐题溯源：孤儿题仍逐题落库（推理不变）
-        rows = harness_log.query("SELECT * FROM predictions WHERE run_id = ?", ("test-run",))
+        # 逐题溯源：孤儿题仍逐题落库（推理不变；record 显式携带传入的 run_id）
+        rows = harness_log.query("SELECT * FROM predictions WHERE run_id = ?", ("run-orphan-e2e",))
         assert {r["question_id"] for r in rows} == {"s1", "po"}

tests/unit/test_runner_diag_tree_inject.py

@@ -64,7 +64,6 @@ def _base_config(workspace_dir: Path, store_dir: Path) -> RunConfig:
         gate_delta_min=0.02,
         gate_lambda_dir=-0.642,
         gate_e_rollback=10.0,
-        gate_block=8,
         gate_n_max=40,
         gate_p_low=0.05,
         gate_p_high=0.95,

tests/unit/test_step_rerun_idempotent.py

@@ -0,0 +1,92 @@
+"""step 重跑幂等：gate 派生行必须随 step 清理，否则崩溃重跑累积重复。"""
+
+from __future__ import annotations
+
+import sqlite3
+from typing import TYPE_CHECKING
+
+from app.harness.runner import _clear_step_rows
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+
+def _mk_db(tmp_path: Path) -> Path:
+    """构造含 rollout 行、gate 派生行、他 step 行与前缀陷阱行的最小 harness.db。
+
+    参数:
+        tmp_path: pytest 临时目录。
+
+    返回:
+        harness.db 路径。
+    """
+    db = tmp_path / "harness.db"
+    conn = sqlite3.connect(db)
+    conn.execute("CREATE TABLE predictions (run_id TEXT, question_id TEXT)")
+    conn.execute("CREATE TABLE traces (run_id TEXT, question_id TEXT)")
+    conn.execute("CREATE TABLE gate_evidence (run_id TEXT, epoch INTEGER, step INTEGER)")
+    conn.execute("CREATE TABLE quadrant_pair (run_id TEXT, epoch INTEGER, step INTEGER)")
+    rows = [
+        ("infer_adhoc_e1_s0", "q1"),  # rollout 行
+        ("infer_adhoc_e1_s0_gate_action-reasoning_base", "q2"),  # gate base 臂
+        ("infer_adhoc_e1_s0_gate_action-reasoning_cand", "q3"),  # gate cand 臂
+        ("infer_adhoc_e1_s1", "q4"),  # 其他 step，不许误删
+        ("infer_adhoc_e1_s10_gate_x_base", "q5"),  # s10 前缀陷阱，不许误删
+    ]
+    conn.executemany("INSERT INTO predictions VALUES (?, ?)", rows)
+    conn.executemany("INSERT INTO traces VALUES (?, ?)", rows)
+    conn.execute("INSERT INTO gate_evidence VALUES ('infer_adhoc', 1, 0)")
+    conn.execute("INSERT INTO gate_evidence VALUES ('infer_adhoc', 1, 1)")
+    conn.execute("INSERT INTO quadrant_pair VALUES ('infer_adhoc', 1, 0)")
+    conn.commit()
+    conn.close()
+    return db
+
+
+def test_clear_step_rows_removes_rollout_and_gate_rows(tmp_path) -> None:
+    """rollout 行 + 本 step 全部 gate 派生行被清；他 step 与 s10 前缀陷阱不动。"""
+    db = _mk_db(tmp_path)
+    _clear_step_rows(str(db), baseline_run_id="infer_adhoc", epoch=1, step=0)
+    conn = sqlite3.connect(db)
+    left = {r[0] for r in conn.execute("SELECT run_id FROM predictions")}
+    assert left == {"infer_adhoc_e1_s1", "infer_adhoc_e1_s10_gate_x_base"}
+    left_t = {r[0] for r in conn.execute("SELECT run_id FROM traces")}
+    assert left_t == left
+    ge = list(conn.execute("SELECT step FROM gate_evidence"))
+    assert ge == [(1,)]  # 只剩 step=1 的行
+    assert list(conn.execute("SELECT COUNT(*) FROM quadrant_pair"))[0][0] == 0
+    conn.close()
+
+
+def test_clear_step_rows_missing_tables_is_noop(tmp_path) -> None:
+    """gate_evidence/quadrant_pair 表尚未建（首个 step）时不报错。"""
+    db = tmp_path / "harness.db"
+    conn = sqlite3.connect(db)
+    conn.execute("CREATE TABLE predictions (run_id TEXT)")
+    conn.execute("CREATE TABLE traces (run_id TEXT)")
+    conn.commit()
+    conn.close()
+    _clear_step_rows(str(db), baseline_run_id="infer_adhoc", epoch=1, step=0)
+
+
+def test_clear_step_rows_like_specials_in_run_id(tmp_path) -> None:
+    """run_id 含 % 与反斜杠时不通配误删他 run 行(LIKE 全特殊字符转义回归锁)。"""
+    db = tmp_path / "harness.db"
+    conn = sqlite3.connect(db)
+    conn.execute("CREATE TABLE predictions (run_id TEXT, question_id TEXT)")
+    conn.execute("CREATE TABLE traces (run_id TEXT, question_id TEXT)")
+    rows = [
+        (r"we%ird\run_e1_s0", "q1"),  # 本 step rollout
+        (r"we%ird\run_e1_s0_gate_x_base", "q2"),  # 本 step gate 行
+        (r"weXird\run_e1_s0_gate_x_base", "q3"),  # % 若未转义会误匹配此行
+        (r"we%irdXrun_e1_s0_gate_x_base", "q4"),  # \ 若未转义会误匹配此行
+    ]
+    conn.executemany("INSERT INTO predictions VALUES (?, ?)", rows)
+    conn.executemany("INSERT INTO traces VALUES (?, ?)", rows)
+    conn.commit()
+    conn.close()
+    _clear_step_rows(str(db), baseline_run_id=r"we%ird\run", epoch=1, step=0)
+    conn = sqlite3.connect(db)
+    left = {r[0] for r in conn.execute("SELECT run_id FROM predictions")}
+    conn.close()
+    assert left == {r"weXird\run_e1_s0_gate_x_base", r"we%irdXrun_e1_s0_gate_x_base"}