From 2296134f733c3728443a277488e51060494892d9 Mon Sep 17 00:00:00 2001 From: iomgaa Date: Tue, 7 Jul 2026 13:43:20 -0400 Subject: [PATCH] =?UTF-8?q?feat(harness):=20runner.py=20=E2=80=94=20train?= =?UTF-8?q?=20loop=20orchestrator=20(#13=20algorithm=20fidelity)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three-level nesting (epoch -> step -> per-skill), slow update 10-step sequence, checkpoint/resume, early stop, probation accept/reject/rollback. Key TRM4->TRM5 changes: - sync -> async (all inference/diagnosis/evolve/validate awaited) - LLMClient.from_env -> injected LLMProvider (DI via constructor) - Direct DB/file access -> module functions (workspace/store/log) - _TrainState as train() local, explicit param passing to helpers Module-level pure functions extracted for testability: resume_plan, _guard_infra_failures, _apply_batch_correctness, _compute_total_steps, _should_early_stop, _format_applied_edits, _fallback_summary, _write_skip_report, _outcome_to_quadrant_pairs, _build_comparison_pairs, _batch_from_ids, _snapshot_current_skills. Tests: 34 unit tests covering 13a-13e sub-tasks. Radon: all functions Grade B or better. --- app/harness/runner.py | 2146 +++++++++++++++++++++++++++++ tests/unit/test_harness_runner.py | 682 +++++++++ 2 files changed, 2828 insertions(+) create mode 100644 app/harness/runner.py create mode 100644 tests/unit/test_harness_runner.py diff --git a/app/harness/runner.py b/app/harness/runner.py new file mode 100644 index 0000000..2cc4e28 --- /dev/null +++ b/app/harness/runner.py @@ -0,0 +1,2146 @@ +"""实验运行器（瘦编排器），对标 PyTorch Trainer。 + +三级嵌套（epoch → step → per-skill）训练循环 + 慢更新十步序 + 断点续训。 +算法保真 #13：训练循环编排从 TRM4 runner.py（2273 行）迁移，逻辑不可简化。 + +关键重构（TRM4 → TRM5）： +- sync → async（await run_inference / run_diagnosis / evolve_* / validate_*） +- LLMClient.from_env → 注入 LLMProvider（self._llm / self._evolve_llm） +- 直接 DB/文件操作 → 通过模块函数（workspace / store / log / observation） +- 瘦身 2273 → ~500 行（推理/诊断/进化/验证全委托模块函数） +""" + +from __future__ import annotations + +import json +import math +import random +import shutil +import sqlite3 +import tempfile +from dataclasses import dataclass, field +from pathlib import Path +from typing import TYPE_CHECKING, Any + +from loguru import logger + +from app.harness.batching import build_batches +from app.harness.checkpoint import ( + check_fingerprint, + deserialize_state_fields, + load_checkpoint, + write_checkpoint, +) +from app.harness.config import RunConfig # noqa: TC001 — 运行时 _compute_total_steps 使用 +from app.harness.gate_ladder import BaselineCache, GatePools, build_or_load_gate_pools +from app.harness.observation import ( + write_dual_metric, + write_epoch_report, + write_gate_evidence, + write_holdout_eval, + write_quadrant_pairs, + write_shadow_gate, + write_step_report, +) +from app.harness.store import advance_version +from app.harness.validate import Probation, ValidationOutcome +from app.harness.workspace import ( + ResolvedPaths, + archive_workspace, + init_workspace, + init_workspace_from_seed, + load_manifest, + read_best, + resolve_paths, + update_best, + update_manifest, +) +from core.evolution import ( + DiagnosisResult, + GateParams, + RejectedEdit, + edit_budget_at, + momentum_inner, + probation_verdict, + replace_momentum, + resolve_skill_file, +) +from core.evolution.diagnose import merge_system_packs, merge_tool_packs + +if TYPE_CHECKING: + from app.harness.inference import InferenceResult + from app.harness.pools import Pools + from core.evolution.types import ( + EvolutionRecord, + SystemCasePack, + ToolCasePack, + ) + from core.protocols import LLMProvider, TelemetryRecorder, VLMProvider + from core.types import GeneratedQuestion + + +class _InterruptError(RuntimeError): + """测试用中断注入信号：_run_step 末尾可选抛出以模拟进程中断。""" + + +# --------------------------------------------------------------------------- +# _TrainState: 19 个可变字段 +# --------------------------------------------------------------------------- + + +@dataclass +class _TrainState: + """一次 train() 的跨 step 可变状态（训练循环的"权重/缓冲"）。 + + 字段说明见 TRM4 同名 dataclass（完整保留 19 字段语义）。 + TRM5 移除 evolve_client（改走构造注入），其余 18 字段 + gate_epoch_observed 不变。 + """ + + correctness: dict[str, bool] + gate_pools: GatePools + baseline_cache: BaselineCache + eval_prev_acc: float + eval_prev_run_id: str + best_val_acc: float + best_skills_version: str + best_prompts_version: str + baseline_skills_version: str = "" + baseline_prompts_version: str = "" + rejected_buffer: dict[str, list[RejectedEdit]] = field(default_factory=dict) + system_packs: list[SystemCasePack] = field(default_factory=list) + tool_packs: list[ToolCasePack] = field(default_factory=list) + global_step: int = 0 + changed_task_types_this_epoch: set[str] = field(default_factory=set) + epoch_start_skills: dict[str, str] = field(default_factory=dict) + steps_since_best_improved: int = 0 + gate_epoch_observed: bool = False + probations: dict[str, Probation] = field(default_factory=dict) + gate_cooldown: dict[str, int] = field(default_factory=dict) + + +# --------------------------------------------------------------------------- +# 纯函数辅助（不依赖 self） +# --------------------------------------------------------------------------- + + +def resume_plan(epoch: int, phase: str, step_completed: int) -> dict: + """据 checkpoint 进度算续跑计划（纯函数，便于单测）。 + + 参数: + epoch: checkpoint 落库时的 epoch 序号。 + phase: "in_epoch" 或 "epoch_done"。 + step_completed: 该 epoch 内最后完整完成的 step 序号。 + + 返回: + {"first_epoch": int, "resume_epoch": int | None, "resume_step_from": int}。 + """ + if phase == "epoch_done": + return {"first_epoch": epoch + 1, "resume_epoch": None, "resume_step_from": 0} + return { + "first_epoch": epoch, + "resume_epoch": epoch, + "resume_step_from": step_completed + 1, + } + + +def _guard_infra_failures(result: InferenceResult, context: str) -> None: + """基础设施失败护栏：stop_reason="error" 占比 > 10% 即硬终止。 + + 参数: + result: 推理聚合结果。 + context: 出错时报错的推理路径名（仅诊断用）。 + + 异常: + RuntimeError: error 占比 > 10%。 + """ + error_rate = result.stop_reason_counts.get("error", 0) / max(result.total, 1) + if error_rate > 0.1: + raise RuntimeError( + f"{context} 推理基础设施失败率过高 {error_rate:.0%}（stop_reason=error），中止本轮" + ) + + +def _apply_batch_correctness( + correctness: dict[str, bool], + log: Any, + run_id: str, + batch: list[GeneratedQuestion], +) -> None: + """从该 run 的 predictions 读 batch 各题新对错，就地增量更新 correctness。 + + 参数: + correctness: question_id -> 是否答对，就地更新。 + log: HarnessLog 实例。 + run_id: rollout 的 run_id。 + batch: 本 step 的题目列表。 + + 异常: + RuntimeError: rollout 不完整（缺预测行）。 + """ + from app.harness.validate import _load_run_rows + + rows = _load_run_rows(log, run_id) + missing = [q.question_id for q in batch if q.question_id not in rows] + if missing: + raise RuntimeError( + f"rollout 不完整：run_id={run_id} 缺 {len(missing)} 道题预测行 {missing}，中止本步" + ) + for q in batch: + correctness[q.question_id] = rows[q.question_id]["_correct"] + + +def _accumulate_slow_packs(diagnosis: DiagnosisResult, state: _TrainState) -> None: + """把本 step 诊断的 system/tool 案例包只累加不更新，留给 epoch 末慢更新消费。""" + if diagnosis.system_case_pack is not None: + state.system_packs.append(diagnosis.system_case_pack) + state.tool_packs.extend(diagnosis.tool_case_packs.values()) + + +def _batch_from_ids(pools: Pools, ids: list[str]) -> list[GeneratedQuestion]: + """按 question_id 从诊断池重建一个 batch（保持原 epoch 划分）。 + + 参数: + pools: 三池容器。 + ids: 一个 batch 的 question_id 列表。 + + 返回: + 按 ids 顺序取出的 GeneratedQuestion 列表。 + """ + by_id = {q.question_id: q for q in pools.diagnosis} + return [by_id[i] for i in ids] + + +def _snapshot_current_skills(skills_dir: Path) -> dict[str, str]: + """快照当前 skills 版本目录下各 skill 文件的正文（文件名 -> 全文）。 + + 参数: + skills_dir: 当前 skills 版本目录。 + + 返回: + {文件名: 全文}，作 momentum 的上一版基准。 + """ + snapshot: dict[str, str] = {} + for path in sorted(skills_dir.glob("*.md")): + snapshot[path.name] = path.read_text(encoding="utf-8") + return snapshot + + +def _should_early_stop( + workspace_dir: Path, + epoch: int, + steps_this_epoch: int, + state: _TrainState, + patience: int, +) -> bool: + """步粒度 early stop：本 epoch best 未刷新则累加本 epoch 步数。 + + 参数: + workspace_dir: workspace 目录（读 manifest best）。 + epoch: 当前 epoch。 + steps_this_epoch: 本 epoch 的 step 总数。 + state: 训练状态（steps_since_best_improved 就地更新）。 + patience: early_stop_patience。 + + 返回: + 是否触发 early stop。 + """ + best = read_best(workspace_dir) + improved_this_epoch = best is not None and best.get("epoch") == epoch + if improved_this_epoch: + state.steps_since_best_improved = 0 + return False + state.steps_since_best_improved += steps_this_epoch + return state.steps_since_best_improved >= patience + + +def _compute_total_steps(pools: Pools, correctness: dict[str, bool], config: RunConfig) -> int: + """退火地平线：用 build_batches 试切一轮拿 selected_count，再乘 epochs。""" + _, selected_count = build_batches( + pools.diagnosis, + correctness, + config.batch_size, + config.min_class_per_batch, + seed=1, + correct_ratio=config.batch_correct_ratio, + ) + steps_per_epoch = max(1, math.ceil(selected_count / config.batch_size)) + return config.epochs * steps_per_epoch + + +def _outcome_to_quadrant_pairs(task_type: str, outcome: ValidationOutcome) -> list[dict]: + """把 ValidationOutcome 的四象限拍平为逐题 pair（供 quadrant_pair 表落库观测）。 + + 参数: + task_type: 该批 gate 的任务类型。 + outcome: 局部验证决策结果。 + + 返回: + 每条含 question_id/task_type/prev_correct/curr_correct/category 的 dict 列表。 + """ + from app.harness.momentum import ( + IMPROVED, + PERSISTENT_FAIL, + REGRESSED, + STABLE_SUCCESS, + ) + + spec = [ + (outcome.improvements, IMPROVED, False, True), + (outcome.regressions, REGRESSED, True, False), + (outcome.persistent_fails, PERSISTENT_FAIL, False, False), + (outcome.stable_successes, STABLE_SUCCESS, True, True), + ] + pairs: list[dict] = [] + for qids, category, prev_ok, curr_ok in spec: + for qid in qids: + pairs.append( + { + "question_id": qid, + "task_type": task_type, + "prev_correct": prev_ok, + "curr_correct": curr_ok, + "category": category, + } + ) + return pairs + + +def _build_comparison_pairs( + sampled: list[GeneratedQuestion], + prev_rows: dict[str, dict], + curr_rows: dict[str, dict], +) -> list[dict]: + """为采样好的诊断池题目构造 momentum 纵向对比对。""" + pairs: list[dict] = [] + for q in sampled: + prev = prev_rows.get(q.question_id, {}) + curr = curr_rows.get(q.question_id, {}) + pairs.append( + { + "question": q.question, + "prev_prediction": prev.get("prediction", ""), + "curr_prediction": curr.get("prediction", ""), + "correct_prev": prev.get("_correct", False), + "correct_curr": curr.get("_correct", False), + } + ) + return pairs + + +def _filter_applied_edits(edits: list[dict], reports: list[dict]) -> list[dict] | str: + """按 apply_report 过滤出真正 applied 的 edit。 + + 参数: + edits: EvolutionRecord.edits 列表。 + reports: EvolutionRecord.apply_report 列表（与 edits 同序对齐）。 + + 返回: + 过滤后的 edit 列表；0 applied 时返回信息性消息字符串。 + reports 为空时返回原 edits 不过滤。 + """ + if not reports: + return edits + applied = [ + edit + for edit, report in zip(edits, reports, strict=True) + if str(report.get("status", "")).startswith("applied") + ] + if not applied: + return "上轮改法全部未成功应用(0 applied)，本条无已验证信息" + return applied + + +def _format_applied_edits(record: Any) -> str | None: + """从 EvolutionRecord 中提取真正 applied 的 edit 并格式化为摘要。 + + 参数: + record: EvolutionRecord（duck-typed，需含 edits / apply_report）。 + + 返回: + 已 applied edit 的格式化摘要；无 edit 或无 applied 时返回 None， + 0 applied 时返回信息性消息（非 None）。 + """ + rec_edits = getattr(record, "edits", []) or [] + if not rec_edits: + return None + filtered = _filter_applied_edits(rec_edits, getattr(record, "apply_report", []) or []) + if isinstance(filtered, str): + return filtered + summary = "; ".join( + f"[{edit.get('op')}]{(edit.get('target') or edit.get('content', ''))[:40]}" + for edit in filtered + if isinstance(edit, dict) + ) + return summary or None + + +def _fallback_summary(record: Any, outcome: Any) -> str: + """从 suggestions 或跌幅信息构造兜底黑名单摘要。 + + 参数: + record: EvolutionRecord（duck-typed，需含 suggestions）。 + outcome: ValidationOutcome（duck-typed，需含 delta_hat）。 + + 返回: + 兜底摘要字符串。 + """ + return "; ".join(s.get("change", "") for s in record.suggestions) or ( + f"上轮对本文件改写被拒(delta {outcome.delta_hat:+.2f})，换方向" + ) + + +def _write_skip_report( + workspace_dir: Path, + epoch: int, + step: int, + global_step: int, + task_type: str, + action: str, + baseline_acc: float, + budget: int, + rank_clip_triggered: bool = False, +) -> None: + """为 cooldown / skipped 路径写 step_report（无 gate 证据）。 + + 参数: + workspace_dir: 工作区目录。 + epoch: 轮次。 + step: epoch 内 step。 + global_step: 全局步计数。 + task_type: 任务类型。 + action: "cooldown" 或 "skipped"。 + baseline_acc: 当前类基线准确率。 + budget: 编辑预算。 + rank_clip_triggered: 是否触发 rank 裁剪（skipped 路径需要）。 + """ + write_step_report( + workspace_dir, + epoch=epoch, + step=step, + global_step=global_step, + task_type=task_type, + gate_action=action, + candidate_acc=baseline_acc, + class_baseline_acc=baseline_acc, + edit_budget=budget, + rank_clip_triggered=rank_clip_triggered, + gate_w=None, + gate_l=None, + gate_e_value=None, + gate_n_used=None, + gate_stop_reason=None, + ) + + +# --------------------------------------------------------------------------- +# Runner 主类 +# --------------------------------------------------------------------------- + + +class Runner: + """实验运行器（瘦编排器），通过 RunConfig 驱动训练/推理/诊断/评估等模式。 + + DI 纪律：self 只持注入依赖 + _paths。_TrainState 是 train() 内局部变量， + 显式传参给模块函数。 + + 参数: + config: 运行配置。 + llm: 推理用 LLMProvider。 + evolve_llm: 进化用 LLMProvider（thinking=True）。 + vlm: VLMProvider。 + telemetry: 遥测记录端口。 + """ + + def __init__( + self, + config: RunConfig, + *, + llm: LLMProvider, + evolve_llm: LLMProvider, + vlm: VLMProvider, + telemetry: TelemetryRecorder, + ) -> None: + self._config = config + self._llm = llm + self._evolve_llm = evolve_llm + self._vlm = vlm + self._telemetry = telemetry + self._ensure_workspace() + self._paths: ResolvedPaths = resolve_paths(config.workspace_dir) + + # ----------------------------------------------------------------------- + # workspace 三态逻辑 + # ----------------------------------------------------------------------- + + def _ensure_workspace(self) -> None: + """train 模式按 --resume/--fresh 分三态；其余模式仅要求 ws 已存在并复用。 + + 三态逻辑: + resume+fresh → ValueError + fresh+已有 → archive + init_from_seed + resume+无进度 → RuntimeError + 无flag+已有 → SystemExit + """ + manifest = self._config.workspace_dir / "manifest.json" + has_progress = manifest.exists() + if self._config.mode != "train": + if not has_progress: + raise RuntimeError( + f"{self._config.mode} 模式要求 workspace 已存在: {self._config.workspace_dir}" + ) + return + if self._config.resume and self._config.fresh: + raise ValueError("--resume 与 --fresh 互斥") + if self._config.fresh: + if has_progress: + logger.info( + "旧 workspace 已归档: {}", + archive_workspace(self._config.workspace_dir), + ) + init_workspace_from_seed( + self._config.workspace_dir, + self._config.store_dir, + self._config.seed, + self._config.questions, + ) + return + if self._config.resume: + if not has_progress: + raise RuntimeError("--resume 但 workspace 无已有进度") + return + if has_progress: + raise SystemExit("workspace 已有进度；用 --resume 续训 或 --fresh 归档重开") + init_workspace( + self._config.workspace_dir, + self._config.store_dir, + self._config.questions, + self._config.skills_version, + self._config.prompts_version, + ) + + # ----------------------------------------------------------------------- + # 公共入口：infer / eval / diagnose / promote + # ----------------------------------------------------------------------- + + async def infer(self, task_types: list[str] | None = None) -> InferenceResult: + """执行单次推理（forward-only）。 + + 参数: + task_types: 若非 None，只保留指定题型。 + + 返回: + InferenceResult 冻结实例。 + """ + from app.harness.inference import run_inference + from app.harness.log import HarnessLog + from app.question_gen import load_benchmark + + questions = load_benchmark(self._paths.questions_dir) + if task_types: + allowed = set(task_types) + questions = [q for q in questions if q.task_type in allowed] + if self._config.n_samples > 0: + questions = questions[: self._config.n_samples] + + run_id = f"infer_{self._config.run_id}" if self._config.run_id else "infer_adhoc" + record_run_dir = self._record_run(run_id) # noqa: F841 + + logger.info( + "启动推理: {} 道题, concurrency={}, max_steps={}, skill_mode={}", + len(questions), + self._config.concurrency, + self._config.max_steps, + self._config.skill_mode, + ) + + with HarnessLog(str(self._paths.db_path), run_id) as log: + return await run_inference( + questions=questions, + llm=self._llm, + tool_dispatch_fn=self._make_tool_dispatch_fn(), + prompt_builder=self._make_prompt_builder(), + log=log, + run_id=run_id, + concurrency=self._config.concurrency, + max_steps=self._config.max_steps, + skill_mode=self._config.skill_mode, + ) + + async def eval(self, version: str) -> InferenceResult: + """用指定 skills 版本跑完整题库，全量记录落 db + 版本回填。 + + 参数: + version: skills 版本号。 + + 返回: + InferenceResult。 + """ + from datetime import UTC, datetime + + from app.harness.inference import run_inference + from app.harness.log import HarnessLog + from app.question_gen import load_benchmark + + cur = load_manifest(self._config.workspace_dir)["current"] + prompts_v = cur["prompts"].split("/")[-1] + skills_dir = self._paths.workspace_dir / "skills" / version + prompts_dir = self._paths.workspace_dir / "prompts" / prompts_v + if not skills_dir.is_dir(): + raise FileNotFoundError(f"skills 版本目录不存在: {skills_dir}") + if not prompts_dir.is_dir(): + raise FileNotFoundError(f"prompts 版本目录不存在: {prompts_dir}") + + questions = load_benchmark(self._paths.questions_dir) + if self._config.n_samples > 0: + questions = questions[: self._config.n_samples] + + ts = datetime.now(UTC).strftime("%Y%m%d_%H%M%S") + run_id = f"eval_{version}-{prompts_v}_{ts}" + self._record_run(run_id) + logger.info( + "eval: 版本 skills/{}+prompts/{} 跑 {} 题 (run_id={})", + version, + prompts_v, + len(questions), + run_id, + ) + + with HarnessLog(str(self._paths.db_path), run_id) as log: + result = 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=prompts_dir + ), + log=log, + run_id=run_id, + concurrency=self._config.concurrency, + max_steps=self._config.max_steps, + skill_mode=self._config.skill_mode, + ) + + # C4 回填 + self._backfill_run_versions(run_id, version, prompts_v) + self._write_eval_report(run_id, version, prompts_v, result) + return result + + async def diagnose(self, run_id: str) -> DiagnosisResult: + """执行指定 run 的两阶段诊断。 + + 参数: + run_id: 待诊断的 run_id。 + + 返回: + DiagnosisResult。 + """ + + return await self._run_diagnosis(run_id) + + def promote(self, version: str, eval_run_id: str, name: str) -> None: + """把当前 ws 的指定版本提升为 Store 新种子。 + + 参数: + version: 要提升的 skills 版本号。 + eval_run_id: canonical eval run。 + name: 新种子名。 + """ + from app.harness.store import promote_to_seed + + seed_dir = promote_to_seed( + self._config.workspace_dir, + self._config.store_dir, + version, + eval_run_id, + name, + description=f"promote from {self._config.workspace_dir.name} {version}", + ) + logger.info("已提升为种子: {}", seed_dir) + + # ----------------------------------------------------------------------- + # train() 三级嵌套（核心） + # ----------------------------------------------------------------------- + + async def train(self, pools: Pools) -> None: + """mini-batch 快慢双速闭环：epoch 内多 step、每 step 按类 per-skill gate。 + + 三级嵌套：epoch → batch(step) → per-skill。 + epoch 末 _slow_update_cycle 十步序。 + 训练收尾 _deliver_best + _final_test_eval。 + """ + state, total_steps, plan, saved_batches = await self._setup_train_run(pools) + for epoch in range(plan["first_epoch"], self._config.epochs + 1): + if epoch == plan["resume_epoch"]: + batches = [_batch_from_ids(pools, ids) for ids in saved_batches] + step_from = plan["resume_step_from"] + else: + logger.info("=== Epoch {} ===", epoch) + state.system_packs = [] + state.tool_packs = [] + state.changed_task_types_this_epoch = set() + state.epoch_start_skills = _snapshot_current_skills(self._paths.skills_dir) + batches, _ = build_batches( + pools.diagnosis, + state.correctness, + self._config.batch_size, + self._config.min_class_per_batch, + seed=epoch, + correct_ratio=self._config.batch_correct_ratio, + ) + step_from = 0 + batch_ids = [[q.question_id for q in b] for b in batches] + for step in range(step_from, len(batches)): + await self._run_step(epoch, step, total_steps, batches[step], pools, state) + state.global_step += 1 + write_checkpoint( + self._config.workspace_dir, + state=state, + epoch=epoch, + step_completed=step, + phase="in_epoch", + global_step=state.global_step, + total_steps=total_steps, + version_snapshot=self._current_version_snapshot(), + epoch_batches=batch_ids, + config=self._config, + ) + await self._slow_update_cycle(epoch, pools, state) + state.system_packs = [] + state.tool_packs = [] + state.changed_task_types_this_epoch = set() + write_checkpoint( + self._config.workspace_dir, + state=state, + epoch=epoch, + step_completed=len(batches) - 1, + phase="epoch_done", + global_step=state.global_step, + total_steps=total_steps, + version_snapshot=self._current_version_snapshot(), + epoch_batches=batch_ids, + config=self._config, + ) + if _should_early_stop( + self._config.workspace_dir, + epoch, + len(batches), + state, + self._config.early_stop_patience, + ): + logger.info("Epoch {} 触发 early stop（best 连续无新高）", epoch) + break + self._deliver_best(state.best_skills_version, state.best_prompts_version) + await self._final_test_eval(pools) + + # ----------------------------------------------------------------------- + # 训练初始化 + # ----------------------------------------------------------------------- + + async def _setup_train_run(self, pools: Pools) -> tuple[_TrainState, int, dict, list | None]: + """据是否 --resume 准备训练起点。 + + 返回: + (state, total_steps, plan, saved_batches)。 + """ + ckpt = load_checkpoint(self._config.workspace_dir) if self._config.resume else None + if self._config.resume and ckpt is None: + raise RuntimeError("--resume 但 checkpoint.json 不存在，拒绝静默从头重训") + gate_pools, baseline_cache = self._init_gate_pools(pools) + if not ckpt: + state = self._init_train_state(pools, gate_pools, baseline_cache) + total_steps = _compute_total_steps(pools, state.correctness, self._config) + plan = {"first_epoch": 1, "resume_epoch": None, "resume_step_from": 0} + return state, total_steps, plan, None + struct, decision = check_fingerprint(ckpt["config_fingerprint"], self._config) + if struct: + raise RuntimeError(f"结构性配置变化，拒绝 resume: {struct}") + if decision: + logger.warning("决策性配置变化，继续 resume: {}", decision) + state = self._restore_train_state(ckpt, pools, gate_pools, baseline_cache) + state.global_step = ckpt["progress"]["global_step"] + update_manifest( + self._config.workspace_dir, + skills=ckpt["version_snapshot"]["skills"], + prompts=ckpt["version_snapshot"]["prompts"], + ) + self._paths = resolve_paths(self._config.workspace_dir) + plan = resume_plan( + ckpt["progress"]["epoch"], + ckpt["progress"]["phase"], + ckpt["progress"]["step_completed"], + ) + return state, ckpt["progress"]["total_steps"], plan, ckpt["epoch_batches"] + + def _init_gate_pools(self, pools: Pools) -> tuple[GatePools, BaselineCache]: + """构建/加载 CE-Gate 信息量阶梯与基线缓存。 + + 副作用：设置 self._gate_questions_by_id（不进 checkpoint）。 + + 参数: + pools: 冻结三池。 + + 返回: + (GatePools, BaselineCache)。 + """ + from app.harness.log import HarnessLog + from app.question_gen import load_benchmark + + questions = load_benchmark(self._paths.questions_dir) + self._gate_questions_by_id: dict[str, GeneratedQuestion] = { + q.question_id: q for q in questions + } + with HarnessLog(str(self._paths.db_path), pools.baseline_run_id) as log: + rows = log.query( + "SELECT question_id, prediction, answer FROM predictions WHERE run_id=?", + (pools.baseline_run_id,), + ) + if not rows: + raise RuntimeError( + f"基线 run {pools.baseline_run_id} 在 predictions 表无任何行，" + "无法构建 gate 阶梯（检查种子基线是否完整落库）" + ) + baseline_correctness = {r["question_id"]: r["prediction"] == r["answer"] for r in rows} + logger.info("gate 阶梯基线对错覆盖 {} 题", len(baseline_correctness)) + gate_task_types = sorted({q.task_type for q in pools.diagnosis}) + gate_pools = build_or_load_gate_pools( + workspace_dir=self._config.workspace_dir, + questions=questions, + test_qids={q.question_id for q in pools.test}, + baseline_correctness=baseline_correctness, + task_types=gate_task_types, + probe_quota=self._config.gate_probe_quota, + seed=1, + baseline_run_id=pools.baseline_run_id, + ) + baseline_cache = BaselineCache(self._config.workspace_dir / "baseline_cache.json") + return gate_pools, baseline_cache + + def _init_train_state( + self, pools: Pools, gate_pools: GatePools, baseline_cache: BaselineCache + ) -> _TrainState: + """初始化跨 step 训练状态。""" + skills_v = self._current_version("skills") + prompts_v = self._current_version("prompts") + update_best( + self._config.workspace_dir, + skills=f"skills/{skills_v}", + prompts=f"prompts/{prompts_v}", + val_acc=pools.baseline_val_accuracy, + run_id=pools.baseline_run_id, + epoch=0, + ) + if read_best(self._config.workspace_dir) is None: + raise RuntimeError("best 指针初始化失败") + return _TrainState( + correctness=dict(pools.correctness), + gate_pools=gate_pools, + baseline_cache=baseline_cache, + eval_prev_acc=pools.baseline_val_accuracy, + eval_prev_run_id=pools.baseline_run_id, + best_val_acc=pools.baseline_val_accuracy, + best_skills_version=skills_v, + best_prompts_version=prompts_v, + baseline_skills_version=skills_v, + baseline_prompts_version=prompts_v, + ) + + def _restore_train_state( + self, + ckpt: dict, + pools: Pools, + gate_pools: GatePools, + baseline_cache: BaselineCache, + ) -> _TrainState: + """从 checkpoint 重建 _TrainState。""" + fields = deserialize_state_fields(ckpt["state"]) + best = read_best(self._config.workspace_dir) or {} + return _TrainState( + gate_pools=gate_pools, + baseline_cache=baseline_cache, + best_val_acc=best.get("val_acc", pools.baseline_val_accuracy), + best_skills_version=best.get("skills", "skills/v1").split("/")[-1], + best_prompts_version=best.get("prompts", "prompts/v1").split("/")[-1], + **fields, + ) + + # ----------------------------------------------------------------------- + # _run_step：rollout → correctness → diagnose → accumulate → gate → cooldown + # ----------------------------------------------------------------------- + + async def _run_step( + self, + epoch: int, + step: int, + total_steps: int, + batch: list[GeneratedQuestion], + pools: Pools, + state: _TrainState, + ) -> None: + """单 step：rollout → correctness 增量 → 诊断 → 累加 system/tool → 按类 gate。""" + run_id = f"{pools.baseline_run_id}_e{epoch}_s{step}" + await self._rollout_batch(batch, run_id) + + from app.harness.log import HarnessLog + + with HarnessLog(str(self._paths.db_path), run_id) as log: + _apply_batch_correctness(state.correctness, log, run_id, batch) + + diagnosis = await self._run_diagnosis(run_id, question_ids=[q.question_id for q in batch]) + _accumulate_slow_packs(diagnosis, state) + await self._gate_batch_skills(epoch, step, diagnosis, total_steps, pools, state) + # 冷却计数每 step 递减、归零剔除 + state.gate_cooldown = {t: n - 1 for t, n in state.gate_cooldown.items() if n - 1 > 0} + # 测试中断注入点 + if getattr(self, "_interrupt_after_step", None) == step: + raise _InterruptError(f"模拟中断于 epoch{epoch} step{step}") + + async def _rollout_batch(self, batch: list[GeneratedQuestion], run_id: str) -> None: + """用当前 skill 版本重推该 batch。""" + result = await self._run_inference_on_pool( + batch, run_id, self._paths.skills_dir, self._paths.prompts_dir + ) + _guard_infra_failures(result, context="rollout") + + # ----------------------------------------------------------------------- + # _gate_batch_skills：per task_type gate + # ----------------------------------------------------------------------- + + async def _gate_batch_skills( + self, + epoch: int, + step: int, + diagnosis: DiagnosisResult, + total_steps: int, + pools: Pools, + state: _TrainState, + ) -> None: + """按 task_type 独立 evolve → 局部验证 → accept/reject。""" + 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, + ) + for task_type in sorted(diagnosis.skill_case_packs): + # 冷却 admission control + if state.gate_cooldown.get(task_type, 0) > 0: + _write_skip_report( + self._config.workspace_dir, + epoch, + step, + state.global_step, + task_type, + action="cooldown", + baseline_acc=self._class_baseline_acc( + task_type, pools.validation, state.correctness + ), + budget=budget, + ) + continue + + pack = diagnosis.skill_case_packs[task_type] + skill_store = VersionedSkillStore(self._paths.skills_dir) + evolve_prompts = self._load_evolve_prompts() + record = await evolve_single_skill( + self._evolve_llm, + pack, + skill_store, + evolve_prompts, + self._current_version("skills"), + budget, + self._config.appendix_consolidate_threshold, + skill_update_mode=self._config.skill_update_mode, + rejected=state.rejected_buffer.get(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 + + outcome = await self._run_gate_validation( + epoch, step, task_type, pack, record, pools, state + ) + # 观测落库 + 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 + ) + + async def _run_gate_validation( + self, + epoch: int, + step: int, + task_type: str, + pack: Any, + record: EvolutionRecord, + pools: Pools, + state: _TrainState, + ) -> ValidationOutcome: + """CE-Gate 块序贯配对验证：阶梯出题 → 基线/候选逐块配对 → e-process 四出口。 + + 参数: + epoch: 轮次。 + step: epoch 内 step。 + task_type: 待验证题型。 + pack: SkillCasePack。 + record: 进化产物。 + pools: 冻结三池。 + state: 训练状态。 + + 返回: + ValidationOutcome。 + """ + from app.harness.log import HarnessLog + from app.harness.validate import validate_skill_local + + exclude_qids = {c.question_id for c in pack.failure_cases + pack.success_cases} + ladder_qids = state.gate_pools.ladder_for( + task_type, + exclude_qids, + p_low=self._config.gate_p_low, + p_high=self._config.gate_p_high, + cold=not state.gate_epoch_observed, + ) + missing = [qid for qid in ladder_qids if qid not in self._gate_questions_by_id] + if missing: + raise ValueError( + f"gate 阶梯[{task_type}] 含 benchmark 中不存在的题: " + f"{missing[:5]}（gate_pools.json 与题库失配）" + ) + ladder_items = [self._gate_questions_by_id[qid] for qid in ladder_qids] + base_skill_content = (self._paths.skills_dir / record.target_file).read_text( + encoding="utf-8" + ) + slug = task_type.lower().replace(" ", "-") + run_inference_fn = self._make_validate_run_inference_fn() + with HarnessLog(str(self._paths.db_path), f"gate_{slug}") as gate_log: + return await validate_skill_local( + workspace_dir=self._config.workspace_dir, + base_skills_version=self._current_version("skills"), + task_type=task_type, + target_file=record.target_file, + candidate_content=record.evolved_content, + base_skill_content=base_skill_content, + ladder_items=ladder_items, + gate_params=GateParams( + e_confirm=self._config.gate_e_confirm, + e_provisional=self._config.gate_e_provisional, + w_net_min=self._config.gate_w_net_min, + delta_min=self._config.gate_delta_min, + lambda_dir=self._config.gate_lambda_dir, + e_rollback=self._config.gate_e_rollback, + ), + gate_block=self._config.gate_block, + gate_n_max=self._config.gate_n_max, + gate_guard_err=self._config.gate_guard_err, + baseline_cache=state.baseline_cache, + prompts_version=self._current_version("prompts"), + run_inference=run_inference_fn, + log=gate_log, + gate_run_prefix=(f"{pools.baseline_run_id}_e{epoch}_s{step}_gate_{slug}"), + ) + + # ----------------------------------------------------------------------- + # accept / reject / probation + # ----------------------------------------------------------------------- + + def _accept_skill( + self, + task_type: str, + record: EvolutionRecord, + outcome: ValidationOutcome, + state: _TrainState, + pools: Pools, + ) -> None: + """accept：写候选为新 skills 版本 → manifest → 路径 → 前移 correctness。 + + probation 分岔：provisional + 无现有试用 + 非 default-strategy.md → 开账。 + 试用中追加 pending_edits。 + """ + pre_accept_version = self._current_version("skills") + # 开账快照在合并前拍取 + pre_merge_snapshot = { + q.question_id: state.correctness.get(q.question_id, False) + for q in pools.validation + if q.task_type == task_type + } + new_version = self._promote_skill_version(record.evolved_content, record.target_file) + update_manifest(self._config.workspace_dir, skills=f"skills/{new_version}") + self._paths = resolve_paths(self._config.workspace_dir) + # correctness 二轨合并（只合并已观测题） + state.correctness.update(outcome.candidate_correctness) + # 清该类黑名单 + state.rejected_buffer.pop(task_type, None) + state.changed_task_types_this_epoch.add(task_type) + # probation 分岔 + if ( + outcome.action == "accept_provisional" + and task_type not in state.probations + and record.target_file != "default-strategy.md" + ): + state.probations[task_type] = Probation( + task_type=task_type, + anchor_skills_version=pre_accept_version, + target_file=record.target_file, + correctness_snapshot=pre_merge_snapshot, + opened_step=state.global_step, + ) + elif outcome.action == "accept_provisional" and record.target_file == "default-strategy.md": + logger.warning( + "按类 gate[{}] provisional 落在共享 default-strategy.md，跳过试用直接转正", + task_type, + ) + if task_type in state.probations: + state.probations[task_type].pending_edits.append( + RejectedEdit( + target_file=record.target_file, + target_type=record.target_type, + change_summary=self._rejected_summary(record, outcome), + delta=outcome.delta_hat, + source_version=record.source_version, + epoch=state.global_step, + gate_w=outcome.w, + gate_l=outcome.l, + gate_e_value=outcome.e_value, + gate_delta_shrunk=outcome.delta_shrunk, + ) + ) + logger.info( + "按类 gate[{}] accept: 候选{:.1%} (观测基线{:.1%}) → skills/{}", + task_type, + outcome.candidate_acc, + outcome.baseline_acc, + new_version, + ) + + def _rollback_probation(self, probation: Probation, state: _TrainState) -> None: + """试用期回滚：文件级 revert 到锚版本 + 恢复快照 + 冷却 + 证据入黑名单。""" + anchor_content = ( + self._config.workspace_dir + / "skills" + / probation.anchor_skills_version + / probation.target_file + ).read_text(encoding="utf-8") + new_version = self._promote_skill_version(anchor_content, probation.target_file) + update_manifest(self._config.workspace_dir, skills=f"skills/{new_version}") + self._paths = resolve_paths(self._config.workspace_dir) + state.correctness.update(probation.correctness_snapshot) + state.gate_cooldown[probation.task_type] = self._config.gate_cooldown_steps + state.rejected_buffer.setdefault(probation.task_type, []).extend(probation.pending_edits) + logger.info( + "probation 回滚[{}]: skills 文件 {} 恢复至锚版本 {} → 新版本 {}，冷却 {} step", + probation.task_type, + probation.target_file, + probation.anchor_skills_version, + new_version, + self._config.gate_cooldown_steps, + ) + + @staticmethod + def _record_rejected_skill( + rejected_buffer: dict[str, list], + task_type: str, + record: EvolutionRecord, + outcome: ValidationOutcome, + global_step: int, + ) -> None: + """reject：按 task_type 累加 A5 黑名单。""" + rejected_buffer.setdefault(task_type, []).append( + RejectedEdit( + target_file=record.target_file, + target_type=record.target_type, + change_summary=Runner._rejected_summary_static(record, outcome), + delta=outcome.delta_hat, + source_version=record.source_version, + epoch=global_step, + gate_w=outcome.w, + gate_l=outcome.l, + gate_e_value=outcome.e_value, + gate_delta_shrunk=outcome.delta_shrunk, + ) + ) + logger.info( + "按类 gate[{}] reject: 候选{:.1%} (观测基线{:.1%}) 回退该 skill", + task_type, + outcome.candidate_acc, + outcome.baseline_acc, + ) + + def _rejected_summary(self, record: EvolutionRecord, outcome: ValidationOutcome) -> str: + """为被拒进化记录生成黑名单摘要：只拼真正 applied 的 edit。""" + return Runner._rejected_summary_static(record, outcome) + + @staticmethod + def _rejected_summary_static(record: EvolutionRecord, outcome: ValidationOutcome) -> str: + """黑名单摘要实现（静态方法，供 accept / reject 两侧复用）。 + + 为何只记 applied：未 applied 的 edit 从未写进候选正文、从未被 gate + 验证过，进黑名单会污染「已验证无效」语义。 + """ + applied_summary = _format_applied_edits(record) + if applied_summary is not None: + return applied_summary + return _fallback_summary(record, outcome) + + # ----------------------------------------------------------------------- + # _slow_update_cycle 十步序 + # ----------------------------------------------------------------------- + + async def _slow_update_cycle(self, epoch: int, pools: Pools, state: _TrainState) -> None: + """epoch 末慢更新十步序。 + + 1. 捕获版本快照 → 全 val 重跑 R + 2. soft score + dual_metric 落库 + 3. R 逐题对错无条件回写 + 4. probation 结算（回滚者覆盖 step 3） + 5. best argmax（严格大于） + 6. momentum（不可变新版本，按 skill 文件分组） + 7. system/tool 慢更新（edit_budget_end） + 8. R2 闭环 + 9. 三态标签 + epoch_report + 四向 held-out + 10. gate 阶梯刷新 + """ + # Phase 1 + eval_skills_version = self._current_version("skills") + eval_prompts_version = self._current_version("prompts") + eval_r = await self._eval_full_val(epoch, pools) + + # Phase 2: soft + dual_metric + eval_soft = await self._try_soft_score(eval_r.run_id, pools.validation) + write_dual_metric( + str(self._paths.db_path), + run_id=self._config.run_id, + epoch=epoch, + version_kind="final", + skills_version=eval_skills_version, + prompts_version=eval_prompts_version, + pool="val", + hard_acc=eval_r.accuracy, + soft_score=eval_soft, + mixed_score=(None if eval_soft is None else 0.5 * eval_r.accuracy + 0.5 * eval_soft), + ) + + # Phase 3: 无条件回写 + self._writeback_val_correctness(eval_r.run_id, pools, state) + + # Phase 4: probation 结算 + self._settle_probations(eval_r.run_id, state) + + # Phase 5: best argmax + self._maybe_promote_best( + eval_skills_version, + eval_prompts_version, + eval_r.accuracy, + eval_r.run_id, + epoch, + state, + ) + + # Phase 6: momentum + momentum_task_types = await self._write_momentum_for_changed_skills( + state, pools, epoch, eval_skills_version + ) + + # Phase 7: system/tool 慢更新 + pre_prompts_version = self._current_version("prompts") + system_tool_updated = await self._update_system_tool(epoch, state) + system_tool_reverted = False + + # Phase 8: R2 闭环 + r2_kept_run_ids: list[str] | None = None + if system_tool_updated: + r2_skills_version = self._current_version("skills") + new_prompts_version = self._current_version("prompts") + eval_r2 = await self._eval_full_val(epoch, pools, run_suffix="_p2") + write_dual_metric( + str(self._paths.db_path), + run_id=self._config.run_id, + epoch=epoch, + version_kind="final", + skills_version=r2_skills_version, + prompts_version=new_prompts_version, + pool="val", + hard_acc=eval_r2.accuracy, + soft_score=None, + mixed_score=None, + ) + system_tool_reverted = eval_r2.accuracy < eval_r.accuracy + if system_tool_reverted: + self._revert_system_tool(pre_prompts_version) + else: + self._writeback_val_correctness(eval_r2.run_id, pools, state) + self._maybe_promote_best( + r2_skills_version, + new_prompts_version, + eval_r2.accuracy, + eval_r2.run_id, + epoch, + state, + ) + state.eval_prev_acc = eval_r2.accuracy + state.eval_prev_run_id = eval_r2.run_id + r2_kept_run_ids = [eval_r2.run_id] + if (not system_tool_updated) or system_tool_reverted: + state.eval_prev_acc = eval_r.accuracy + state.eval_prev_run_id = eval_r.run_id + + # Phase 9: 三态标签 + epoch_report + held-out + if system_tool_reverted: + system_tool_action = "reverted" + elif system_tool_updated: + system_tool_action = "updated" + else: + system_tool_action = "none" + write_epoch_report( + self._config.workspace_dir, + epoch=epoch, + system_tool_action=system_tool_action, + momentum_updated_task_types=momentum_task_types, + best_val_acc=state.best_val_acc, + ) + await self._holdout_four_way(epoch, pools, state, eval_skills_version, eval_prompts_version) + + # Phase 10: gate 阶梯刷新 + self._refresh_gate_ladder( + epoch, pools.baseline_run_id, state, extra_run_ids=r2_kept_run_ids + ) + + # ----------------------------------------------------------------------- + # 慢更新内部方法 + # ----------------------------------------------------------------------- + + def _settle_probations(self, eval_run_id: str, state: _TrainState) -> None: + """epoch 末试用期一次性结算：全 val 重跑逐题结果与锚快照配对。 + + 参数: + eval_run_id: 本 epoch 全 val 重跑（R）的 run_id。 + state: 训练状态（probations 结算后清空）。 + + 异常: + RuntimeError: 重跑缺某快照题的预测行。 + """ + if not state.probations: + return + from app.harness.log import HarnessLog + from app.harness.validate import _load_run_rows + + with HarnessLog(str(self._paths.db_path), eval_run_id) as log: + rows = _load_run_rows(log, eval_run_id) + + 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, + ) + for task_type in sorted(state.probations): + probation = state.probations[task_type] + w = l = 0 # noqa: E741 + for qid, snap_correct in probation.correctness_snapshot.items(): + row = rows.get(qid) + if row is None: + raise RuntimeError( + f"probation 结算缺预测行: {task_type}/{qid}（run={eval_run_id}）" + ) + cur = row["_correct"] + if not snap_correct and cur: + w += 1 + elif snap_correct and not cur: + l += 1 # noqa: E741 + verdict = probation_verdict(w, l, params=params) + logger.info("probation 结算[{}]: W={} L={} → {}", task_type, w, l, verdict) + if verdict == "rollback": + self._rollback_probation(probation, state) + state.probations.clear() + + async def _eval_full_val( + self, epoch: int, pools: Pools, run_suffix: str = "" + ) -> InferenceResult: + """全验证池重跑一次并护栏。""" + run_id = f"{self._config.run_id}_slow_e{epoch}{run_suffix}" + result = await self._run_inference_on_pool( + pools.validation, run_id, self._paths.skills_dir, self._paths.prompts_dir + ) + _guard_infra_failures(result, context="全 val 重跑") + return result + + def _writeback_val_correctness( + self, eval_run_id: str, pools: Pools, state: _TrainState + ) -> None: + """把全 val 重跑逐题对错回写进 state.correctness。""" + from app.harness.log import HarnessLog + from app.harness.validate import _load_run_rows + + with HarnessLog(str(self._paths.db_path), eval_run_id) as log: + rows = _load_run_rows(log, eval_run_id) + for q in pools.validation: + row = rows.get(q.question_id) + if row is not None: + state.correctness[q.question_id] = row["_correct"] + + def _maybe_promote_best( + self, + skills_v: str, + prompts_v: str, + eval_acc: float, + run_id: str, + epoch: int, + state: _TrainState, + ) -> None: + """全局 best argmax（严格大于才推进）。""" + if eval_acc <= state.best_val_acc: + return + state.best_val_acc = eval_acc + state.best_skills_version = skills_v + state.best_prompts_version = prompts_v + state.steps_since_best_improved = 0 + update_best( + self._config.workspace_dir, + skills=f"skills/{skills_v}", + prompts=f"prompts/{prompts_v}", + val_acc=eval_acc, + run_id=run_id, + epoch=epoch, + ) + logger.info( + "全局 best argmax 刷新: {:.1%} → skills/{} prompts/{}", + eval_acc, + skills_v, + prompts_v, + ) + + async def _write_momentum_for_changed_skills( + self, + state: _TrainState, + pools: Pools, + epoch: int, + eval_skills_version: str, + ) -> list[str]: + """为本 epoch 改过的题型写 momentum：推进不可变新版本。 + + 返回: + 实际写过 momentum 的题型列表。 + """ + + if not self._config.use_slow_momentum: + return [] + if not state.changed_task_types_this_epoch: + return [] + + file_to_task_types = self._group_changed_task_types_by_file( + state.changed_task_types_this_epoch + ) + with tempfile.TemporaryDirectory() as tmp: + staged_skills = Path(tmp) / "skills" + shutil.copytree(self._paths.skills_dir, staged_skills, dirs_exist_ok=True) + for target_file in sorted(file_to_task_types): + await self._stage_momentum_for_file( + target_file, + file_to_task_types[target_file], + state, + pools, + staged_skills, + epoch, + ) + new_version = advance_version( + self._paths.workspace_dir, + "skills", + staged_skills, + { + "source": "slow_momentum", + "parent": eval_skills_version, + "description": "epoch 末 momentum（不可变新版本）", + }, + ) + update_manifest(self._config.workspace_dir, skills=f"skills/{new_version}") + self._paths = resolve_paths(self._config.workspace_dir) + logger.info( + "Epoch 末 momentum → skills/{}（不改 eval 版本 {}）", + new_version, + eval_skills_version, + ) + return sorted(state.changed_task_types_this_epoch) + + async def _stage_momentum_for_file( + self, + target_file: str, + task_types: list[str], + state: _TrainState, + pools: Pools, + staged_skills: Path, + epoch: int, + ) -> None: + """单个 skill 文件的 momentum 生成：诊断池采样 → 两版 rollout → 纵向对比。""" + from app.harness.log import HarnessLog + from app.harness.momentum import run_slow_momentum + from app.harness.validate import _load_run_rows + + skill_path = staged_skills / target_file + skill_content = skill_path.read_text(encoding="utf-8") + prev_skill = state.epoch_start_skills.get(target_file, skill_content) + prev_guidance = momentum_inner(skill_content) + + # 采样 + allowed = set(task_types) + candidates = [q for q in pools.diagnosis if q.task_type in allowed] + rng = random.Random(epoch) + n = min(self._config.momentum_samples, len(candidates)) + sampled = rng.sample(candidates, n) if n > 0 else [] + + if not sampled: + skill_path.write_text( + replace_momentum(skill_content, prev_guidance or ""), + encoding="utf-8", + ) + logger.debug( + "Epoch {} momentum 跳过 {}：诊断池无匹配题型 {} 的样本", + epoch, + target_file, + sorted(task_types), + ) + return + + # 两版 rollout + prev_run_id = f"momentum_prev_e{epoch}_{target_file.replace('.md', '')}" + curr_run_id = f"momentum_curr_e{epoch}_{target_file.replace('.md', '')}" + + with tempfile.TemporaryDirectory() as prev_tmp: + prev_skills_dir = Path(prev_tmp) / "skills" + shutil.copytree(self._paths.skills_dir, prev_skills_dir, dirs_exist_ok=True) + (prev_skills_dir / target_file).write_text(prev_skill, encoding="utf-8") + await self._run_inference_on_pool( + sampled, prev_run_id, prev_skills_dir, self._paths.prompts_dir + ) + + await self._run_inference_on_pool( + sampled, curr_run_id, self._paths.skills_dir, self._paths.prompts_dir + ) + + with HarnessLog(str(self._paths.db_path), prev_run_id) as log: + prev_rows = _load_run_rows(log, prev_run_id) + with HarnessLog(str(self._paths.db_path), curr_run_id) as log: + curr_rows = _load_run_rows(log, curr_run_id) + + comparison_pairs = _build_comparison_pairs(sampled, prev_rows, curr_rows) + guidance = await run_slow_momentum( + llm=self._evolve_llm, + diagnose_prompts_dir=Path("prompts"), + skill_content=skill_content, + prev_skill=prev_skill, + prev_guidance=prev_guidance, + comparison_pairs=comparison_pairs, + ) + new_content = replace_momentum(skill_content, guidance) + skill_path.write_text(new_content, encoding="utf-8") + logger.info( + "Epoch {} momentum 写入 skill 文件 {}（题型 {}，采样 {} 题）", + epoch, + target_file, + sorted(task_types), + len(sampled), + ) + + async def _update_system_tool(self, epoch: int, state: _TrainState) -> bool: + """merge 本 epoch 累加的 system/tool 案例包 → 进化 → accept 写新 prompts 版本。 + + 返回: + 是否实际写了新 prompts 版本。 + """ + from app.harness.workspace import VersionedPromptStore + from core.evolution import evolve_single_tool, evolve_system_prompt + + merged_system = merge_system_packs(state.system_packs) + merged_tools = merge_tool_packs(state.tool_packs) + source_version = self._current_version("prompts") + max_edits = self._config.edit_budget_end + evolve_prompts = self._load_evolve_prompts() + prompt_store = VersionedPromptStore(self._paths.prompts_dir) + + records: list[EvolutionRecord] = [] + if merged_system is not None: + records.append( + await evolve_system_prompt( + self._evolve_llm, + merged_system, + prompt_store, + evolve_prompts, + source_version, + max_edits, + ) + ) + for tool_name in sorted(merged_tools): + records.append( + await evolve_single_tool( + self._evolve_llm, + merged_tools[tool_name], + prompt_store, + evolve_prompts, + source_version, + max_edits, + ) + ) + accepted = [r for r in records if r.status == "accepted"] + if not accepted: + logger.debug("Epoch {} 慢更新：无 system/tool 改动被接受", epoch) + return False + + new_version = self._write_accepted_prompts_version(accepted, source_version) + if new_version is None: + return False + update_manifest(self._config.workspace_dir, prompts=f"prompts/{new_version}") + self._paths = resolve_paths(self._config.workspace_dir) + logger.info("Epoch {} 慢更新：system/tool → prompts/{}", epoch, new_version) + return True + + def _write_accepted_prompts_version( + self, accepted: list[EvolutionRecord], source_version: str + ) -> str | None: + """将 accepted system/tool records 写成新 prompts 版本。 + + 参数: + accepted: 状态为 accepted 的 EvolutionRecord 列表。 + source_version: 改写前 prompts 版本。 + + 返回: + 新版本号，或 None（无实际变化时）。 + """ + with tempfile.TemporaryDirectory() as tmp: + staged = Path(tmp) / "prompts" + shutil.copytree(self._paths.prompts_dir, staged, dirs_exist_ok=True) + any_changed = False + for rec in accepted: + if rec.target_type == "tool": + # tool: evolved_content = json.dumps({"extract": ..., "verify": ...}) + combined = json.loads(rec.evolved_content) + for key in ("extract", "verify"): + fname = rec.target_file.replace("_extract.md", f"_{key}.md") + (staged / fname).write_text(combined[key], encoding="utf-8") + any_changed = True + else: + (staged / rec.target_file).write_text(rec.evolved_content, encoding="utf-8") + any_changed = True + if not any_changed: + return None + return advance_version( + self._paths.workspace_dir, + "prompts", + staged, + { + "source": "evolution", + "parent": source_version, + "description": "epoch 末 system/tool 慢更新", + }, + ) + + def _revert_system_tool(self, pre_prompts_version: str) -> None: + """prompts-only delta 退步时回退到更新前版本。""" + update_manifest(self._config.workspace_dir, prompts=f"prompts/{pre_prompts_version}") + self._paths = resolve_paths(self._config.workspace_dir) + logger.info( + "慢更新 prompts-only delta 退步：system/tool 回退到 prompts/{}", + pre_prompts_version, + ) + + def _refresh_gate_ladder( + self, + epoch: int, + base_run_id: str, + state: _TrainState, + extra_run_ids: list[str] | None = None, + ) -> None: + """用本 epoch 非 gate run 的逐题观测 γ-EMA 更新阶梯 p-hat 并落盘。 + + 精确三源：step rollout GLOB 排除 _gate_ + slow R + kept R2。 + """ + db_path = resolve_paths(self._config.workspace_dir).db_path + conn = sqlite3.connect(str(db_path)) + conn.row_factory = sqlite3.Row + try: + rows = conn.execute( + "SELECT question_id, prediction, answer FROM predictions " + "WHERE run_id GLOB ? AND run_id NOT GLOB '*_gate_*' " + "ORDER BY rowid", + (f"{base_run_id}_e{epoch}_s*",), + ).fetchall() + slow_rows = conn.execute( + "SELECT question_id, prediction, answer FROM predictions " + "WHERE run_id=? ORDER BY rowid", + (f"{self._config.run_id}_slow_e{epoch}",), + ).fetchall() + extra_rows_lists = [ + conn.execute( + "SELECT question_id, prediction, answer FROM predictions " + "WHERE run_id=? ORDER BY rowid", + (rid,), + ).fetchall() + for rid in (extra_run_ids or []) + ] + finally: + conn.close() + + obs = {r["question_id"]: r["prediction"] == r["answer"] for r in rows} + obs.update({r["question_id"]: r["prediction"] == r["answer"] for r in slow_rows}) + for extra_rows in extra_rows_lists: + obs.update({r["question_id"]: r["prediction"] == r["answer"] for r in extra_rows}) + state.gate_pools.update_probs(obs, gamma=self._config.gate_gamma_decay) + state.gate_pools.save(self._config.workspace_dir / "gate_pools.json") + state.gate_epoch_observed = True + + async def _holdout_four_way( + self, + epoch: int, + pools: Pools, + state: _TrainState, + eval_skills_version: str, + eval_prompts_version: str, + ) -> None: + """四向 held-out：baseline/best_hard/final/best_mixed 各在 test 池评估。 + + test 池仅观测落库，绝不进 gate/best/early-stop/调参。 + """ + best_mixed = await self._pick_mixed_best( + epoch, pools, state, eval_skills_version, eval_prompts_version + ) + versions: dict[str, tuple[str, str] | None] = { + "baseline": (state.baseline_skills_version, state.baseline_prompts_version), + "best_hard": (state.best_skills_version, state.best_prompts_version), + "final": (eval_skills_version, eval_prompts_version), + "best_mixed": best_mixed, + } + for version_kind, version in versions.items(): + if version is None: + continue + sv, pv = version + run_id = f"{self._config.run_id}_holdout_{version_kind}_e{epoch}" + res = await self._eval_version_on_pool( + sv, pv, pools.test, run_id, context=f"held-out {version_kind}" + ) + soft = await self._try_soft_score(run_id, pools.test) + mixed = None if soft is None else 0.5 * res.accuracy + 0.5 * soft + write_holdout_eval( + str(self._paths.db_path), + run_id=self._config.run_id, + epoch=epoch, + version_kind=version_kind, + hard_acc=res.accuracy, + soft_score=soft, + mixed_score=mixed, + per_task_type_json=json.dumps(res.per_task_type, ensure_ascii=False), + ) + + async def _pick_mixed_best( + self, + epoch: int, + pools: Pools, + state: _TrainState, + eval_skills_version: str, + eval_prompts_version: str, + ) -> tuple[str, str] | None: + """在 val 池对候选集算 mixed，落 shadow_gate，返回 argmax mixed 版本。 + + 只观测落库，绝不改 manifest/best/early-stop。 + """ + candidates = { + "best_hard": (state.best_skills_version, state.best_prompts_version), + "final": (eval_skills_version, eval_prompts_version), + } + best_kind: str | None = None + best_mixed: float | None = None + for kind, (sv, pv) in candidates.items(): + run_id = f"{self._config.run_id}_shadow_{kind}_e{epoch}" + res = await self._eval_version_on_pool( + sv, pv, pools.validation, run_id, context=f"mixed 影子 {kind}" + ) + soft = await self._try_soft_score(run_id, pools.validation) + mixed = None if soft is None else 0.5 * res.accuracy + 0.5 * soft + write_shadow_gate( + str(self._paths.db_path), + run_id=self._config.run_id, + epoch=epoch, + candidate_version=f"skills/{sv}+prompts/{pv}", + hard_acc=res.accuracy, + soft_score=soft, + mixed_score=mixed, + is_mixed_best=False, + ) + if mixed is not None and (best_mixed is None or mixed > best_mixed): + best_mixed, best_kind = mixed, kind + if best_kind is None: + return None + self._mark_shadow_best(epoch, candidates[best_kind]) + return candidates[best_kind] + + def _mark_shadow_best(self, epoch: int, best_version: tuple[str, str]) -> None: + """回标 shadow_gate 中 argmax mixed 选中的版本 is_mixed_best=1。""" + sv, pv = best_version + candidate_version = f"skills/{sv}+prompts/{pv}" + conn = sqlite3.connect(str(self._paths.db_path)) + try: + conn.execute( + "UPDATE shadow_gate SET is_mixed_best=1 WHERE rowid = (" + " SELECT rowid FROM shadow_gate " + " WHERE run_id=? AND epoch=? AND candidate_version=? LIMIT 1" + ")", + (self._config.run_id, epoch, candidate_version), + ) + conn.commit() + finally: + conn.close() + + # ----------------------------------------------------------------------- + # 收尾 + # ----------------------------------------------------------------------- + + def _deliver_best(self, best_skills_version: str, best_prompts_version: str) -> None: + """若当前 current 不是历史最优，回滚 manifest 到 best 并刷新路径。""" + cur = load_manifest(self._config.workspace_dir)["current"] + if ( + cur["skills"] != f"skills/{best_skills_version}" + or cur["prompts"] != f"prompts/{best_prompts_version}" + ): + update_manifest( + self._config.workspace_dir, + skills=f"skills/{best_skills_version}", + prompts=f"prompts/{best_prompts_version}", + ) + self._paths = resolve_paths(self._config.workspace_dir) + logger.info( + "收尾交付 best：current → skills/{} prompts/{}", + best_skills_version, + best_prompts_version, + ) + + async def _final_test_eval(self, pools: Pools) -> None: + """收尾在 held-out test 池跑一次评估。""" + run_id = f"{self._config.run_id}_final_test" + result = await self._run_inference_on_pool( + pools.test, run_id, self._paths.skills_dir, self._paths.prompts_dir + ) + _guard_infra_failures(result, context="held-out test 评估") + report = { + "run_id": result.run_id, + "accuracy": result.accuracy, + "total": result.total, + "correct": result.correct, + "per_task_type": result.per_task_type, + } + path = self._config.workspace_dir / "analyses" / "final_test_eval.json" + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8") + logger.info("held-out test 评估写入: {} (acc={:.1%})", path, result.accuracy) + + # ----------------------------------------------------------------------- + # 私有辅助 + # ----------------------------------------------------------------------- + + def _current_version(self, kind: str) -> str: + """读取 manifest current 指针中某类资源的当前版本名。""" + return load_manifest(self._config.workspace_dir)["current"][kind].split("/")[-1] + + def _current_version_snapshot(self) -> dict[str, str]: + """读 manifest.current 的 skills/prompts 指针。""" + cur = load_manifest(self._config.workspace_dir)["current"] + return {"skills": cur["skills"], "prompts": cur["prompts"]} + + def _class_baseline_acc( + self, + task_type: str, + validation: list[GeneratedQuestion], + correctness: dict[str, bool], + ) -> float: + """该 task_type 验证子集在当前 correctness 下的准确率。""" + class_items = [q for q in validation if q.task_type == task_type] + assert class_items, f"task_type={task_type} 在验证池中无对应题目" + correct = sum(1 for q in class_items if correctness.get(q.question_id, False)) + return correct / len(class_items) + + def _promote_skill_version(self, content: str, target_file: str) -> str: + """把候选 skill 内容写成新正式 skills 版本。""" + with tempfile.TemporaryDirectory() as tmp: + src = Path(tmp) / "skills" + shutil.copytree(self._paths.skills_dir, src, dirs_exist_ok=True) + (src / target_file).write_text(content, encoding="utf-8") + return advance_version( + self._paths.workspace_dir, + "skills", + src, + { + "source": "evolution", + "parent": self._current_version("skills"), + "description": f"按类 gate accept {target_file}", + }, + ) + + def _group_changed_task_types_by_file( + self, changed_task_types: set[str] + ) -> dict[str, list[str]]: + """把改过的 task_type 集合经 fallback 解析映射到 skill 文件，按文件分组。""" + from app.harness.workspace import VersionedSkillStore + + skill_store = VersionedSkillStore(self._paths.skills_dir) + grouped: dict[str, list[str]] = {} + for task_type in changed_task_types: + skill_file = resolve_skill_file(skill_store, task_type) + grouped.setdefault(skill_file, []).append(task_type) + return grouped + + def _record_run(self, run_id: str) -> Path: + """将 current 版本快照追加到 manifest history，创建 run 目录。""" + from app.harness.workspace import record_run + + return record_run(self._config.workspace_dir, run_id) + + def _backfill_run_versions( + self, run_id: str, skills_version: str, prompts_version: str + ) -> None: + """eval run 的 skills/prompts 版本对 + questions_ref 回填进 _runs。""" + from app.harness.log import HarnessLog + + with HarnessLog(str(self._paths.db_path), run_id) as log: + log.execute( + "UPDATE _runs SET skills_version = ?, prompts_version = ?, " + "questions_ref = ? WHERE run_id = ?", + (skills_version, prompts_version, self._config.questions, run_id), + ) + + def _write_eval_report( + self, + run_id: str, + skills_version: str, + prompts_version: str, + result: InferenceResult, + ) -> None: + """写 eval 评测报告 analyses/eval_{run_id}.json。""" + report = { + "run_id": run_id, + "skills_version": skills_version, + "prompts_version": prompts_version, + "accuracy": result.accuracy, + "total": result.total, + "correct": result.correct, + "stop_reason_counts": result.stop_reason_counts, + } + analyses_dir = self._config.workspace_dir / "analyses" + analyses_dir.mkdir(parents=True, exist_ok=True) + path = analyses_dir / f"eval_{run_id}.json" + path.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8") + logger.info("eval 报告写入: {} (acc={:.1%})", path, result.accuracy) + + async def _run_inference_on_pool( + self, + questions: list[GeneratedQuestion], + run_id: str, + skills_dir: Path, + prompts_dir: Path, + ) -> InferenceResult: + """用指定版本在给定题池跑一次 run_inference。""" + from app.harness.inference import run_inference + from app.harness.log import HarnessLog + + self._record_run(run_id) + with HarnessLog(str(self._paths.db_path), run_id) as log: + return await run_inference( + questions=questions, + llm=self._llm, + tool_dispatch_fn=self._make_tool_dispatch_fn(skills_dir=skills_dir), + prompt_builder=self._make_prompt_builder( + skills_dir=skills_dir, prompts_dir=prompts_dir + ), + log=log, + run_id=run_id, + concurrency=self._config.concurrency, + max_steps=self._config.max_steps, + skill_mode=self._config.skill_mode, + ) + + async def _eval_version_on_pool( + self, + skills_version: str, + prompts_version: str, + questions: list[GeneratedQuestion], + run_id: str, + context: str, + ) -> InferenceResult: + """用指定版本在给定池跑一次推理并护栏。""" + skills_dir = self._paths.workspace_dir / "skills" / skills_version + prompts_dir = self._paths.workspace_dir / "prompts" / prompts_version + result = await self._run_inference_on_pool(questions, run_id, skills_dir, prompts_dir) + _guard_infra_failures(result, context=context) + return result + + async def _run_diagnosis( + self, run_id: str, *, question_ids: list[str] | None = None + ) -> DiagnosisResult: + """执行两阶段诊断。""" + from app.harness.log import RunLogImpl + from app.harness.workspace import VersionedSkillStore + from app.question_gen import load_benchmark + from core.evolution.diagnose import run_diagnosis + + questions = load_benchmark(self._paths.questions_dir) + run_log = RunLogImpl(str(self._paths.db_path)) + skill_store = VersionedSkillStore(self._paths.skills_dir) + diagnose_prompts = self._load_diagnose_prompts() + + return await run_diagnosis( + run_id=run_id, + questions=questions, + tree_data={}, # tree_data 由诊断管线内部按需加载 + llm=self._llm, + run_log=run_log, + skill_store=skill_store, + prompts=diagnose_prompts, + concurrency=self._config.concurrency, + question_ids=question_ids, + ) + + async def _try_soft_score( + self, run_id: str, questions: list[GeneratedQuestion] + ) -> float | None: + """尝试计算 soft score，失败降级为 None。""" + try: + # soft score 暂不实现（需诊断 span_evaluations 表），降级为 None + return None + except Exception: + logger.warning("soft score 计算失败（run={}），降级为 None", run_id) + return None + + # ----------------------------------------------------------------------- + # 注入工厂（暂用占位，由 main.py 绑定实际实现） + # ----------------------------------------------------------------------- + + def _make_tool_dispatch_fn(self, *, skills_dir: Path | None = None): + """构造工具调度函数（由子类或 main.py 覆盖）。""" + + async def _noop_dispatch(tool_name: str, args: dict, *, context: dict) -> str: + raise NotImplementedError( + f"工具 {tool_name} 调度未配置（需由 main.py 注入 tool_dispatch_fn）" + ) + + return _noop_dispatch + + def _make_prompt_builder( + self, *, skills_dir: Path | None = None, prompts_dir: Path | None = None + ): + """构造 prompt 构建函数（由子类或 main.py 覆盖）。""" + + def _noop_builder(qa: GeneratedQuestion) -> tuple[str, str]: + raise NotImplementedError("prompt_builder 未配置（需由 main.py 注入）") + + return _noop_builder + + def _make_validate_run_inference_fn(self): + """构造 validate 用的 RunInferenceFn（绑定共享依赖）。""" + from app.harness.inference import run_inference + from app.harness.log import HarnessLog + + async def _run( + questions: list[GeneratedQuestion], + *, + run_id: str, + skills_dir: Path, + ) -> InferenceResult: + self._record_run(run_id) + with HarnessLog(str(self._paths.db_path), run_id) as log: + return await run_inference( + questions=questions, + llm=self._llm, + 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=log, + run_id=run_id, + concurrency=self._config.concurrency, + max_steps=self._config.max_steps, + skill_mode=self._config.skill_mode, + ) + + return _run + + def _load_evolve_prompts(self): + """加载进化模板束（从项目根 prompts/ 读取诊断标尺模板）。""" + from core.evolution.types import EvolvePrompts + + def _read(name: str) -> str: + p = Path("prompts") / name + return p.read_text(encoding="utf-8") if p.exists() else "" + + return EvolvePrompts( + evolve_skill=_read("evolve_skill.md"), + evolve_system=_read("evolve_system.md"), + evolve_tool=_read("evolve_tool.md"), + evolve_rank=_read("evolve_rank.md"), + consolidate_system=_read("consolidate_system.md"), + ) + + def _load_diagnose_prompts(self): + """加载诊断模板束（从项目根 prompts/ 读取）。""" + from core.evolution.types import DiagnosePrompts + + def _read(name: str) -> str: + p = Path("prompts") / name + return p.read_text(encoding="utf-8") if p.exists() else "" + + return DiagnosePrompts( + defect_vs_lapse=_read("defect_vs_lapse.md"), + reasoning_sub=_read("reasoning_sub.md"), + span_eval_system=_read("span_eval_system.md"), + span_eval_user=_read("span_eval_user.md"), + missed_nodes=_read("missed_nodes.md"), + skill_adherence=_read("skill_adherence.md"), + confirmation_bias=_read("confirmation_bias.md"), + evidence_sufficiency=_read("evidence_sufficiency.md"), + ) diff --git a/tests/unit/test_harness_runner.py b/tests/unit/test_harness_runner.py new file mode 100644 index 0000000..76e0211 --- /dev/null +++ b/tests/unit/test_harness_runner.py @@ -0,0 +1,682 @@ +"""runner.py 单元测试（算法保真 #13）。 + +覆盖 13a-13e 五个子任务，测试 Runner 骨架、三级嵌套、gate/accept/reject/probation、 +慢更新十步序、deliver_best + early stop。大部分测试用纯函数或 mock 构造避免真实推理。 +""" + +from __future__ import annotations + +import json +from dataclasses import dataclass, field +from pathlib import Path # noqa: TC003 — 运行时 tmp_path 标注使用 +from unittest.mock import MagicMock, patch + +import pytest + +from app.harness.runner import ( + _apply_batch_correctness, + _batch_from_ids, + _build_comparison_pairs, + _compute_total_steps, + _fallback_summary, + _format_applied_edits, + _guard_infra_failures, + _outcome_to_quadrant_pairs, + _should_early_stop, + _snapshot_current_skills, + _TrainState, + _write_skip_report, + resume_plan, +) +from app.harness.validate import Probation, ValidationOutcome +from core.evolution import RejectedEdit + +# --------------------------------------------------------------------------- +# 测试辅助 +# --------------------------------------------------------------------------- + + +@dataclass(frozen=True) +class _FakeInferenceResult: + """InferenceResult 替身。""" + + run_id: str = "test_run" + accuracy: float = 0.5 + total: int = 10 + correct: int = 5 + per_task_type: dict = field(default_factory=dict) + steps_mean: float = 3.0 + token_usage: dict = field(default_factory=lambda: {"prompt_tokens": 0, "completion_tokens": 0}) + stop_reason_counts: dict = field(default_factory=lambda: {"finished": 10}) + + +@dataclass(frozen=True) +class _FakeQuestion: + """GeneratedQuestion 替身。""" + + question_id: str + video_id: str = "v1" + task_type: str = "Action Reasoning" + question: str = "问题" + options: tuple = ("A", "B", "C", "D") + answer: str = "A" + source_nodes: tuple = () + difficulty: str = "medium" + + +@dataclass +class _FakePools: + """Pools 替身。""" + + diagnosis: list = field(default_factory=list) + validation: list = field(default_factory=list) + test: list = field(default_factory=list) + baseline_run_id: str = "baseline_run" + baseline_val_accuracy: float = 0.5 + correctness: dict = field(default_factory=dict) + + +# ========================================================================= +# 13a: Runner 骨架 + 纯函数 +# ========================================================================= + + +class TestResumePlan: + """resume_plan 纯函数。""" + + def test_epoch_done_advances_epoch(self) -> None: + """epoch_done 阶段：下一 epoch 从头开始。""" + plan = resume_plan(epoch=3, phase="epoch_done", step_completed=5) + assert plan["first_epoch"] == 4 + assert plan["resume_epoch"] is None + assert plan["resume_step_from"] == 0 + + def test_in_epoch_resumes_same_epoch(self) -> None: + """in_epoch 阶段：从同 epoch 的下一个 step 续跑。""" + plan = resume_plan(epoch=2, phase="in_epoch", step_completed=3) + assert plan["first_epoch"] == 2 + assert plan["resume_epoch"] == 2 + assert plan["resume_step_from"] == 4 + + def test_in_epoch_step_zero(self) -> None: + """in_epoch step_completed=0：从 step 1 续跑。""" + plan = resume_plan(epoch=1, phase="in_epoch", step_completed=0) + assert plan["resume_step_from"] == 1 + + +class TestGuardInfraFailures: + """_guard_infra_failures 基础设施护栏。""" + + def test_low_error_rate_passes(self) -> None: + """error 率 <= 10% 不抛异常。""" + result = _FakeInferenceResult( + total=100, + stop_reason_counts={"finished": 95, "error": 5}, + ) + _guard_infra_failures(result, context="test") # 不应抛异常 + + def test_high_error_rate_raises(self) -> None: + """error 率 > 10% 抛 RuntimeError。""" + result = _FakeInferenceResult( + total=10, + stop_reason_counts={"finished": 8, "error": 2}, + ) + with pytest.raises(RuntimeError, match="基础设施失败率过高"): + _guard_infra_failures(result, context="test") + + def test_zero_total_does_not_crash(self) -> None: + """total=0 时不除零崩溃。""" + result = _FakeInferenceResult(total=0, stop_reason_counts={}) + _guard_infra_failures(result, context="test") # 不应抛异常 + + def test_no_error_key_passes(self) -> None: + """stop_reason_counts 无 error 键时正常通过。""" + result = _FakeInferenceResult( + total=10, + stop_reason_counts={"finished": 10}, + ) + _guard_infra_failures(result, context="test") + + +# ========================================================================= +# 13b: _apply_batch_correctness + _compute_total_steps +# ========================================================================= + + +class TestApplyBatchCorrectness: + """_apply_batch_correctness rollout 完整性护栏。""" + + def test_complete_batch_updates_correctness(self) -> None: + """完整 rollout 正常更新 correctness。""" + batch = [_FakeQuestion(question_id="q1"), _FakeQuestion(question_id="q2")] + correctness: dict[str, bool] = {} + + # mock HarnessLog + mock_log = MagicMock() + mock_log.query.return_value = [ + {"question_id": "q1", "prediction": "A", "answer": "A", "steps_json": "[]"}, + {"question_id": "q2", "prediction": "B", "answer": "A", "steps_json": "[]"}, + ] + + with patch("app.harness.validate._load_run_rows") as mock_load: + mock_load.return_value = { + "q1": {"prediction": "A", "answer": "A", "_correct": True, "steps": []}, + "q2": {"prediction": "B", "answer": "A", "_correct": False, "steps": []}, + } + _apply_batch_correctness(correctness, mock_log, "run_1", batch) + + assert correctness["q1"] is True + assert correctness["q2"] is False + + def test_missing_prediction_raises(self) -> None: + """rollout 缺预测行时抛 RuntimeError。""" + batch = [_FakeQuestion(question_id="q1"), _FakeQuestion(question_id="q2")] + correctness: dict[str, bool] = {} + + mock_log = MagicMock() + with patch("app.harness.validate._load_run_rows") as mock_load: + mock_load.return_value = { + "q1": {"prediction": "A", "answer": "A", "_correct": True, "steps": []}, + # q2 缺失 + } + with pytest.raises(RuntimeError, match="rollout 不完整"): + _apply_batch_correctness(correctness, mock_log, "run_1", batch) + + +class TestComputeTotalSteps: + """_compute_total_steps 退火地平线。""" + + def test_basic_calculation(self) -> None: + """基本退火地平线计算。""" + questions = [ + _FakeQuestion(question_id=f"q{i}", task_type="Action Reasoning") for i in range(20) + ] + # 全错题 + correctness = {q.question_id: False for q in questions} + + config = MagicMock() + config.batch_size = 5 + config.min_class_per_batch = 1 + config.batch_correct_ratio = 0.0 + config.epochs = 3 + + pools = _FakePools(diagnosis=questions, correctness=correctness) + total = _compute_total_steps(pools, correctness, config) + # 20 题 / batch_size 5 = 4 步/epoch * 3 epochs = 12 + assert total == 12 + + +# ========================================================================= +# 13b: _batch_from_ids +# ========================================================================= + + +class TestBatchFromIds: + """_batch_from_ids 按 ID 重建 batch。""" + + def test_preserves_order(self) -> None: + """按 ids 顺序取出，保持原 batch 划分。""" + q1 = _FakeQuestion(question_id="q1") + q2 = _FakeQuestion(question_id="q2") + q3 = _FakeQuestion(question_id="q3") + pools = _FakePools(diagnosis=[q1, q2, q3]) + + batch = _batch_from_ids(pools, ["q3", "q1"]) + assert [q.question_id for q in batch] == ["q3", "q1"] + + +# ========================================================================= +# 13b: _snapshot_current_skills +# ========================================================================= + + +class TestSnapshotCurrentSkills: + """_snapshot_current_skills 快照 skill 文件。""" + + def test_snapshots_md_files(self, tmp_path: Path) -> None: + """只快照 .md 文件。""" + (tmp_path / "action-reasoning.md").write_text("skill content 1") + (tmp_path / "temporal.md").write_text("skill content 2") + (tmp_path / "meta.json").write_text("{}") + + snapshot = _snapshot_current_skills(tmp_path) + assert "action-reasoning.md" in snapshot + assert "temporal.md" in snapshot + assert "meta.json" not in snapshot + assert snapshot["action-reasoning.md"] == "skill content 1" + + +# ========================================================================= +# 13c: _outcome_to_quadrant_pairs +# ========================================================================= + + +class TestOutcomeToQuadrantPairs: + """_outcome_to_quadrant_pairs 四象限拍平。""" + + def test_all_quadrants(self) -> None: + """四象限各有一个 qid 时生成 4 条 pair。""" + outcome = ValidationOutcome( + action="accept_confirmed", + accepted=True, + stop_reason="confirmed", + e_value=10.0, + w=3, + l=0, + n_used=10, + delta_hat=0.3, + delta_shrunk=0.2, + baseline_acc=0.7, + candidate_acc=0.9, + improvements=["q1"], + regressions=["q2"], + persistent_fails=["q3"], + stable_successes=["q4"], + ) + pairs = _outcome_to_quadrant_pairs("Action Reasoning", outcome) + assert len(pairs) == 4 + by_qid = {p["question_id"]: p for p in pairs} + assert by_qid["q1"]["category"] == "improved" + assert by_qid["q1"]["prev_correct"] is False + assert by_qid["q1"]["curr_correct"] is True + assert by_qid["q2"]["category"] == "regressed" + assert by_qid["q3"]["category"] == "persistent_fail" + assert by_qid["q4"]["category"] == "stable_success" + + def test_empty_outcome(self) -> None: + """四象限全空时返回空列表。""" + outcome = ValidationOutcome( + action="reject", + accepted=False, + stop_reason="directional", + e_value=0.5, + w=0, + l=2, + n_used=5, + delta_hat=-0.1, + delta_shrunk=-0.05, + baseline_acc=0.8, + candidate_acc=0.6, + ) + assert _outcome_to_quadrant_pairs("Any", outcome) == [] + + +# ========================================================================= +# 13c: _build_comparison_pairs +# ========================================================================= + + +class TestBuildComparisonPairs: + """_build_comparison_pairs momentum 纵向对比对。""" + + def test_builds_pairs(self) -> None: + """正确构造对比对。""" + sampled = [_FakeQuestion(question_id="q1", question="问题1")] + prev_rows = { + "q1": {"prediction": "A", "_correct": True}, + } + curr_rows = { + "q1": {"prediction": "B", "_correct": False}, + } + pairs = _build_comparison_pairs(sampled, prev_rows, curr_rows) + assert len(pairs) == 1 + assert pairs[0]["question"] == "问题1" + assert pairs[0]["prev_prediction"] == "A" + assert pairs[0]["curr_prediction"] == "B" + assert pairs[0]["correct_prev"] is True + assert pairs[0]["correct_curr"] is False + + def test_missing_rows_use_defaults(self) -> None: + """缺失行时使用默认值。""" + sampled = [_FakeQuestion(question_id="q1")] + pairs = _build_comparison_pairs(sampled, {}, {}) + assert pairs[0]["prev_prediction"] == "" + assert pairs[0]["correct_prev"] is False + + +# ========================================================================= +# 13e: _should_early_stop +# ========================================================================= + + +class TestShouldEarlyStop: + """_should_early_stop 步粒度 early stop。""" + + def test_improved_this_epoch_resets(self, tmp_path: Path) -> None: + """本 epoch best 刷新时重置计数器。""" + # 写 manifest + best + manifest = { + "name": "test", + "store": ".", + "current": {"videos": "v", "questions": "q", "skills": "s/v1", "prompts": "p/v1"}, + "best": {"epoch": 2, "val_acc": 0.9}, + "history": [], + } + (tmp_path / "manifest.json").write_text(json.dumps(manifest)) + + state = MagicMock() + state.steps_since_best_improved = 10 + + result = _should_early_stop(tmp_path, epoch=2, steps_this_epoch=5, state=state, patience=20) + assert result is False + assert state.steps_since_best_improved == 0 + + def test_no_improvement_accumulates(self, tmp_path: Path) -> None: + """未刷新时累加步数。""" + manifest = { + "name": "test", + "store": ".", + "current": {"videos": "v", "questions": "q", "skills": "s/v1", "prompts": "p/v1"}, + "best": {"epoch": 1, "val_acc": 0.5}, + "history": [], + } + (tmp_path / "manifest.json").write_text(json.dumps(manifest)) + + state = MagicMock() + state.steps_since_best_improved = 15 + + result = _should_early_stop(tmp_path, epoch=3, steps_this_epoch=5, state=state, patience=20) + assert result is True # 15 + 5 = 20 >= 20 + assert state.steps_since_best_improved == 20 + + def test_below_patience_continues(self, tmp_path: Path) -> None: + """累计步数未达阈值时继续。""" + manifest = { + "name": "test", + "store": ".", + "current": {"videos": "v", "questions": "q", "skills": "s/v1", "prompts": "p/v1"}, + "best": {"epoch": 1, "val_acc": 0.5}, + "history": [], + } + (tmp_path / "manifest.json").write_text(json.dumps(manifest)) + + state = MagicMock() + state.steps_since_best_improved = 10 + + result = _should_early_stop(tmp_path, epoch=3, steps_this_epoch=5, state=state, patience=20) + assert result is False + assert state.steps_since_best_improved == 15 + + def test_step_granularity(self, tmp_path: Path) -> None: + """步粒度而非 epoch 粒度。""" + manifest = { + "name": "test", + "store": ".", + "current": {"videos": "v", "questions": "q", "skills": "s/v1", "prompts": "p/v1"}, + "best": {"epoch": 1, "val_acc": 0.5}, + "history": [], + } + (tmp_path / "manifest.json").write_text(json.dumps(manifest)) + + state = MagicMock() + # 连续 3 个 epoch，每个 3 步 + state.steps_since_best_improved = 0 + for ep in range(2, 5): + stopped = _should_early_stop( + tmp_path, epoch=ep, steps_this_epoch=3, state=state, patience=10 + ) + if ep < 4: + assert stopped is False + else: + # 3+3+3=9 < 10 但第三轮后 9+3=12>=10 在 ep=5 触发 + # 实际：ep=2 → 3, ep=3 → 6, ep=4 → 9 + assert stopped is False + stopped = _should_early_stop( + tmp_path, epoch=5, steps_this_epoch=3, state=state, patience=10 + ) + assert stopped is True # 9+3=12>=10 + + +# ========================================================================= +# 13c: Probation 数据结构测试 +# ========================================================================= + + +class TestProbation: + """Probation 数据结构。""" + + def test_probation_fields(self) -> None: + """Probation 必须具备全部字段。""" + p = Probation( + task_type="Action Reasoning", + anchor_skills_version="v1", + target_file="action-reasoning.md", + correctness_snapshot={"q1": True}, + opened_step=5, + ) + assert p.task_type == "Action Reasoning" + assert p.pending_edits == [] + + def test_pending_edits_append(self) -> None: + """pending_edits 可追加 RejectedEdit。""" + p = Probation( + task_type="Action Reasoning", + anchor_skills_version="v1", + target_file="action-reasoning.md", + correctness_snapshot={}, + opened_step=0, + ) + edit = RejectedEdit( + target_file="action-reasoning.md", + target_type="skill", + change_summary="test", + delta=0.1, + source_version="v1", + epoch=0, + gate_w=3, + gate_l=1, + gate_e_value=2.5, + gate_delta_shrunk=0.05, + ) + p.pending_edits.append(edit) + assert len(p.pending_edits) == 1 + + +# ========================================================================= +# 13c: RejectedSummary 黑名单防污染 +# ========================================================================= + + +class TestFormatAppliedEdits: + """_format_applied_edits 只拼 applied 的 edit。""" + + def test_only_applied_edits_in_summary(self) -> None: + """只有 applied 状态的 edit 进入摘要。""" + record = MagicMock() + record.edits = [ + {"op": "replace", "target": "section1"}, + {"op": "insert", "content": "new_content"}, + {"op": "delete", "target": "old_stuff"}, + ] + record.apply_report = [ + {"status": "applied_exact"}, + {"status": "skipped_not_found"}, + {"status": "applied_fuzzy"}, + ] + + summary = _format_applied_edits(record) + assert summary is not None + assert "section1" in summary + assert "old_stuff" in summary + assert "new_content" not in summary + + def test_zero_applied_returns_info_message(self) -> None: + """0 applied 返回信息性消息（非 None）。""" + record = MagicMock() + record.edits = [{"op": "replace", "target": "sec"}] + record.apply_report = [{"status": "skipped_not_found"}] + + summary = _format_applied_edits(record) + assert summary is not None + assert "0 applied" in summary + + def test_no_edits_returns_none(self) -> None: + """无 edit 时返回 None。""" + record = MagicMock() + record.edits = [] + + assert _format_applied_edits(record) is None + + def test_no_report_includes_all_edits(self) -> None: + """无 apply_report 时包含所有 edit。""" + record = MagicMock() + record.edits = [{"op": "replace", "target": "foo"}] + record.apply_report = [] + + summary = _format_applied_edits(record) + assert summary is not None + assert "foo" in summary + + +class TestFallbackSummary: + """_fallback_summary 兜底黑名单摘要。""" + + def test_from_suggestions(self) -> None: + """有 suggestions 时拼接 change 字段。""" + record = MagicMock() + record.suggestions = [{"change": "改 A"}, {"change": "改 B"}] + outcome = MagicMock() + outcome.delta_hat = -0.1 + + summary = _fallback_summary(record, outcome) + assert "改 A" in summary + assert "改 B" in summary + + def test_no_suggestions_uses_delta(self) -> None: + """无 suggestions 时使用 delta 信息。""" + record = MagicMock() + record.suggestions = [] + outcome = MagicMock() + outcome.delta_hat = -0.15 + + summary = _fallback_summary(record, outcome) + assert "delta" in summary + assert "-0.15" in summary + + +class TestRejectedSummaryIntegration: + """_rejected_summary_static 集成：两个子函数组合。""" + + def test_static_delegates_to_format_applied(self) -> None: + """有 applied edit 时 static 方法返回 _format_applied_edits 结果。""" + from app.harness.runner import Runner + + record = MagicMock() + record.edits = [{"op": "replace", "target": "section1"}] + record.apply_report = [{"status": "applied_exact"}] + record.suggestions = [] + outcome = MagicMock() + outcome.delta_hat = 0.1 + + summary = Runner._rejected_summary_static(record, outcome) + assert "section1" in summary + + def test_static_falls_back_to_suggestions(self) -> None: + """无 edit 时 static 方法使用 _fallback_summary。""" + from app.harness.runner import Runner + + record = MagicMock() + record.edits = [] + record.suggestions = [{"change": "尝试 X"}] + outcome = MagicMock() + outcome.delta_hat = -0.2 + + summary = Runner._rejected_summary_static(record, outcome) + assert "尝试 X" in summary + + +class TestWriteSkipReport: + """_write_skip_report 辅助函数。""" + + def test_writes_cooldown_report(self, tmp_path: Path) -> None: + """cooldown 路径写 step_report JSON 文件。""" + (tmp_path / "analyses").mkdir() + _write_skip_report( + tmp_path, + epoch=1, + step=0, + global_step=5, + task_type="Action Reasoning", + action="cooldown", + baseline_acc=0.75, + budget=3, + ) + report_path = tmp_path / "analyses" / "step_report_e1_s0_action-reasoning.json" + assert report_path.exists() + data = json.loads(report_path.read_text()) + assert data["gate_action"] == "cooldown" + assert data["candidate_acc"] == 0.75 + assert data["gate_w"] is None + + def test_writes_skipped_report(self, tmp_path: Path) -> None: + """skipped 路径写 step_report 并传递 rank_clip_triggered。""" + (tmp_path / "analyses").mkdir() + _write_skip_report( + tmp_path, + epoch=2, + step=1, + global_step=10, + task_type="Temporal Reasoning", + action="skipped", + baseline_acc=0.6, + budget=2, + rank_clip_triggered=True, + ) + report_path = tmp_path / "analyses" / "step_report_e2_s1_temporal-reasoning.json" + assert report_path.exists() + data = json.loads(report_path.read_text()) + assert data["gate_action"] == "skipped" + assert data["rank_clip_triggered"] is True + + +# ========================================================================= +# 13a: _TrainState 基本构造 +# ========================================================================= + + +class TestTrainState: + """_TrainState dataclass 基本构造与字段默认值。""" + + def test_default_fields(self) -> None: + """默认字段值正确。""" + state = _TrainState( + correctness={"q1": True}, + gate_pools=MagicMock(), + baseline_cache=MagicMock(), + eval_prev_acc=0.5, + eval_prev_run_id="run1", + best_val_acc=0.5, + best_skills_version="v1", + best_prompts_version="v1", + ) + assert state.global_step == 0 + assert state.gate_epoch_observed is False + assert state.probations == {} + assert state.gate_cooldown == {} + assert state.rejected_buffer == {} + assert state.system_packs == [] + assert state.tool_packs == [] + assert state.changed_task_types_this_epoch == set() + assert state.steps_since_best_improved == 0 + + +# ========================================================================= +# 13c: cooldown 递减测试 +# ========================================================================= + + +class TestCooldownDecrement: + """gate_cooldown 每 step 递减、归零剔除。""" + + def test_decrement_and_remove(self) -> None: + """冷却值递减，归零剔除。""" + cooldown = {"type_a": 3, "type_b": 1, "type_c": 2} + # 模拟 _run_step 末尾的冷却递减 + cooldown = {t: n - 1 for t, n in cooldown.items() if n - 1 > 0} + assert cooldown == {"type_a": 2, "type_c": 1} + + cooldown = {t: n - 1 for t, n in cooldown.items() if n - 1 > 0} + assert cooldown == {"type_a": 1} + + cooldown = {t: n - 1 for t, n in cooldown.items() if n - 1 > 0} + assert cooldown == {}