From 77fd35830c62a53cd55d0f5891ace61c36ee0db0 Mon Sep 17 00:00:00 2001 From: iomgaa Date: Thu, 16 Jul 2026 06:27:38 -0400 Subject: [PATCH] fix: idempotent _run_step (DELETE stale) + checkpoint after gate save --- app/harness/runner.py | 69 +++++++++++++++++----- tests/integration/test_checkpoint_pair.py | 1 + tests/unit/test_runner_diag_tree_inject.py | 41 +++++++++++++ 3 files changed, 95 insertions(+), 16 deletions(-) diff --git a/app/harness/runner.py b/app/harness/runner.py index 8f05956..6d04cf2 100644 --- a/app/harness/runner.py +++ b/app/harness/runner.py @@ -887,21 +887,15 @@ class Runner: epoch_batches=batch_unit_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, + # checkpoint 落盘移入 _slow_update_cycle 末尾（gate_pools.save 之后立即写）， + # 消除 gate_epoch_observed 在 gate_pools.json 与 checkpoint 间的双计窗口。 + await self._slow_update_cycle( + epoch, + pools, + state, total_steps=total_steps, - version_snapshot=self._current_version_snapshot(), + step_completed=len(batches) - 1, epoch_batches=batch_unit_ids, - config=self._config, ) if _should_early_stop( self._config.workspace_dir, @@ -1081,10 +1075,21 @@ class Runner: ) -> 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.inference import PREDICTIONS_SCHEMA, TRACES_SCHEMA from app.harness.log import HarnessLog + # 幂等：重跑同一 step 前先清旧行，避免断点续跑重复累计双计。 + # 先 CREATE TABLE IF NOT EXISTS（fresh workspace 首跑时表尚未由 run_inference 建）， + # register_run=False 避免只读清理污染 _runs 运行状态。 + with HarnessLog(str(self._paths.db_path), run_id, register_run=False) as log: + log.create_table("predictions", PREDICTIONS_SCHEMA) + log.create_table("traces", TRACES_SCHEMA) + log.execute("DELETE FROM predictions WHERE run_id=?", (run_id,)) + log.execute("DELETE FROM traces WHERE run_id=?", (run_id,)) + + await self._rollout_batch(batch, run_id) + with HarnessLog(str(self._paths.db_path), run_id) as log: _apply_batch_correctness(state.correctness, log, run_id, batch) @@ -1454,7 +1459,16 @@ class Runner: # _slow_update_cycle 十步序 # ----------------------------------------------------------------------- - async def _slow_update_cycle(self, epoch: int, pools: Pools, state: _TrainState) -> None: + async def _slow_update_cycle( + self, + epoch: int, + pools: Pools, + state: _TrainState, + *, + total_steps: int, + step_completed: int, + epoch_batches: list[list[str]], + ) -> None: """epoch 末慢更新十步序。 1. 捕获版本快照 → 全 val 重跑 R @@ -1466,7 +1480,12 @@ class Runner: 7. system/tool 慢更新（edit_budget_end） 8. R2 闭环 9. 三态标签 + epoch_report + 四向 held-out - 10. gate 阶梯刷新 + 10. gate 阶梯刷新 → 重置 epoch 累加器 → 立即落 epoch_done checkpoint + + 参数: + total_steps: 全局总 step 数（checkpoint 用）。 + step_completed: 本 epoch 已完成 step 数（checkpoint 用）。 + epoch_batches: 本 epoch batch 的 unit_id 划分（checkpoint 用）。 """ # Phase 1 eval_skills_version = self._current_version("skills") @@ -1578,6 +1597,24 @@ class Runner: epoch, pools.baseline_run_id, state, extra_run_ids=r2_kept_run_ids ) + # 重置 epoch 累加器 + 立即落 epoch_done checkpoint：与 _refresh_gate_ladder 内的 + # gate_pools.save + gate_epoch_observed=True 同刻一致，消除断点续跑的双计窗口。 + 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=step_completed, + phase="epoch_done", + global_step=state.global_step, + total_steps=total_steps, + version_snapshot=self._current_version_snapshot(), + epoch_batches=epoch_batches, + config=self._config, + ) + # ----------------------------------------------------------------------- # 慢更新内部方法 # ----------------------------------------------------------------------- diff --git a/tests/integration/test_checkpoint_pair.py b/tests/integration/test_checkpoint_pair.py index c6b28dc..c62fd9b 100644 --- a/tests/integration/test_checkpoint_pair.py +++ b/tests/integration/test_checkpoint_pair.py @@ -111,6 +111,7 @@ class _FakeConfig: diag_size: int = 30 val_size: int = 50 batch_correct_ratio: float = 0.0 + trainable_min_units: int = 8 edit_budget_start: int = 6 edit_budget_end: int = 3 early_stop_patience: int = 3 diff --git a/tests/unit/test_runner_diag_tree_inject.py b/tests/unit/test_runner_diag_tree_inject.py index 5c9a05b..b608644 100644 --- a/tests/unit/test_runner_diag_tree_inject.py +++ b/tests/unit/test_runner_diag_tree_inject.py @@ -212,6 +212,47 @@ async def test_diagnosis_reads_traces_from_steps_json( assert traces[0]["tool_name"] == "search_tree" +@pytest.mark.asyncio +async def test_run_step_deletes_stale_rows_before_rerun( + runner_with_real_store: Runner, monkeypatch: pytest.MonkeyPatch +) -> None: + """同 run_id 重跑前先清 predictions/traces，避免重复行双计（幂等）。""" + from unittest.mock import AsyncMock + + from app.harness.inference import PREDICTIONS_SCHEMA, TRACES_SCHEMA + from app.harness.log import HarnessLog + + run_id = "infer_adhoc_e1_s0" + # 预置该 step run_id 的旧 predictions/traces 行 + with HarnessLog(str(runner_with_real_store._paths.db_path), run_id) as log: + log.create_table("predictions", PREDICTIONS_SCHEMA) + log.create_table("traces", TRACES_SCHEMA) + log.insert("predictions", {"video_id": "vA", "question_id": "q1", "prediction": "A"}) + log.insert("traces", {"video_id": "vA", "question_id": "q1", "step": 0}) + + batch = [_fake_question("q1", "vA")] + runner_with_real_store._rollout_batch = AsyncMock() + monkeypatch.setattr("app.harness.runner._apply_batch_correctness", lambda *a, **k: None) + runner_with_real_store._run_diagnosis = AsyncMock(return_value=DiagnosisResult(run_id=run_id)) + runner_with_real_store._gate_batch_skills = AsyncMock() + + state = MagicMock() + state.correctness = {"q1": True} + state.gate_cooldown = {} + pools = MagicMock() + pools.baseline_run_id = "infer_adhoc" + + await runner_with_real_store._run_step(1, 0, 10, batch, pools, state) + + with HarnessLog( + str(runner_with_real_store._paths.db_path), run_id, register_run=False + ) as log: + preds = log.query("SELECT * FROM predictions WHERE run_id=?", (run_id,)) + traces = log.query("SELECT * FROM traces WHERE run_id=?", (run_id,)) + assert preds == [] + assert traces == [] + + @pytest.mark.asyncio async def test_run_step_aborts_on_high_degrade_rate( runner_with_real_store: Runner, monkeypatch: pytest.MonkeyPatch