# 连续并发 gate + Redis 复用 实现计划

> **For agentic workers:** REQUIRED SUB-SKILL: Use subagent-driven-development to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.

**Goal:** 把 gate 验证从"题型串行 × 块串行 × 双臂串行"重构为"推理乱序并发 × 统计阶梯序前缀有序"的连续调度器,单 step gate 墙钟 ~9-18h → ~1.5-2.5h;顺手修复 step 重跑不清 gate 行的幂等 bug;Redis TTL 提到 7 天。

**Architecture:** 设计文档 `research-wiki/designs/2026-07-16-gate-speedup-design.md`(v3,已批准)。统计内核(`core/evolution/gate.py` 的 e-值/四出口、`pair_block` 配对、信息量阶梯、试用期)**一行不动**;只重构 `app/harness/validate.py` 的推理编排层与 `app/harness/runner.py` 的 `_gate_batch_skills` 装配层。核心不变量:**(W,L) 只按预声明阶梯序的"已配齐前缀"推进,严禁按完成到达序消费**(到达序消费会因两臂延迟不对称产生假接受偏差,Codex 审查 C1)。

**Tech Stack:** Python 3.11 asyncio(Semaphore/Lock/gather),pytest-asyncio,SQLite(HarnessLog 单连接+锁模式)。

**保真声明:** 本计划涉及核心算法 #6(块顺序验证)的**已批准语义修订**:块序贯 → 阶梯序前缀逐对序贯,判据不变。#4(CE-Gate e-process)、#5(信息阶梯)不动。每个触及 #6 的 commit message 须标注 `(algo #6)`。

---

## 全局约束(每个 Task 都适用)

- 所有命令在 `Video-Tree-TRM` conda 环境:`conda run -n Video-Tree-TRM pytest ...`
- 禁止 `except Exception: pass`;中文 docstring;loguru 日志
- 现有纯函数复用,不重写:`_ladder_units` / `_load_run_rows` / `_infra_question_ids_from_db` / `_candidate_correctness_from_db` / `_count_infra_units` / `_check_infra_guard` / `_finalize_outcome` / `materialize_candidate_skill` / `unit_correctness_view` / `pair_block` / `gate_decision`
- 每个 Task 结束跑 `conda run -n Video-Tree-TRM pytest tests/unit -x -q` 保持全绿后才 commit

---

### Task 1: 前缀消费纯逻辑 + 数据结构(validate.py 新增,不删旧码)

**Files:**
- Modify: `app/harness/validate.py`(文件末尾追加新 section)
- Test: `tests/unit/test_gate_prefix.py`(新建)

- [ ] **Step 1: 写失败测试**(核心不变量:乱序到达下统计严格按阶梯序)

```python
"""连续并发 gate 的前缀消费纯逻辑测试。"""
from __future__ import annotations

import pytest

from app.harness.validate import GateSpec, _advance_prefix, _GateRun, _UnitSlot
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",
    )


def _mk_unit(qid: str, task_type: str = "Action Reasoning") -> QuestionUnit:
    return QuestionUnit(unit_id=qid, questions=[_mk_question(qid, task_type)])


def _mk_run(n_units: int) -> _GateRun:
    spec = GateSpec(
        task_type="Action Reasoning", target_file="action-reasoning.md",
        candidate_content="cand", base_skill_content="base",
        units=[_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:
    """过线(此处 8 连胜 E≥20 confirmed)即冻结,后续已配齐单元不再消费。"""
    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 < 12  # 早停,没吃满


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_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
    assert run.verdict is not None and run.verdict.decision != "continue"
    assert run.w == 0 and run.l == 0
```

- [ ] **Step 2: 跑测试确认失败**

Run: `conda run -n Video-Tree-TRM pytest tests/unit/test_gate_prefix.py -x -q`
Expected: FAIL(ImportError: cannot import name 'GateSpec')
注意:若 `GeneratedQuestion` / `QuestionUnit` 构造字段与上述不符,先 `grep -n "class GeneratedQuestion\|class QuestionUnit" core/types.py` 按真实字段修 fixture,再确认失败原因是 ImportError。

- [ ] **Step 3: 在 `app/harness/validate.py` 末尾实现**(新 section,旧块循环暂不删)

