diff --git a/adapters/baseline_diagnosis_store.py b/adapters/baseline_diagnosis_store.py new file mode 100644 index 0000000..75bc44a --- /dev/null +++ b/adapters/baseline_diagnosis_store.py @@ -0,0 +1,154 @@ +"""SqliteDiagnosisSignalStore：baseline 逐题诊断信号的 SQLite 持久化适配器。 + +实现 core/evolution/protocols.py::DiagnosisSignalStore 端口。信号行以 +(question_id, baseline_run_id, diag_fingerprint) 为主键，INSERT OR REPLACE +保证逐题幂等 upsert；bool 字段以 0/1 存储，可空字段以 NULL 存储。表建在 +harness.db，供离线诊断编排写入、视频级切分选择器读取。 +""" + +from __future__ import annotations + +import sqlite3 +from pathlib import Path + +from core.evolution.types import DiagnosisSignalRow + +# 表列顺序即 DiagnosisSignalRow 字段顺序（question_id..session_id）， +# upsert 写入与 load 还原共用，避免手写列名两处漂移。 +_COLUMNS: tuple[str, ...] = ( + "question_id", + "video_id", + "baseline_run_id", + "diag_fingerprint", + "task_type", + "error_type", + "cause_category", + "tier", + "evolution_target", + "degraded", + "infra", + "session_id", +) + + +class SqliteDiagnosisSignalStore: + """逐题诊断信号的 SQLite 存储实现。 + + 构造时按需建表（CREATE TABLE IF NOT EXISTS），主键 + (question_id, baseline_run_id, diag_fingerprint) 保证同键覆盖。 + 每次写操作单事务 commit，保证原子落盘与断点续跑。 + + 参数: + db_path: SQLite 数据库文件路径（通常为 harness.db）。 + """ + + def __init__(self, db_path: str) -> None: + Path(db_path).parent.mkdir(parents=True, exist_ok=True) + self._conn = sqlite3.connect(db_path) + self._conn.row_factory = sqlite3.Row + self._init_table() + + def _init_table(self) -> None: + """创建 baseline_diagnosis 表（若不存在）。""" + self._conn.execute( + """ + CREATE TABLE IF NOT EXISTS baseline_diagnosis ( + question_id TEXT NOT NULL, + video_id TEXT NOT NULL, + baseline_run_id TEXT NOT NULL, + diag_fingerprint TEXT NOT NULL, + task_type TEXT NOT NULL, + error_type TEXT, + cause_category TEXT, + tier TEXT NOT NULL, + evolution_target TEXT, + degraded INTEGER NOT NULL, + infra INTEGER NOT NULL, + session_id TEXT, + PRIMARY KEY (question_id, baseline_run_id, diag_fingerprint) + ) + """ + ) + self._conn.commit() + + def upsert(self, row: DiagnosisSignalRow) -> None: + """写入或覆盖单题诊断信号（按主键幂等，单事务）。 + + 参数: + row: 待持久化的诊断信号行。bool 字段转 0/1，None 存 NULL。 + + 关键实现: + 用 INSERT OR REPLACE 按主键覆盖，避免重复行；commit 保证原子。 + """ + placeholders = ", ".join("?" for _ in _COLUMNS) + col_names = ", ".join(_COLUMNS) + values = ( + row.question_id, + row.video_id, + row.baseline_run_id, + row.diag_fingerprint, + row.task_type, + row.error_type, + row.cause_category, + row.tier, + row.evolution_target, + int(row.degraded), + int(row.infra), + row.session_id, + ) + self._conn.execute( + f"INSERT OR REPLACE INTO baseline_diagnosis ({col_names}) VALUES ({placeholders})", + values, + ) + self._conn.commit() + + def done_question_ids(self, baseline_run_id: str, diag_fingerprint: str) -> set[str]: + """查询指定 run 与诊断指纹下已完成的 question_id 集合。 + + 参数: + baseline_run_id: baseline run 标识。 + diag_fingerprint: 诊断口径指纹。 + + 返回: + 已落盘信号的 question_id 集合；无匹配时为空集，供断点续跑跳过。 + """ + cursor = self._conn.execute( + "SELECT DISTINCT question_id FROM baseline_diagnosis" + " WHERE baseline_run_id = ? AND diag_fingerprint = ?", + (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]: + """加载指定 run 与诊断指纹下的全部诊断信号行。 + + 参数: + baseline_run_id: baseline run 标识。 + diag_fingerprint: 诊断口径指纹。 + + 返回: + 还原后的 DiagnosisSignalRow 列表（0/1→bool，NULL→None）。 + """ + col_names = ", ".join(_COLUMNS) + cursor = self._conn.execute( + f"SELECT {col_names} FROM baseline_diagnosis" + " WHERE baseline_run_id = ? AND diag_fingerprint = ?", + (baseline_run_id, diag_fingerprint), + ) + return [ + DiagnosisSignalRow( + question_id=r["question_id"], + video_id=r["video_id"], + baseline_run_id=r["baseline_run_id"], + diag_fingerprint=r["diag_fingerprint"], + task_type=r["task_type"], + error_type=r["error_type"], + cause_category=r["cause_category"], + tier=r["tier"], + evolution_target=r["evolution_target"], + degraded=bool(r["degraded"]), + infra=bool(r["infra"]), + session_id=r["session_id"], + ) + for r in cursor.fetchall() + ] diff --git a/core/evolution/protocols.py b/core/evolution/protocols.py index 088217f..ecadb06 100644 --- a/core/evolution/protocols.py +++ b/core/evolution/protocols.py @@ -7,7 +7,10 @@ SkillStore / PromptStore 为同步（文件读取量小且快），RunLog 为异 from __future__ import annotations -from typing import Any, Protocol, runtime_checkable +from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable + +if TYPE_CHECKING: + from core.evolution.types import DiagnosisSignalRow @runtime_checkable @@ -104,3 +107,44 @@ class RunLog(Protocol): 轨迹记录字典列表。 """ ... + + +@runtime_checkable +class DiagnosisSignalStore(Protocol): + """逐题诊断信号存储端口。 + + 隔离 SQLite 实现细节，app/core 不写裸 SQL。逐题 upsert 落盘、 + 支持断点续跑（done_question_ids 查已完成集合）。 + """ + + def upsert(self, row: DiagnosisSignalRow) -> None: + """写入或覆盖单题诊断信号（按主键幂等）。 + + 参数: + row: 待持久化的诊断信号行。 + """ + ... + + def done_question_ids(self, baseline_run_id: str, diag_fingerprint: str) -> set[str]: + """查询指定 run 与诊断指纹下已完成的 question_id 集合。 + + 参数: + baseline_run_id: baseline run 标识。 + diag_fingerprint: 诊断口径指纹。 + + 返回: + 已落盘信号的 question_id 集合，用于断点续跑跳过。 + """ + ... + + def load(self, baseline_run_id: str, diag_fingerprint: str) -> list[DiagnosisSignalRow]: + """加载指定 run 与诊断指纹下的全部诊断信号行。 + + 参数: + baseline_run_id: baseline run 标识。 + diag_fingerprint: 诊断口径指纹。 + + 返回: + 还原后的 DiagnosisSignalRow 列表。 + """ + ... diff --git a/core/evolution/types.py b/core/evolution/types.py index 96b5d21..fc8b765 100644 --- a/core/evolution/types.py +++ b/core/evolution/types.py @@ -300,6 +300,44 @@ class DiagnosisResult: degraded_question_ids: list[str] = field(default_factory=list) +@dataclass(frozen=True) +class DiagnosisSignalRow: + """单题诊断信号行，即 baseline run 逐题诊断的持久化单元。 + + 供后续视频级切分选择器消费；由 (question_id, baseline_run_id, + diag_fingerprint) 唯一确定，逐题 upsert 支持断点续跑。 + + 属性: + question_id: 题目唯一标识。 + video_id: 对应视频唯一标识。 + baseline_run_id: 产出该信号的 baseline run 标识。 + diag_fingerprint: 诊断口径指纹，隔离不同诊断配置的信号。 + task_type: 题目任务类型。 + error_type: 错误类别（extraction/search/reasoning/mixed）； + T0/uncertain 行为 None。 + cause_category: 病因类别（defect/lapse）；不适用为 None。 + tier: 诊断分层（T0/T1/T2/uncertain）。 + evolution_target: 进化目标（tool/skill/system）； + error_type 为 None 时亦为 None。 + degraded: 是否为降级信号（judge 解析失败时生成）。 + infra: 是否为 INFRA 护栏排除行。 + session_id: 关联的会话标识；不适用为 None。 + """ + + question_id: str + video_id: str + baseline_run_id: str + diag_fingerprint: str + task_type: str + error_type: str | None + cause_category: str | None + tier: str + evolution_target: str | None + degraded: bool + infra: bool + session_id: str | None + + # ========================================================================= # 3. 进化类型 # ========================================================================= diff --git a/tests/unit/test_baseline_diagnosis_store.py b/tests/unit/test_baseline_diagnosis_store.py new file mode 100644 index 0000000..0702704 --- /dev/null +++ b/tests/unit/test_baseline_diagnosis_store.py @@ -0,0 +1,80 @@ +"""SqliteDiagnosisSignalStore 单元测试。 + +验证逐题诊断信号的 upsert 幂等、断点续跑（done_question_ids）、 +load 往返还原，以及 INFRA 行的可空字段处理。 +""" + +from adapters.baseline_diagnosis_store import DiagnosisSignalRow, SqliteDiagnosisSignalStore + + +def _store(tmp_path): + """构造指向临时 harness.db 的信号存储。""" + return SqliteDiagnosisSignalStore(str(tmp_path / "h.db")) + + +def test_upsert_and_get_done_ids(tmp_path): + s = _store(tmp_path) + row = DiagnosisSignalRow( + question_id="q1", + video_id="v1", + baseline_run_id="r", + diag_fingerprint="fp", + task_type="Counting", + error_type="search_failure", + cause_category="defect", + tier="T2", + evolution_target="skill", + degraded=False, + infra=False, + session_id="s", + ) + s.upsert(row) + assert s.done_question_ids("r", "fp") == {"q1"} + s.upsert(row) # 同键覆盖，不重复 + assert s.done_question_ids("r", "fp") == {"q1"} + assert s.done_question_ids("r", "other_fp") == set() # 不同 fingerprint 隔离 + + +def test_load_rows_roundtrip(tmp_path): + s = _store(tmp_path) + s.upsert( + DiagnosisSignalRow( + "q2", + "v2", + "r", + "fp", + "OCR Problems", + "extraction_failure", + "lapse", + "T1", + "tool", + False, + False, + "s", + ) + ) + rows = s.load("r", "fp") + assert len(rows) == 1 and rows[0].tier == "T1" and rows[0].evolution_target == "tool" + + +def test_null_fields_for_infra_row(tmp_path): + s = _store(tmp_path) + # T0 INFRA 行：error_type/cause_category/evolution_target 为 None + s.upsert( + DiagnosisSignalRow( + "q3", + "v3", + "r", + "fp", + "Counting", + None, + None, + "T0", + None, + degraded=False, + infra=True, + session_id=None, + ) + ) + rows = s.load("r", "fp") + assert rows[0].error_type is None and rows[0].evolution_target is None and rows[0].infra is True