@classmethod
def from_yaml(cls, path: str | Path) -> "EvalConfig":
"""Load config from a YAML file."""
path = Path(path)
if not path.exists():
raise FileNotFoundError(f"Config not found: {path}")
with open(path) as f:
raw = yaml.safe_load(f) or {}
# Deprecation: top-level `skill:` is auto-normalized into
# execution.skill (below) but the canonical home is the execution
# block, symmetric with execution.prompt. Warn once per load; only
# for a non-empty value that isn't already mirrored in execution.
exec_raw = raw.get("execution", {})
if raw.get("skill") and not (exec_raw.get("skill") or "").strip():
import warnings
warnings.warn(
f"Top-level 'skill:' in {path} is deprecated; move it under "
"execution.skill (it is auto-normalized for now and will be "
"removed in a future release).",
DeprecationWarning,
stacklevel=2,
)
# Dataset
dataset = raw.get("dataset", {})
# Execution config — including an optional multi-step pipeline.
steps = []
for i, s in enumerate(exec_raw.get("steps") or []):
if not isinstance(s, dict):
raise ValueError(f"execution.steps[{i}] must be a mapping")
step_runner = None
if s.get("runner"):
step_runner = _parse_runner_config(
s.get("runner"), context=f"execution.steps[{i}].runner")
step_env = s.get("env") or {}
if not isinstance(step_env, dict):
raise ValueError(
f"execution.steps[{i}].env must be a mapping")
step_timeout = s.get("timeout")
if step_timeout is not None and (
not isinstance(step_timeout, int)
or isinstance(step_timeout, bool)
or step_timeout <= 0):
raise ValueError(
f"execution.steps[{i}].timeout must be a positive integer")
step_budget = s.get("max_budget_usd")
if step_budget is not None and (
not isinstance(step_budget, (int, float))
or isinstance(step_budget, bool)
or step_budget < 0):
raise ValueError(
f"execution.steps[{i}].max_budget_usd must be a "
"non-negative number")
step = StepConfig(
id=s.get("id", "") or "",
name=s.get("name", "") or "",
skill=s.get("skill", "") or "",
prompt=s.get("prompt", "") or "",
arguments=s.get("arguments", "") or "",
env=step_env,
timeout=step_timeout,
max_budget_usd=step_budget,
runner=step_runner,
on_failure=s.get("on_failure", "fail"),
)
if not ((step.skill and step.skill.strip())
or (step.prompt and step.prompt.strip())):
raise ValueError(
f"execution.steps[{i}] ('{step.id}') must set either "
"skill or prompt")
steps.append(step)
execution = ExecutionConfig(
mode=exec_raw.get("mode", "case"),
skill=exec_raw.get("skill", "") or raw.get("skill", ""),
prompt=exec_raw.get("prompt", ""),
arguments=exec_raw.get("arguments", ""),
timeout=exec_raw.get("timeout"),
max_budget_usd=exec_raw.get("max_budget_usd"),
parallelism=exec_raw.get("parallelism"),
env=exec_raw.get("env") or {},
steps=steps,
)
# Runner config (block form)
runner = _parse_runner_config(raw.get("runner"), context="runner")
# Models block
models_raw = raw.get("models", {}) or {}
models = ModelsConfig(
skill=models_raw.get("skill"),
subagent=models_raw.get("subagent"),
judge=models_raw.get("judge"),
hook=models_raw.get("hook"),
)
# MLflow block. Experiment defaults to the eval's top-level
# `name` only when an `mlflow:` block is present — so omitting
# the block entirely leaves MLflow off (no accidental experiment
# creation on shared tracking servers).
has_mlflow_block = "mlflow" in raw and raw["mlflow"] is not None
mlflow_raw = raw.get("mlflow") or {}
if has_mlflow_block:
experiment = mlflow_raw.get("experiment") or raw.get("name", "")
else:
experiment = ""
mlflow = MlflowConfig(
experiment=experiment,
tracking_uri=mlflow_raw.get("tracking_uri"),
tags=mlflow_raw.get("tags", {}) or {},
)
# Dataset — path, schema, and workspace file provisioning
ws_raw = dataset.get("workspace", {}) or {}
ws_files_raw = ws_raw.get("files", []) or []
ws_files = []
for i, f in enumerate(ws_files_raw):
if not isinstance(f, str):
raise ValueError(
f"dataset.workspace.files[{i}] must be a string, got {type(f).__name__}"
)
ws_files.append(
_validate_relative_path(f.rstrip("/"), "dataset.workspace.files")
)
dataset_config = DatasetConfig(
path=_validate_relative_path(
dataset.get("path", ""), "dataset.path", allow_absolute=True
),
schema=dataset.get("schema", ""),
workspace=WorkspaceConfig(files=ws_files),
)
# Generation — synthetic test-case generation (optional) with validation
gen_raw = raw.get("generation") or {}
seeds = []
for i, s in enumerate(gen_raw.get("seeds") or []):
category = s.get("category", "")
count = s.get("count")
if not category or not isinstance(category, str):
raise ValueError(
f"generation.seeds[{i}].category must be a non-empty string, got: {category!r}")
# count is required — a silent default would swallow a mistyped field name
if not isinstance(count, int) or count < 1:
raise ValueError(
f"generation.seeds[{i}].count must be an integer >= 1, got: {count!r}")
# Exactly one prompt discriminator (mirrors judges: builtin/prompt_file/prompt)
discriminators = [
k for k in ("builtin", "prompt_file", "prompt") if s.get(k)
]
if len(discriminators) != 1:
raise ValueError(
f"generation.seeds[{i}] ('{category}') must set exactly one of "
f"builtin / prompt_file / prompt, got: {discriminators or 'none'}")
seeds.append(GenerationSeed(
category=category,
count=count,
builtin=s.get("builtin", ""),
prompt_file=s.get("prompt_file", ""),
prompt=s.get("prompt", ""),
description=s.get("description", ""),
))
# Provenance: absent normalizes to 'skill' (the default source).
strategy = gen_raw.get("strategy") or "skill"
if strategy not in GENERATION_STRATEGIES:
raise ValueError(
f"generation.strategy must be one of "
f"{', '.join(GENERATION_STRATEGIES)}, got: {strategy!r}")
if strategy == "synthetic" and not seeds:
raise ValueError(
"generation.strategy is 'synthetic' but generation.seeds is empty.")
if seeds and strategy != "synthetic":
raise ValueError(
f"generation.seeds are only valid with strategy: synthetic "
f"(got strategy: {strategy}).")
generation_config = GenerationConfig(
strategy=strategy,
context=gen_raw.get("context", {}),
seeds=seeds,
)
config = cls(
name=raw.get("name", path.stem),
description=raw.get("description", ""),
skill=raw.get("skill") or None, # Convert empty string to None
permissions=raw.get("permissions", {}),
execution=execution,
runner=runner,
models=models,
mlflow=mlflow,
config_dir=path.resolve().parent,
config_path=path.resolve(),
dataset=dataset_config,
generation=generation_config,
)
# Outputs (path or tool)
for i, o in enumerate(raw.get("outputs", [])):
config.outputs.append(
OutputConfig(
path=_validate_relative_path(
o.get("path", ""), f"outputs[{i}].path", reject_root=True
),
tool=o.get("tool", ""),
schema=o.get("schema", ""),
batch_pattern=o.get("batch_pattern", ""),
types=o.get("types") or None,
)
)
# Inputs (tool interception)
inputs_raw = raw.get("inputs", {})
for t in inputs_raw.get("tools") or []:
config.inputs.tools.append(
ToolInputConfig(
match=t.get("match", ""),
prompt=t.get("prompt", ""),
prompt_file=t.get("prompt_file", ""),
)
)
# Traces
traces = raw.get("traces", {})
if traces:
config.traces = TracesConfig(
stdout=traces.get("stdout", True),
stderr=traces.get("stderr", True),
events=traces.get("events", True),
metrics=traces.get("metrics", True),
)
# Judges
for j in raw.get("judges", []):
builtin_val = j.get("builtin", "")
if builtin_val is None:
builtin_val = ""
if not isinstance(builtin_val, str):
raise ValueError(
f"Judge '{j.get('name', '')}': 'builtin' must be a string"
)
args_val = j.get("arguments")
if args_val is None:
args_val = {}
elif not isinstance(args_val, dict):
raise ValueError(
f"Judge '{j.get('name', '')}': 'arguments' must be a mapping"
)
agent_val = j.get("agent")
if agent_val is None:
agent_val = {}
elif not isinstance(agent_val, dict):
raise ValueError(
f"Judge '{j.get('name', '')}': 'agent' must be a mapping"
)
elif agent_val.get("runner") is not None:
# Parse the nested runner: sub-block with the SAME block-parsing
# logic as the top-level runner, so a judge's runner is fully
# validated. Shallow-copy so the raw YAML isn't mutated.
if not isinstance(agent_val["runner"], dict):
raise ValueError(
f"Judge '{j.get('name', '')}': 'agent.runner' must be a mapping"
)
agent_val = dict(agent_val)
agent_val["runner"] = _parse_runner_config(
agent_val["runner"],
context=f"Judge '{j.get('name', '')}': agent.runner",
)
score_range_val = j.get("score_range")
if score_range_val is not None:
jname = j.get("name", "")
if (not isinstance(score_range_val, list)
or len(score_range_val) != 2):
raise ValueError(
f"Judge '{jname}': 'score_range' must be a [min, max] list")
try:
lo, hi = float(score_range_val[0]), float(score_range_val[1])
except (TypeError, ValueError) as exc:
raise ValueError(
f"Judge '{jname}': 'score_range' values must be numeric") from exc
if not (math.isfinite(lo) and math.isfinite(hi)) or lo >= hi:
raise ValueError(
f"Judge '{jname}': 'score_range' must be finite and "
"increasing [min, max]")
score_range_val = [lo, hi]
config.judges.append(
JudgeConfig(
name=j.get("name", ""),
description=j.get("description", ""),
condition=j.get("if", ""),
check=j.get("check", ""),
prompt=j.get("prompt", ""),
prompt_file=j.get("prompt_file", ""),
llm_rubric=j.get("llm_rubric", ""),
context=j.get("context", []),
feedback_type=j.get("feedback_type", ""),
score_range=score_range_val,
model=j.get("model", ""),
module=j.get("module", ""),
function=j.get("function", ""),
builtin=builtin_val,
arguments=args_val,
step=j.get("step", "") or "",
samples=int(j.get("samples", 1)),
agent=agent_val,
)
)
# Per-step judge scoping: a judge's `step:` must name a defined
# execution step (fail loud on typos, like reward.judge validation).
step_ids = {s.id for s in execution.steps}
for jc in config.judges:
if not jc.step:
continue
if not execution.steps:
raise ValueError(
f"Judge '{jc.name}': 'step: {jc.step}' requires an "
"execution.steps pipeline")
if jc.step not in step_ids:
raise ValueError(
f"Judge '{jc.name}': 'step: {jc.step}' does not match any "
f"execution step id ({sorted(step_ids)})")
# Scale coherence: a judge's declared scale has to agree with its
# feedback_type and with the scorer that will actually run it. Each of
# these used to be accepted and then quietly ignored at scoring time,
# which is how a judge shipped scoring on a scale nobody declared.
from agent_eval.judges import builtin_judge_kind, builtin_judge_names
for jc in config.judges:
builtin_kind = builtin_judge_kind(jc.builtin) if jc.builtin else None
if jc.builtin and builtin_kind is None:
raise ValueError(
f"Judge '{jc.name}': unknown builtin judge '{jc.builtin}' "
f"(available: {', '.join(builtin_judge_names())})")
if jc.feedback_type == "bool" and jc.score_range:
raise ValueError(
f"Judge '{jc.name}': 'score_range' has no meaning with "
"'feedback_type: bool' (the verdict is pass/fail) — "
"drop one of the two")
if (jc.feedback_type == "int" and jc.score_range
and any(float(b) != int(b) for b in jc.score_range)):
raise ValueError(
f"Judge '{jc.name}': 'feedback_type: int' cannot express "
f"the fractional 'score_range' {jc.score_range} — use "
"'feedback_type: float'")
if (builtin_kind == "llm"
and (jc.feedback_type not in ("", "bool") or jc.score_range)):
raise ValueError(
f"Judge '{jc.name}': builtin LLM judge '{jc.builtin}' is "
"always scored as pass/fail, so 'feedback_type'/"
"'score_range' would be silently ignored")
# `feedback_type` is optional, and score.py's `_numeric_bounds`
# treats anything that is not "bool" as numeric — so the judge that
# most needs this warning is the one that declares neither field,
# and gating on ("int", "float") alone never reached it.
# The judge named "pairwise" is exempt: score.py routes it past
# the numeric path by that name into the A/B/tie verdict flow, so
# no scale — declared or defaulted — is ever applied to it.
if (jc.feedback_type in ("int", "float", "") and not jc.score_range
and not jc.builtin and jc.name != "pairwise"
and (jc.prompt or jc.prompt_file or jc.llm_rubric)):
import warnings
warnings.warn(
f"Judge '{jc.name}': numeric judge has no 'score_range', "
"so it is scored on the unenforced [1, 5] default — "
"declare one to have the returned value checked",
stacklevel=2)
# Reward composition
if "reward" in raw:
reward_raw = raw.get("reward")
if not isinstance(reward_raw, dict):
raise ValueError("reward must be a mapping when provided")
sr = reward_raw.get("score_range")
reward_score_range = None
if sr is not None:
if not isinstance(sr, list) or len(sr) != 2:
raise ValueError(
"reward.score_range must be a [min, max] list")
try:
score_min = float(sr[0])
score_max = float(sr[1])
except (TypeError, ValueError) as exc:
raise ValueError(
"reward.score_range values must be numeric") from exc
if (not (math.isfinite(score_min) and math.isfinite(score_max))
or score_min >= score_max):
raise ValueError(
"reward.score_range must be finite and increasing "
"[min, max]")
reward_score_range = [score_min, score_max]
weights = reward_raw.get("weights", {}) or {}
if not isinstance(weights, dict):
raise ValueError("reward.weights must be a mapping")
try:
weights = {str(k): float(v) for k, v in weights.items()}
except (TypeError, ValueError) as exc:
raise ValueError(
"reward.weights values must be numeric") from exc
if any(v < 0 for v in weights.values()):
raise ValueError("reward.weights values must be non-negative")
raw_list = reward_raw.get("raw", []) or []
if not isinstance(raw_list, list):
raw_list = [raw_list]
# Single-judge mode: one judge's value is the reward. Mutually
# exclusive with the composition inputs.
judge = reward_raw.get("judge")
if judge is not None:
if not isinstance(judge, str) or not judge.strip():
raise ValueError(
"reward.judge must be a non-empty judge name")
conflicting = [k for k in ("formula", "weights", "raw")
if k in reward_raw]
if conflicting:
raise ValueError(
"reward.judge cannot be combined with "
f"{'/'.join(conflicting)}")
judge_names = {j.name for j in config.judges if j.name}
if judge not in judge_names:
raise ValueError(
f"reward.judge '{judge}' does not match any defined "
"judge")
normalize = reward_raw.get("normalize", False)
if not isinstance(normalize, bool):
raise ValueError("reward.normalize must be a boolean")
# gate defaults to False in judge mode, True for composition.
gate = reward_raw.get("gate", judge is None)
if not isinstance(gate, bool):
raise ValueError("reward.gate must be a boolean")
formula = str(reward_raw.get("formula", "weighted"))
# Validate expression formulas now so a typo or unsafe construct
# fails loudly here, not silently as reward 0.0 on every case at
# run time. Bare references ("weighted") are resolved at compute
# time, so skip the expression check for them. Skipped in judge
# mode, where formula is unused.
if judge is None and not re.fullmatch(
r"[A-Za-z_][\w.\-]*", formula.strip()):
from agent_eval.harbor.reward import validate_formula
try:
validate_formula(formula)
except ValueError as exc:
raise ValueError(
f"reward.formula is invalid: {exc}") from exc
config.reward = RewardConfig(
formula=formula,
weights=weights,
gate=gate,
score_range=reward_score_range,
raw=[str(r) for r in raw_list],
judge=judge,
normalize=normalize,
)
if sr is not None:
_warn_reward_range_precedence(config)
_warn_reward_judge_clamp(config)
# Thresholds
config.thresholds = raw.get("thresholds", {})
# Hooks
hooks_raw = raw.get("hooks", {}) or {}
phases = ["before_all", "before_each", "after_each", "before_step",
"after_step", "before_scoring", "after_all", "before_report"]
for phase in phases:
entries = []
for h in (hooks_raw.get(phase) or []):
on_failure_val = h.get("on_failure", "fail")
if on_failure_val not in ("fail", "continue"):
raise ValueError(
f"hooks.{phase}: on_failure must be 'fail' or "
f"'continue', got '{on_failure_val}'")
timeout_val = h.get("timeout", 120)
if not isinstance(timeout_val, int) or timeout_val <= 0:
raise ValueError(
f"hooks.{phase}: timeout must be a positive "
f"integer, got {timeout_val}")
entries.append(HookEntry(
command=h.get("command", ""),
timeout=timeout_val,
description=h.get("description", ""),
on_failure=on_failure_val,
condition=h.get("condition", ""),
))
setattr(config.hooks, phase, entries)
if config.execution.mode == "batch":
per_case = []
if config.hooks.before_each:
per_case.append("before_each")
if config.hooks.after_each:
per_case.append("after_each")
if config.hooks.before_step:
per_case.append("before_step")
if config.hooks.after_step:
per_case.append("after_step")
if per_case:
import warnings
warnings.warn(
f"hooks.{', '.join(per_case)} ignored in batch mode "
f"(per-case hooks only run in case/prompt mode)",
stacklevel=2,
)
resolved_skill = config.resolve_skill()
if resolved_skill:
try:
_validate_path_segment(resolved_skill, f"skill name in {path}")
except ValueError as e:
raise ValueError(str(e)) from e
codex_runners = [config.runner]
codex_runners.extend(
step.runner for step in config.execution.steps if step.runner)
codex_runners = [runner for runner in codex_runners
if runner.type == "codex"]
if codex_runners and config.inputs.tools:
raise ValueError(
"runner.type 'codex' does not support inputs.tools interception; "
"use claude-code or remove the tool interceptors")
if any(runner.workspace_mode == "repo" for runner in codex_runners):
raise ValueError(
"runner.type 'codex' does not support workspace_mode: repo "
"because repository answer-key protections cannot be enforced")
return config