diff --git a/adapters/baseline_diagnosis_store.py b/adapters/baseline_diagnosis_store.py index a4cb9ca..c744e71 100644 --- a/adapters/baseline_diagnosis_store.py +++ b/adapters/baseline_diagnosis_store.py @@ -106,21 +106,31 @@ class SqliteDiagnosisSignalStore: ) self._conn.commit() - def done_question_ids(self, baseline_run_id: str, diag_fingerprint: str) -> set[str]: + def done_question_ids( + self, + baseline_run_id: str, + diag_fingerprint: str, + *, + retry_uncertain: bool = False, + ) -> set[str]: """查询指定 run 与诊断指纹下已完成的 question_id 集合。 参数: baseline_run_id: baseline run 标识。 diag_fingerprint: 诊断口径指纹。 + retry_uncertain: True 时追加 `AND tier != 'uncertain'`，把 uncertain + （信号不可信降级）题排除出已完成集，令其被重新诊断；默认 False。 返回: 已落盘信号的 question_id 集合；无匹配时为空集，供断点续跑跳过。 """ - cursor = self._conn.execute( + sql = ( "SELECT DISTINCT question_id FROM baseline_diagnosis" - " WHERE baseline_run_id = ? AND diag_fingerprint = ?", - (baseline_run_id, diag_fingerprint), + " WHERE baseline_run_id = ? AND diag_fingerprint = ?" ) + if retry_uncertain: + sql += " AND tier != 'uncertain'" + cursor = self._conn.execute(sql, (baseline_run_id, diag_fingerprint)) return {r["question_id"] for r in cursor.fetchall()} def load(self, baseline_run_id: str, diag_fingerprint: str) -> list[DiagnosisSignalRow]: diff --git a/app/harness/baseline_diagnosis.py b/app/harness/baseline_diagnosis.py index 5632f90..6002911 100644 --- a/app/harness/baseline_diagnosis.py +++ b/app/harness/baseline_diagnosis.py @@ -6,7 +6,9 @@ 包装内层 RunLog，兼容 traces 未落表的历史 run）； 3. 把 error_attributions / infra / degraded 三类产物确定性投影为 DiagnosisSignalRow（tier 由 split_selection.score_signal 判定）； - 4. 逐行 store.upsert 落盘，单行单事务 → 崩溃最多丢正在写的一行。 + 4. run 末（Phase 3）逐行 store.upsert 落库——诊断在 Phase 2 全部跑完后才落库， + 故崩溃丢本次 run 未落库的全部结果（不是"仅一行"）；靠 GovernedLLMClient 的 + Redis 缓存缓解重跑时的 LLM 重烧，下次调用命中缓存直接续。 错误处理诚实标注（不谎称全传播）: - run_diagnosis 的 C1/C2 阶段（指标计算、错误归因）网络/API 失败经 @@ -70,8 +72,9 @@ async def run_baseline_diagnosis( questions: dict[str, GeneratedQuestion], store: DiagnosisSignalStore, deps: DiagnosisDeps, + retry_uncertain: bool = False, ) -> None: - """对 baseline run 的错题跑离线诊断并把信号逐行落库（断点续跑幂等）。 + """对 baseline run 的错题跑离线诊断并把信号落库（断点续跑幂等）。 参数: baseline_run_id: baseline run 标识（如 "infer_adhoc"），信号行主键之一。 @@ -79,8 +82,10 @@ async def run_baseline_diagnosis( wrong_ids: 本次待诊断的可诊断错题 question_id 列表（保序）。 questions: question_id → GeneratedQuestion 映射，需覆盖 wrong_ids 全部题 及 run_diagnosis 返回的所有 infra/degraded 题（用于取 video_id/task_type）。 - store: 诊断信号存储端口，逐行 upsert 落盘并提供 done_question_ids 续跑查询。 + store: 诊断信号存储端口，upsert 落盘并提供 done_question_ids 续跑查询。 deps: 外部依赖束（见 DiagnosisDeps）。 + retry_uncertain: True 时把已落 tier='uncertain'（信号不可信降级）的题也纳入 + remaining 重新诊断，透传给 store.done_question_ids；默认 False。 返回: None。副作用为把逐题 DiagnosisSignalRow 写入 store。 @@ -93,7 +98,9 @@ async def run_baseline_diagnosis( （T0）、degraded_question_ids（uncertain）。 """ # Phase 1: 算 remaining（续跑幂等） - done = store.done_question_ids(baseline_run_id, diag_fingerprint) + done = store.done_question_ids( + baseline_run_id, diag_fingerprint, retry_uncertain=retry_uncertain + ) remaining = [qid for qid in wrong_ids if qid not in done] if not remaining: logger.info( @@ -169,7 +176,8 @@ def _project_and_persist( {tier: 行数} 计数字典（T2/T1/T0/uncertain），供上层日志与 manifest。 关键实现: - 逐行 upsert（单行单事务），中途崩溃最多丢正在写的一行。三桶**非互斥**： + 本函数在 run 末（Phase 3）逐行 upsert（单行单事务）；诊断已在 Phase 2 全部 + 跑完，故本阶段中途崩溃丢本次 run 未落库的余下行。三桶**非互斥**： 同一 degraded 错题可能同时出现在 error_attributions（judge 解析失败仍建 attribution）里，故按 **degraded > infra > attribution** 优先级去重——先落 degraded/infra，再在 attribution 循环跳过已落题，保证**每题恰写一行、 diff --git a/app/harness/video_split_cli.py b/app/harness/video_split_cli.py index 754e8f3..1c7e21f 100644 --- a/app/harness/video_split_cli.py +++ b/app/harness/video_split_cli.py @@ -522,6 +522,7 @@ async def run_pipeline( out_dir: Path, generated_at: str, force: bool = False, + retry_uncertain: bool = False, ) -> SplitBuildResult: """内联三阶段：Phase 0 INFRA T0 补录 → Phase 1 诊断 → Phase 2 冻结切分 → McNemar 护栏。 @@ -538,6 +539,7 @@ async def run_pipeline( out_dir: 冻结产物目录（pools.json + split_manifest.json）。 generated_at: 生成时间戳（ISO 字符串，由调用方传入；见模块 C-2 复现锚点约定）。 force: 覆盖已存在冻结产物开关，透传给 build_split（False 时已存在即报错）。 + retry_uncertain: 透传给 run_baseline_diagnosis，令已落 uncertain 题被重新诊断。 返回: SplitBuildResult（冻结三池 + manifest + assignment）。 @@ -557,6 +559,7 @@ async def run_pipeline( questions=questions, store=signal_store, deps=diagnosis_deps, + retry_uncertain=retry_uncertain, ) # Phase 2: 冻结切分（读诊断信号 → 贪心选择 → 视频组原子切三池 → 冻结 + 六条断言）。 @@ -656,6 +659,7 @@ def _execute_real(config: VideoSplitConfig, fingerprint: str, args: argparse.Nam out_dir=out_dir, generated_at=generated_at, force=args.force, + retry_uncertain=args.retry_uncertain, ) ) finally: @@ -784,6 +788,12 @@ def build_arg_parser() -> argparse.ArgumentParser: action="store_true", help="覆盖已存在的冻结 pools.json/manifest（旧产物备份为 .bak.*）", ) + parser.add_argument( + "--retry-uncertain", + action="store_true", + dest="retry_uncertain", + help="把已落 tier='uncertain'（信号不可信降级）的题重新诊断，而非当作已完成跳过", + ) return parser diff --git a/core/evolution/protocols.py b/core/evolution/protocols.py index df4039c..650f9c2 100644 --- a/core/evolution/protocols.py +++ b/core/evolution/protocols.py @@ -127,12 +127,20 @@ class DiagnosisSignalStore(Protocol): """ ... - def done_question_ids(self, baseline_run_id: str, diag_fingerprint: str) -> set[str]: + def done_question_ids( + self, + baseline_run_id: str, + diag_fingerprint: str, + *, + retry_uncertain: bool = False, + ) -> set[str]: """查询指定 run 与诊断指纹下已完成的 question_id 集合。 参数: baseline_run_id: baseline run 标识。 diag_fingerprint: 诊断口径指纹。 + retry_uncertain: True 时把 tier='uncertain'（信号不可信降级）题视为 + 未完成，令其被重新诊断；默认 False（uncertain 也算完成，不重诊）。 返回: 已落盘信号的 question_id 集合，用于断点续跑跳过。 diff --git a/tests/unit/test_baseline_diagnosis_store.py b/tests/unit/test_baseline_diagnosis_store.py index 00a654d..afee6b9 100644 --- a/tests/unit/test_baseline_diagnosis_store.py +++ b/tests/unit/test_baseline_diagnosis_store.py @@ -85,6 +85,33 @@ def test_null_fields_for_infra_row(tmp_path): assert rows[0].error_type is None and rows[0].evolution_target is None and rows[0].infra is True +def test_done_question_ids_retry_uncertain_excludes(tmp_path): + """retry_uncertain=True 时 uncertain 题不算完成（会被重诊），非 uncertain 仍算完成。""" + s = _store(tmp_path) + # T2（defect）题：已完成 + s.upsert( + DiagnosisSignalRow( + "t2_qid", "v1", "run", "fp", "Counting", + "search_failure", "defect", "T2", "skill", + degraded=False, infra=False, session_id="s", + ) + ) + # uncertain（degraded）题：信号不可信 + s.upsert( + DiagnosisSignalRow( + "uncertain_qid", "v2", "run", "fp", "Counting", + None, None, "uncertain", None, + degraded=True, infra=False, session_id=None, + ) + ) + # 默认：两者都算完成 + assert s.done_question_ids("run", "fp") == {"t2_qid", "uncertain_qid"} + # retry_uncertain：uncertain 被排除，T2 仍算完成 + done = s.done_question_ids("run", "fp", retry_uncertain=True) + assert "uncertain_qid" not in done + assert "t2_qid" in done + + def test_context_manager_closes_connection(tmp_path): # with 块退出后连接关闭，再操作应报 ProgrammingError with SqliteDiagnosisSignalStore(str(tmp_path / "h.db")) as s: