From 8958eee11b1fb4c9e6c890802bd93da7d919cf21 Mon Sep 17 00:00:00 2001 From: iomgaa Date: Fri, 17 Jul 2026 04:40:14 -0400 Subject: [PATCH] refactor: remove block-sequential gate path and gate_block knob (algo #6) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit config/train_videomme.yaml 同时收录待入库的实验配置变更(run_id v2 / concurrency 32 / batch_size 40)。tests/integration/test_v3_contract_e2e.py 的 run_id 断言按 Task 5 显式契约同步修正(原断言依赖旧隐式实例注入)。 --- app/harness/checkpoint.py | 1 - app/harness/config.py | 13 +- app/harness/validate.py | 435 +--------------- config/default.yaml | 1 - config/question_gen_180_补.yaml | 1 - config/question_gen_360.yaml | 1 - config/train_action_recognition.yaml | 1 - config/train_ar30.yaml | 1 - config/train_videomme.yaml | 11 +- tests/integration/test_checkpoint_pair.py | 1 - tests/integration/test_v3_contract_e2e.py | 2 +- ..._block_unit.py => test_gate_unit_scope.py} | 146 +++--- tests/unit/test_harness_checkpoint.py | 2 - tests/unit/test_harness_config.py | 17 +- tests/unit/test_harness_pools.py | 3 - tests/unit/test_harness_runner.py | 1 - tests/unit/test_harness_validate.py | 475 ++++++++---------- tests/unit/test_runner_diag_tree_inject.py | 1 - 18 files changed, 322 insertions(+), 791 deletions(-) rename tests/unit/{test_gate_block_unit.py => test_gate_unit_scope.py} (72%) diff --git a/app/harness/checkpoint.py b/app/harness/checkpoint.py index c0849d6..f892962 100644 --- a/app/harness/checkpoint.py +++ b/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", diff --git a/app/harness/config.py b/app/harness/config.py index 824735d..cc24033 100644 --- a/app/harness/config.py +++ b/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，" diff --git a/app/harness/validate.py b/app/harness/validate.py index b2228c0..f4644d8 100644 --- a/app/harness/validate.py +++ b/app/harness/validate.py @@ -1,15 +1,15 @@ -"""async 块序贯验证编排 — CE-Gate 局部验证的唯一独立子编排器。 +"""async 连续并发 gate 验证编排 — CE-Gate 局部验证的唯一独立子编排器。 -从 TRM4 core/harness/validate.py (626 行) 迁移，重大重构： -- 同步 → async（run_inference 注入为 async callable） -- _classify_quadrants → core.evolution.classify_quadrants 纯函数 -- 配对逻辑 → 复用 core.evolution.pair_block + 本地证据行组装 -- _load_run_rows / _candidate_correctness_from_db → 共享 log.query() -- materialize_candidate_skill 保持同步（纯文件操作） +多题型全部 (单元, 臂) 任务共享题槽并发（validate_skills_concurrent）， +统计推进不按到达序，而按预声明的阶梯序前缀消费（_advance_prefix）： +base 臂缓存命中瞬间返回、cand 臂必新鲜跑，两臂延迟不对称，按到达序判定 +会系统性偏向早到翻转；前缀消费把判定顺序钉回阶梯序，anytime-valid 无条件 +成立（核心算法保真 #6，语义修订：块序贯 → 阶梯序前缀逐对序贯）。 -基线与候选在同一阶梯前缀上逐块配对，只数翻转（基线错→候选对 = W， -基线对→候选错 = L），每块结束调 gate_decision 做四出口判定。 -基线侧逐题对错走 BaselineCache 内容寻址缓存，miss 才新鲜跑。 +基线与候选在同一阶梯前缀上逐单元配对，只数翻转（基线错→候选对 = W， +基线对→候选错 = L），每消费一个单元调一次 gate_decision 做四出口判定， +过线即冻结、τ 之后的 in-flight 结果整体丢弃。基线侧单元级对错走 +BaselineCache 内容寻址缓存，miss 才新鲜跑；INFRA 单元不写缓存、从配对剔除。 判定逻辑全部在 core/evolution/gate，本模块只负责推理编排与证据收集。 """ @@ -26,7 +26,7 @@ from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable from loguru import logger from app.harness.gate_ladder import BaselineCache, skill_hash -from app.harness.question_units import build_units, flatten_units, unit_correctness_view +from app.harness.question_units import build_units, unit_correctness_view from core.evolution import ( INFRA_STOP_REASONS, GateParams, @@ -68,7 +68,7 @@ class RunInferenceFn(Protocol): 调用方（runner）负责绑定 llm、tool_dispatch_fn、prompt_builder、 log、concurrency、max_steps、skill_mode 等共享依赖。 - validate 侧只传 questions、run_id、skills_dir 三个逐块变化的参数。 + validate 侧只传 questions、run_id、skills_dir 三个逐任务变化的参数。 """ async def __call__( @@ -85,21 +85,6 @@ class RunInferenceFn(Protocol): # --------------------------------------------------------------------------- -@dataclass(frozen=True) -class InferenceRunConfig: - """一次推理运行的配置三元组，把"如何跑推理"内聚成一组。 - - 字段: - concurrency: 推理并发度。 - max_steps: 单题最大推理步数。 - skill_mode: 推理 skill 模式（"auto" / "manual" / "none"）。 - """ - - concurrency: int - max_steps: int - skill_mode: str - - @dataclass class ValidationOutcome: """CE-Gate 局部验证结果：三态动作 + e-process 证据（单元口径）+ 逐题溯源对错。 @@ -260,23 +245,6 @@ def _infra_question_ids_from_db( } -def _count_infra_units(units: list[QuestionUnit], infra_qids: set[str]) -> int: - """统计含 INFRA record 的 unit 数（一个 unit 任一题 INFRA 即计 1）。 - - 使护栏分子与分母（r.total，unit 粒度）同口径：AR pair 一 unit 含两 record， - 逐 record 计数会放大分子致 gate_guard_err 误触发，破坏 unit 粒度一致性 - （核心算法保真 #5/#6）。 - - 参数: - units: 当前块的单元列表（single 或 AR pair）。 - infra_qids: 本 run 中 stop_reason 属 INFRA 故障族的 question_id 集合。 - - 返回: - 含至少一题 INFRA 的 unit 数。 - """ - return sum(1 for u in units if any(q.question_id in infra_qids for q in u.questions)) - - def _candidate_correctness_from_db( log: HarnessLog, run_id: str, @@ -296,164 +264,13 @@ def _candidate_correctness_from_db( return {q.question_id: rows.get(q.question_id, {}).get("_correct", False) for q in chunk} -# --------------------------------------------------------------------------- -# 块级 async 函数 -# --------------------------------------------------------------------------- - - -async def _resolve_baseline_block( - units: list[QuestionUnit], - task_type: str, - s_hash: str, - prompts_version: str, - baseline_cache: BaselineCache, - base_skills_dir: Path, - run_inference: RunInferenceFn, - log: HarnessLog, - run_id: str, -) -> tuple[dict[str, bool], list[QuestionUnit], int, int]: - """基线侧处理一个块：缓存优先（unit 键），miss 的单元新鲜跑基线版本并回写缓存。 - - 缓存以 unit_id 为键、存单元级对错（AR pair 双向 AND 折叠后一个布尔）。 - miss 的单元展开为逐题送推理，读回逐题预测后经 unit_correctness_view 折叠成 - 单元级对错再写缓存（核心算法保真 #5）。逐题 predictions 仍逐题落库溯源。 - - INFRA 隔离（算法 #6）：miss 单元内**任一题** stop_reason ∈ {error, parse_error} - 即判定该单元为 INFRA 故障——**不写 BaselineCache**（否则瞬时故障永久污染基线 - 快照）、**不入 b_units**、并从返回的有效单元集中剔除，避免污染 W/L 翻转与配对。 - 命中缓存的单元恒为有效（此前已成功验证过）。 - - 参数: - units: 当前块的单元列表（single 或 AR pair）。 - task_type: 当前验证题型（缓存键成分）。 - s_hash: 基线侧生效 skill 的内容哈希（缓存键成分）。 - prompts_version: 当前 prompts 版本（缓存键成分）。 - baseline_cache: 基线侧单元级对错缓存（键含 unit_id）。 - base_skills_dir: 基线 skills 版本目录。 - run_inference: 注入的 async 推理函数。 - log: HarnessLog 共享实例（推理后读预测）。 - run_id: 本块基线 run_id。 - - 返回: - (b_units, valid_units, errors_inc, denom_inc)：块内有效 unit_id -> 基线单元 - 对错、剔除 INFRA 后的有效单元列表、本块新增的 INFRA error 计数与推理题次 - 分母增量（全命中时为 0, 0）。 - """ - miss_units = [ - u - for u in units - if baseline_cache.get(task_type, s_hash, prompts_version, u.unit_id) is None - ] - errors_inc = 0 - denom_inc = 0 - infra_qids: set[str] = set() - if miss_units: - miss_questions = flatten_units(miss_units) - r_b = await run_inference(miss_questions, run_id=run_id, skills_dir=base_skills_dir) - infra_qids = _infra_question_ids_from_db(log, r_b.run_id, miss_questions) - # 护栏分子与分母（r.total，unit 粒度）同口径：含 INFRA record 的 unit 计 1， - # 避免 AR pair（一 unit 两 record）逐 record 计数放大分子致误触发；仍涵盖 - # error + parse_error（_infra_question_ids_from_db 口径），parse_error 风暴不被绕过。 - errors_inc = _count_infra_units(miss_units, infra_qids) - denom_inc = r_b.total - fresh_per_q = _candidate_correctness_from_db(log, r_b.run_id, miss_questions) - fresh_units = unit_correctness_view(miss_units, fresh_per_q) - # 只回写非 INFRA 单元；INFRA 单元不入缓存（不永久污染基线快照） - for u in miss_units: - if any(q.question_id in infra_qids for q in u.questions): - continue - baseline_cache.put(task_type, s_hash, prompts_version, u.unit_id, fresh_units[u.unit_id]) - - valid_units = [ - u for u in units if not any(q.question_id in infra_qids for q in u.questions) - ] - - b_units: dict[str, bool] = {} - for u in valid_units: - val = baseline_cache.get(task_type, s_hash, prompts_version, u.unit_id) - assert val is not None, f"基线缓存补齐后仍有 miss: unit={u.unit_id} run_id={run_id}" - b_units[u.unit_id] = val - return b_units, valid_units, errors_inc, denom_inc - - -async def _run_candidate_block( - units: list[QuestionUnit], - cand_dir: Path, - run_inference: RunInferenceFn, - log: HarnessLog, - run_id: str, -) -> tuple[dict[str, bool], int, int]: - """候选侧处理一个块：单元展开为逐题全块新鲜跑候选版本并从 db 读逐题对错。 - - 返回逐题对错映射（question_id -> bool），折叠为单元视图交由调用方完成， - 逐题结果同时用于 candidate_correctness 溯源与二轨 correctness 合并。 - - 参数: - units: 当前块的单元列表。 - cand_dir: 已物化的候选 skills 目录。 - run_inference: 注入的 async 推理函数。 - log: HarnessLog 共享实例（推理后读预测）。 - run_id: 本块候选 run_id。 - - 返回: - (c_per_q, errors_inc, denom_inc)：块内 question_id -> 候选对错。 - """ - questions = flatten_units(units) - r_c = await run_inference(questions, run_id=run_id, skills_dir=cand_dir) - c_per_q = _candidate_correctness_from_db(log, r_c.run_id, questions) - infra_qids = _infra_question_ids_from_db(log, r_c.run_id, questions) - # 护栏分子与分母（r.total，unit 粒度）同口径：含 INFRA record 的 unit 计 1 - # （见 _count_infra_units），涵盖 error + parse_error。 - errors_inc = _count_infra_units(units, infra_qids) - return c_per_q, errors_inc, r_c.total - - -def _build_evidence_rows( - units: list[QuestionUnit], - b_units: dict[str, bool], - c_units: dict[str, bool], - task_type: str, - block_idx: int, -) -> list[dict]: - """组装一个块的 gate_evidence 单元级证据行。 - - 证据行按 unit 口径（question_id 字段存 unit_id、correct 存单元级对错）， - 与 e-process 判定同粒度；逐题预测明细仍在 predictions 表逐题溯源。 - e_value 留 None 待块判定后回填，stop_reason 留空串待终态回填。 - - 参数: - units: 当前块的单元列表。 - b_units: 块内 unit_id -> 基线单元对错。 - c_units: 块内 unit_id -> 候选单元对错。 - task_type: 当前验证题型。 - block_idx: 当前块序号。 - - 返回: - 单元级证据行列表。 - """ - return [ - { - "question_id": u.unit_id, - "task_type": task_type, - # 落库列已更名 ladder_rank(阶梯序号);旧块路径此处值仍为块号, - # 仅键名对齐 gate_evidence 表结构以保持落库兼容。 - "ladder_rank": block_idx, - "baseline_correct": b_units[u.unit_id], - "candidate_correct": c_units[u.unit_id], - "e_value": None, - "stop_reason": "", - } - for u in units - ] - - # --------------------------------------------------------------------------- # INFRA 护栏 # --------------------------------------------------------------------------- def _check_infra_guard(errors: int, infra_denom: int, gate_guard_err: float) -> None: - """跨块累计 INFRA 错误率护栏：分母 >=10 且超阈值时 raise。 + """累计 INFRA 错误率护栏：分母 >=10 且超阈值时 raise。 参数: errors: 两侧累计 error 计数。 @@ -484,13 +301,13 @@ def _finalize_outcome( evidence_rows: list[dict], task_type: str, ) -> ValidationOutcome: - """将块循环终态判定组装为 ValidationOutcome。 + """将终态判定组装为 ValidationOutcome。 四象限/准确率/W/L 均按单元口径（base_obs/cand_obs 为 unit_id -> bool）， candidate_correctness 独立保留逐题溯源（供 runner 二轨合并进 state.correctness）。 参数: - verdict: 最后一块的 gate 判定结果。 + verdict: 终态 gate 判定结果。 w: 累计 W（基线错→候选对单元翻转）。 l: 累计 L（基线对→候选错单元翻转）。 n_used: 已消费的阶梯单元数。 @@ -576,228 +393,6 @@ def _ladder_units(ladder_items: list[GeneratedQuestion]) -> list[QuestionUnit]: return units -async def _run_local_validation( - workspace_dir: Path, - cand_dir: Path, - base_skills_version: str, - task_type: str, - base_skill_content: str, - units: list[QuestionUnit], - gate_params: GateParams, - gate_block: int, - gate_guard_err: float, - baseline_cache: BaselineCache, - prompts_version: str, - run_inference: RunInferenceFn, - log: HarnessLog, - gate_run_prefix: str, -) -> ValidationOutcome: - """块序贯循环主体：逐块基线（缓存优先）/候选按单元配对推理，块间 e-process 判定。 - - 按 gate_block 切**单元**前缀（AR pair 整锁在同一块，不跨块拆分），每块先补齐 - 基线侧缓存 miss（新鲜跑基线版本并按 unit_id 写 BaselineCache），再全块跑候选， - 折叠成单元视图后配对累计 W/L 调 gate_decision；非 continue 即早停。单元尽时 - 最后一块的判定即终态（n_remaining=0 走 provisional/inertia 分支），无循环外补判。 - - 参数: - workspace_dir: Workspace 根目录。 - cand_dir: 已物化的候选 skills 目录。 - base_skills_version: 基线 skills 版本名。 - task_type: 当前验证题型。 - base_skill_content: 基线侧生效 skill 全文（skill_hash 作缓存键成分）。 - units: 已截断到 gate_n_max 的阶梯单元序（single 或 AR pair）。 - gate_params: e-process 判据阈值组。 - gate_block: 块大小（单位为**单元数**）。 - gate_guard_err: 跨块累计 INFRA 错误率护栏（分母 >=10 才触发）。 - baseline_cache: 基线侧单元级对错缓存（键含 unit_id）。 - prompts_version: 当前 prompts 版本（缓存键成分）。 - run_inference: 注入的 async 推理函数。 - log: HarnessLog 共享实例。 - gate_run_prefix: 块 run_id 前缀（含 "_gate_" 标记）。 - - 返回: - ValidationOutcome。 - - 关键实现: - INFRA 护栏跨块累计基线+候选两侧的 error 计数，分母（总推理题次，仍逐题计） - >=10 且错误率超 gate_guard_err 时直接 raise，避免坏批次污染判定。 - """ - w = 0 - l = 0 # noqa: E741 - n_used = 0 - n_excluded = 0 # 累计被 INFRA 隔离剔除的单元数（从阶梯分母扣除） - errors = 0 - infra_denom = 0 - evidence_rows: list[dict] = [] - base_obs: dict[str, bool] = {} - cand_obs: dict[str, bool] = {} - candidate_per_q: dict[str, bool] = {} - s_hash = skill_hash(base_skill_content) - base_skills_dir = workspace_dir / "skills" / base_skills_version - unit_chunks = [units[i : i + gate_block] for i in range(0, len(units), gate_block)] - n_plan = len(units) - verdict: GateVerdict | None = None - - for block_idx, unit_chunk in enumerate(unit_chunks): - # Phase 1: 基线侧（缓存优先，miss 新鲜跑，INFRA 单元剔除） - b_units, valid_chunk, err_b, den_b = await _resolve_baseline_block( - units=unit_chunk, - task_type=task_type, - s_hash=s_hash, - prompts_version=prompts_version, - baseline_cache=baseline_cache, - base_skills_dir=base_skills_dir, - run_inference=run_inference, - log=log, - run_id=f"{gate_run_prefix}_b{block_idx}_base", - ) - # 本块全 INFRA：无有效单元可配对——候选无需空跑，仅把基线侧错误计入护栏后 - # 累计剔除数进入下一块（护栏仍能在整轮 INFRA 错误率超阈值时熔断）。 - n_excluded += len(unit_chunk) - len(valid_chunk) - if not valid_chunk: - errors += err_b - infra_denom += den_b - _check_infra_guard(errors, infra_denom, gate_guard_err) - continue - - # 候选侧只跑基线侧判定有效（非 INFRA）的单元，保证配对 unit_ids 两侧一致 - c_per_q, err_c, den_c = await _run_candidate_block( - units=valid_chunk, - cand_dir=cand_dir, - run_inference=run_inference, - log=log, - run_id=f"{gate_run_prefix}_b{block_idx}_cand", - ) - - # Phase 2: INFRA 护栏（跨块累计，分母 >=10 才触发）——写缓存前置于此已由 - # _resolve_baseline_block 保证 INFRA 单元不落缓存，此处仅做整轮错误率熔断。 - errors += err_b + err_c - infra_denom += den_b + den_c - _check_infra_guard(errors, infra_denom, gate_guard_err) - - # Phase 3: 折叠成单元视图 + 配对 + 证据行 + 块间判定（均用有效单元） - c_units = unit_correctness_view(valid_chunk, c_per_q) - candidate_per_q.update(c_per_q) - unit_ids = [u.unit_id for u in valid_chunk] - pair_result = pair_block(b_units, c_units, unit_ids) - for uid, (b, c) in pair_result.observed.items(): - base_obs[uid] = b - cand_obs[uid] = c - - block_rows = _build_evidence_rows(valid_chunk, b_units, c_units, task_type, block_idx) - - w += pair_result.w - l += pair_result.l # noqa: E741 - n_used += len(valid_chunk) - # 阶梯剩余按扣除 INFRA 后的有效分母计：n_remaining = (n_plan - n_excluded) - n_used - verdict = gate_decision(w, l, n_used, (n_plan - n_excluded) - n_used, params=gate_params) - - for row in block_rows: - row["e_value"] = verdict.e_value - evidence_rows.extend(block_rows) - - if verdict.decision != "continue": - break - - # verdict 仍为 None ⟺ 全部单元被 INFRA 排除（空 ladder 已在入口拒绝）。 - # 明确失败，避免落到误导性的"空阶梯"断言而无法定位为 INFRA 原因。 - if verdict is None: - raise RuntimeError("gate 阶梯所有 unit 被判为 INFRA 排除，无法验证（检查推理基础设施）") - # 最后一块判定即终态（n_remaining=0 → provisional/inertia） - return _finalize_outcome( - verdict=verdict, - w=w, - l=l, - n_used=n_used, - n_plan=n_plan, - base_obs=base_obs, - cand_obs=cand_obs, - candidate_per_q=candidate_per_q, - evidence_rows=evidence_rows, - task_type=task_type, - ) - - -async def validate_skill_local( - workspace_dir: Path, - base_skills_version: str, - task_type: str, - target_file: str, - candidate_content: str, - base_skill_content: str, - ladder_items: list[GeneratedQuestion], - gate_params: GateParams, - gate_block: int, - gate_n_max: int, - gate_guard_err: float, - baseline_cache: BaselineCache, - prompts_version: str, - run_inference: RunInferenceFn, - log: HarnessLog, - gate_run_prefix: str, -) -> ValidationOutcome: - """块序贯配对验证：阶梯出题，基线/候选逐块配对，e-process 四出口早停。 - - 参数: - workspace_dir: workspace 根目录。 - base_skills_version: 基线 skills 版本名（候选物化复制源）。 - task_type: 待验证题型。 - target_file: fallback 解析后该题型的真实生效 skill 文件名 - （record.target_file，可能是共享 default-strategy.md）； - 候选物化写此文件，与 accept 路径同源。 - candidate_content: 候选 skill 全文。 - base_skill_content: 基线侧该题型解析后生效 skill 文件全文 - （skill_hash(base_skill_content) 作 BaselineCache 键成分）。 - ladder_items: 阶梯序题目列表（已排除本 step 案例包题）。 - gate_params: e-process 判据阈值组。 - gate_block: 块大小（单位为**单元数**，AR pair 整锁不跨块拆）。 - gate_n_max: 单 gate 单元数上限（阶梯截断到此数量个单元）。 - gate_guard_err: 跨块累计 INFRA 错误率护栏（分母 >=10 才触发）。 - baseline_cache: 基线侧单元级对错缓存（键含 unit_id）。 - prompts_version: 当前 prompts 版本（缓存键成分）。 - run_inference: 注入的 async 推理函数（RunInferenceFn 协议）。 - log: HarnessLog 共享实例（供 DB 回读逐题对错）。 - gate_run_prefix: gate 内推理 run_id 前缀，必须含 "_gate_" - （防泄露过滤靠它识别）。块 run_id = f"{prefix}_b{block_idx}_{arm}"。 - - 返回: - ValidationOutcome。单元级证据记入 outcome.evidence_rows 随结果返回， - gate_evidence 落库由调用方（runner）负责。 - """ - if "_gate_" not in gate_run_prefix: - raise ValueError(f"gate_run_prefix 必须含 '_gate_'（防泄露过滤依赖）: {gate_run_prefix!r}") - if not ladder_items: - raise ValueError(f"task_type={task_type} 阶梯为空，无法验证") - - # 阶梯题序聚合为单元并按信息阶梯序截断到 gate_n_max 个单元（AR pair 整锁不拆） - units = _ladder_units(ladder_items)[:gate_n_max] - cand_dir = materialize_candidate_skill( - workspace_dir, base_skills_version, target_file, candidate_content - ) - try: - return await _run_local_validation( - workspace_dir=workspace_dir, - cand_dir=cand_dir, - base_skills_version=base_skills_version, - task_type=task_type, - base_skill_content=base_skill_content, - units=units, - gate_params=gate_params, - gate_block=gate_block, - gate_guard_err=gate_guard_err, - baseline_cache=baseline_cache, - prompts_version=prompts_version, - run_inference=run_inference, - log=log, - gate_run_prefix=gate_run_prefix, - ) - finally: - try: - shutil.rmtree(cand_dir) - except OSError as e: - logger.warning("候选临时目录清理失败 {}: {}", cand_dir, e) - - # --------------------------------------------------------------------------- # 连续并发 gate:数据结构 + 前缀消费(algo #6 语义修订:块序贯 → 阶梯序前缀逐对序贯) # --------------------------------------------------------------------------- diff --git a/config/default.yaml b/config/default.yaml index 0048741..40c0105 100644 --- a/config/default.yaml +++ b/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 diff --git a/config/question_gen_180_补.yaml b/config/question_gen_180_补.yaml index e0bd0c6..225701c 100644 --- a/config/question_gen_180_补.yaml +++ b/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 diff --git a/config/question_gen_360.yaml b/config/question_gen_360.yaml index 1673fe5..d785dcf 100644 --- a/config/question_gen_360.yaml +++ b/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 diff --git a/config/train_action_recognition.yaml b/config/train_action_recognition.yaml index a80a3db..d09da4a 100644 --- a/config/train_action_recognition.yaml +++ b/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 diff --git a/config/train_ar30.yaml b/config/train_ar30.yaml index a5c406e..736852b 100644 --- a/config/train_ar30.yaml +++ b/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 diff --git a/config/train_videomme.yaml b/config/train_videomme.yaml index 8be6868..a20b165 100644 --- a/config/train_videomme.yaml +++ b/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 diff --git a/tests/integration/test_checkpoint_pair.py b/tests/integration/test_checkpoint_pair.py index c62fd9b..81e42fc 100644 --- a/tests/integration/test_checkpoint_pair.py +++ b/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 diff --git a/tests/integration/test_v3_contract_e2e.py b/tests/integration/test_v3_contract_e2e.py index 939d14f..e187220 100644 --- a/tests/integration/test_v3_contract_e2e.py +++ b/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}, "逐题落库题数与输入不符" diff --git a/tests/unit/test_gate_block_unit.py b/tests/unit/test_gate_unit_scope.py similarity index 72% rename from tests/unit/test_gate_block_unit.py rename to tests/unit/test_gate_unit_scope.py index b58cdd5..8795eee 100644 --- a/tests/unit/test_gate_block_unit.py +++ b/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() diff --git a/tests/unit/test_harness_checkpoint.py b/tests/unit/test_harness_checkpoint.py index e1fc36f..51d4dbc 100644 --- a/tests/unit/test_harness_checkpoint.py +++ b/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", diff --git a/tests/unit/test_harness_config.py b/tests/unit/test_harness_config.py index 7a689bd..a54c943 100644 --- a/tests/unit/test_harness_config.py +++ b/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: diff --git a/tests/unit/test_harness_pools.py b/tests/unit/test_harness_pools.py index 48cc3d6..8c17131 100644 --- a/tests/unit/test_harness_pools.py +++ b/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, diff --git a/tests/unit/test_harness_runner.py b/tests/unit/test_harness_runner.py index 44f0711..7d88af5 100644 --- a/tests/unit/test_harness_runner.py +++ b/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, diff --git a/tests/unit/test_harness_validate.py b/tests/unit/test_harness_validate.py index 97fd936..052af8a 100644 --- a/tests/unit/test_harness_validate.py +++ b/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", - ) - # 分子按 unit 计：1 个 INFRA unit（不是 2 条 record）；分母同粒度 = r.total = 1 - assert errors_inc == 1 - assert denom_inc == 1 - # 整对 INFRA → 从有效单元剔除 - assert valid_units == [] + 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 去重：双臂 × 两 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: diff --git a/tests/unit/test_runner_diag_tree_inject.py b/tests/unit/test_runner_diag_tree_inject.py index bc9b259..2d62e24 100644 --- a/tests/unit/test_runner_diag_tree_inject.py +++ b/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,