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

This commit is contained in:
2026-07-17 00:57:53 -04:00
parent 9e8a254fbb
commit ea6bec5421
2 changed files with 123 additions and 9 deletions
+54 -9
View File
@@ -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_idgate 派生 run_id
`{step_run_id}_gate_%`)从不清理,重跑会累积重复 predictionsHarnessLog
无主键去重),_load_run_rows 的 dict 覆盖使结果依赖 SELECT 顺序。
gate_evidence / quadrant_pair 以 (run_id, epoch, step) 过滤删除;
表不存在(首个 step)时跳过。step_report 为按文件名覆盖写的 JSON,天然幂等。
参数:
db_path: harness.db 路径。
baseline_run_id: 基线 rungate_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 主类 # Runner 主类
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -1096,17 +1142,16 @@ class Runner:
"""单 steprollout → correctness 增量 → 诊断 → 累加 system/tool → 按类 gate。""" """单 steprollout → correctness 增量 → 诊断 → 累加 system/tool → 按类 gate。"""
run_id = f"{pools.baseline_run_id}_e{epoch}_s{step}" 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 from app.harness.log import HarnessLog
# 幂等:重跑同一 step 前先清旧行,避免断点续跑重复累计双计。 # 幂等:重跑同一 step 前清 rollout + 全部 gate 派生旧行(修复潜伏 bug:
# 先 CREATE TABLE IF NOT EXISTSfresh 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) await self._rollout_batch(batch, run_id)
+69
View File
@@ -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)