iomgaa/Video-Tree-TRM5
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(harness): observation.py — 五张观测表 + step/epoch 报告

Browse Source
This commit is contained in:
iomgaa
2026-07-07 12:55:12 -04:00
parent bd4e438c6c
commit a550d39e1c
2 changed files with 813 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/harness/observation.py
+459
View File
@@ -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
tests/unit/test_harness_observation.py
+354
View File
@@ -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)