```python
# ---------------------------------------------------------------------------
# 连续并发 gate:数据结构 + 前缀消费(algo #6 语义修订:块序贯 → 阶梯序前缀逐对序贯)
# ---------------------------------------------------------------------------


@dataclass(frozen=True)
class GateSpec:
    """单题型 gate 验证规格(runner 装配阶段产物,调度器输入)。

    字段:
        task_type: 题型。
        target_file: 解析后生效 skill 文件名(候选物化写此文件)。
        candidate_content: 候选 skill 全文。
        base_skill_content: 基线侧生效 skill 全文(skill_hash 作缓存键)。
        units: 阶梯序单元列表(已排除案例单元、截断 gate_n_max)。
        gate_run_prefix: run_id 前缀,必须含 "_gate_"(防泄露过滤依赖)。
    """

    task_type: str
    target_file: str
    candidate_content: str
    base_skill_content: str
    units: list[QuestionUnit]
    gate_run_prefix: str


@dataclass
class _UnitSlot:
    """单个阶梯单元的双臂到达状态。

    base 为单元级对错(AR pair 已折叠);cand_per_q 为逐题对错(折叠交给消费时,
    以复用 unit_correctness_view 并保留逐题溯源)。INFRA 标志与结果互斥。
    """

    unit: QuestionUnit
    base: bool | None = None
    cand_per_q: dict[str, bool] | None = None
    base_infra: bool = False
    cand_infra: bool = False

    def resolved(self) -> bool:
        """双臂均已出结果(含 INFRA 判定)。"""
        base_done = self.base is not None or self.base_infra
        cand_done = self.cand_per_q is not None or self.cand_infra
        return base_done and cand_done

    def excluded(self) -> bool:
        """任一臂 INFRA 即整单元剔除(不入配对)。"""
        return self.base_infra or self.cand_infra


@dataclass
class _GateRun:
    """单题型 gate 的运行时状态(计数器 + 前缀指针 + 证据)。"""

    spec: GateSpec
    slots: list[_UnitSlot]
    s_hash: str
    prefix_ptr: int = 0
    w: int = 0
    l: int = 0  # noqa: E741
    n_used: int = 0
    n_excluded: int = 0
    errors: int = 0
    infra_denom: int = 0
    frozen: bool = False
    verdict: GateVerdict | None = None
    base_obs: dict[str, bool] = field(default_factory=dict)
    cand_obs: dict[str, bool] = field(default_factory=dict)
    candidate_per_q: dict[str, bool] = field(default_factory=dict)
    evidence_rows: list[dict] = field(default_factory=list)

    @classmethod
    def from_spec(cls, spec: GateSpec) -> _GateRun:
        """由规格构造初始状态(slots 与阶梯序一一对应)。"""
        return cls(
            spec=spec,
            slots=[_UnitSlot(unit=u) for u in spec.units],
            s_hash=skill_hash(spec.base_skill_content),
        )


def _advance_prefix(run: _GateRun, params: GateParams) -> None:
    """沿阶梯序消费"已配齐前缀",逐单元更新 (W,L) 并判定,过线即冻结。

    统计合法性关键(设计 v3 §1 / Codex C1):严禁按到达序消费——base 臂缓存命中
    瞬间返回、cand 臂必新鲜跑,两臂延迟不对称,若 cand 延迟与对错相关,早到翻转
    对系统性偏向 W 型 → e-值虚高假接受。前缀消费把判定顺序钉回预声明阶梯序,
    anytime-valid 无条件成立;INFRA 单元视为"已解决(剔除)"不阻塞前缀。
    """
    while not run.frozen and run.prefix_ptr < len(run.slots):
        slot = run.slots[run.prefix_ptr]
        if not slot.resolved():
            return
        rank = run.prefix_ptr
        run.prefix_ptr += 1
        if slot.excluded():
            run.n_excluded += 1
            continue
        uid = slot.unit.unit_id
        assert slot.base is not None and slot.cand_per_q is not None
        c_units = unit_correctness_view([slot.unit], slot.cand_per_q)
        pair_result = pair_block({uid: slot.base}, c_units, [uid])
        run.candidate_per_q.update(slot.cand_per_q)
        for u, (b, c) in pair_result.observed.items():
            run.base_obs[u] = b
            run.cand_obs[u] = c
        run.w += pair_result.w
        run.l += pair_result.l
        run.n_used += 1
        n_remaining = (len(run.slots) - run.n_excluded) - run.n_used
        run.verdict = gate_decision(run.w, run.l, run.n_used, n_remaining, params=params)
        run.evidence_rows.append(
            {
                "question_id": uid,
                "task_type": run.spec.task_type,
                "ladder_rank": rank,
                "baseline_correct": slot.base,
                "candidate_correct": c_units[uid],
                "e_value": run.verdict.e_value,
                "stop_reason": "",
            }
        )
        if run.verdict.decision != "continue":
            run.frozen = True
```

- [ ] **Step 4: 跑测试确认通过**

Run: `conda run -n Video-Tree-TRM pytest tests/unit/test_gate_prefix.py -x -q`
Expected: 5 passed

- [ ] **Step 5: Commit**

```bash
git add app/harness/validate.py tests/unit/test_gate_prefix.py
git commit -m "feat: gate prefix-ordered consumption core (algo #6)"
```

---

### Task 2: 题槽并发闸 + 单元臂执行任务

**Files:**
- Modify: `app/harness/validate.py`(继续追加)
- Test: `tests/unit/test_gate_unit_arm.py`(新建)

- [ ] **Step 1: 写失败测试**

```python
"""单元臂执行任务测试:缓存命中/新鲜跑/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 core.evolution import GateParams

from tests.unit.test_gate_prefix import _mk_unit, _PARAMS  # 复用 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]
        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]:
    spec = GateSpec(
        task_type="Action Reasoning", target_file="action-reasoning.md",
        candidate_content="cand", base_skill_content="base",
        units=[_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
```

- [ ] **Step 2: 跑测试确认失败**

Run: `conda run -n Video-Tree-TRM pytest tests/unit/test_gate_unit_arm.py -x -q`
Expected: FAIL(ImportError: cannot import name '_QuestionSlots')

- [ ] **Step 3: 实现 `_QuestionSlots` 与 `_run_unit_arm`**(validate.py 追加)

