@@ -0,0 +1,882 @@
# app/harness/ 训练循环编排层 — 实现计划
> **For agentic workers:** REQUIRED SUB-SKILL: Use subagent-driven-development to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
**Goal: ** 实现 app/harness/ 训练循环编排层（14 个文件），组合 core/evolution/ + core/agent/ + adapters/ 完成自进化闭环。
**Architecture: ** 瘦 Runner class（DI 容器 + 顶层编排）调用扁平模块函数。_TrainState 显式传参，不藏在 self。全异步（asyncio.Semaphore + gather）。store.py/workspace.py 分离（SRP + SDP）。算法保真 #6/ #10/ #13 。
**Tech Stack: ** Python 3.11, asyncio, sqlite3, pydantic-settings, loguru, pytest, pytest-asyncio
**设计规格: ** `research-wiki/designs/2026-07-07-app-harness-design.md`
**TRM4 参考: ** `/home/iomgaa/Projects/Video-Tree-TRM4/core/harness/` (全部 .py) + `/home/iomgaa/Projects/Video-Tree-TRM4/core/workspace.py`
---
## Task 0: core/evolution/__init__.py 补导出 types
**Files: **
- Modify: `core/evolution/__init__.py`
- Test: `tests/unit/test_evolution_types.py` （已存在，验证导入）
- [ ] **Step 1: 在 `__init__.py` 补导出全部 dataclass **
``` python
from core . evolution . types import (
CaseSample , DiagnosePrompts , DiagnosisResult , ErrorAttribution ,
EvolutionRecord , EvolutionResult , EvolvePrompts , GateParams ,
GateVerdict , PairResult , QuadrantClassification , QuestionMetrics ,
RejectedEdit , SkillCasePack , SkillStepAdherence , SpanMetrics ,
SystemCasePack , ToolCasePack ,
)
```
将上述名称加入 `__all__` 。
- [ ] **Step 2: 验证导入 **
Run: `conda activate Video-Tree-TRM && python -c "from core.evolution import GateParams, EvolutionRecord, DiagnosisResult; print('ok')"`
- [ ] **Step 3: 跑全量测试确认无回归 **
Run: `conda activate Video-Tree-TRM && pytest tests/unit/test_evolution_types.py tests/unit/test_gate.py tests/unit/test_evolve.py -q`
- [ ] **Step 4: Commit **
```
feat(evolution): export dataclass types from __init__.py
```
---
## Task 1: config.py — RunConfig + 四层校验
**Files: **
- Create: `app/harness/config.py`
- Test: `tests/unit/test_harness_config.py`
**参考: ** TRM4 `core/harness/config.py` （
**关键差异: **
- 新增 .env 层：工程配置（workspace_dir, store_dir 等路径）可从 .env 读取，优先级 CLI > .env > YAML
- `--resume` /`--fresh` 互斥校验移到 runner.py（不在 config 校验）
- import 路径：`from core.types import GeneratedQuestion`
- [ ] **Step 1: 写测试 ** — 测试 RunConfig 构造、四层校验（valid/invalid）、load_config YAML+CLI 合并
核心测试用例：
``` python
def test_valid_config ( ) : . . .
def test_mode_validation ( ) : . . .
def test_edit_budget_validation ( ) : . . .
def test_minibatch_validation ( ) : . . .
def test_gate_validation ( ) : . . .
def test_load_config_cli_overrides ( ) : . . .
def test_val_size_floor ( ) : . . . # val_size >= eval_min_per_class * 11
```
- [ ] **Step 2: 实现 config.py ** — 从 TRM4 `config.py` 迁移，逐行比对保留全部校验逻辑
- [ ] **Step 3: 测试通过 + lint **
Run: `conda activate Video-Tree-TRM && pytest tests/unit/test_harness_config.py -v && ruff check app/harness/config.py`
- [ ] **Step 4: Commit **
```
feat(harness): config.py — RunConfig frozen dataclass + 四层校验
```
---
## Task 2: log.py — HarnessLog + RunLogImpl
**Files: **
- Create: `app/harness/log.py`
- Test: `tests/unit/test_harness_log.py`
**参考: ** TRM4 `core/harness/log.py` （
- [ ] **Step 1: 写测试 **
``` python
# HarnessLog 测试（TRM4 行为保持）
def test_create_table_and_insert ( ) : . . .
def test_query_thread_safety ( ) : . . . # Lock 保护
def test_context_manager_status ( ) : . . .
def test_insert_or_ignore_idempotent ( ) : . . .
def test_wal_mode ( ) : . . .
# RunLogImpl 测试（新增）
@pytest.mark.asyncio
async def test_run_log_get_predictions ( ) : . . .
@pytest.mark.asyncio
async def test_run_log_get_traces ( ) : . . .
@pytest.mark.asyncio
async def test_run_log_readonly ( ) : . . . # 不触发 _runs INSERT
```
- [ ] **Step 2: 实现 log.py **
HarnessLog: 从 TRM4 直搬。关键保留：WAL + Lock + INSERT OR IGNORE + query 也持锁。
RunLogImpl: 新增，实现 `core.evolution.protocols.RunLog` ：
``` python
class RunLogImpl :
def __init__ ( self , db_path : Path ) - > None :
self . _db_path = db_path
async def get_predictions ( self , run_id , * , question_ids = None ) :
return await asyncio . to_thread ( self . _query_predictions , run_id , question_ids )
async def get_traces ( self , run_id , * , question_ids = None ) :
return await asyncio . to_thread ( self . _query_traces , run_id , question_ids )
def _query_predictions ( self , run_id , question_ids ) :
# 独立 sqlite3.connect（只读，不经 HarnessLog 生命周期）
. . .
```
- [ ] **Step 3: 测试通过 + lint + Protocol 兼容性验证 **
``` python
assert isinstance ( RunLogImpl ( tmp_path / " test.db " ) , RunLog )
```
- [ ] **Step 4: Commit **
```
feat(harness): log.py — HarnessLog SQLite wrapper + RunLogImpl Protocol impl
```
---
## Task 3: store.py — Store 版本操作 + Seed 管理
**Files: **
- Create: `app/harness/store.py`
- Test: `tests/unit/test_harness_store.py`
**参考: ** TRM4 `core/workspace.py` 中 Store + Seed 相关函数（~300 行）。
- [ ] **Step 1: 写测试 **
``` python
def test_parse_version ( ) : . . .
def test_list_versions_numeric_sort ( ) : . . . # v10 排在 v2 后
def test_next_version ( ) : . . .
def test_advance_version ( ) : . . .
def test_init_store ( ) : . . .
def test_init_seed ( ) : . . .
def test_read_seed_not_found ( ) : . . .
def test_extract_run_db_preserves_pk ( ) : . . . # 原始 CREATE 保留主键
def test_promote_to_seed_version_mismatch ( ) : . . . # 强校验
def test_promote_to_seed_null_version ( ) : . . .
```
- [ ] **Step 2: 实现 store.py ** — 从 TRM4 workspace.py 拆出
函数集：`_parse_version` , `list_versions` , `next_version` , `advance_version` , `_write_meta` , `init_store` , `init_seed` , `list_seeds` , `read_seed` , `extract_run_db` , `promote_to_seed` 。
关键保留：numeric 排序（非字典序）；extract_run_db 用原始 CREATE 语句；promote_to_seed 强校验版本一致+非NULL；临时 db finally 清理。
- [ ] **Step 3: 测试通过 + lint **
- [ ] **Step 4: Commit **
```
feat(harness): store.py — Store 版本操作 + Seed 管理
```
---
## Task 4: workspace.py — Workspace 生命周期 + Protocol 实现
**Files: **
- Create: `app/harness/workspace.py`
- Test: `tests/unit/test_harness_workspace.py`
**参考: ** TRM4 `core/workspace.py` 中 Workspace 相关函数（~350 行）。
- [ ] **Step 1: 写测试 **
``` python
def test_resolved_paths_frozen ( ) : . . .
def test_init_workspace ( ) : . . .
def test_init_workspace_from_seed ( ) : . . .
def test_init_workspace_from_seed_missing_questions ( ) : . . . # fail-fast
def test_load_manifest ( ) : . . .
def test_update_manifest_invalid_key ( ) : . . .
def test_record_run_idempotent ( ) : . . . # 同 run_id 不重复 + exist_ok
def test_update_best_independent_of_current ( ) : . . .
def test_archive_workspace ( ) : . . .
# Protocol 实现测试
def test_versioned_skill_store_read ( ) : . . .
def test_versioned_skill_store_list ( ) : . . .
def test_versioned_prompt_store ( ) : . . .
def test_skill_store_protocol_compliance ( ) : . . . # isinstance check
```
- [ ] **Step 2: 实现 workspace.py **
从 TRM4 workspace.py 拆出 Workspace 相关函数。新增 VersionedSkillStore / VersionedPromptStore。
关键保留：
- skills_dir/prompts_dir 解析到 workspace（非 store）
- init_workspace_from_seed 创建前校验 questions ref
- record_run 幂等（history + 目录 exist_ok）
- best 指针独立于 current
依赖 store.py： `from app.harness.store import advance_version, read_seed, ...`
- [ ] **Step 3: 测试通过 + Protocol 兼容性 **
``` python
from core . evolution . protocols import SkillStore , PromptStore
assert isinstance ( VersionedSkillStore ( tmp_path ) , SkillStore )
assert isinstance ( VersionedPromptStore ( tmp_path ) , PromptStore )
```
- [ ] **Step 4: Commit **
```
feat(harness): workspace.py — Workspace lifecycle + SkillStore/PromptStore Protocol impl
```
---
## Task 5: pools.py — 三池切分
**Files: **
- Create: `app/harness/pools.py`
- Test: `tests/unit/test_harness_pools.py`
**参考: ** TRM4 `core/harness/pools.py` （
- [ ] **Step 1: 写测试 **
``` python
def test_build_pools_mutual_exclusion ( ) : . . . # 三池 question_id 无交集
def test_build_pools_test_natural_distribution ( ) : . . . # test 池 correct_ratio=None
def test_save_load_pools_roundtrip ( ) : . . .
def test_load_pools_old_format_reject ( ) : . . . # 无 test → ValueError
def test_build_or_load_pools_frozen ( ) : . . . # 已存在则加载不重切
```
- [ ] **Step 2: 实现 pools.py ** — 从 TRM4 直搬
import 路径变更：`from core.types import GeneratedQuestion` ， `from app.question_gen import stratified_sample` 。
GeneratedQuestion 序列化适配：TRM5 的 options 是 `tuple[str, ...]` + 多了 source_nodes/difficulty 字段，`_q_to_dict` / `_dict_to_q` 需适配。
- [ ] **Step 3: 测试通过 + lint **
- [ ] **Step 4: Commit **
```
feat(harness): pools.py — 三池切分（test→validation→diagnosis）
```
---
## Task 6: batching.py — 算法保真 #10
**Files: **
- Create: `app/harness/batching.py`
- Test: `tests/unit/test_harness_batching.py`
**参考: ** TRM4 `core/harness/batching.py` （241 行），1:1 迁移。**算法保真 #10 **。
- [ ] **Step 1: 写测试 ** — 从 TRM4 测试迁移 + 新增保真校验
``` python
def test_build_batches_deterministic ( ) : . . . # 相同 seed 产出相同结果
def test_small_class_not_split ( ) : . . . # 小类整组不拆
def test_large_class_round_robin ( ) : . . . # 大类全局指针散布
def test_correct_ratio_mixing ( ) : . . . # 正确题按比例混入
def test_no_wrong_answers_empty ( ) : . . . # 无错题 → ([], 0)
def test_validate_params_strict ( ) : . . . # min_class < batch_size
def test_correctness_false_vs_none ( ) : . . . # is False 精确匹配
```
- [ ] **Step 2: 实现 batching.py ** — 从 TRM4 直搬，逐行比对
import 变更：`from core.types import GeneratedQuestion` 。
**保真校验点 ** ：与 TRM4 batching.py 逐函数比对——`build_batches` , `_validate_params` , `_split_by_size` , `_select_mixed_by_task_type` , `_small_groups_decreasing` , `_pack_small_class` , `_distribute_large_classes` , `_place_round_robin` 所有 8 个函数的逻辑完全一致。
- [ ] **Step 3: 测试通过 + lint **
- [ ] **Step 4: Commit **
```
feat(harness): batching.py — FFD + round-robin mini-batch (#10 算法保真)
```
---
## Task 7: gate_ladder.py — 算法保真 #6
**Files: **
- Create: `app/harness/gate_ladder.py`
- Test: `tests/unit/test_harness_gate_ladder.py`
**参考: ** TRM4 `core/harness/gate_ladder.py` （343 行），1:1 迁移。**算法保真 #6 **。
- [ ] **Step 1: 写测试 **
``` python
def test_cold_start_interleaving ( ) : . . . # 2:1 错对交错 + probe 尾
def test_cold_start_p_hat_beta ( ) : . . . # 错=1/3, 对=2/3
def test_warm_ordering_information ( ) : . . . # p̂(1-p̂) 降序
def test_warm_filter_bounds ( ) : . . . # 剔除 p̂ ∉ [p_low, p_high]
def test_gate_pools_save_load_atomic ( ) : . . . # .tmp + os.replace
def test_gate_pools_fingerprint_mismatch ( ) : . . . # RuntimeError
def test_baseline_cache_content_addressed ( ) : . . . # 四维键
def test_baseline_cache_disk_first ( ) : . . . # 先盘后存
def test_ladder_for_excludes_qids ( ) : . . . # 排除案例包题
def test_gamma_ema_update ( ) : . . . # p̂ ← γ·p̂ + (1-γ
def test_update_probs_excludes_gate_runs ( ) : . . . # 调用方须过滤 _gate_ run，
```
- [ ] **Step 2: 实现 gate_ladder.py ** — 从 TRM4 直搬，逐行比对
import 变更：`from core.types import GeneratedQuestion` 。
全部类型和函数保留：`LadderEntry` , `GatePools` , `BaselineCache` , `skill_hash` , `build_cold_entries` , `order_ladder` , `build_or_load_gate_pools` 。
**保真校验点 ** ：冷启动 2:1 交错逻辑、probe 探针抽取、信息量排序公式、γ-EMA 更新、BaselineCache 四维键 + 先盘后存、指纹 sha1 计算。
- [ ] **Step 3: 测试通过 + lint **
- [ ] **Step 4: Commit **
```
feat(harness): gate_ladder.py — 信息阶梯 + BaselineCache (#6 算法保真)
```
---
## Task 8: observation.py — 五表 + 报告
**Files: **
- Create: `app/harness/observation.py`
- Test: `tests/unit/test_harness_observation.py`
**参考: ** TRM4 `core/harness/metric_log.py` （ `loop_report.py` （
- [ ] **Step 1: 写测试 **
``` python
# 五表写入/回读（全覆盖）
def test_write_read_dual_metric ( ) : . . .
def test_write_read_shadow_gate ( ) : . . .
def test_write_read_holdout_eval ( ) : . . .
def test_write_read_gate_evidence ( ) : . . .
def test_write_read_quadrant_pairs ( ) : . . .
def test_null_not_zero_for_soft ( ) : . . . # None → NULL, 不存 0
def test_read_only_connection ( ) : . . . # read 不触发 _runs INSERT
# 报告
def test_write_step_report ( ) : . . .
def test_write_step_report_skipped_null_fields ( ) : . . . # gate 字段 None
def test_write_epoch_report ( ) : . . .
```
- [ ] **Step 2: 实现 observation.py ** — 合并 metric_log + loop_report
全部 5 张表 schema 保留。write_*/read_* 函数保留。新增 write_step_report/write_epoch_report。
关键保留：read 走独立只读连接；soft/mixed None → NULL；
- [ ] **Step 3: 测试通过 + lint **
- [ ] **Step 4: Commit **
```
feat(harness): observation.py — 五张观测表 + step/epoch 报告
```
---
## Task 9: inference.py — async 推理编排
**Files: **
- Create: `app/harness/inference.py`
- Test: `tests/unit/test_harness_inference.py`
**参考: ** TRM4 `core/harness/inference.py` （~560 行）。**重大重构：同步→异步 + DI**。
- [ ] **Step 1: 写测试 **
``` python
@pytest.mark.asyncio
async def test_run_inference_basic ( ) : . . . # mock LLM + ToolDispatcher
@pytest.mark.asyncio
async def test_run_inference_concurrency ( ) : . . . # Semaphore 限制
@pytest.mark.asyncio
async def test_prediction_always_written ( ) : . . . # 异常时仍落库 stop_reason=error
@pytest.mark.asyncio
async def test_to_text_field ( ) : . . . # list/dict → JSON str
@pytest.mark.asyncio
async def test_run_id_empty_raises ( ) : . . . # 空串 ValueError
@pytest.mark.asyncio
async def test_aggregate_results_from_memory ( ) : . . . # 不从 DB 回读
@pytest.mark.asyncio
async def test_inference_result_frozen ( ) : . . .
```
- [ ] **Step 2: 实现 inference.py **
关键签名：
``` python
async def run_inference (
questions : list [ GeneratedQuestion ] ,
* ,
llm : LLMProvider ,
tool_dispatch_fn : Callable ,
log : HarnessLog ,
run_id : str ,
concurrency : int ,
max_steps : int ,
skill_mode : str ,
plugins_factory : Callable [ [ str , str ] , list [ object ] ] | None = None ,
) - > InferenceResult :
```
从 TRM4 迁移：InferenceResult dataclass、5 张表 schema、_to_text_field、_run_single_question（→ async）、_aggregate_results（从内存聚合）。
新增：asyncio.Semaphore + gather 替代 ThreadPoolExecutor。plugins_factory 接收 (video_id, question_id) 返回插件列表（TracePlugin 等由调用方装配）。
关键保留：悲观默认值（stop_reason="error"）；prediction 必落库（try 外）；_to_text_field 归一化。
- [ ] **Step 3: 测试通过 + lint **
- [ ] **Step 4: Commit **
```
feat(harness): inference.py — async run_inference + DI (#11 loop integration)
```
---
## Task 10: validate.py — 块序贯验证编排
**Files: **
- Create: `app/harness/validate.py`
- Test: `tests/unit/test_harness_validate.py`
**参考: ** TRM4 `core/harness/validate.py` （626 行）。**重大重构：sync→async + 调 core/evolution 纯函数**。
- [ ] **Step 1: 写测试 **
``` python
# 类型测试
def test_validation_outcome_fields ( ) : . . .
def test_probation_fields ( ) : . . .
# materialize
def test_materialize_candidate_skill ( ) : . . .
def test_materialize_cleanup_on_failure ( ) : . . .
# 块序贯验证
@pytest.mark.asyncio
async def test_validate_skill_local_accept ( ) : . . . # mock inference
@pytest.mark.asyncio
async def test_validate_skill_local_reject ( ) : . . .
@pytest.mark.asyncio
async def test_gate_prefix_must_contain_gate ( ) : . . . # ValueError
@pytest.mark.asyncio
async def test_infra_guard_threshold ( ) : . . . # 分母≥10 才触发
@pytest.mark.asyncio
async def test_baseline_cache_hit ( ) : . . . # miss 才推理
@pytest.mark.asyncio
async def test_block_level_barrier ( ) : . . . # 块间顺序执行
@pytest.mark.asyncio
async def test_last_block_terminal ( ) : . . . # 无循环外补判
```
- [ ] **Step 2: 实现 validate.py **
类型定义：InferenceRunConfig, ValidationOutcome, Probation（从 TRM4 迁移，Probation 的 pending_edits 用 `core.evolution.types.RejectedEdit` ）。
函数 async 化：validate_skill_local, _run_local_validation, _resolve_baseline_block, _run_candidate_block 加 async。
替换 core/ 调用：`from core.evolution import gate_decision, pair_block, classify_quadrants` （纯函数，直接调用）。替代 TRM4 的 `_pair_block` 和 `_classify_quadrants` 本地实现。
关键保留：gate_run_prefix 含 "_gate_" 入口校验；materialize 用 .cand_tmp/ + finally 清理；INFRA 护栏跨块累计分母≥10；块级屏障顺序执行；最后一块判定即终态；只有终态题携带 stop_reason。
- [ ] **Step 3: 测试通过 + lint **
- [ ] **Step 4: Commit **
```
feat(harness): validate.py — async 块序贯验证编排
```
---
## Task 11: momentum.py — 慢更新动量
**Files: **
- Create: `app/harness/momentum.py`
- Test: `tests/unit/test_harness_momentum.py`
**参考: ** TRM4 `core/harness/momentum.py` （
- [ ] **Step 1: 写测试 **
``` python
def test_categorize_pair_all_four ( ) : . . .
def test_categorize_pair_missing_key ( ) : . . . # KeyError 不掩盖
def test_format_comparison_pairs_order ( ) : . . . # REGRESSED 优先
def test_format_comparison_pairs_empty ( ) : . . .
@pytest.mark.asyncio
async def test_run_slow_momentum_basic ( ) : . . . # mock LLM
@pytest.mark.asyncio
async def test_run_slow_momentum_parse_failure ( ) : . . . # 保留 prev_guidance
```
- [ ] **Step 2: 实现 momentum.py ** — 从 TRM4 迁移 + async 化
关键保留：
- 四类常量单一真源
- 展示顺序 REGRESSED 优先
- `_format_comparison_pairs` 在 try 外（KeyError 不被 ValueError 吞）
- 解析失败保留 prev_guidance
- `client.chat` → `await llm.chat` ，返回 LLMResponse
- `extract_json_from_response` 从 `core.evolution.diagnose` 导入（TRM5 已有）
- [ ] **Step 3: 测试通过 + lint **
- [ ] **Step 4: Commit **
```
feat(harness): momentum.py — async 慢更新动量生成
```
---
## Task 12: checkpoint.py — 状态序列化
**Files: **
- Create: `app/harness/checkpoint.py`
- Test: `tests/unit/test_harness_checkpoint.py`
**参考: ** TRM4 `core/harness/checkpoint.py` （
- [ ] **Step 1: 写测试 **
``` python
def test_serialize_deserialize_roundtrip ( ) : . . .
def test_serialize_set_to_sorted_list ( ) : . . . # changed_task_types
def test_deserialize_nested_system_pack ( ) : . . . # CaseSample 重建
def test_deserialize_nested_probation ( ) : . . . # RejectedEdit 重建
def test_deserialize_missing_key_raises ( ) : . . . # 硬失败不 .get
def test_fingerprint_structural_vs_decision ( ) : . . .
def test_check_fingerprint_structural_reject ( ) : . . .
def test_check_fingerprint_decision_warn ( ) : . . .
def test_write_checkpoint_atomic ( ) : . . . # .tmp + os.replace
def test_load_checkpoint_missing ( ) : . . . # 返回 None
```
- [ ] **Step 2: 实现 checkpoint.py ** — 从 TRM4 直搬
import 路径变更：
- `from core.evolution.types import CaseSample, SystemCasePack, ToolCasePack, RejectedEdit`
- `from app.harness.validate import Probation`
完整持久化字段集：correctness, eval_prev_acc, eval_prev_run_id, baseline_skills_version, baseline_prompts_version, steps_since_best_improved, epoch_start_skills, changed_task_types_this_epoch, rejected_buffer, system_packs, tool_packs, probations, gate_cooldown, gate_epoch_observed。
嵌套复活规则保留：SystemCasePack→CaseSample, Probation→RejectedEdit。
关键保留：反序列化 d[...] 不用 .get；结构性 vs 决策性指纹；原子写。
- [ ] **Step 3: 测试通过 + lint **
- [ ] **Step 4: Commit **
```
feat(harness): checkpoint.py — TrainState 序列化 + 原子写 + 指纹校验
```
---
## Task 13: runner.py — 训练循环编排（算法保真 #13）
**Files: **
- Create: `app/harness/runner.py`
- Test: `tests/unit/test_harness_runner.py`
**参考: ** TRM4 `core/harness/runner.py` （
本 Task 分为多个 sub-step，按训练循环层级组织。
### Sub-task 13a: Runner 类骨架 + _TrainState + 模式路由
- [ ] **Step 1: 写测试 ** — Runner 构造、infer 模式、eval 模式
``` python
@pytest.mark.asyncio
async def test_runner_init ( ) : . . .
@pytest.mark.asyncio
async def test_runner_infer ( ) : . . . # mock inference
@pytest.mark.asyncio
async def test_runner_eval_backfill_versions ( ) : . . . # _runs 版本回填
```
- [ ] **Step 2: 实现 Runner 类骨架 **
``` python
class Runner :
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 . _paths = resolve_paths ( config . workspace_dir )
async def infer ( self , . . . ) - > InferenceResult : . . .
async def eval ( self , version : str ) - > InferenceResult : . . .
async def diagnose ( self , run_id : str ) - > DiagnosisResult : . . .
def promote ( self , version , eval_run_id , name ) - > None : . . .
```
_TrainState dataclass（
关键保留：_ensure_workspace 的 resume+fresh 互斥 → ValueError； ；
- [ ] **Step 3: 测试通过 **
### Sub-task 13b: train() 三级嵌套 + _run_step
- [ ] **Step 4: 写测试 ** — train 循环骨架、rollout、correctness 增量
``` python
@pytest.mark.asyncio
async def test_train_epoch_step_nesting ( ) : . . . # mock 全链路
@pytest.mark.asyncio
async def test_run_step_sequence ( ) : . . . # rollout→correctness→diagnose→gate
@pytest.mark.asyncio
async def test_guard_infra_failures ( ) : . . . # >10% error rate
@pytest.mark.asyncio
async def test_apply_batch_correctness_complete ( ) : . . . # 缺行 → RuntimeError
@pytest.mark.asyncio
async def test_apply_batch_correctness_incremental ( ) : . . . # 增量更新
@pytest.mark.asyncio
async def test_init_gate_pools_empty_baseline_raises ( ) : . . . # baseline run 零 prediction 行 → RuntimeError
```
- [ ] **Step 5: 实现 train() + _run_step + 辅助函数 **
模块级函数（不依赖 self）：`resume_plan` , `_guard_infra_failures` , `_apply_batch_correctness` , `_accumulate_slow_packs` , `_batch_from_ids` , `_snapshot_current_skills` , `_compute_total_steps` , `_should_early_stop` 。
train() 三级嵌套：epoch 循环 + batch 切分 + step 循环 + checkpoint。
关键保留：resume 用 saved_batches；新 epoch 清空累加器；每 step 后 global_step++；
- [ ] **Step 6: 测试通过 **
### Sub-task 13c: _gate_batch_skills + accept/reject/probation
- [ ] **Step 7: 写测试 ** — per-skill gate、accept/reject、probation
``` python
@pytest.mark.asyncio
async def test_gate_cooldown_skip ( ) : . . .
@pytest.mark.asyncio
async def test_gate_no_change_skip ( ) : . . . # evolved==original
@pytest.mark.asyncio
async def test_gate_accept_confirmed ( ) : . . .
@pytest.mark.asyncio
async def test_gate_accept_provisional_open_probation ( ) : . . .
@pytest.mark.asyncio
async def test_gate_default_strategy_no_probation ( ) : . . . # 共享文件不开账
@pytest.mark.asyncio
async def test_gate_reject_blacklist ( ) : . . .
@pytest.mark.asyncio
async def test_rejected_summary_only_applied ( ) : . . . # 黑名单防污染
@pytest.mark.asyncio
async def test_rollback_probation_file_level ( ) : . . . # revert-commit 式
@pytest.mark.asyncio
async def test_cooldown_decrement ( ) : . . . # 每 step 递减
```
- [ ] **Step 8: 实现 _gate_batch_skills + accept/reject/rollback **
关键保留：cooldown admission control； ；
- [ ] **Step 9: 测试通过 **
### Sub-task 13d: _slow_update_cycle 十步序
- [ ] **Step 10: 写测试 ** — 十步序关键路径
``` python
@pytest.mark.asyncio
async def test_slow_update_r_before_prompts ( ) : . . . # Phase 1 先于 7
@pytest.mark.asyncio
async def test_slow_update_probation_settle ( ) : . . . # Phase 4
@pytest.mark.asyncio
async def test_slow_update_best_strict_greater ( ) : . . . # Phase 5
@pytest.mark.asyncio
async def test_slow_update_momentum_immutable_version ( ) : . . . # Phase 6
@pytest.mark.asyncio
async def test_slow_update_r2_revert ( ) : . . . # Phase 8 退步
@pytest.mark.asyncio
async def test_slow_update_r2_keep ( ) : . . . # Phase 8 保留
@pytest.mark.asyncio
async def test_slow_update_r2_version_binding ( ) : . . . # R2 用 r2_skills_version
@pytest.mark.asyncio
async def test_slow_update_gate_refresh_sources ( ) : . . . # 精确三源
@pytest.mark.asyncio
async def test_slow_update_reverted_r2_not_absorbed ( ) : . . . # 防泄露
@pytest.mark.asyncio
async def test_soft_score_missing_table_raises ( ) : . . . # span_evaluations 表不存在 → RuntimeError
@pytest.mark.asyncio
async def test_probation_settle_missing_prediction_raises ( ) : . . . # 快照题缺预测行 → RuntimeError
@pytest.mark.asyncio
async def test_momentum_samples_from_diagnosis_pool ( ) : . . . # 从诊断池采样，不从 val 池
```
- [ ] **Step 11: 实现 _slow_update_cycle **
十步序完整实现，~150 行。逐步比对 TRM4 runner.py:1363-1523。
关键保留：版本快照在 R 前捕获；R 无条件回写；probation 结算覆盖回写；best argmax 严格大于；momentum 不可变新版本；system/tool 用 edit_budget_end； _ gate _ + R 精确 run_id + kept R2）；reverted R2 不吸收。
- [ ] **Step 12: 测试通过 **
### Sub-task 13e: deliver_best + final_test + held-out
- [ ] **Step 13: 写测试 **
``` python
@pytest.mark.asyncio
async def test_deliver_best_rollback ( ) : . . .
@pytest.mark.asyncio
async def test_final_test_eval ( ) : . . .
@pytest.mark.asyncio
async def test_early_stop_step_granularity ( ) : . . . # 步粒度累加
@pytest.mark.asyncio
async def test_epoch_report_system_tool_action_tristate ( ) : . . . # updated/reverted/none
@pytest.mark.asyncio
async def test_holdout_no_decision_side_effect ( ) : . . . # held-out 仅观测落库
```
- [ ] **Step 14: 实现 deliver_best + final_test + held-out **
- [ ] **Step 15: 全量测试通过 + lint **
Run: `conda activate Video-Tree-TRM && pytest tests/unit/test_harness_runner.py -v`
- [ ] **Step 16: Commit **
```
feat(harness): runner.py — 瘦编排器 + 训练循环三级嵌套 (#13 算法保真)
```
---
## Task 14: __init__.py + 集成验证
**Files: **
- Modify: `app/harness/__init__.py`
- Test: `tests/unit/test_harness_init.py`
- [ ] **Step 1: 写 __init__.py 公开 API **
``` python
""" app/harness/ — 训练循环编排层。 """
from app . harness . config import RunConfig , load_config
from app . harness . inference import InferenceResult , run_inference
from app . harness . log import HarnessLog , RunLogImpl
from app . harness . pools import Pools , build_or_load_pools , build_pools , load_pools , save_pools
from app . harness . runner import Runner
from app . harness . workspace import (
ResolvedPaths , VersionedPromptStore , VersionedSkillStore ,
resolve_paths ,
)
__all__ = [
" HarnessLog " , " InferenceResult " , " Pools " , " ResolvedPaths " ,
" RunConfig " , " RunLogImpl " , " Runner " , " VersionedPromptStore " ,
" VersionedSkillStore " , " build_or_load_pools " , " build_pools " ,
" load_config " , " load_pools " , " resolve_paths " , " run_inference " ,
" save_pools " ,
]
```
- [ ] **Step 2: 集成测试 ** — 验证全模块 import + Runner 构造
``` python
def test_public_api_imports ( ) : . . .
def test_runner_construction ( ) : . . . # 全依赖注入
```
- [ ] **Step 3: 跑全量 harness 测试 **
Run: `conda activate Video-Tree-TRM && pytest tests/unit/test_harness_*.py -v --tb=short`
- [ ] **Step 4: 跑全量项目测试确认无回归 **
Run: `conda activate Video-Tree-TRM && pytest tests/ -q`
- [ ] **Step 5: lint 全量 **
Run: `conda activate Video-Tree-TRM && ruff check app/harness/ --fix && ruff format app/harness/`
- [ ] **Step 6: Commit **
```
feat(harness): __init__.py public API + integration verification
```
---
## 算法保真校验
本计划涉及 3 项核心算法迁移：
| # | 算法 | 保真 Task | 校验方式 |
|---|------|----------|---------|
| 6 | 信息阶梯 | Task 7 | 逐函数比对 TRM4 gate_ladder.py（
| 10 | mini-batch | Task 6 | 逐函数比对 TRM4 batching.py（
| 13 | 训练循环编排 | Task 13 | 逐方法比对 TRM4 runner.py（十步序 + 三级嵌套 + probation）
不涉及的核心算法（#1 -#5 , #7 -#9 , #11 -#12 ）：已在 core/evolution/ (Design A) 或 app/tree/ 中实现，本计划不触及。
---
## TRM4 Tips 覆盖索引
120 条 tips 按 Task 归属：
| Task | Tips |
|------|------|
| 0 | — |
| 1 (config) | T48-T53 |
| 2 (log) | T87-T90 |
| 3 (store) | T114, T118-T120 |
| 4 (workspace) | T113, T115-T117 |
| 5 (pools) | T69-T72 |
| 6 (batching) | T62-T68 |
| 7 (gate_ladder) | T73-T81 |
| 8 (observation) | T106-T109, T110-T112 |
| 9 (inference) | T54-T61 |
| 10 (validate) | T97-T105 |
| 11 (momentum) | T82-T86 |
| 12 (checkpoint) | T91-T96 |
| 13 (runner) | T1-T47 |
| 14 (__init__) | — |
T6-T7（ （
T8（ （
T9（ （
