fix: idempotent _run_step (DELETE stale) + checkpoint after gate save

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,
+        )
+
     # -----------------------------------------------------------------------
     # 慢更新内部方法
     # -----------------------------------------------------------------------

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

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