```python
class _QuestionSlots:
    """按题数计数的共享并发闸:峰值在飞请求恒 ≤ width(设计 v3 §2.4)。

    多槽获取(AR pair 一单元两题)经内部锁串行化,防多任务半持有交错死锁。
    asyncio.Semaphore 等待队列 FIFO,任务按创建序(题型 round-robin)获得槽,
    即公平调度的实现载体(Codex I2)。
    """

    def __init__(self, width: int) -> None:
        assert width > 0, f"并发宽度必须为正: {width}"
        self._sem = asyncio.Semaphore(width)
        self._acquire_lock = asyncio.Lock()

    async def acquire(self, n: int) -> None:
        """原子获取 n 个题槽。"""
        async with self._acquire_lock:
            for _ in range(n):
                await self._sem.acquire()

    def release(self, n: int) -> None:
        """归还 n 个题槽。"""
        for _ in range(n):
            self._sem.release()


async def _run_unit_arm(
    run: _GateRun,
    slot_idx: int,
    arm: str,
    slots: _QuestionSlots,
    run_inference: RunInferenceFn,
    log: HarnessLog,
    baseline_cache: BaselineCache,
    prompts_version: str,
    base_skills_dir: Path,
    cand_dir: Path,
    gate_params: GateParams,
    gate_guard_err: float,
) -> None:
    """执行一个 (单元, 臂) 任务:缓存/推理 → 到达登记 → 前缀消费推进。

    冻结检查两次:启动时(排队任务撤销点)与获得题槽后(获槽期间被冻结)。
    base 臂缓存命中不占题槽(零推理);INFRA 单元不写缓存(不永久污染基线快照)。
    护栏在每次臂完成时检查(等价迁移自跨块累计,设计 v3 §2.3),超阈值 raise
    中止整轮(与现行行为一致)。
    """
    assert arm in ("base", "cand")
    if run.frozen:
        return
    slot = run.slots[slot_idx]
    spec = run.spec

    if arm == "base":
        cached = baseline_cache.get(
            spec.task_type, run.s_hash, prompts_version, slot.unit.unit_id
        )
        if cached is not None:
            slot.base = cached
            _advance_prefix(run, gate_params)
            return

    questions = list(slot.unit.questions)
    await slots.acquire(len(questions))
    try:
        if run.frozen:
            return
        run_id = f"{spec.gate_run_prefix}_{arm}"
        skills_dir = base_skills_dir if arm == "base" else cand_dir
        r = await run_inference(questions, run_id=run_id, skills_dir=skills_dir)
        infra_qids = _infra_question_ids_from_db(log, r.run_id, questions)
        run.infra_denom += r.total
        if infra_qids:
            run.errors += 1
            if arm == "base":
                slot.base_infra = True
            else:
                slot.cand_infra = True
        else:
            per_q = _candidate_correctness_from_db(log, r.run_id, questions)
            if arm == "base":
                folded = unit_correctness_view([slot.unit], per_q)
                slot.base = folded[slot.unit.unit_id]
                baseline_cache.put(
                    spec.task_type, run.s_hash, prompts_version,
                    slot.unit.unit_id, slot.base,
                )
            else:
                slot.cand_per_q = per_q
        _check_infra_guard(run.errors, run.infra_denom, gate_guard_err)
    finally:
        slots.release(len(questions))
    _advance_prefix(run, gate_params)
```

- [ ] **Step 4: 跑测试确认通过**

Run: `conda run -n Video-Tree-TRM pytest tests/unit/test_gate_unit_arm.py tests/unit/test_gate_prefix.py -x -q`
Expected: 10 passed

- [ ] **Step 5: Commit**

```bash
git add app/harness/validate.py tests/unit/test_gate_unit_arm.py
git commit -m "feat: gate unit-arm tasks with question-slot gate (algo #6)"
```

---

### Task 3: 调度编排 `validate_skills_concurrent` + evidence 表 ladder_rank

**Files:**
- Modify: `app/harness/validate.py`(追加编排函数)
- Modify: `app/harness/observation.py`(`_GATE_EVIDENCE_COLS` 的 `block_idx` → `ladder_rank`;同步改 docstring)
- Test: `tests/unit/test_gate_concurrent.py`(新建)

- [ ] **Step 1: 写失败测试**

```python
"""连续并发 gate 编排测试:乱序到达/多题型隔离/round-robin/终态组装。"""
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, validate_skills_concurrent
from tests.unit.test_gate_prefix import _mk_unit, _PARAMS
from tests.unit.test_gate_unit_arm import _FakeLog


def _mk_spec(task_type: str, slug: str, n: int) -> GateSpec:
    return GateSpec(
        task_type=task_type, target_file=f"{slug}.md",
        candidate_content=f"cand-{slug}", base_skill_content=f"base-{slug}",
        units=[_mk_unit(f"{slug}-q{i}", task_type) for i in range(n)],
        gate_run_prefix=f"r_e1_s0_gate_{slug}",
    )


def _scripted_inference(log: _FakeLog, script: dict[str, tuple[bool, float]]):
    """脚本化假推理:按 question_id 决定 (对错, 延迟秒),制造乱序到达。"""

    class _R:
        def __init__(self, run_id: str, total: int) -> None:
            self.run_id = run_id
            self.total = total

    async def _run(questions, *, run_id: str, skills_dir: Path):
        arm = "cand" if run_id.endswith("_cand") else "base"
        correct, delay = script[f"{questions[0].question_id}|{arm}"]
        await asyncio.sleep(delay)
        for q in questions:
            log.rows.append({
                "run_id": run_id, "question_id": q.question_id,
                "prediction": "A" if correct else "B", "answer": "A",
                "stop_reason": "finished", "steps_json": "[]",
            })
        return _R(run_id, len(questions))

    return _run


@pytest.mark.asyncio
async def test_out_of_order_arrival_still_ladder_order(tmp_path, monkeypatch) -> None:
    """尾部先到、头部后到:判定结果与顺序到达完全相同(前缀有序性端到端)。"""
    spec = _mk_spec("Action Reasoning", "action-reasoning", 4)
    log = _FakeLog()
    # 头部 q0 最慢;全部翻转为 W(base 错 cand 对)
    script = {}
    for i in range(4):
        script[f"action-reasoning-q{i}|base"] = (False, 0.05 if i == 0 else 0.0)
        script[f"action-reasoning-q{i}|cand"] = (True, 0.05 if i == 0 else 0.0)
    monkeypatch.setattr(
        "app.harness.validate.materialize_candidate_skill",
        lambda *a, **k: tmp_path / "cand",
    )
    outcomes = await validate_skills_concurrent(
        workspace_dir=tmp_path, base_skills_version="v1", specs=[spec],
        gate_params=_PARAMS, gate_guard_err=0.10,
        baseline_cache=BaselineCache(tmp_path / "bc.json"),
        prompts_version="v1", run_inference=_scripted_inference(log, script),
        log=log, concurrency=8,
    )
    o = outcomes["Action Reasoning"]
    assert o.w == 4 and o.l == 0
    assert [r["ladder_rank"] for r in o.evidence_rows] == [0, 1, 2, 3]


@pytest.mark.asyncio
async def test_two_types_isolated(tmp_path, monkeypatch) -> None:
    """两题型并行:计数互不污染,各自独立判定。

    A 型 4 单元全 W(题尽 accept_provisional);B 型 2 单元全平
    (futility 早停,W=L=0)——两型结果都不受对方污染。
    """
    spec_a = _mk_spec("Action Reasoning", "action-reasoning", 4)
    spec_b = _mk_spec("Counting Problem", "counting-problem", 2)
    log = _FakeLog()
    script = {}
    for i in range(4):
        script[f"action-reasoning-q{i}|base"] = (False, 0.0)
        script[f"action-reasoning-q{i}|cand"] = (True, 0.0)
    for i in range(2):
        script[f"counting-problem-q{i}|base"] = (True, 0.0)
        script[f"counting-problem-q{i}|cand"] = (True, 0.0)
    monkeypatch.setattr(
        "app.harness.validate.materialize_candidate_skill",
        lambda *a, **k: tmp_path / "cand",
    )
    outcomes = await validate_skills_concurrent(
        workspace_dir=tmp_path, base_skills_version="v1",
        specs=[spec_a, spec_b], gate_params=_PARAMS, gate_guard_err=0.10,
        baseline_cache=BaselineCache(tmp_path / "bc.json"),
        prompts_version="v1", run_inference=_scripted_inference(log, script),
        log=log, concurrency=8,
    )
    assert outcomes["Action Reasoning"].w == 4
    assert outcomes["Counting Problem"].w == 0
    assert outcomes["Counting Problem"].l == 0


@pytest.mark.asyncio
async def test_all_infra_raises(tmp_path, monkeypatch) -> None:
    """全单元 INFRA:保留现行 RuntimeError 语义(检查推理基础设施)。"""
    spec = _mk_spec("Action Reasoning", "action-reasoning", 2)
    log = _FakeLog()

    class _R:
        def __init__(self, run_id, total):
            self.run_id, self.total = run_id, total

    async def _infra_run(questions, *, run_id, skills_dir):
        for q in questions:
            log.rows.append({
                "run_id": run_id, "question_id": q.question_id,
                "prediction": "", "answer": "A",
                "stop_reason": "error", "steps_json": "[]",
            })
        return _R(run_id, len(questions))

    monkeypatch.setattr(
        "app.harness.validate.materialize_candidate_skill",
        lambda *a, **k: tmp_path / "cand",
    )
    with pytest.raises(RuntimeError):
        await validate_skills_concurrent(
            workspace_dir=tmp_path, base_skills_version="v1", specs=[spec],
            gate_params=_PARAMS, gate_guard_err=0.99,  # 护栏放宽,逼出全 INFRA 分支
            baseline_cache=BaselineCache(tmp_path / "bc.json"),
            prompts_version="v1", run_inference=_infra_run,
            log=log, concurrency=8,
        )
```

- [ ] **Step 2: 跑测试确认失败**

Run: `conda run -n Video-Tree-TRM pytest tests/unit/test_gate_concurrent.py -x -q`
Expected: FAIL(ImportError: cannot import name 'validate_skills_concurrent')

- [ ] **Step 3: 实现编排函数**(validate.py 追加)

