From 21c360bc8711cea1067aa2ee81369aff0d6ccb82 Mon Sep 17 00:00:00 2001 From: iomgaa Date: Thu, 16 Jul 2026 23:44:32 -0400 Subject: [PATCH] feat: gate prefix-ordered consumption core (algo #6) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CE-Gate 语义修订获批:块序贯 → 阶梯序前缀逐对序贯。新增 GateSpec/_UnitSlot/ _GateRun 数据结构与 _advance_prefix 纯逻辑(乱序到达下统计严格按预声明阶梯序 消费,INFRA 剔除后重判防 continue 悬置,过线即冻结)。旧块路径共存,Task 6 删。 --- app/harness/validate.py | 135 +++++++++++++++++++++++++++++++++ tests/unit/test_gate_prefix.py | 134 ++++++++++++++++++++++++++++++++ 2 files changed, 269 insertions(+) create mode 100644 tests/unit/test_gate_prefix.py diff --git a/app/harness/validate.py b/app/harness/validate.py index 3e37756..0e2adf7 100644 --- a/app/harness/validate.py +++ b/app/harness/validate.py @@ -793,3 +793,138 @@ async def validate_skill_local( shutil.rmtree(cand_dir) except OSError as e: logger.warning("候选临时目录清理失败 {}: {}", cand_dir, e) + + +# --------------------------------------------------------------------------- +# 连续并发 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 + # 剔除使 n_remaining 缩小,必须重判(Codex plan 审 C1):否则尾部全 INFRA + # 时 verdict 停留在 "continue",绕过题尽第四出口且 _finalize_outcome + # 查 stop_reason 映射 KeyError。n_used==0(纯前导 INFRA)时无证据可判,跳过。 + if run.n_used > 0: + 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) + if run.verdict.decision != "continue": + run.frozen = True + 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 diff --git a/tests/unit/test_gate_prefix.py b/tests/unit/test_gate_prefix.py new file mode 100644 index 0000000..bf68ad6 --- /dev/null +++ b/tests/unit/test_gate_prefix.py @@ -0,0 +1,134 @@ +"""连续并发 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=[_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_ties_hit_futility_early_and_freeze() -> None: + """全打平(无翻转对)时 futility 出口尽早触发并冻结——早停语义(数值:W=L=0 + 时乐观 E = E(n_remaining, 0),n 小易