feat(harness): RunConfig frozen dataclass + 四层校验 + YAML/CLI/.env 三层加载

- RunConfig: 46 字段 frozen dataclass,从 TRM4 core/harness/config.py 迁移
- 四层校验链:_validate → _validate_edit_budget + _validate_minibatch + _validate_gate
- 新增 .env 覆盖层:工程配置(workspace_dir, store_dir)可通过 HARNESS_* 环境变量注入
- 合并优先级:CLI > .env > YAML(CLAUDE.md §4.5)
- load_config 支持嵌套 harness 段和扁平 YAML 两种格式
- run_id 改为默认空字符串(CLI-only 字段,YAML 不提供)
- resume/fresh 互斥校验不在 config 层(移至 runner.py)
- 70 个单元测试全部通过

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-07-07 11:57:35 -04:00
parent d84cd679b4
commit 09a385addc
2 changed files with 883 additions and 0 deletions
+343
View File
@@ -0,0 +1,343 @@
"""运行配置:RunConfig frozen dataclass 与 YAML + CLI + .env 三层加载。
三层合并优先级:CLI > .env > YAML(遵循 CLAUDE.md §4.5 配置管理规范)。
- YAML:科研实验配置(会在实验中反复扫动的参数),存放于 config/ 下。
- .env:工程配置(少变路径如 workspace_dir、store_dir),通过环境变量注入。
- CLI:单次临时覆盖。
"""
from __future__ import annotations
import dataclasses
import os
from dataclasses import dataclass
from pathlib import Path
import yaml
_VALID_MODES = {"infer", "train", "diagnose", "evolve", "eval", "promote"}
_VALID_SKILL_MODES = {"auto", "manual", "none"}
_VALID_SKILL_UPDATE_MODES = {"patch", "rewrite"}
_PATH_FIELDS = {"workspace_dir", "store_dir"}
# Video-MME 的任务类型数量:验证池每类至少保底 eval_min_per_class 题,共 11 类。
_VIDEO_MME_TASK_TYPE_COUNT = 11
# .env 工程配置字段映射(环境变量名 → RunConfig 字段名)。
# 仅路径类工程配置走 .env,科研实验参数走 YAML。
_ENV_FIELD_MAP: dict[str, str] = {
"HARNESS_WORKSPACE_DIR": "workspace_dir",
"HARNESS_STORE_DIR": "store_dir",
}
@dataclass(frozen=True)
class RunConfig:
"""实验运行配置,所有参数的唯一归口。
frozen=True 确保配置在创建后不可变,防止运行中被意外修改。
三层合并优先级:CLI > .env > YAML。
字段:
workspace_dir: Workspace 根目录。
store_dir: Store 根目录。
mode: 运行模式,"infer" / "train" / "diagnose" / "evolve" / "eval" / "promote"
concurrency: 并行 worker 数。
max_steps: AgentLoop 单题最大步数。
skill_mode: Skill 加载模式,"auto" / "manual" / "none"
n_samples: 题目截取数,0 表示全量。
questions: 题目在 questions/ 下的相对路径。
skills_version: Skills 版本号。
prompts_version: Prompts 版本号。
epochs: 训练轮数。
diag_size: 诊断池题目数。
diag_correct_ratio: 诊断池中正确题目占比。
val_size: 验证池题目数。
val_correct_ratio: 验证池中正确题目占比。
edit_budget_start: 编辑预算前期上限。
edit_budget_end: 编辑预算后期下限。
batch_size: mini-batch 单批题目数。
min_class_per_batch: 单批中每个任务类型至少保留的题目数(< batch_size)。
eval_min_per_class: 验证池中每个任务类型至少保底的题目数。
early_stop_patience: 全局 best 连续未提升的容忍轮数,达到即早停。
test_size: held-out 测试池题目数。
use_slow_momentum: 是否启用快慢双速进化中的慢速 momentum 更新。
gate_e_confirm: CE-Gate CONFIRMED 接受的 e 值门槛(1/alphaVille 界假阳率 alpha)。
gate_e_provisional: 题尽暂定接受门 + futility 提前止损的代数界。
gate_w_net_min: 题尽暂定接受要求的最小净胜数(win - loss)。
gate_delta_min: 最小点估计效应量下限(承接旧 margin 语义)。
gate_lambda_dir: Wald 方向拒绝的对数似然比阈值(必须为负)。
gate_e_rollback: 试用期对称回滚门(回滚 e 值门槛)。
gate_block: 块序贯验证的块大小(=推理并发度,块内跑满)。
gate_n_max: 单次 gate 消耗的题数上限。
gate_p_low: 信息量阶梯 p-hat 保留区间下界(剔除必错零信息题)。
gate_p_high: 信息量阶梯 p-hat 保留区间上界(剔除必对零信息题)。
gate_probe_quota: 冷启动探针集比例(全错题中插尾的比例)。
gate_gamma_decay: 逐题正确率估计 p-hat 的 EMA 衰减系数。
gate_cooldown_steps: 回滚后该题型跳过进化的冷却 step 数。
gate_guard_err: gate 内跨块累计 INFRA 错误率护栏。
skill_update_mode: skill 进化模式,"patch"(局部 edit/ "rewrite"(整篇重写)。
appendix_consolidate_threshold: appendix note 条数达此值触发 LLM consolidation。
run_id: diagnose/evolve 模式要分析的运行 ID,默认空字符串。
batch_correct_ratio: 单批中正确题目占比,范围 [0, 1)。
momentum_samples: 慢速 momentum 更新时从诊断池采样的题目数,必须 >= 1。
seed: fresh 训练的种子名(对应 seed.json),默认 "initial"
version: eval/promote 模式指定的 store 版本号(如 "v3")。
resume: train 模式是否从已有 checkpoint 续训。
fresh: train 模式是否从种子全新开始。
"""
# ── 必填字段(无默认值,来自 YAML 或 CLI) ──
workspace_dir: Path
store_dir: Path
mode: str
concurrency: int
max_steps: int
skill_mode: str
n_samples: int
questions: str
skills_version: str
prompts_version: str
epochs: int
diag_size: int
diag_correct_ratio: float
val_size: int
val_correct_ratio: float
edit_budget_start: int
edit_budget_end: int
batch_size: int
min_class_per_batch: int
eval_min_per_class: int
early_stop_patience: int
test_size: int
use_slow_momentum: bool
gate_e_confirm: float
gate_e_provisional: float
gate_w_net_min: int
gate_delta_min: float
gate_lambda_dir: float
gate_e_rollback: float
gate_block: int
gate_n_max: int
gate_p_low: float
gate_p_high: float
gate_probe_quota: float
gate_gamma_decay: float
gate_cooldown_steps: int
gate_guard_err: float
skill_update_mode: str
appendix_consolidate_threshold: int
# ── 有默认值的字段(通常由 CLI 传入或可选) ──
run_id: str = ""
batch_correct_ratio: float = 0.5
momentum_samples: int = 20
seed: str = "initial"
version: str = ""
resume: bool = False
fresh: bool = False
def _validate(config: RunConfig) -> None:
"""校验 RunConfig 全部字段约束。
四层校验链:基础字段 → 编辑预算 → mini-batch → CE-Gate。
参数:
config: 待校验的配置实例。
异常:
ValueError: 任一字段值不合法。
"""
# ── 基础字段校验 ──
if config.mode not in _VALID_MODES:
raise ValueError(f"mode 必须为 {_VALID_MODES} 之一,实际: {config.mode!r}")
if config.mode in ("diagnose", "evolve") and not config.run_id:
raise ValueError(f"mode 为 {config.mode!r} 时必须提供 run_id。")
if config.mode in ("eval", "promote") and not config.version:
raise ValueError(f"mode 为 {config.mode!r} 时必须提供 --version。")
if config.mode == "promote" and not config.run_id:
raise ValueError("promote 必须提供 --run-id(指定 canonical eval run)。")
if config.skill_mode not in _VALID_SKILL_MODES:
raise ValueError(
f"skill_mode 必须为 {_VALID_SKILL_MODES} 之一,实际: {config.skill_mode!r}"
)
if config.concurrency <= 0:
raise ValueError(f"concurrency 必须 > 0,实际: {config.concurrency}")
if config.max_steps <= 0:
raise ValueError(f"max_steps 必须 > 0,实际: {config.max_steps}")
if config.n_samples < 0:
raise ValueError(f"n_samples 必须 >= 0,实际: {config.n_samples}")
if config.epochs <= 0:
raise ValueError(f"epochs 必须 > 0,实际: {config.epochs}")
# ── 编辑预算校验 ──
_validate_edit_budget(config)
if config.skill_update_mode not in _VALID_SKILL_UPDATE_MODES:
raise ValueError(
f"skill_update_mode 必须为 {_VALID_SKILL_UPDATE_MODES} 之一,"
f"实际: {config.skill_update_mode!r}"
)
if config.appendix_consolidate_threshold < 1:
raise ValueError(
f"appendix_consolidate_threshold 必须 >= 1"
f"实际: {config.appendix_consolidate_threshold}"
)
# ── mini-batch 校验 ──
_validate_minibatch(config)
# ── CE-Gate 校验 ──
_validate_gate(config)
def _validate_edit_budget(config: RunConfig) -> None:
"""校验编辑预算退火的前期/后期上限约束。
参数:
config: 待校验的配置实例。
异常:
ValueError: edit_budget_start < edit_budget_end,或 end <= 0。
"""
if config.edit_budget_start < config.edit_budget_end:
raise ValueError(
f"edit_budget_start({config.edit_budget_start}) 必须 >= "
f"edit_budget_end({config.edit_budget_end})"
)
if config.edit_budget_end <= 0:
raise ValueError(f"edit_budget_end 必须 > 0,实际: {config.edit_budget_end}")
def _validate_minibatch(config: RunConfig) -> None:
"""校验 mini-batch 自进化闭环参数约束。
参数:
config: 待校验的 RunConfig 配置对象。
异常:
ValueError: 任一约束被违反。
关键实现细节:
val_size 必须 >= eval_min_per_class * _VIDEO_MME_TASK_TYPE_COUNT,保证验证池
能为 Video-MME 的全部 11 个任务类型各保底 eval_min_per_class 题。
"""
if config.batch_size <= 0:
raise ValueError(f"batch_size 必须 > 0,实际: {config.batch_size}")
if not (1 <= config.min_class_per_batch < config.batch_size):
raise ValueError(
f"min_class_per_batch 必须满足 1 <= 值 < batch_size"
f"({config.batch_size}),实际: {config.min_class_per_batch}"
)
if config.eval_min_per_class < 1:
raise ValueError(f"eval_min_per_class 必须 >= 1,实际: {config.eval_min_per_class}")
floor = config.eval_min_per_class * _VIDEO_MME_TASK_TYPE_COUNT
if config.val_size < floor:
raise ValueError(
f"val_size 必须 >= eval_min_per_class * {_VIDEO_MME_TASK_TYPE_COUNT}"
f"(={floor})Video-MME 共 {_VIDEO_MME_TASK_TYPE_COUNT} 个任务类型,"
f"每类需 eval_min_per_class 题保底,故验证池下限为 {floor}"
f"实际: {config.val_size}"
)
if config.early_stop_patience <= 0:
raise ValueError(f"early_stop_patience 必须 > 0,实际: {config.early_stop_patience}")
if config.test_size <= 0:
raise ValueError(f"test_size 必须 > 0,实际: {config.test_size}")
if not (0 <= config.batch_correct_ratio < 1):
raise ValueError(
f"batch_correct_ratio 必须满足 0 <= 值 < 1,实际: {config.batch_correct_ratio}"
)
if config.momentum_samples < 1:
raise ValueError(f"momentum_samples 必须 >= 1,实际: {config.momentum_samples}")
def _validate_gate(config: RunConfig) -> None:
"""校验 CE-Gate 判据与阶梯参数约束。
参数:
config: 待校验的配置实例。
异常:
ValueError: 任一 gate 参数不合法。
"""
if config.gate_e_confirm <= 1:
raise ValueError(f"gate_e_confirm 必须 > 1,实际: {config.gate_e_confirm}")
if not (1 < config.gate_e_provisional <= config.gate_e_confirm):
raise ValueError(
f"gate_e_provisional 必须在 (1, gate_e_confirm] 内,实际: {config.gate_e_provisional}"
)
if config.gate_e_rollback <= 1:
raise ValueError(f"gate_e_rollback 必须 > 1,实际: {config.gate_e_rollback}")
if config.gate_w_net_min < 1:
raise ValueError(f"gate_w_net_min 必须 >= 1,实际: {config.gate_w_net_min}")
if config.gate_lambda_dir >= 0:
raise ValueError(f"gate_lambda_dir 必须 < 0,实际: {config.gate_lambda_dir}")
if config.gate_block <= 0 or config.gate_n_max < config.gate_block:
raise ValueError(
f"需 0 < gate_block <= gate_n_max"
f"实际: block={config.gate_block}, n_max={config.gate_n_max}"
)
if not (0 <= config.gate_p_low < config.gate_p_high <= 1):
raise ValueError(
f"需 0 <= gate_p_low < gate_p_high <= 1"
f"实际: [{config.gate_p_low}, {config.gate_p_high}]"
)
if not (0 <= config.gate_probe_quota <= 1):
raise ValueError(f"gate_probe_quota 须在 [0,1],实际: {config.gate_probe_quota}")
if not (0 < config.gate_gamma_decay < 1):
raise ValueError(f"gate_gamma_decay 须在 (0,1),实际: {config.gate_gamma_decay}")
if config.gate_cooldown_steps < 1:
raise ValueError(f"gate_cooldown_steps 必须 >= 1,实际: {config.gate_cooldown_steps}")
if not (0 < config.gate_guard_err < 1):
raise ValueError(f"gate_guard_err 须在 (0,1),实际: {config.gate_guard_err}")
def load_config(
yaml_path: Path,
cli_overrides: dict[str, object] | None = None,
) -> RunConfig:
"""从 YAML 加载配置,叠加 .env 和 CLI 覆盖层后构造 RunConfig。
三层合并优先级:CLI > .env > YAML。
参数:
yaml_path: YAML 配置文件路径,需包含 ``harness`` 段。
cli_overrides: CLI 参数字典,值为 None 表示未传入(不覆盖)。
返回:
构造并校验后的 RunConfig 实例。
异常:
FileNotFoundError: YAML 文件不存在。
ValueError: 校验失败。
"""
# Phase 1: 加载 YAML 基础层
with open(yaml_path, encoding="utf-8") as f:
raw: dict = yaml.safe_load(f)
# 支持嵌套 harness 段和扁平 YAML 两种格式
yaml_data: dict = raw.get("harness", raw)
# Phase 2: .env 覆盖层(仅工程配置字段)
for env_key, field_name in _ENV_FIELD_MAP.items():
env_val = os.environ.get(env_key)
if env_val is not None:
yaml_data[field_name] = env_val
# Phase 3: CLI 覆盖层(最高优先级)
valid_fields = {f.name for f in dataclasses.fields(RunConfig)}
if cli_overrides:
for key, value in cli_overrides.items():
if value is not None and key in valid_fields:
yaml_data[key] = value
# Phase 4: 类型转换 — 路径字段转 Path
for field_name in _PATH_FIELDS:
if field_name in yaml_data:
yaml_data[field_name] = Path(yaml_data[field_name])
# Phase 5: 构造并校验
config = RunConfig(**{k: v for k, v in yaml_data.items() if k in valid_fields})
_validate(config)
return config
+540
View File
@@ -0,0 +1,540 @@
"""app/harness/config.py 单元测试。
覆盖 RunConfig 构造、四层校验(_validate → 三个子校验)、
YAML + CLI + .env 三层加载优先级。
"""
from __future__ import annotations
from pathlib import Path
import pytest
import yaml
from app.harness.config import RunConfig, _validate, load_config
# ────────────────────────────── 测试数据工厂 ──────────────────────────────
def _valid_kwargs() -> dict:
"""构造一组完整合法的 RunConfig 字段值(使用真实 default.yaml 数据)。"""
return {
"workspace_dir": Path("workspaces/default"),
"store_dir": Path("store"),
"mode": "infer",
"run_id": "",
"concurrency": 12,
"max_steps": 15,
"skill_mode": "auto",
"n_samples": 0,
"questions": "benchmarks/Video-MME",
"skills_version": "v1",
"prompts_version": "v1",
"epochs": 1,
"diag_size": 200,
"diag_correct_ratio": 0.5,
"val_size": 30,
"val_correct_ratio": 0.5,
"edit_budget_start": 5,
"edit_budget_end": 2,
"batch_size": 15,
"min_class_per_batch": 2,
"eval_min_per_class": 2,
"early_stop_patience": 8,
"test_size": 60,
"use_slow_momentum": True,
"gate_e_confirm": 20.0,
"gate_e_provisional": 3.0,
"gate_w_net_min": 2,
"gate_delta_min": 0.02,
"gate_lambda_dir": -0.642,
"gate_e_rollback": 10.0,
"gate_block": 8,
"gate_n_max": 40,
"gate_p_low": 0.05,
"gate_p_high": 0.95,
"gate_probe_quota": 0.2,
"gate_gamma_decay": 0.9,
"gate_cooldown_steps": 2,
"gate_guard_err": 0.10,
"skill_update_mode": "patch",
"appendix_consolidate_threshold": 6,
"batch_correct_ratio": 0.5,
"momentum_samples": 20,
}
def _make_config(**overrides: object) -> RunConfig:
"""用 _valid_kwargs 构造 RunConfig,支持字段覆盖。"""
kwargs = _valid_kwargs()
kwargs.update(overrides)
return RunConfig(**kwargs)
def _yaml_harness_dict() -> dict:
"""构造可序列化为 YAML 的合法 harness 配置字典(路径用字符串)。"""
d = _valid_kwargs()
d["workspace_dir"] = str(d["workspace_dir"])
d["store_dir"] = str(d["store_dir"])
return d
def _write_yaml(tmp_path: Path, harness_data: dict) -> Path:
"""将 harness 配置写入临时 YAML 文件,返回文件路径。"""
yaml_path = tmp_path / "experiment.yaml"
with open(yaml_path, "w", encoding="utf-8") as f:
yaml.dump({"harness": harness_data}, f)
return yaml_path
# ──────────────────────────── test_valid_config ────────────────────────────
class TestValidConfig:
"""合法参数应能正常构造并通过校验。"""
def test_default_yaml_values_pass_validation(self) -> None:
"""使用 default.yaml 真实默认值构造的 RunConfig 应通过全部校验。"""
cfg = _make_config()
_validate(cfg)
assert cfg.mode == "infer"
assert cfg.workspace_dir == Path("workspaces/default")
assert cfg.store_dir == Path("store")
def test_frozen_immutability(self) -> None:
"""frozen=True 应禁止字段赋值。"""
cfg = _make_config()
with pytest.raises(AttributeError):
cfg.mode = "train" # type: ignore[misc]
def test_default_field_values(self) -> None:
"""有默认值的字段不传时应使用默认值。"""
kwargs = _valid_kwargs()
# 不传 run_id / seed / version / resume / fresh,使用默认值
kwargs.pop("run_id", None)
cfg = RunConfig(**kwargs)
assert cfg.run_id == ""
assert cfg.seed == "initial"
assert cfg.version == ""
assert cfg.resume is False
assert cfg.fresh is False
# ──────────────────────────── test_mode_validation ────────────────────────
class TestModeValidation:
"""mode 字段校验。"""
@pytest.mark.parametrize("bad_mode", ["unknown", "test", "", "INFER", "Train"])
def test_invalid_mode_rejected(self, bad_mode: str) -> None:
"""非法 mode 应抛出 ValueError。"""
cfg = _make_config(mode=bad_mode)
with pytest.raises(ValueError, match="mode"):
_validate(cfg)
@pytest.mark.parametrize(
"valid_mode", ["infer", "train", "diagnose", "evolve", "eval", "promote"]
)
def test_all_valid_modes_accepted(self, valid_mode: str) -> None:
"""全部合法 mode 应通过校验(diagnose/evolve 需 run_id)。"""
overrides: dict = {"mode": valid_mode}
if valid_mode in ("diagnose", "evolve"):
overrides["run_id"] = "run-001"
if valid_mode in ("eval", "promote"):
overrides["version"] = "v1"
if valid_mode == "promote":
overrides["run_id"] = "run-001"
cfg = _make_config(**overrides)
_validate(cfg)
def test_diagnose_requires_run_id(self) -> None:
"""diagnose 模式缺少 run_id 应报错。"""
cfg = _make_config(mode="diagnose", run_id="")
with pytest.raises(ValueError, match="run_id"):
_validate(cfg)
def test_evolve_requires_run_id(self) -> None:
"""evolve 模式缺少 run_id 应报错。"""
cfg = _make_config(mode="evolve", run_id="")
with pytest.raises(ValueError, match="run_id"):
_validate(cfg)
def test_eval_requires_version(self) -> None:
"""eval 模式缺少 version 应报错。"""
cfg = _make_config(mode="eval", version="")
with pytest.raises(ValueError, match="version"):
_validate(cfg)
def test_promote_requires_run_id_and_version(self) -> None:
"""promote 模式需同时提供 run_id 和 version。"""
cfg = _make_config(mode="promote", run_id="", version="v1")
with pytest.raises(ValueError, match="run.id"):
_validate(cfg)
def test_concurrency_positive(self) -> None:
"""concurrency <= 0 应报错。"""
cfg = _make_config(concurrency=0)
with pytest.raises(ValueError, match="concurrency"):
_validate(cfg)
def test_max_steps_positive(self) -> None:
"""max_steps <= 0 应报错。"""
cfg = _make_config(max_steps=-1)
with pytest.raises(ValueError, match="max_steps"):
_validate(cfg)
def test_n_samples_non_negative(self) -> None:
"""n_samples < 0 应报错。"""
cfg = _make_config(n_samples=-1)
with pytest.raises(ValueError, match="n_samples"):
_validate(cfg)
def test_epochs_positive(self) -> None:
"""epochs <= 0 应报错。"""
cfg = _make_config(epochs=0)
with pytest.raises(ValueError, match="epochs"):
_validate(cfg)
@pytest.mark.parametrize("bad_skill_mode", ["Auto", "disabled", ""])
def test_invalid_skill_mode(self, bad_skill_mode: str) -> None:
"""非法 skill_mode 应报错。"""
cfg = _make_config(skill_mode=bad_skill_mode)
with pytest.raises(ValueError, match="skill_mode"):
_validate(cfg)
@pytest.mark.parametrize("bad_update_mode", ["append", "delete", ""])
def test_invalid_skill_update_mode(self, bad_update_mode: str) -> None:
"""非法 skill_update_mode 应报错。"""
cfg = _make_config(skill_update_mode=bad_update_mode)
with pytest.raises(ValueError, match="skill_update_mode"):
_validate(cfg)
def test_appendix_consolidate_threshold_positive(self) -> None:
"""appendix_consolidate_threshold < 1 应报错。"""
cfg = _make_config(appendix_consolidate_threshold=0)
with pytest.raises(ValueError, match="appendix_consolidate_threshold"):
_validate(cfg)
# ────────────────────── test_edit_budget_validation ───────────────────────
class TestEditBudgetValidation:
"""编辑预算退火校验(_validate_edit_budget)。"""
def test_start_less_than_end_rejected(self) -> None:
"""edit_budget_start < edit_budget_end 应抛出 ValueError。"""
cfg = _make_config(edit_budget_start=1, edit_budget_end=5)
with pytest.raises(ValueError, match="edit_budget_start"):
_validate(cfg)
def test_end_zero_rejected(self) -> None:
"""edit_budget_end <= 0 应抛出 ValueError。"""
cfg = _make_config(edit_budget_start=1, edit_budget_end=0)
with pytest.raises(ValueError, match="edit_budget_end"):
_validate(cfg)
def test_equal_values_accepted(self) -> None:
"""edit_budget_start == edit_budget_end 应通过。"""
cfg = _make_config(edit_budget_start=3, edit_budget_end=3)
_validate(cfg)
# ─────────────────────── test_minibatch_validation ───────────────────────
class TestMinibatchValidation:
"""mini-batch 自进化闭环参数校验(_validate_minibatch)。"""
def test_batch_size_zero_rejected(self) -> None:
"""batch_size <= 0 应抛出 ValueError。"""
cfg = _make_config(batch_size=0)
with pytest.raises(ValueError, match="batch_size"):
_validate(cfg)
def test_min_class_per_batch_equals_batch_size_rejected(self) -> None:
"""min_class_per_batch >= batch_size 应抛出 ValueError。"""
cfg = _make_config(batch_size=5, min_class_per_batch=5)
with pytest.raises(ValueError, match="min_class_per_batch"):
_validate(cfg)
def test_min_class_per_batch_zero_rejected(self) -> None:
"""min_class_per_batch < 1 应抛出 ValueError。"""
cfg = _make_config(min_class_per_batch=0)
with pytest.raises(ValueError, match="min_class_per_batch"):
_validate(cfg)
def test_eval_min_per_class_zero_rejected(self) -> None:
"""eval_min_per_class < 1 应抛出 ValueError。"""
cfg = _make_config(eval_min_per_class=0)
with pytest.raises(ValueError, match="eval_min_per_class"):
_validate(cfg)
def test_early_stop_patience_zero_rejected(self) -> None:
"""early_stop_patience <= 0 应抛出 ValueError。"""
cfg = _make_config(early_stop_patience=0)
with pytest.raises(ValueError, match="early_stop_patience"):
_validate(cfg)
def test_test_size_zero_rejected(self) -> None:
"""test_size <= 0 应抛出 ValueError。"""
cfg = _make_config(test_size=0)
with pytest.raises(ValueError, match="test_size"):
_validate(cfg)
def test_batch_correct_ratio_one_rejected(self) -> None:
"""batch_correct_ratio >= 1 应抛出 ValueError。"""
cfg = _make_config(batch_correct_ratio=1.0)
with pytest.raises(ValueError, match="batch_correct_ratio"):
_validate(cfg)
def test_batch_correct_ratio_negative_rejected(self) -> None:
"""batch_correct_ratio < 0 应抛出 ValueError。"""
cfg = _make_config(batch_correct_ratio=-0.1)
with pytest.raises(ValueError, match="batch_correct_ratio"):
_validate(cfg)
def test_momentum_samples_zero_rejected(self) -> None:
"""momentum_samples < 1 应抛出 ValueError。"""
cfg = _make_config(momentum_samples=0)
with pytest.raises(ValueError, match="momentum_samples"):
_validate(cfg)
# ─────────────────────────── test_gate_validation ────────────────────────
class TestGateValidation:
"""CE-Gate 参数校验(_validate_gate)。"""
def test_e_confirm_at_one_rejected(self) -> None:
"""gate_e_confirm <= 1 应抛出 ValueError。"""
cfg = _make_config(gate_e_confirm=1.0, gate_e_provisional=1.0)
with pytest.raises(ValueError, match="gate_e_confirm"):
_validate(cfg)
def test_e_provisional_exceeds_confirm_rejected(self) -> None:
"""gate_e_provisional > gate_e_confirm 应抛出 ValueError。"""
cfg = _make_config(gate_e_confirm=10.0, gate_e_provisional=15.0)
with pytest.raises(ValueError, match="gate_e_provisional"):
_validate(cfg)
def test_e_provisional_at_one_rejected(self) -> None:
"""gate_e_provisional <= 1 应抛出 ValueError。"""
cfg = _make_config(gate_e_provisional=0.5)
with pytest.raises(ValueError, match="gate_e_provisional"):
_validate(cfg)
def test_e_rollback_at_one_rejected(self) -> None:
"""gate_e_rollback <= 1 应抛出 ValueError。"""
cfg = _make_config(gate_e_rollback=1.0)
with pytest.raises(ValueError, match="gate_e_rollback"):
_validate(cfg)
def test_w_net_min_zero_rejected(self) -> None:
"""gate_w_net_min < 1 应抛出 ValueError。"""
cfg = _make_config(gate_w_net_min=0)
with pytest.raises(ValueError, match="gate_w_net_min"):
_validate(cfg)
def test_lambda_dir_positive_rejected(self) -> None:
"""gate_lambda_dir >= 0 应抛出 ValueError。"""
cfg = _make_config(gate_lambda_dir=0.5)
with pytest.raises(ValueError, match="gate_lambda_dir"):
_validate(cfg)
def test_lambda_dir_zero_rejected(self) -> None:
"""gate_lambda_dir == 0 也应报错。"""
cfg = _make_config(gate_lambda_dir=0.0)
with pytest.raises(ValueError, match="gate_lambda_dir"):
_validate(cfg)
def test_block_exceeds_n_max_rejected(self) -> None:
"""gate_block > gate_n_max 应抛出 ValueError。"""
cfg = _make_config(gate_block=50, gate_n_max=40)
with pytest.raises(ValueError, match="gate_block"):
_validate(cfg)
def test_block_zero_rejected(self) -> None:
"""gate_block <= 0 应抛出 ValueError。"""
cfg = _make_config(gate_block=0)
with pytest.raises(ValueError, match="gate_block"):
_validate(cfg)
def test_p_low_exceeds_p_high_rejected(self) -> None:
"""gate_p_low >= gate_p_high 应抛出 ValueError。"""
cfg = _make_config(gate_p_low=0.9, gate_p_high=0.1)
with pytest.raises(ValueError, match="gate_p_low"):
_validate(cfg)
def test_probe_quota_negative_rejected(self) -> None:
"""gate_probe_quota < 0 应抛出 ValueError。"""
cfg = _make_config(gate_probe_quota=-0.1)
with pytest.raises(ValueError, match="gate_probe_quota"):
_validate(cfg)
def test_gamma_decay_zero_rejected(self) -> None:
"""gate_gamma_decay <= 0 应抛出 ValueError。"""
cfg = _make_config(gate_gamma_decay=0.0)
with pytest.raises(ValueError, match="gate_gamma_decay"):
_validate(cfg)
def test_gamma_decay_one_rejected(self) -> None:
"""gate_gamma_decay >= 1 应抛出 ValueError。"""
cfg = _make_config(gate_gamma_decay=1.0)
with pytest.raises(ValueError, match="gate_gamma_decay"):
_validate(cfg)
def test_cooldown_steps_zero_rejected(self) -> None:
"""gate_cooldown_steps < 1 应抛出 ValueError。"""
cfg = _make_config(gate_cooldown_steps=0)
with pytest.raises(ValueError, match="gate_cooldown_steps"):
_validate(cfg)
def test_guard_err_zero_rejected(self) -> None:
"""gate_guard_err <= 0 应抛出 ValueError。"""
cfg = _make_config(gate_guard_err=0.0)
with pytest.raises(ValueError, match="gate_guard_err"):
_validate(cfg)
def test_guard_err_one_rejected(self) -> None:
"""gate_guard_err >= 1 应抛出 ValueError。"""
cfg = _make_config(gate_guard_err=1.0)
with pytest.raises(ValueError, match="gate_guard_err"):
_validate(cfg)
# ─────────────────────── test_load_config_cli_overrides ──────────────────
class TestLoadConfigCliOverrides:
"""load_config 的 CLI 覆盖层测试。"""
def test_cli_overrides_yaml_values(self, tmp_path: Path) -> None:
"""CLI 参数应覆盖 YAML 中的同名字段。"""
harness_data = _yaml_harness_dict()
yaml_path = _write_yaml(tmp_path, harness_data)
cfg = load_config(yaml_path, cli_overrides={"concurrency": 4, "max_steps": 30})
assert cfg.concurrency == 4
assert cfg.max_steps == 30
def test_cli_none_values_ignored(self, tmp_path: Path) -> None:
"""CLI 中值为 None 的字段不应覆盖 YAML。"""
harness_data = _yaml_harness_dict()
yaml_path = _write_yaml(tmp_path, harness_data)
cfg = load_config(yaml_path, cli_overrides={"concurrency": None, "max_steps": 20})
assert cfg.concurrency == 12 # YAML 默认值
assert cfg.max_steps == 20
def test_cli_run_id_override(self, tmp_path: Path) -> None:
"""CLI 可通过 run_id 覆盖默认空字符串。"""
harness_data = _yaml_harness_dict()
yaml_path = _write_yaml(tmp_path, harness_data)
cfg = load_config(
yaml_path,
cli_overrides={"mode": "diagnose", "run_id": "run-abc-123"},
)
assert cfg.run_id == "run-abc-123"
assert cfg.mode == "diagnose"
def test_path_fields_converted_to_path(self, tmp_path: Path) -> None:
"""workspace_dir 和 store_dir 应被转换为 Path 对象。"""
harness_data = _yaml_harness_dict()
yaml_path = _write_yaml(tmp_path, harness_data)
cfg = load_config(yaml_path)
assert isinstance(cfg.workspace_dir, Path)
assert isinstance(cfg.store_dir, Path)
def test_unknown_cli_keys_ignored(self, tmp_path: Path) -> None:
"""YAML 和 RunConfig 中不存在的 CLI key 应被忽略。"""
harness_data = _yaml_harness_dict()
yaml_path = _write_yaml(tmp_path, harness_data)
cfg = load_config(yaml_path, cli_overrides={"nonexistent_field": 42})
assert cfg.concurrency == 12 # 正常字段不受影响
def test_validation_runs_after_loading(self, tmp_path: Path) -> None:
"""load_config 加载后应运行校验,非法值应报错。"""
harness_data = _yaml_harness_dict()
harness_data["mode"] = "invalid_mode"
yaml_path = _write_yaml(tmp_path, harness_data)
with pytest.raises(ValueError, match="mode"):
load_config(yaml_path)
# ──────────────────── test_load_config_env_overrides ─────────────────────
class TestLoadConfigEnvOverrides:
"""load_config 的 .env 环境变量覆盖层测试。"""
def test_env_overrides_yaml_workspace_dir(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""环境变量 HARNESS_WORKSPACE_DIR 应覆盖 YAML 中的 workspace_dir。"""
harness_data = _yaml_harness_dict()
yaml_path = _write_yaml(tmp_path, harness_data)
monkeypatch.setenv("HARNESS_WORKSPACE_DIR", "/custom/workspace")
cfg = load_config(yaml_path)
assert cfg.workspace_dir == Path("/custom/workspace")
def test_env_overrides_yaml_store_dir(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""环境变量 HARNESS_STORE_DIR 应覆盖 YAML 中的 store_dir。"""
harness_data = _yaml_harness_dict()
yaml_path = _write_yaml(tmp_path, harness_data)
monkeypatch.setenv("HARNESS_STORE_DIR", "/custom/store")
cfg = load_config(yaml_path)
assert cfg.store_dir == Path("/custom/store")
def test_cli_overrides_env(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""CLI 优先级高于 .envCLI > .env > YAML。"""
harness_data = _yaml_harness_dict()
yaml_path = _write_yaml(tmp_path, harness_data)
monkeypatch.setenv("HARNESS_WORKSPACE_DIR", "/env/workspace")
cfg = load_config(yaml_path, cli_overrides={"workspace_dir": "/cli/workspace"})
assert cfg.workspace_dir == Path("/cli/workspace")
# ─────────────────────────── test_val_size_floor ─────────────────────────
class TestValSizeFloor:
"""val_size >= eval_min_per_class * 11 的下限校验。"""
def test_val_size_below_floor_rejected(self) -> None:
"""val_size < eval_min_per_class * 11 应抛出 ValueError。
eval_min_per_class=3 → 下限 = 3 * 11 = 33val_size=30 不足。
"""
cfg = _make_config(eval_min_per_class=3, val_size=30)
with pytest.raises(ValueError, match="val_size"):
_validate(cfg)
def test_val_size_at_floor_accepted(self) -> None:
"""val_size == eval_min_per_class * 11 应通过。"""
cfg = _make_config(eval_min_per_class=3, val_size=33)
_validate(cfg)
def test_val_size_above_floor_accepted(self) -> None:
"""val_size > eval_min_per_class * 11 应通过。"""
cfg = _make_config(eval_min_per_class=2, val_size=100)
_validate(cfg)
def test_default_yaml_values_satisfy_floor(self) -> None:
"""default.yaml 的默认值(val_size=30, eval_min_per_class=2)应满足下限。
下限 = 2 * 11 = 22val_size=30 >= 22,通过。
"""
cfg = _make_config()
_validate(cfg) # 不应抛出异常