fix: clear gate-derived rows on step rerun (idempotency)

app/harness/runner.py

@@ -602,6 +602,52 @@ def _write_skip_report(
     )
 
 
+def _clear_step_rows(db_path: str, *, baseline_run_id: str, epoch: int, step: int) -> None:
+    """清空一个 step 的全部旧行（rollout + gate 派生），保证崩溃重跑幂等。
+
+    修复前序潜伏 bug：旧实现只清 rollout run_id，gate 派生 run_id
+    （`{step_run_id}_gate_%`）从不清理，重跑会累积重复 predictions（HarnessLog
+    无主键去重），_load_run_rows 的 dict 覆盖使结果依赖 SELECT 顺序。
+    gate_evidence / quadrant_pair 以 (run_id, epoch, step) 过滤删除；
+    表不存在（首个 step）时跳过。step_report 为按文件名覆盖写的 JSON，天然幂等。
+
+    参数:
+        db_path: harness.db 路径。
+        baseline_run_id: 基线 run（gate_evidence/quadrant_pair 的 run_id 维度）。
+        epoch: 轮次（1-based）。
+        step: epoch 内 step 序号（0-based）。
+
+    返回:
+        无。
+
+    关键实现细节:
+        predictions/traces 的 gate 行按 LIKE 前缀删除，'_' 通配显式转义
+        （ESCAPE）钉死字面匹配，避免 `..._s1` 误匹配 `..._s10` 类前缀陷阱。
+    """
+    from app.harness.inference import PREDICTIONS_SCHEMA, TRACES_SCHEMA
+    from app.harness.log import HarnessLog
+
+    step_run_id = f"{baseline_run_id}_e{epoch}_s{step}"
+    with HarnessLog(db_path, step_run_id, register_run=False) as log:
+        log.create_table("predictions", PREDICTIONS_SCHEMA)
+        log.create_table("traces", TRACES_SCHEMA)
+        for table in ("predictions", "traces"):
+            log.execute(f"DELETE FROM {table} WHERE run_id=?", (step_run_id,))
+            log.execute(
+                f"DELETE FROM {table} WHERE run_id LIKE ? ESCAPE '\\'",
+                (step_run_id.replace("_", r"\_") + r"\_gate\_%",),
+            )
+        for table in ("gate_evidence", "quadrant_pair"):
+            exists = log.query(
+                "SELECT name FROM sqlite_master WHERE type='table' AND name=?", (table,)
+            )
+            if exists:
+                log.execute(
+                    f"DELETE FROM {table} WHERE run_id=? AND epoch=? AND step=?",
+                    (baseline_run_id, epoch, step),
+                )
+
+
 # ---------------------------------------------------------------------------
 # Runner 主类
 # ---------------------------------------------------------------------------
@@ -1096,17 +1142,16 @@ class Runner:
         """单 step：rollout → correctness 增量 → 诊断 → 累加 system/tool → 按类 gate。"""
         run_id = f"{pools.baseline_run_id}_e{epoch}_s{step}"
 
-        from app.harness.inference import PREDICTIONS_SCHEMA, TRACES_SCHEMA
         from app.harness.log import HarnessLog
 
-        # 幂等：重跑同一 step 前先清旧行，避免断点续跑重复累计双计。
+        # 幂等：重跑同一 step 前清 rollout + 全部 gate 派生旧行（修复潜伏 bug：
-        # 先 CREATE TABLE IF NOT EXISTS（fresh workspace 首跑时表尚未由 run_inference 建），
+        # 旧实现只清 rollout，gate 行崩溃重跑会累积重复）。
-        # register_run=False 避免只读清理污染 _runs 运行状态。
+        _clear_step_rows(
-        with HarnessLog(str(self._paths.db_path), run_id, register_run=False) as log:
+            str(self._paths.db_path),
-            log.create_table("predictions", PREDICTIONS_SCHEMA)
+            baseline_run_id=pools.baseline_run_id,
-            log.create_table("traces", TRACES_SCHEMA)
+            epoch=epoch,
-            log.execute("DELETE FROM predictions WHERE run_id=?", (run_id,))
+            step=step,
-            log.execute("DELETE FROM traces WHERE run_id=?", (run_id,))
+        )
 
         await self._rollout_batch(batch, run_id)
 

tests/unit/test_step_rerun_idempotent.py

@@ -0,0 +1,69 @@
+"""step 重跑幂等：gate 派生行必须随 step 清理，否则崩溃重跑累积重复。"""
+
+from __future__ import annotations
+
+import sqlite3
+from typing import TYPE_CHECKING
+
+from app.harness.runner import _clear_step_rows
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+
+def _mk_db(tmp_path: Path) -> Path:
+    """构造含 rollout 行、gate 派生行、他 step 行与前缀陷阱行的最小 harness.db。
+
+    参数:
+        tmp_path: pytest 临时目录。
+
+    返回:
+        harness.db 路径。
+    """
+    db = tmp_path / "harness.db"
+    conn = sqlite3.connect(db)
+    conn.execute("CREATE TABLE predictions (run_id TEXT, question_id TEXT)")
+    conn.execute("CREATE TABLE traces (run_id TEXT, question_id TEXT)")
+    conn.execute("CREATE TABLE gate_evidence (run_id TEXT, epoch INTEGER, step INTEGER)")
+    conn.execute("CREATE TABLE quadrant_pair (run_id TEXT, epoch INTEGER, step INTEGER)")
+    rows = [
+        ("infer_adhoc_e1_s0", "q1"),  # rollout 行
+        ("infer_adhoc_e1_s0_gate_action-reasoning_base", "q2"),  # gate base 臂
+        ("infer_adhoc_e1_s0_gate_action-reasoning_cand", "q3"),  # gate cand 臂
+        ("infer_adhoc_e1_s1", "q4"),  # 其他 step，不许误删
+        ("infer_adhoc_e1_s10_gate_x_base", "q5"),  # s10 前缀陷阱，不许误删
+    ]
+    conn.executemany("INSERT INTO predictions VALUES (?, ?)", rows)
+    conn.executemany("INSERT INTO traces VALUES (?, ?)", rows)
+    conn.execute("INSERT INTO gate_evidence VALUES ('infer_adhoc', 1, 0)")
+    conn.execute("INSERT INTO gate_evidence VALUES ('infer_adhoc', 1, 1)")
+    conn.execute("INSERT INTO quadrant_pair VALUES ('infer_adhoc', 1, 0)")
+    conn.commit()
+    conn.close()
+    return db
+
+
+def test_clear_step_rows_removes_rollout_and_gate_rows(tmp_path) -> None:
+    """rollout 行 + 本 step 全部 gate 派生行被清；他 step 与 s10 前缀陷阱不动。"""
+    db = _mk_db(tmp_path)
+    _clear_step_rows(str(db), baseline_run_id="infer_adhoc", epoch=1, step=0)
+    conn = sqlite3.connect(db)
+    left = {r[0] for r in conn.execute("SELECT run_id FROM predictions")}
+    assert left == {"infer_adhoc_e1_s1", "infer_adhoc_e1_s10_gate_x_base"}
+    left_t = {r[0] for r in conn.execute("SELECT run_id FROM traces")}
+    assert left_t == left
+    ge = list(conn.execute("SELECT step FROM gate_evidence"))
+    assert ge == [(1,)]  # 只剩 step=1 的行
+    assert list(conn.execute("SELECT COUNT(*) FROM quadrant_pair"))[0][0] == 0
+    conn.close()
+
+
+def test_clear_step_rows_missing_tables_is_noop(tmp_path) -> None:
+    """gate_evidence/quadrant_pair 表尚未建（首个 step）时不报错。"""
+    db = tmp_path / "harness.db"
+    conn = sqlite3.connect(db)
+    conn.execute("CREATE TABLE predictions (run_id TEXT)")
+    conn.execute("CREATE TABLE traces (run_id TEXT)")
+    conn.commit()
+    conn.close()
+    _clear_step_rows(str(db), baseline_run_id="infer_adhoc", epoch=1, step=0)