```python
async def validate_skills_concurrent(
    workspace_dir: Path,
    base_skills_version: str,
    specs: list[GateSpec],
    gate_params: GateParams,
    gate_guard_err: float,
    baseline_cache: BaselineCache,
    prompts_version: str,
    run_inference: RunInferenceFn,
    log: HarnessLog,
    concurrency: int,
) -> dict[str, ValidationOutcome]:
    """连续并发 gate:多题型全部臂共享题槽并发,统计按阶梯序前缀有序推进。

    发射顺序 = 题型 round-robin × 题型内阶梯序(base 先 cand 后),asyncio.Semaphore
    FIFO 等待队列保证获槽顺序即发射顺序 → 公平调度,大题型不饿死小题型。
    题型过线即冻结:其排队任务启动时自查冻结标志撤销;in-flight 跑完落库但
    不计入(τ 之后样本,合法丢弃)。全部题型判定后统一组装 ValidationOutcome。

    参数:
        workspace_dir: workspace 根目录(候选物化用)。
        base_skills_version: 基线 skills 版本名。
        specs: 各题型 gate 规格(units 已阶梯序 + 截断 n_max)。
        gate_params / gate_guard_err: e-process 判据与 INFRA 护栏阈值。
        baseline_cache / prompts_version: 基线缓存及其键成分。
        run_inference: 注入推理函数(调用方须绑定共享 HarnessLog,见 runner)。
        log: HarnessLog 共享实例(推理后读预测,与 run_inference 同库)。
        concurrency: 题槽宽度(峰值在飞题数上限)。

    返回:
        {task_type: ValidationOutcome}。

    异常:
        RuntimeError: INFRA 护栏超阈值,或某题型全部单元被 INFRA 排除。
        ValueError: spec 校验失败(空阶梯 / run_prefix 缺 "_gate_")。
    """
    for spec in specs:
        if "_gate_" not in spec.gate_run_prefix:
            raise ValueError(f"gate_run_prefix 必须含 '_gate_': {spec.gate_run_prefix!r}")
        if not spec.units:
            raise ValueError(f"task_type={spec.task_type} 阶梯为空,无法验证")

    base_skills_dir = workspace_dir / "skills" / base_skills_version
    runs = [_GateRun.from_spec(s) for s in specs]
    cand_dirs = {
        r.spec.task_type: materialize_candidate_skill(
            workspace_dir, base_skills_version, r.spec.target_file, r.spec.candidate_content
        )
        for r in runs
    }
    slots_gate = _QuestionSlots(concurrency)
    try:
        # 发射队列:round-robin 交错(rank 0 各题型 → rank 1 各题型 → ...)
        coros = []
        max_rank = max(len(r.slots) for r in runs)
        for rank in range(max_rank):
            for r in runs:
                if rank < len(r.slots):
                    for arm in ("base", "cand"):
                        coros.append(
                            _run_unit_arm(
                                r, rank, arm, slots_gate, run_inference, log,
                                baseline_cache, prompts_version,
                                base_skills_dir, cand_dirs[r.spec.task_type],
                                gate_params, gate_guard_err,
                            )
                        )
        await asyncio.gather(*coros)
    finally:
        for d in cand_dirs.values():
            try:
                shutil.rmtree(d)
            except OSError as e:
                logger.warning("候选临时目录清理失败 {}: {}", d, e)

    outcomes: dict[str, ValidationOutcome] = {}
    for r in runs:
        if r.verdict is None:
            raise RuntimeError(
                f"gate[{r.spec.task_type}] 全部 unit 被判为 INFRA 排除,无法验证(检查推理基础设施)"
            )
        outcomes[r.spec.task_type] = _finalize_outcome(
            verdict=r.verdict, w=r.w, l=r.l, n_used=r.n_used, n_plan=len(r.slots),
            base_obs=r.base_obs, cand_obs=r.cand_obs,
            candidate_per_q=r.candidate_per_q, evidence_rows=r.evidence_rows,
            task_type=r.spec.task_type,
        )
    return outcomes
```

顶部 import 需补 `import asyncio`。

- [ ] **Step 4: `observation.py` 改 evidence 列**

找到 `_GATE_EVIDENCE_COLS` 定义(`grep -n "_GATE_EVIDENCE_COLS" app/harness/observation.py`),把 `"block_idx"` 列改名 `"ladder_rank"`(类型不变 INTEGER),`write_gate_evidence` docstring 中 `block_idx` 描述同步改为 `ladder_rank(阶梯序号)`。

- [ ] **Step 5: 跑测试确认通过**

Run: `conda run -n Video-Tree-TRM pytest tests/unit/test_gate_concurrent.py tests/unit/test_gate_prefix.py tests/unit/test_gate_unit_arm.py -x -q`
Expected: 13 passed
注意:`test_out_of_order_arrival_still_ladder_order` 里 cand 臂识别依赖 `run_id.endswith("_cand")`,与实现的 run_id 命名 `{prefix}_{arm}` 一致;若失败先核对 run_id 拼接。

- [ ] **Step 6: Commit**

```bash
git add app/harness/validate.py app/harness/observation.py tests/unit/test_gate_concurrent.py
git commit -m "feat: continuous concurrent gate orchestrator (algo #6)"
```

---

### Task 4: step 重跑幂等修复(现行潜伏 bug)

**Files:**
- Modify: `app/harness/runner.py:1105-1109`(`_run_step` 开头的清行块)
- Test: `tests/unit/test_step_rerun_idempotent.py`(新建)

- [ ] **Step 1: 写失败测试**

```python
"""step 重跑幂等:gate 派生行必须随 step 清理,否则崩溃重跑累积重复。"""
from __future__ import annotations

import sqlite3
from pathlib import Path

from app.harness.runner import _clear_step_rows


def _mk_db(tmp_path: Path) -> Path:
    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:
    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)
```

- [ ] **Step 2: 跑测试确认失败**

