diff --git a/app/harness/observation.py b/app/harness/observation.py new file mode 100644 index 0000000..f9ddd92 --- /dev/null +++ b/app/harness/observation.py @@ -0,0 +1,459 @@ +"""五张观测表的落库写入与回读 + step/epoch 报告文件输出。 + +合并 TRM4 的 metric_log.py（五表）和 loop_report.py（报告）。 + +五张表均经 structured-logging 定义，DDL 与之逐列一致： + dual_metric_eval / shadow_gate / holdout_eval / quadrant_pair / gate_evidence。 + +公共契约（守 P5）：soft/mixed 为 None（invalid，无 span / 诊断失败）时存 NULL，**绝不存 0**—— +SQLite 对 dict 中 None 值写入即 NULL，分析时按 NULL 跳过。每个写函数内幂等建表 +（``HarnessLog.create_table`` 用 CREATE TABLE IF NOT EXISTS），run_id/timestamp 列由 +HarnessLog 自动补。 + +报告函数输出 JSON 到 workspace 的 analyses/ 目录，供人工审查诊断 prompt 与进化 prompt。 +""" + +from __future__ import annotations + +import json +import sqlite3 +from typing import TYPE_CHECKING, Any + +if TYPE_CHECKING: + from pathlib import Path + + +def _read_table(db_path: str, table: str, run_id: str) -> list[dict[str, Any]]: + """纯读某表指定 run 的全部行——不经 HarnessLog 生命周期，避免回读污染 _runs 运行状态。 + + HarnessLog.__enter__/__exit__ 会对 run_id 做 INSERT OR IGNORE 并在退出时标 completed； + 回读指标绝不应改运行状态，故 read_* 一律走本只读连接（仅 SELECT）。 + + 参数: + db_path: SQLite 路径。 + table: 表名（内部固定常量，非外部输入，无注入风险）。 + run_id: 过滤的 run ID。 + + 返回: + 行 dict 列表；表尚未建（没写过）视为无数据返 []。 + """ + conn = sqlite3.connect(db_path) + conn.row_factory = sqlite3.Row + try: + exists = conn.execute( + "SELECT name FROM sqlite_master WHERE type='table' AND name=?", (table,) + ).fetchone() + if exists is None: + return [] + rows = conn.execute(f"SELECT * FROM {table} WHERE run_id=?", (run_id,)).fetchall() + return [dict(r) for r in rows] + finally: + conn.close() + + +# --------------------------------------------------------------------------- +# 列定义严格对齐 research-wiki/schemas/*.md（run_id/timestamp 由 create_table 自动补） +# --------------------------------------------------------------------------- + +_DUAL_COLS: dict[str, str] = { + "epoch": "INTEGER", + "version_kind": "TEXT", + "skills_version": "TEXT", + "prompts_version": "TEXT", + "pool": "TEXT", + "hard_acc": "REAL", + "soft_score": "REAL", + "mixed_score": "REAL", +} + +_SHADOW_COLS: dict[str, str] = { + "epoch": "INTEGER", + "candidate_version": "TEXT", + "hard_acc": "REAL", + "soft_score": "REAL", + "mixed_score": "REAL", + "is_mixed_best": "INTEGER", +} + +_HOLDOUT_COLS: dict[str, str] = { + "epoch": "INTEGER", + "version_kind": "TEXT", + "hard_acc": "REAL", + "soft_score": "REAL", + "mixed_score": "REAL", + "per_task_type_json": "TEXT", +} + +_QUADRANT_COLS: dict[str, str] = { + "epoch": "INTEGER", + "step": "INTEGER", + "question_id": "TEXT", + "task_type": "TEXT", + "prev_correct": "INTEGER", + "curr_correct": "INTEGER", + "category": "TEXT", +} + +_GATE_EVIDENCE_COLS: dict[str, str] = { + "epoch": "INTEGER", + "step": "INTEGER", + "task_type": "TEXT", + "question_id": "TEXT", + "block_idx": "INTEGER", + "baseline_correct": "INTEGER", + "candidate_correct": "INTEGER", + "e_value": "REAL", + "stop_reason": "TEXT", +} + + +# --------------------------------------------------------------------------- +# dual_metric_eval +# --------------------------------------------------------------------------- + + +def write_dual_metric( + db_path: str, + *, + run_id: str, + epoch: int, + version_kind: str, + skills_version: str, + prompts_version: str, + pool: str, + hard_acc: float, + soft_score: float | None, + mixed_score: float | None, +) -> None: + """落 dual_metric_eval 一行：epoch 末关键版本的 hard+soft+mixed 双轨度量。 + + 参数: + db_path: SQLite 路径。 + run_id: 训练 run ID。 + epoch: 轮次（1-based）。 + version_kind: baseline / best_hard / best_mixed / final。 + skills_version / prompts_version: 评估的资源版本。 + pool: val / test。 + hard_acc: hard 准确率。 + soft_score: soft 连续分；invalid 传 None -> 存 NULL。 + mixed_score: 0.5*hard+0.5*soft；soft 缺失传 None -> 存 NULL。 + """ + from app.harness.log import HarnessLog + + with HarnessLog(db_path, run_id) as log: + log.create_table("dual_metric_eval", _DUAL_COLS) + log.insert( + "dual_metric_eval", + { + "epoch": epoch, + "version_kind": version_kind, + "skills_version": skills_version, + "prompts_version": prompts_version, + "pool": pool, + "hard_acc": hard_acc, + "soft_score": soft_score, + "mixed_score": mixed_score, + }, + ) + + +def read_dual_metric(db_path: str, *, run_id: str) -> list[dict[str, Any]]: + """回读指定 run 的 dual_metric_eval 全部行（纯读，不污染运行状态）。""" + return _read_table(db_path, "dual_metric_eval", run_id) + + +# --------------------------------------------------------------------------- +# shadow_gate +# --------------------------------------------------------------------------- + + +def write_shadow_gate( + db_path: str, + *, + run_id: str, + epoch: int, + candidate_version: str, + hard_acc: float, + soft_score: float | None, + mixed_score: float | None, + is_mixed_best: bool, +) -> None: + """落 shadow_gate 一行：mixed 影子 best 候选的 hard/soft/mixed 及是否 argmax 选中。 + + 参数: + db_path: SQLite 路径。 + run_id: 训练 run ID。 + epoch: 轮次（1-based）。 + candidate_version: 候选版本标识（如 skills/vX+prompts/vY）。 + hard_acc: hard 准确率。 + soft_score: soft 连续分；invalid 传 None -> 存 NULL（该版本不进 argmax）。 + mixed_score: 0.5*hard+0.5*soft；soft 缺失传 None -> 存 NULL。 + is_mixed_best: 是否本 epoch mixed argmax 选中（存 1/0）。 + """ + from app.harness.log import HarnessLog + + with HarnessLog(db_path, run_id) as log: + log.create_table("shadow_gate", _SHADOW_COLS) + log.insert( + "shadow_gate", + { + "epoch": epoch, + "candidate_version": candidate_version, + "hard_acc": hard_acc, + "soft_score": soft_score, + "mixed_score": mixed_score, + "is_mixed_best": int(is_mixed_best), + }, + ) + + +def read_shadow_gate(db_path: str, *, run_id: str) -> list[dict[str, Any]]: + """回读指定 run 的 shadow_gate 全部行（纯读，不污染运行状态）。""" + return _read_table(db_path, "shadow_gate", run_id) + + +# --------------------------------------------------------------------------- +# holdout_eval +# --------------------------------------------------------------------------- + + +def write_holdout_eval( + db_path: str, + *, + run_id: str, + epoch: int, + version_kind: str, + hard_acc: float, + soft_score: float | None, + mixed_score: float | None, + per_task_type_json: str, +) -> None: + """落 holdout_eval 一行：四向 held-out 在 test 池的 hard+soft+mixed 及按题型细分。 + + 参数: + db_path: SQLite 路径。 + run_id: 训练 run ID。 + epoch: 轮次（1-based）。 + version_kind: baseline / best_hard / best_mixed / final。 + hard_acc: hard 准确率。 + soft_score: soft 连续分；invalid 传 None -> 存 NULL。 + mixed_score: 0.5*hard+0.5*soft；soft 缺失传 None -> 存 NULL。 + per_task_type_json: 按 task_type 的 {accuracy,total,correct} JSON 串。 + """ + from app.harness.log import HarnessLog + + with HarnessLog(db_path, run_id) as log: + log.create_table("holdout_eval", _HOLDOUT_COLS) + log.insert( + "holdout_eval", + { + "epoch": epoch, + "version_kind": version_kind, + "hard_acc": hard_acc, + "soft_score": soft_score, + "mixed_score": mixed_score, + "per_task_type_json": per_task_type_json, + }, + ) + + +def read_holdout_eval(db_path: str, *, run_id: str) -> list[dict[str, Any]]: + """回读指定 run 的 holdout_eval 全部行（纯读，不污染运行状态）。""" + return _read_table(db_path, "holdout_eval", run_id) + + +# --------------------------------------------------------------------------- +# quadrant_pair +# --------------------------------------------------------------------------- + + +def write_quadrant_pairs( + db_path: str, + *, + run_id: str, + epoch: int, + step: int, + pairs: list[dict[str, Any]], +) -> None: + """落 quadrant_pair 多行：fast gate 后逐题四象限（prev/curr 翻转 + category）落库。 + + 参数: + db_path: SQLite 路径。 + run_id: 训练 run ID。 + epoch: 轮次（1-based）。 + step: epoch 内 step 序号（0-based）。 + pairs: 每条含 question_id/task_type/prev_correct/curr_correct/category； + prev_correct/curr_correct 为 bool，写库前转 0/1。 + + 关键实现: + 用 insert_many 批量落库；pairs 为空时只建表不插入（fast gate 无翻转的极端情况）。 + """ + records = [ + { + "epoch": epoch, + "step": step, + "question_id": pair["question_id"], + "task_type": pair["task_type"], + "prev_correct": int(pair["prev_correct"]), + "curr_correct": int(pair["curr_correct"]), + "category": pair["category"], + } + for pair in pairs + ] + from app.harness.log import HarnessLog + + with HarnessLog(db_path, run_id) as log: + log.create_table("quadrant_pair", _QUADRANT_COLS) + if records: + log.insert_many("quadrant_pair", records) + + +def read_quadrant_pairs(db_path: str, *, run_id: str) -> list[dict[str, Any]]: + """回读指定 run 的 quadrant_pair 全部行（纯读，不污染运行状态）。""" + return _read_table(db_path, "quadrant_pair", run_id) + + +# --------------------------------------------------------------------------- +# gate_evidence +# --------------------------------------------------------------------------- + + +def write_gate_evidence( + db_path: str, + *, + run_id: str, + epoch: int, + step: int, + rows: list[dict[str, Any]], +) -> None: + """落 gate_evidence 逐题行：CE-Gate 每次决策的可回放审计记录。 + + 参数: + db_path: SQLite 路径。 + run_id: 训练 run ID。 + epoch: 该 gate 所属的轮次（1-based）。 + step: epoch 内 step 序号（0-based）。 + rows: 每题一行，含 question_id/task_type/block_idx/baseline_correct/ + candidate_correct/e_value（该题所在块判定后的累计 e 值）/ + stop_reason（仅最后一题携带最终 stop_reason，其余空串）。 + + 关键实现: + 逐行 insert（非 insert_many），保证每行独立事务。 + """ + from app.harness.log import HarnessLog + + with HarnessLog(db_path, run_id) as log: + log.create_table("gate_evidence", _GATE_EVIDENCE_COLS) + for row in rows: + log.insert("gate_evidence", {"epoch": epoch, "step": step, **row}) + + +def read_gate_evidence(db_path: str, *, run_id: str) -> list[dict[str, Any]]: + """回读指定 run 的 gate_evidence 全部行（纯读，不污染运行状态）。""" + return _read_table(db_path, "gate_evidence", run_id) + + +# --------------------------------------------------------------------------- +# 报告函数（从 TRM4 loop_report.py 迁移） +# --------------------------------------------------------------------------- + + +def write_step_report( + workspace_dir: Path, + epoch: int, + step: int, + global_step: int, + task_type: str, + gate_action: str, + candidate_acc: float, + class_baseline_acc: float, + edit_budget: int, + rank_clip_triggered: bool, + gate_w: int | None, + gate_l: int | None, + gate_e_value: float | None, + gate_n_used: int | None, + gate_stop_reason: str | None, +) -> Path: + """写单个 (step, task_type) 快路径 gate 的最小观测记录 JSON。 + + 文件名按 (epoch, step, task_type) 命名，slug 由 task_type 规范化（小写、空格转 '-'）得到。 + + 参数: + workspace_dir: 实验工作区目录。 + epoch: 当前轮次（1-based）。 + step: epoch 内 step 序号（0-based）。 + global_step: 全局步计数（驱动 edit_budget 退火）。 + task_type: 本条 gate 的任务类型。 + gate_action: 闸门动作（accept_confirmed / accept_provisional / reject / + skipped / cooldown）。 + candidate_acc: 候选在 gate 已观测题上的准确率（观测口径）。 + class_baseline_acc: 基线在 gate 已观测题上的准确率（观测口径）。 + edit_budget: 该 step 按 global_step 退火得到的 per-target 编辑预算上限。 + rank_clip_triggered: 该 skill 进化是否触发了 rank 裁剪。 + gate_w: e-process 累计 W（基线错->候选对翻转数）；skipped/cooldown 路径传 None。 + gate_l: e-process 累计 L（基线对->候选错翻转数）；skipped/cooldown 路径传 None。 + gate_e_value: 停时的 e 值；skipped/cooldown 路径传 None。 + gate_n_used: gate 实际消费的阶梯题数；skipped/cooldown 路径传 None。 + gate_stop_reason: e-process 停止原因；skipped/cooldown 路径传 None。 + + 返回: + 写入的 step_report 文件路径。 + """ + report = { + "epoch": epoch, + "step": step, + "global_step": global_step, + "task_type": task_type, + "gate_action": gate_action, + "candidate_acc": candidate_acc, + "class_baseline_acc": class_baseline_acc, + "edit_budget": edit_budget, + "rank_clip_triggered": rank_clip_triggered, + "gate_w": gate_w, + "gate_l": gate_l, + "gate_e_value": gate_e_value, + "gate_n_used": gate_n_used, + "gate_stop_reason": gate_stop_reason, + } + slug = task_type.lower().replace(" ", "-") + out_dir = workspace_dir / "analyses" + out_dir.mkdir(parents=True, exist_ok=True) + path = out_dir / f"step_report_e{epoch}_s{step}_{slug}.json" + path.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8") + return path + + +def write_epoch_report( + workspace_dir: Path, + epoch: int, + system_tool_action: str, + momentum_updated_task_types: list[str], + best_val_acc: float, +) -> Path: + """写 epoch 末慢更新汇总 JSON。 + + 慢更新无单一 ValidationOutcome，故本函数只落慢更新可观测的最小集： + system/tool gate 动作、本 epoch 写过 momentum 的题型、慢更新后的全局 best。 + + 参数: + workspace_dir: 实验工作区目录。 + epoch: 当前轮次（1-based）。 + system_tool_action: 慢更新 system/tool 动作（updated / reverted / none）。 + momentum_updated_task_types: 本 epoch 写过 momentum 的题型列表。 + best_val_acc: 慢更新后（含 best argmax）的全局 best 验证准确率。 + + 返回: + 写入的 epoch_report 文件路径。 + """ + report = { + "epoch": epoch, + "system_tool_action": system_tool_action, + "momentum_updated_task_types": momentum_updated_task_types, + "best_val_acc": best_val_acc, + } + out_dir = workspace_dir / "analyses" + out_dir.mkdir(parents=True, exist_ok=True) + path = out_dir / f"epoch_report_{epoch}.json" + path.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8") + return path diff --git a/tests/unit/test_harness_observation.py b/tests/unit/test_harness_observation.py new file mode 100644 index 0000000..d37cc40 --- /dev/null +++ b/tests/unit/test_harness_observation.py @@ -0,0 +1,354 @@ +"""五张观测表 + step/epoch 报告的单元测试。""" + +from __future__ import annotations + +import json +import sqlite3 +from typing import TYPE_CHECKING + +import pytest + +if TYPE_CHECKING: + from pathlib import Path + +from app.harness.observation import ( + read_dual_metric, + read_gate_evidence, + read_holdout_eval, + read_quadrant_pairs, + read_shadow_gate, + write_dual_metric, + write_epoch_report, + write_gate_evidence, + write_holdout_eval, + write_quadrant_pairs, + write_shadow_gate, + write_step_report, +) + + +@pytest.fixture() +def db_path(tmp_path: Path) -> str: + """返回临时 SQLite 路径。""" + return str(tmp_path / "test_obs.db") + + +@pytest.fixture() +def run_id() -> str: + return "run-obs-001" + + +# --------------------------------------------------------------------------- +# dual_metric_eval +# --------------------------------------------------------------------------- + + +def test_write_read_dual_metric(db_path: str, run_id: str) -> None: + """写入 dual_metric_eval 后回读应与输入一致。""" + write_dual_metric( + db_path, + run_id=run_id, + epoch=1, + version_kind="baseline", + skills_version="v0", + prompts_version="v0", + pool="val", + hard_acc=0.75, + soft_score=0.80, + mixed_score=0.775, + ) + rows = read_dual_metric(db_path, run_id=run_id) + assert len(rows) == 1 + row = rows[0] + assert row["epoch"] == 1 + assert row["version_kind"] == "baseline" + assert row["skills_version"] == "v0" + assert row["prompts_version"] == "v0" + assert row["pool"] == "val" + assert row["hard_acc"] == pytest.approx(0.75) + assert row["soft_score"] == pytest.approx(0.80) + assert row["mixed_score"] == pytest.approx(0.775) + assert row["run_id"] == run_id + + +# --------------------------------------------------------------------------- +# shadow_gate +# --------------------------------------------------------------------------- + + +def test_write_read_shadow_gate(db_path: str, run_id: str) -> None: + """写入 shadow_gate 后回读应与输入一致，is_mixed_best 布尔转 int。""" + write_shadow_gate( + db_path, + run_id=run_id, + epoch=2, + candidate_version="skills/v1+prompts/v1", + hard_acc=0.82, + soft_score=0.78, + mixed_score=0.80, + is_mixed_best=True, + ) + rows = read_shadow_gate(db_path, run_id=run_id) + assert len(rows) == 1 + row = rows[0] + assert row["epoch"] == 2 + assert row["candidate_version"] == "skills/v1+prompts/v1" + assert row["hard_acc"] == pytest.approx(0.82) + assert row["is_mixed_best"] == 1 + + +# --------------------------------------------------------------------------- +# holdout_eval +# --------------------------------------------------------------------------- + + +def test_write_read_holdout_eval(db_path: str, run_id: str) -> None: + """写入 holdout_eval 后回读应与输入一致。""" + per_task = json.dumps({"temporal": {"accuracy": 0.9, "total": 10, "correct": 9}}) + write_holdout_eval( + db_path, + run_id=run_id, + epoch=1, + version_kind="best_hard", + hard_acc=0.85, + soft_score=0.70, + mixed_score=0.775, + per_task_type_json=per_task, + ) + rows = read_holdout_eval(db_path, run_id=run_id) + assert len(rows) == 1 + row = rows[0] + assert row["version_kind"] == "best_hard" + assert row["hard_acc"] == pytest.approx(0.85) + parsed = json.loads(row["per_task_type_json"]) + assert parsed["temporal"]["correct"] == 9 + + +# --------------------------------------------------------------------------- +# gate_evidence +# --------------------------------------------------------------------------- + + +def test_write_read_gate_evidence(db_path: str, run_id: str) -> None: + """写入 gate_evidence 后回读应与输入一致，逐行插入。""" + evidence_rows = [ + { + "task_type": "temporal", + "question_id": "q1", + "block_idx": 0, + "baseline_correct": 1, + "candidate_correct": 1, + "e_value": 1.0, + "stop_reason": "", + }, + { + "task_type": "temporal", + "question_id": "q2", + "block_idx": 0, + "baseline_correct": 0, + "candidate_correct": 1, + "e_value": 2.0, + "stop_reason": "confirmed", + }, + ] + write_gate_evidence(db_path, run_id=run_id, epoch=1, step=0, rows=evidence_rows) + rows = read_gate_evidence(db_path, run_id=run_id) + assert len(rows) == 2 + assert rows[0]["question_id"] == "q1" + assert rows[1]["stop_reason"] == "confirmed" + assert rows[1]["e_value"] == pytest.approx(2.0) + + +# --------------------------------------------------------------------------- +# quadrant_pair +# --------------------------------------------------------------------------- + + +def test_write_read_quadrant_pairs(db_path: str, run_id: str) -> None: + """写入 quadrant_pair 后回读，bool -> int 转换正确。""" + pairs = [ + { + "question_id": "q1", + "task_type": "causal", + "prev_correct": True, + "curr_correct": False, + "category": "regression", + }, + { + "question_id": "q2", + "task_type": "causal", + "prev_correct": False, + "curr_correct": True, + "category": "improvement", + }, + ] + write_quadrant_pairs(db_path, run_id=run_id, epoch=1, step=0, pairs=pairs) + rows = read_quadrant_pairs(db_path, run_id=run_id) + assert len(rows) == 2 + # bool -> int 转换 + assert rows[0]["prev_correct"] == 1 + assert rows[0]["curr_correct"] == 0 + assert rows[1]["prev_correct"] == 0 + assert rows[1]["curr_correct"] == 1 + + +# --------------------------------------------------------------------------- +# NULL vs 0 语义 +# --------------------------------------------------------------------------- + + +def test_null_not_zero_for_soft(db_path: str, run_id: str) -> None: + """soft_score/mixed_score 为 None 时存 NULL（非 0），回读也是 None。""" + write_dual_metric( + db_path, + run_id=run_id, + epoch=1, + version_kind="baseline", + skills_version="v0", + prompts_version="v0", + pool="val", + hard_acc=0.75, + soft_score=None, + mixed_score=None, + ) + rows = read_dual_metric(db_path, run_id=run_id) + assert len(rows) == 1 + row = rows[0] + assert row["soft_score"] is None + assert row["mixed_score"] is None + + # 用原生 SQL 确认存的是 NULL 而非 0 + conn = sqlite3.connect(db_path) + cursor = conn.execute( + "SELECT soft_score, mixed_score FROM dual_metric_eval WHERE run_id=?", + (run_id,), + ) + raw = cursor.fetchone() + conn.close() + assert raw[0] is None + assert raw[1] is None + + +# --------------------------------------------------------------------------- +# 只读连接隔离 +# --------------------------------------------------------------------------- + + +def test_read_only_connection(db_path: str, run_id: str) -> None: + """read_* 使用独立只读连接，不向 _runs 表插入新行。""" + write_dual_metric( + db_path, + run_id=run_id, + epoch=1, + version_kind="baseline", + skills_version="v0", + prompts_version="v0", + pool="val", + hard_acc=0.5, + soft_score=None, + mixed_score=None, + ) + + # 用另一个 run_id 回读——不应在 _runs 表中创建新行 + other_run = "run-obs-ghost" + rows = read_dual_metric(db_path, run_id=other_run) + assert rows == [] + + conn = sqlite3.connect(db_path) + cursor = conn.execute("SELECT run_id FROM _runs") + run_ids = [r[0] for r in cursor.fetchall()] + conn.close() + assert other_run not in run_ids + + +# --------------------------------------------------------------------------- +# step_report +# --------------------------------------------------------------------------- + + +def test_write_step_report(tmp_path: Path) -> None: + """step_report 写入 JSON 文件，内容字段完整。""" + workspace = tmp_path / "ws" + workspace.mkdir() + path = write_step_report( + workspace_dir=workspace, + epoch=1, + step=2, + global_step=12, + task_type="Temporal Order", + gate_action="accept_confirmed", + candidate_acc=0.85, + class_baseline_acc=0.70, + edit_budget=5, + rank_clip_triggered=False, + gate_w=3, + gate_l=1, + gate_e_value=4.2, + gate_n_used=8, + gate_stop_reason="confirmed", + ) + assert path.exists() + assert path.name == "step_report_e1_s2_temporal-order.json" + data = json.loads(path.read_text(encoding="utf-8")) + assert data["epoch"] == 1 + assert data["step"] == 2 + assert data["global_step"] == 12 + assert data["task_type"] == "Temporal Order" + assert data["gate_action"] == "accept_confirmed" + assert data["gate_w"] == 3 + assert data["gate_stop_reason"] == "confirmed" + assert data["rank_clip_triggered"] is False + + +def test_write_step_report_skipped_null_fields(tmp_path: Path) -> None: + """skipped/cooldown 路径的 gate 字段应为 null。""" + workspace = tmp_path / "ws" + workspace.mkdir() + path = write_step_report( + workspace_dir=workspace, + epoch=1, + step=0, + global_step=0, + task_type="causal", + gate_action="skipped", + candidate_acc=0.0, + class_baseline_acc=0.0, + edit_budget=10, + rank_clip_triggered=False, + gate_w=None, + gate_l=None, + gate_e_value=None, + gate_n_used=None, + gate_stop_reason=None, + ) + data = json.loads(path.read_text(encoding="utf-8")) + assert data["gate_w"] is None + assert data["gate_l"] is None + assert data["gate_e_value"] is None + assert data["gate_n_used"] is None + assert data["gate_stop_reason"] is None + + +# --------------------------------------------------------------------------- +# epoch_report +# --------------------------------------------------------------------------- + + +def test_write_epoch_report(tmp_path: Path) -> None: + """epoch_report 写入 JSON 文件，内容字段完整。""" + workspace = tmp_path / "ws" + workspace.mkdir() + path = write_epoch_report( + workspace_dir=workspace, + epoch=3, + system_tool_action="updated", + momentum_updated_task_types=["temporal", "causal"], + best_val_acc=0.88, + ) + assert path.exists() + assert path.name == "epoch_report_3.json" + data = json.loads(path.read_text(encoding="utf-8")) + assert data["epoch"] == 3 + assert data["system_tool_action"] == "updated" + assert data["momentum_updated_task_types"] == ["temporal", "causal"] + assert data["best_val_acc"] == pytest.approx(0.88)