From ea6bec542161ee478c05cef164875b143ecb8895 Mon Sep 17 00:00:00 2001 From: iomgaa Date: Fri, 17 Jul 2026 00:57:53 -0400 Subject: [PATCH] fix: clear gate-derived rows on step rerun (idempotency) --- app/harness/runner.py | 63 ++++++++++++++++++---- tests/unit/test_step_rerun_idempotent.py | 69 ++++++++++++++++++++++++ 2 files changed, 123 insertions(+), 9 deletions(-) create mode 100644 tests/unit/test_step_rerun_idempotent.py diff --git a/app/harness/runner.py b/app/harness/runner.py index 3c8b1a1..5b0bb7b 100644 --- a/app/harness/runner.py +++ b/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 前先清旧行，避免断点续跑重复累计双计。 - # 先 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,)) + # 幂等：重跑同一 step 前清 rollout + 全部 gate 派生旧行（修复潜伏 bug： + # 旧实现只清 rollout，gate 行崩溃重跑会累积重复）。 + _clear_step_rows( + str(self._paths.db_path), + baseline_run_id=pools.baseline_run_id, + epoch=epoch, + step=step, + ) await self._rollout_batch(batch, run_id) diff --git a/tests/unit/test_step_rerun_idempotent.py b/tests/unit/test_step_rerun_idempotent.py new file mode 100644 index 0000000..1a966f8 --- /dev/null +++ b/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)