Run: `conda run -n Video-Tree-TRM pytest tests/unit/test_step_rerun_idempotent.py -x -q`
Expected: FAIL(ImportError: cannot import name '_clear_step_rows')

- [ ] **Step 3: 实现 `_clear_step_rows`(runner.py 模块级函数)并替换 `_run_step` 内清行块**

新函数(放在 `_write_skip_report` 附近的模块级函数区):

```python
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,天然幂等。
    """
    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}"
    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,))
            # ESCAPE 显式声明,防 run_id 中出现 '_' 通配歧义(_gate_ 前缀含字面下划线,
            # LIKE 的 '_' 单字符通配在此无害但语义须钉死为字面匹配 + '%' 后缀)
            log.execute(
                f"DELETE FROM {table} WHERE run_id LIKE ? ESCAPE '\\'",
                (step_run_id.replace("_", r"\_") + 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),
                )
```

`_run_step` 内原 `with HarnessLog(...) as log: ... DELETE ...` 五行(runner.py:1105-1109)替换为一行:

```python
        _clear_step_rows(
            str(self._paths.db_path),
            baseline_run_id=pools.baseline_run_id, epoch=epoch, step=step,
        )
```

- [ ] **Step 4: 跑测试确认通过 + 回归**

Run: `conda run -n Video-Tree-TRM pytest tests/unit/test_step_rerun_idempotent.py tests/unit -x -q`
Expected: 全绿(重点:runner 相关既有测试不破)

- [ ] **Step 5: Commit**

```bash
git add app/harness/runner.py tests/unit/test_step_rerun_idempotent.py
git commit -m "fix: clear gate-derived rows on step rerun (idempotency)"
```

---

### Task 5: runner 重接线 —— 并行进化 + 装配 + 连续 gate + 字母序落账

**Files:**
- Modify: `app/harness/runner.py`(`_gate_batch_skills`:1143-1252 重写;`_run_gate_validation`:1254-1334 删除;`_make_validate_run_inference_fn` 改共享 log)
- Test: `tests/unit/test_gate_batch_parallel.py`(新建)

- [ ] **Step 1: 写失败测试**(target_file 冲突 fail-fast 是纯逻辑,可直接测)

```python
"""_gate_batch_skills 并行装配的纯逻辑护栏测试。"""
from __future__ import annotations

import pytest

from app.harness.runner import _assert_disjoint_target_files


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"}
        )
```

- [ ] **Step 2: 跑测试确认失败**

Run: `conda run -n Video-Tree-TRM pytest tests/unit/test_gate_batch_parallel.py -x -q`
Expected: FAIL(ImportError)

- [ ] **Step 3: 实现 runner 侧改造**

3a. 模块级函数(放 `_write_skip_report` 附近):

```python
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 题型均有专属文件,此断言防未来配置漂移)。
    """
    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
```

3b. `_make_validate_run_inference_fn` 改为共享单一 HarnessLog(消除 per-unit 新建连接的锁竞争,对齐单连接+锁模式):

```python
    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。
        """
        from app.harness.inference import run_inference

        recorded: set[str] = set()

        async def _run(
            questions: list[GeneratedQuestion],
            *,
            run_id: str,
            skills_dir: Path,
        ) -> InferenceResult:
            if run_id not in recorded:
                recorded.add(run_id)
                self._record_run(run_id)
            return await run_inference(
                questions=questions,
                llm=self._llm,
                tool_dispatch_fn=self._make_tool_dispatch_fn(skills_dir=skills_dir),
                prompt_builder=self._make_prompt_builder(
                    skills_dir=skills_dir, prompts_dir=self._paths.prompts_dir
                ),
                log=gate_log,
                run_id=run_id,
                concurrency=self._config.concurrency,
                max_steps=self._config.max_steps,
                skill_mode=self._config.skill_mode,
            )

        return _run
```

3c. `_gate_batch_skills` 重写为四阶段(替换 1143-1252 的串行 for;`_run_gate_validation` 整个删除,其阶梯装配逻辑收编进 Phase B):

```python
    async def _gate_batch_skills(
        self,
        epoch: int,
        step: int,
        diagnosis: DiagnosisResult,
        total_steps: int,
        pools: Pools,
        state: _TrainState,
    ) -> None:
        """按 task_type 并行 evolve → 连续并发 gate → 字母序统一落账。

        四阶段(设计 v3 §2.1):Phase A 并行进化(cooldown/无改动照旧跳过);
        Phase B 装配 GateSpec(阶梯出题 + 案例单元排除 + n_max 截断);
        Phase C validate_skills_concurrent(共享题槽,统计按阶梯序前缀);
        Phase D 唯一写 state 阶段——按字母序 accept/reject 落账,与原串行
        语义等价(题型 skill 文件不相交,合并顺序仅为确定性)。
        """
        from app.harness.workspace import VersionedSkillStore
        from core.evolution import evolve_single_skill

        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: 并行进化(冷却/跳过路径先出清) ----
        active_types: list[str] = []
        for task_type in sorted(diagnosis.skill_case_packs):
            if state.gate_cooldown.get(task_type, 0) > 0:
                _write_skip_report(
                    self._config.workspace_dir, epoch, step, state.global_step,
                    task_type, action="cooldown",
                    baseline_acc=self._class_baseline_acc(
                        task_type, pools.validation, state.correctness
                    ),
                    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)
            return await evolve_single_skill(
                self._evolve_llm, pack, skill_store, evolve_prompts,
                skills_version, budget, self._config.appendix_consolidate_threshold,
                skill_update_mode=self._config.skill_update_mode,
                rejected=state.rejected_buffer.get(task_type, []),
            )

        records = dict(
            zip(
                active_types,
                await asyncio.gather(*[_evolve_one(t) for t in active_types]),
                strict=True,
            )
        )

        # 无真实改动的题型照旧写 skipped 后出队
        gated_types: list[str] = []
        for task_type in active_types:
            record = records[task_type]
            if record.status in ("rejected", "skipped") or (
                record.evolved_content == record.original_content
            ):
                _write_skip_report(
                    self._config.workspace_dir, epoch, step, state.global_step,
                    task_type, action="skipped",
                    baseline_acc=self._class_baseline_acc(
                        task_type, pools.validation, state.correctness
                    ),
                    budget=budget,
                    rank_clip_triggered=bool(record.clip_info.get("triggered", False)),
                )
                continue
            gated_types.append(task_type)
        if not gated_types:
            return

        _assert_disjoint_target_files({t: records[t].target_file for t in gated_types})

        # ---- Phase B: 装配 GateSpec(阶梯出题,收编原 _run_gate_validation 前半) ----
        specs: list[GateSpec] = []
        for task_type in gated_types:
            record = records[task_type]
            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=_ladder_units(ladder_items)[: self._config.gate_n_max],
                    gate_run_prefix=f"{pools.baseline_run_id}_e{epoch}_s{step}_gate_{slug}",
                )
            )

        # ---- 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 阶段(字母序确定性落账) ----
        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, epoch=epoch, step=step,
                rows=outcome.evidence_rows,
            )
            write_step_report(
                self._config.workspace_dir,
                epoch=epoch, step=step, global_step=state.global_step,
                task_type=task_type, gate_action=outcome.action,
                candidate_acc=outcome.candidate_acc,
                class_baseline_acc=outcome.baseline_acc,
                edit_budget=budget,
                rank_clip_triggered=bool(record.clip_info.get("triggered", False)),
                gate_w=outcome.w, gate_l=outcome.l, gate_e_value=outcome.e_value,
                gate_n_used=outcome.n_used, gate_stop_reason=outcome.stop_reason,
            )
            write_quadrant_pairs(
                str(self._paths.db_path),
                run_id=pools.baseline_run_id, epoch=epoch, step=step,
                pairs=_outcome_to_quadrant_pairs(task_type, outcome),
            )
            if outcome.accepted:
                self._accept_skill(task_type, record, outcome, state, pools)
            else:
                self._record_rejected_skill(
                    state.rejected_buffer, task_type, record, outcome, state.global_step
                )
```

导入区补:`from app.harness.validate import GateSpec, _ladder_units, validate_skills_concurrent`(与既有 validate 导入合并;`_ladder_units` 若被视为私有,可在 validate.py 里改名 `ladder_units` 导出并同步旧调用点)。

**语义等价性注意(实现者必读):**
- 原串行版中,题型 B 的 `base_skill_content` / `base_skills_version` 读的是"A accept 之后"的版本;新版全部读 step 起点版本。设计 v3 §1 已论证内容等价(文件不相交)+ 用户批准。**但** `_accept_skill` 内部按当前 manifest 版本推进——Phase D 串行调用时每次 accept 后版本前移,第二个 accept 基于第一个 accept 后的版本追加自己的 target_file 内容,文件不相交所以互不覆盖。逐一确认 `_accept_skill`(runner.py:1340 起)满足此性质:它以 `record.evolved_content` 写 `target_file`,复制其余文件自当前版本 → 满足。
- `gate_epoch_observed` / cooldown 递减等 step 级状态,原逻辑不在本函数内改动的,保持不动。

- [ ] **Step 4: 跑测试 + 全量回归**

Run: `conda run -n Video-Tree-TRM pytest tests/unit/test_gate_batch_parallel.py tests/unit -x -q`
Expected: 新测试过;既有 runner/gate 测试中直接调用 `_run_gate_validation` 或 `validate_skill_local` 的会失败——**属预期**,Task 6 处理;本步允许用 `--deselect` 记录清单但不得改产品代码迁就旧测试。

- [ ] **Step 5: Commit**

```bash
git add app/harness/runner.py tests/unit/test_gate_batch_parallel.py
git commit -m "feat: parallel evolve + continuous gate wiring in runner (algo #6)"
```

---

### Task 6: 删除旧块路径 + gate_block 全线移除 + 旧测试迁移

**Files:**
- Modify: `app/harness/validate.py`(删 `_run_local_validation`、`validate_skill_local`、`_resolve_baseline_block`、`_run_candidate_block`、`_build_evidence_rows`)
- Modify: `app/harness/config.py:73,128,372-375`(删 gate_block 字段/文档/校验;校验改为 `gate_n_max > 0`)
- Modify: `app/harness/checkpoint.py:61`(从字段清单删 `"gate_block"`)
- Modify: `config/*.yaml` 全部含 `gate_block:` 的文件(train_videomme / default / question_gen_180_补 / question_gen_360 / train_action_recognition / train_ar30)删该行
- Modify/Delete: `tests/unit/test_gate_block_unit.py`、`tests/unit/test_harness_validate.py` 等引用被删符号的测试

- [ ] **Step 1: 全局定位被删符号的引用**

Run: `grep -rn "validate_skill_local\|_run_local_validation\|_resolve_baseline_block\|_run_candidate_block\|_build_evidence_rows\|gate_block" app/ core/ config/ tests/ main.py --include="*.py" --include="*.yaml"`
逐一处置:产品代码引用应已在 Task 5 清零(若有残留即 Task 5 遗漏,回去补);测试引用见 Step 2。

- [ ] **Step 2: 迁移旧测试**

原则:**测的语义保留,测的载体更新**。
- 纯配对/四象限/单元折叠测试(`pair_block`/`classify_quadrants`/`unit_correctness_view`,在 test_validate.py 等):不动,它们测 core 纯函数。
- 块循环编排测试(`test_gate_block_unit.py`、`test_harness_validate.py` 中调 `validate_skill_local` 的用例):改写为经 `validate_skills_concurrent` 的等价断言(单 spec 输入,断言 action/W/L/E/evidence 与原期望一致;块边界断言删除,替换为 ladder_rank 连续性断言)。改写时保持原测试意图注释。
- INFRA 护栏测试:改为构造假推理触发 `_check_infra_guard` 阈值,断言 RuntimeError(语义同前)。

- [ ] **Step 3: 删除产品代码旧路径与 gate_block**

按 Files 清单逐个删除;`config.py` 校验块 372-375 改为:

```python
    if config.gate_n_max <= 0:
        raise ValueError(f"需 gate_n_max > 0,实际: n_max={config.gate_n_max}")
```

- [ ] **Step 4: 全量回归 + lint**

Run: `conda run -n Video-Tree-TRM pytest tests/ -x -q && conda run -n Video-Tree-TRM ruff check app/ core/ adapters/ && conda run -n Video-Tree-TRM ruff format --check app/ core/ adapters/`
Expected: 全绿、零 lint 错误。覆盖率:`conda run -n Video-Tree-TRM pytest tests/ --cov=app --cov=core -q | tail -3` ≥ 80%。

- [ ] **Step 5: Commit**

```bash
git add -u app/ config/ tests/
git commit -m "refactor: remove block-sequential gate path and gate_block knob (algo #6)"
```

---

### Task 7: Redis TTL + 收尾验证

**Files:**
- Modify: `.env`(仅 `REDIS_CACHE_TTL` 一行,**严禁**触碰其他行)
- Modify: `.env.example`(同键同值,保持模板同步)

- [ ] **Step 1: 改 TTL**

```bash
sed -i 's/^REDIS_CACHE_TTL=86400$/REDIS_CACHE_TTL=604800/' .env
sed -i 's/^REDIS_CACHE_TTL=.*/REDIS_CACHE_TTL=604800/' .env.example
grep -n "^REDIS_CACHE_TTL=" .env .env.example
```

Expected: 两文件均为 604800。

- [ ] **Step 2: 全量测试终验**

Run: `conda run -n Video-Tree-TRM pytest tests/ -q --cov=app --cov=core --cov-report=term-missing | tail -15`
Expected: 全绿,覆盖率 ≥80%。

- [ ] **Step 3: Commit**(.env 不入库,仅 .env.example)

```bash
git add .env.example
git commit -m "chore: raise Redis cache TTL to 7 days"
```

---

## 运维 Runbook(实现完成后、重启训练前,控制器执行,不属实现任务)

1. **一次性续期今日 Redis 键**(scratchpad 脚本,用 dotenv 读 REDIS_URL):
   `for k in scan_iter('llm_cache:*'): 0 < ttl(k) < 604800 → expire(k, 604800)`
2. **重启训练**:
   `tmux new-session -d -s train_videomme "CUDA_VISIBLE_DEVICES=0 bash scripts/train_videomme.sh 2>&1 | tee logs/train_videomme.log"`
   (config 的 run_id 保持 `train_videomme_v2`;rollout/诊断的 Redis 盐派生自 `infer_adhoc` 不变 → 已烧调用命中复用)
3. **验证里程碑**:启动检查点(预检剔除 4 题型 / gate 阶梯 8 题型 / Best=0.7167)→ 首 batch rollout(应大量缓存命中,显著快于 27min)→ 首个 step 的连续 gate(观察多题型交错的 `_gate_` run 记录与判定日志)。

## 保真校验记录

- **#4 CE-Gate e-process**:`core/evolution/gate.py` 零改动;`gate_decision` 调用参数语义不变(n_remaining 等价迁移)。✅
- **#5 信息阶梯**:`ladder_for` / `_ladder_units` 排序与截断逻辑零改动;案例单元排除保留。✅
- **#6 块顺序验证**:**已批准语义修订**(设计 v3):块序贯 → 阶梯序前缀逐对序贯;基线缓存、INFRA 护栏、配对翻转等价迁移;本计划 Task 1-3/5-6 涉及,commit 均标注 `(algo #6)`。⚠️ 已声明
- 其余 #1-#3、#7-#12:不涉及。

## 风险与回滚

- 每 Task 独立 commit,任一集成点失败可 `git revert` 到上一 Task。
- Phase C 若在真实训练暴露调度死锁(理论上 `_QuestionSlots` 的锁序已防),现象为 gate 阶段静默无推理调用 → 立即 `tmux kill` + 回滚 Task 5/6 的 commit,退回块序贯版本重启(旧路径在 Task 6 前仍在)。