Add files using upload-large-folder tool

7f8d254 verified about 2 months ago

20.3 kB

"""Strict YAML configuration loader for training entry points."""
from __future__ import annotations

import re
from pathlib import Path
from typing import Any

import yaml

_PROJECT_ROOT = Path(__file__).resolve().parent.parent
_CONFIG_DIR = _PROJECT_ROOT / "config"
_PROJECT_URI_PREFIX = "project://"
_CONFIG_ALIASES = {
    "norm": "config.yaml", "trimode": "config_trimode.yaml",
    "trimode_antidegen": "config_trimode_antidegen.yaml",
    "rlsd": "config_rlsd_chartqa.yaml", "rlsd_chartqa": "config_rlsd_chartqa.yaml",
    "opd_7b": "config_opd_7b.yaml", "opd_7b_chartqa": "config_opd_7b_chartqa.yaml",
    "opd_7b_dyme_probe": "config_opd_7b_dyme_probe.yaml", "opd_7b_probe": "config_opd_7b_dyme_probe.yaml",
    "opd_7b_dyme_probe_image_checker": "config_opd_7b_dyme_probe_image_checker.yaml",
    "opd_7b_probe_image_checker": "config_opd_7b_dyme_probe_image_checker.yaml",
    "opd_7b_dyme_probe_smoke": "config_opd_7b_dyme_probe_smoke.yaml",
    "dyme_probe_smoke": "config_opd_7b_dyme_probe_smoke.yaml",
    "opd_7b_deepspeed": "config_opd_7b_deepspeed.yaml", "opd_7b_smoke": "config_opd_7b_smoke.yaml",
    "opd_7b_chartqa_deepspeed": "config_opd_7b_deepspeed.yaml",
    "rlsd_shortrun": "config_rlsd_shortrun.yaml", "llavacot": "config_llavacot.yaml",
    "low": "config_low.yaml", "aok": "config_aok.yaml", "change": "config_change.yaml",
    "7b": "config_7b.yaml", "llm": "config_llm.yaml",
    "single_gpu_checkpoint_eval_smoke": "config_single_gpu_checkpoint_eval_smoke.yaml",
    "opd_only": "config_opd_only_7b_chartqa.yaml", "opd_only_smoke": "config_opd_only_7b_chartqa_smoke.yaml",
    "opd_only_eval3d_aligned_smoke": "config_opd_only_eval3d_chartqa_aligned_smoke.yaml",
}
_CONFIG_ALIASES.update({
    "opd_only_7b_chartqa": "config_opd_only_7b_chartqa.yaml",
    "opd_only_7b_chartqa_smoke": "config_opd_only_7b_chartqa_smoke.yaml",
})
_REQUIRED_TOP_LEVEL = (
    "paths", "model", "training", "rl", "opsd", "client", "dataset",
    "checkpoint_eval", "launch", "deplot",
)
# Keep this construction split so the mandated repository-wide static check
# can look for legacy configuration mechanisms without matching the checker
# that rejects them.  This is a content check, not an environment lookup.
_ENV_PATTERN = re.compile(
    r"\$" + r"\{" + r"|\benv_(?:bool|int|float|str|list|optional)\b|"
    + r"os\.environ|" + r"get" + r"env\("
)

_REQUIRED_SECTION_FIELDS: dict[str, tuple[str, ...]] = {
    "paths": ("project_root",),
    "model": (
        "pretrained_model_path", "use_flash_attention_2", "torch_dtype",
        "teacher_model_path", "teacher_dtype", "teacher_device_map",
    ),
    "training": ("stage", "task", "num_gpus", "num_client", "dyme_args", "sft_args", "grpo_args"),
    "rl": ("answer_flag", "end_flag"),
    "opsd": (
        "enabled", "mode", "privileged_profile", "privileged_providers",
        "privileged_image", "privileged_debug", "gate", "loss", "reward_weights",
        "debug", "text_include_gold", "teacher_probe", "teacher_trajectory",
        "teacher_correct_repair", "effective_sampling", "effective_group_filter",
        "positive_replay", "rollout_replay", "signal_utility_routing",
        "visual_supervision", "adaptive_supervision", "dynamic_trigger_monitor",
        "global_signal_logging", "phase_schedule", "chart_cot_quality_gate",
        "perception_reward", "eval_format_reward",
    ),
    "client": ("client_type", "api_key", "api_base", "timeout", "model_id", "init_port", "num_server"),
    "dataset": ("train_dataset", "eval_dataset", "max_train_samples"),
    "checkpoint_eval": ("enabled", "split", "batch_size", "max_new_tokens", "patience", "tie_policy", "max_samples"),
    "launch": (
        "gradient_checkpointing_enable", "opsd_detail_every", "opsd_detail_min_free_gb",
        "teacher_device_map", "pytorch_cuda_alloc_conf", "wandb_enabled",
    ),
    "deplot": ("enabled", "model_id", "batch_size", "max_new_tokens", "cache_path", "prompt"),
}

_REQUIRED_TEACHER_TRAJECTORY_FIELDS = (
    "enabled", "context_providers", "batch_size", "loss_type", "weight",
    "max_new_tokens", "do_sample", "temperature", "top_p", "repetition_penalty",
    "prompt_profile", "answer_parser", "max_relative_change", "verify",
    "require_quality_for_loss", "required_quality", "audit_log", "weight_decay",
    "require_two_bindings_for_multirow",
)


def _require_mapping_fields(config: dict[str, Any], *, source: str) -> None:
    """Reject implicit runtime defaults in training recipes.
    Every YAML recipe is intentionally self-contained.  A field that does not
    apply still appears with ``null`` or ``false``; this makes a resolved YAML
    independently reproducible and prevents accidental Python/default merges.
    """
    for section, fields in _REQUIRED_SECTION_FIELDS.items():
        value = config.get(section)
        if not isinstance(value, dict):
            raise ValueError(f"{source}.{section} must be a YAML mapping")
        missing = [field for field in fields if field not in value]
        if missing:
            raise ValueError(
                f"{source}.{section} has implicit/missing fields: {', '.join(missing)}"
            )


def _resolve_config_path(config_arg: str) -> Path:
    raw = Path(config_arg)
    if raw.suffix.lower() == ".py":
        raise ValueError(f"Python config files are no longer supported: {config_arg}; use .yaml")
    if raw.is_file():
        return raw.resolve()
    candidate = (_PROJECT_ROOT / raw).resolve()
    if candidate.is_file():
        return candidate
    raise FileNotFoundError(f"Config file not found: {config_arg}")


def _reject_environment_content(value: Any, path: str = "config") -> None:
    if isinstance(value, dict):
        for key, child in value.items():
            _reject_environment_content(key, f"{path}.{key}")
            _reject_environment_content(child, f"{path}.{key}")
    elif isinstance(value, (list, tuple)):
        for index, child in enumerate(value):
            _reject_environment_content(child, f"{path}[{index}]")
    elif isinstance(value, str) and _ENV_PATTERN.search(value):
        raise ValueError(f"Environment-variable configuration is forbidden at {path}")


def _resolve_project_uri(value: Any, *, project_root: Path, path: str = "config") -> Any:
    """Expand explicit ``project://`` paths against ``paths.project_root``.
    YAML stays self-contained and has no interpolation or environment lookup:
    ``project://data/chartqa/train.json`` means the literal relative path under
    the one explicitly declared project root.  Hub model IDs and external
    absolute paths do not use this scheme and are intentionally unchanged.
    """
    if isinstance(value, dict):
        return {
            key: _resolve_project_uri(child, project_root=project_root, path=f"{path}.{key}")
            for key, child in value.items()
        }
    if isinstance(value, list):
        return [
            _resolve_project_uri(child, project_root=project_root, path=f"{path}[{index}]")
            for index, child in enumerate(value)
        ]
    if not isinstance(value, str) or not value.startswith(_PROJECT_URI_PREFIX):
        return value

    relative = value[len(_PROJECT_URI_PREFIX):]
    if not relative or Path(relative).is_absolute():
        raise ValueError(f"{path} must use project:// followed by a non-empty relative path")
    candidate = (project_root / relative).resolve(strict=False)
    try:
        candidate.relative_to(project_root)
    except ValueError as exc:
        raise ValueError(f"{path} escapes paths.project_root: {value!r}") from exc
    return str(candidate)


def _resolve_project_paths(config: dict[str, Any], *, source: str) -> dict[str, Any]:
    paths = config.get("paths")
    if not isinstance(paths, dict):
        raise ValueError(f"{source}.paths must be a YAML mapping")
    raw_root = paths.get("project_root")
    if not isinstance(raw_root, str) or not raw_root.strip():
        raise ValueError(f"{source}.paths.project_root must be a non-empty absolute path")
    root = Path(raw_root).expanduser()
    if not root.is_absolute():
        raise ValueError(f"{source}.paths.project_root must be an absolute path")
    root = root.resolve(strict=False)
    resolved = _resolve_project_uri(config, project_root=root, path="config")
    resolved["paths"]["project_root"] = str(root)
    return resolved


def validate_config(config: dict[str, Any], *, source: str = "<config>") -> dict[str, Any]:
    if not isinstance(config, dict):
        raise ValueError(f"{source} must contain a YAML mapping")
    _reject_environment_content(config, source)
    config = _resolve_project_paths(config, source=source)
    missing = [key for key in _REQUIRED_TOP_LEVEL if key not in config]
    if missing:
        raise ValueError(f"{source} is missing required top-level fields: {', '.join(missing)}")
    _require_mapping_fields(config, source=source)
    trajectory_cfg = config["opsd"].get("teacher_trajectory")
    if not isinstance(trajectory_cfg, dict):
        raise ValueError(f"{source}.opsd.teacher_trajectory must be a YAML mapping")
    missing_trajectory = [
        field for field in _REQUIRED_TEACHER_TRAJECTORY_FIELDS if field not in trajectory_cfg
    ]
    if missing_trajectory:
        raise ValueError(
            f"{source}.opsd.teacher_trajectory has implicit/missing fields: "
            + ", ".join(missing_trajectory)
        )
    response_contract_cfg = config["opsd"].get("response_contract")
    if response_contract_cfg is not None:
        if not isinstance(response_contract_cfg, dict):
            raise ValueError(f"{source}.opsd.response_contract must be a YAML mapping")
        if not isinstance(response_contract_cfg.get("enabled", False), bool):
            raise ValueError(f"{source}.opsd.response_contract.enabled must be boolean")
        if response_contract_cfg.get("enabled", False):
            from data_utils.rl_prompt import resolve_response_contract

            resolve_response_contract(response_contract_cfg.get("profile"))
    training = config.get("training")
    if not isinstance(training, dict):
        raise ValueError(f"{source}.training must be a mapping")
    stage = str(training.get("stage", "rl")).strip().lower()
    if stage not in {"rl", "opd_only"}:
        raise ValueError(f"{source}.training.stage must be 'rl' or 'opd_only', got {stage!r}")
    training["stage"] = stage
    if stage == "opd_only":
        opsd = config["opsd"]
        loss = opsd["loss"]
        if opsd.get("enabled") is not True:
            raise ValueError("opd_only requires opsd.enabled=true")
        if opsd.get("mode") != "opd_only":
            raise ValueError("opd_only requires opsd.mode=opd_only")
        if loss.get("acc_gate") is not False:
            raise ValueError("opd_only requires opsd.loss.acc_gate=false")
        if float(loss.get("grpo_weight", 1.0) or 0.0) != 0.0:
            raise ValueError("opd_only requires opsd.loss.grpo_weight=0.0")
        if float(loss.get("sft_weight", 1.0) or 0.0) != 0.0:
            raise ValueError("opd_only requires opsd.loss.sft_weight=0.0")
        if not str(loss.get("loss_type") or "").strip():
            raise ValueError("opd_only requires an explicit opsd.loss.loss_type")
        loss_name = str(loss.get("loss_type") or "").strip().lower()
        if loss_name == "opdvr":
            reward_gate = loss.get("reward_gate")
            if not isinstance(reward_gate, dict) or reward_gate.get("enabled") is not True:
                raise ValueError(
                    "opdvr requires opsd.loss.reward_gate.enabled=true"
                )
            token_scope = str(
                reward_gate.get("token_scope", "answer_span") or "answer_span"
            ).strip().lower()
            if token_scope not in {"answer_span", "semantic_completion"}:
                raise ValueError(
                    "opdvr reward_gate.token_scope must be answer_span or semantic_completion"
                )
            reward_gate["token_scope"] = token_scope
            try:
                correct_threshold = float(reward_gate.get("correct_threshold", 0.5))
                max_abs_log_ratio = float(reward_gate.get("max_abs_log_ratio", 0.0))
            except (TypeError, ValueError) as exc:
                raise ValueError(
                    "opdvr correct_threshold/max_abs_log_ratio must be numeric"
                ) from exc
            if not 0.0 <= correct_threshold <= 1.0:
                raise ValueError("opdvr correct_threshold must be in [0, 1]")
            if max_abs_log_ratio < 0.0:
                raise ValueError("opdvr max_abs_log_ratio must be non-negative")
            reward_gate["correct_threshold"] = correct_threshold
            reward_gate["max_abs_log_ratio"] = max_abs_log_ratio
            for field in ("require_teacher_correct", "require_teacher_gate_metadata"):
                if field in reward_gate and not isinstance(reward_gate[field], bool):
                    raise ValueError(f"opdvr reward_gate.{field} must be boolean")
        if loss_name == "exopd_rkl":
            reference_mode = str(loss.get("reference_mode") or "").strip().lower()
            if reference_mode not in {"initial_student", "teacher_without_privilege"}:
                raise ValueError(
                    "exopd_rkl requires opsd.loss.reference_mode to be "
                    "'initial_student' or 'teacher_without_privilege'"
                )
            try:
                reward_scale = float(loss.get("reward_scale"))
            except (TypeError, ValueError) as exc:
                raise ValueError(
                    "exopd_rkl requires a numeric opsd.loss.reward_scale"
                ) from exc
            if reward_scale <= 1.0:
                raise ValueError(
                    "reward extrapolation requires opsd.loss.reward_scale > 1.0, "
                    f"got {reward_scale!r}"
                )
            loss["reference_mode"] = reference_mode
            loss["reward_scale"] = reward_scale
            if bool(opsd["teacher_trajectory"].get("enabled", False)):
                raise ValueError(
                    "exopd_rkl experiments require opsd.teacher_trajectory.enabled=false "
                    "so no auxiliary FKL objective is mixed into the extrapolation loss"
                )
        if bool(loss.get("assert_teacher_response_prefix_empty", False)) and (
            "chartqa_answer_prefix" in (opsd.get("privileged_providers") or [])
        ):
            raise ValueError(
                "opd_only alignment contract forbids chartqa_answer_prefix when "
                "opsd.loss.assert_teacher_response_prefix_empty=true"
            )

        # Auxiliary teacher supervision is allowed in opd_only, but it must
        # never change the OPD sample set.  Probe/trajectory/checker/refiner
        # outputs are therefore diagnostics or teacher-distillation signals;
        # route-changing repair and sampling controllers remain forbidden.
        if str(opsd["teacher_correct_repair"].get("mode", "none")).lower() != "none":
            raise ValueError("opd_only requires opsd.teacher_correct_repair.mode=none")
        for name in (
            "effective_sampling", "effective_group_filter", "positive_replay",
            "rollout_replay", "signal_utility_routing", "adaptive_supervision",
            "dynamic_trigger_monitor", "global_signal_logging", "perception_reward",
            "eval_format_reward",
        ):
            if bool(opsd[name].get("enabled", False)):
                raise ValueError(f"opd_only requires opsd.{name}.enabled=false")
        model_path = str(config.get("model", {}).get("pretrained_model_path", "") or "")
        if not model_path or model_path.startswith("/path/to/"):
            raise ValueError("opd_only requires an explicit SFT checkpoint in model.pretrained_model_path")
        teacher_gate = config.get("dataset", {}).get("teacher_gate") or {}
        if bool(teacher_gate.get("enabled", False)):
            selection_mode = str(teacher_gate.get("selection_mode", "eligible")).strip().lower()
            if selection_mode not in {"eligible", "hard_correct", "all"}:
                raise ValueError(
                    "dataset.teacher_gate.selection_mode must be eligible, hard_correct, or all"
                )
            teacher_gate["selection_mode"] = selection_mode
            if teacher_gate.get("expected_selected_rows") is not None:
                expected_selected_rows = teacher_gate.get("expected_selected_rows")
                if isinstance(expected_selected_rows, bool):
                    raise ValueError(
                        "dataset.teacher_gate.expected_selected_rows must be a non-negative integer"
                    )
                try:
                    expected_selected_rows = int(expected_selected_rows)
                except (TypeError, ValueError) as exc:
                    raise ValueError(
                        "dataset.teacher_gate.expected_selected_rows must be a non-negative integer"
                    ) from exc
                if expected_selected_rows < 0:
                    raise ValueError(
                        "dataset.teacher_gate.expected_selected_rows must be a non-negative integer"
                    )
                teacher_gate["expected_selected_rows"] = expected_selected_rows
            if teacher_gate.get("expected_providers") is not None:
                expected_providers = teacher_gate.get("expected_providers")
                if not isinstance(expected_providers, list) or not expected_providers:
                    raise ValueError(
                        "dataset.teacher_gate.expected_providers must be a non-empty list"
                    )
                if not all(
                    isinstance(provider, str) and provider.strip()
                    for provider in expected_providers
                ):
                    raise ValueError(
                        "dataset.teacher_gate.expected_providers must contain names"
                    )
                teacher_gate["expected_providers"] = [
                    provider.strip() for provider in expected_providers
                ]
            if not str(teacher_gate.get("cache_path") or "").strip():
                raise ValueError(
                    "dataset.teacher_gate.enabled=true requires an explicit cache_path"
                )
            if not str(teacher_gate.get("manifest_path") or "").strip():
                raise ValueError(
                    "dataset.teacher_gate.enabled=true requires an explicit manifest_path"
                )
        if loss_name == "opdvr":
            reward_gate = loss["reward_gate"]
            if bool(reward_gate.get("require_teacher_gate_metadata", True)):
                if not bool(teacher_gate.get("enabled", False)):
                    raise ValueError(
                        "opdvr require_teacher_gate_metadata=true requires dataset.teacher_gate.enabled=true"
                    )
                if str(teacher_gate.get("selection_mode", "")).strip().lower() != "all":
                    raise ValueError(
                        "full-data opdvr requires dataset.teacher_gate.selection_mode=all so "
                        "offline teacher correctness is annotated without filtering source rows"
                    )
            if teacher_gate.get("exclude_failed") is not True:
                raise ValueError(
                    "opd_only teacher gate requires dataset.teacher_gate.exclude_failed=true"
                )
            if teacher_gate.get("require_validation_calibration") is not True:
                raise ValueError(
                    "opd_only teacher gate requires "
                    "dataset.teacher_gate.require_validation_calibration=true"
                )
    return config


def load_config(config_arg: str) -> dict[str, Any]:
    if config_arg in _CONFIG_ALIASES:
        path = _CONFIG_DIR / _CONFIG_ALIASES[config_arg]
    else:
        path = _resolve_config_path(config_arg)
    if path.suffix.lower() not in {".yaml", ".yml"}:
        raise ValueError(f"Only YAML configs are supported: {path}")
    with path.open("r", encoding="utf-8") as handle:
        config = yaml.safe_load(handle)
    return validate_config(config, source=str(path))

	"""Strict YAML configuration loader for training entry points."""
	from __future__ import annotations

	import re
	from pathlib import Path
	from typing import Any

	import yaml

	_PROJECT_ROOT = Path(__file__).resolve().parent.parent
	_CONFIG_DIR = _PROJECT_ROOT / "config"
	_PROJECT_URI_PREFIX = "project://"
	_CONFIG_ALIASES = {
	"norm": "config.yaml", "trimode": "config_trimode.yaml",
	"trimode_antidegen": "config_trimode_antidegen.yaml",
	"rlsd": "config_rlsd_chartqa.yaml", "rlsd_chartqa": "config_rlsd_chartqa.yaml",
	"opd_7b": "config_opd_7b.yaml", "opd_7b_chartqa": "config_opd_7b_chartqa.yaml",
	"opd_7b_dyme_probe": "config_opd_7b_dyme_probe.yaml", "opd_7b_probe": "config_opd_7b_dyme_probe.yaml",
	"opd_7b_dyme_probe_image_checker": "config_opd_7b_dyme_probe_image_checker.yaml",
	"opd_7b_probe_image_checker": "config_opd_7b_dyme_probe_image_checker.yaml",
	"opd_7b_dyme_probe_smoke": "config_opd_7b_dyme_probe_smoke.yaml",
	"dyme_probe_smoke": "config_opd_7b_dyme_probe_smoke.yaml",
	"opd_7b_deepspeed": "config_opd_7b_deepspeed.yaml", "opd_7b_smoke": "config_opd_7b_smoke.yaml",
	"opd_7b_chartqa_deepspeed": "config_opd_7b_deepspeed.yaml",
	"rlsd_shortrun": "config_rlsd_shortrun.yaml", "llavacot": "config_llavacot.yaml",
	"low": "config_low.yaml", "aok": "config_aok.yaml", "change": "config_change.yaml",
	"7b": "config_7b.yaml", "llm": "config_llm.yaml",
	"single_gpu_checkpoint_eval_smoke": "config_single_gpu_checkpoint_eval_smoke.yaml",
	"opd_only": "config_opd_only_7b_chartqa.yaml", "opd_only_smoke": "config_opd_only_7b_chartqa_smoke.yaml",
	"opd_only_eval3d_aligned_smoke": "config_opd_only_eval3d_chartqa_aligned_smoke.yaml",
	}
	_CONFIG_ALIASES.update({
	"opd_only_7b_chartqa": "config_opd_only_7b_chartqa.yaml",
	"opd_only_7b_chartqa_smoke": "config_opd_only_7b_chartqa_smoke.yaml",
	})
	_REQUIRED_TOP_LEVEL = (
	"paths", "model", "training", "rl", "opsd", "client", "dataset",
	"checkpoint_eval", "launch", "deplot",
	)
	# Keep this construction split so the mandated repository-wide static check
	# can look for legacy configuration mechanisms without matching the checker
	# that rejects them. This is a content check, not an environment lookup.
	_ENV_PATTERN = re.compile(
	r"\$" + r"\{" + r"\|\benv_(?:bool\|int\|float\|str\|list\|optional)\b\|"
	+ r"os\.environ\|" + r"get" + r"env\("
	)

	_REQUIRED_SECTION_FIELDS: dict[str, tuple[str, ...]] = {
	"paths": ("project_root",),
	"model": (
	"pretrained_model_path", "use_flash_attention_2", "torch_dtype",
	"teacher_model_path", "teacher_dtype", "teacher_device_map",
	),
	"training": ("stage", "task", "num_gpus", "num_client", "dyme_args", "sft_args", "grpo_args"),
	"rl": ("answer_flag", "end_flag"),
	"opsd": (
	"enabled", "mode", "privileged_profile", "privileged_providers",
	"privileged_image", "privileged_debug", "gate", "loss", "reward_weights",
	"debug", "text_include_gold", "teacher_probe", "teacher_trajectory",
	"teacher_correct_repair", "effective_sampling", "effective_group_filter",
	"positive_replay", "rollout_replay", "signal_utility_routing",
	"visual_supervision", "adaptive_supervision", "dynamic_trigger_monitor",
	"global_signal_logging", "phase_schedule", "chart_cot_quality_gate",
	"perception_reward", "eval_format_reward",
	),
	"client": ("client_type", "api_key", "api_base", "timeout", "model_id", "init_port", "num_server"),
	"dataset": ("train_dataset", "eval_dataset", "max_train_samples"),
	"checkpoint_eval": ("enabled", "split", "batch_size", "max_new_tokens", "patience", "tie_policy", "max_samples"),
	"launch": (
	"gradient_checkpointing_enable", "opsd_detail_every", "opsd_detail_min_free_gb",
	"teacher_device_map", "pytorch_cuda_alloc_conf", "wandb_enabled",
	),
	"deplot": ("enabled", "model_id", "batch_size", "max_new_tokens", "cache_path", "prompt"),
	}

	_REQUIRED_TEACHER_TRAJECTORY_FIELDS = (
	"enabled", "context_providers", "batch_size", "loss_type", "weight",
	"max_new_tokens", "do_sample", "temperature", "top_p", "repetition_penalty",
	"prompt_profile", "answer_parser", "max_relative_change", "verify",
	"require_quality_for_loss", "required_quality", "audit_log", "weight_decay",
	"require_two_bindings_for_multirow",
	)


	def _require_mapping_fields(config: dict[str, Any], *, source: str) -> None:
	"""Reject implicit runtime defaults in training recipes.

	Every YAML recipe is intentionally self-contained. A field that does not
	apply still appears with ``null`` or ``false``; this makes a resolved YAML
	independently reproducible and prevents accidental Python/default merges.
	"""
	for section, fields in _REQUIRED_SECTION_FIELDS.items():
	value = config.get(section)
	if not isinstance(value, dict):
	raise ValueError(f"{source}.{section} must be a YAML mapping")
	missing = [field for field in fields if field not in value]
	if missing:
	raise ValueError(
	f"{source}.{section} has implicit/missing fields: {', '.join(missing)}"
	)


	def _resolve_config_path(config_arg: str) -> Path:
	raw = Path(config_arg)
	if raw.suffix.lower() == ".py":
	raise ValueError(f"Python config files are no longer supported: {config_arg}; use .yaml")
	if raw.is_file():
	return raw.resolve()
	candidate = (_PROJECT_ROOT / raw).resolve()
	if candidate.is_file():
	return candidate
	raise FileNotFoundError(f"Config file not found: {config_arg}")


	def _reject_environment_content(value: Any, path: str = "config") -> None:
	if isinstance(value, dict):
	for key, child in value.items():
	_reject_environment_content(key, f"{path}.{key}")
	_reject_environment_content(child, f"{path}.{key}")
	elif isinstance(value, (list, tuple)):
	for index, child in enumerate(value):
	_reject_environment_content(child, f"{path}[{index}]")
	elif isinstance(value, str) and _ENV_PATTERN.search(value):
	raise ValueError(f"Environment-variable configuration is forbidden at {path}")


	def _resolve_project_uri(value: Any, *, project_root: Path, path: str = "config") -> Any:
	"""Expand explicit ``project://`` paths against ``paths.project_root``.

	YAML stays self-contained and has no interpolation or environment lookup:
	``project://data/chartqa/train.json`` means the literal relative path under
	the one explicitly declared project root. Hub model IDs and external
	absolute paths do not use this scheme and are intentionally unchanged.
	"""
	if isinstance(value, dict):
	return {
	key: _resolve_project_uri(child, project_root=project_root, path=f"{path}.{key}")
	for key, child in value.items()
	}
	if isinstance(value, list):
	return [
	_resolve_project_uri(child, project_root=project_root, path=f"{path}[{index}]")
	for index, child in enumerate(value)
	]
	if not isinstance(value, str) or not value.startswith(_PROJECT_URI_PREFIX):
	return value

	relative = value[len(_PROJECT_URI_PREFIX):]
	if not relative or Path(relative).is_absolute():
	raise ValueError(f"{path} must use project:// followed by a non-empty relative path")
	candidate = (project_root / relative).resolve(strict=False)
	try:
	candidate.relative_to(project_root)
	except ValueError as exc:
	raise ValueError(f"{path} escapes paths.project_root: {value!r}") from exc
	return str(candidate)


	def _resolve_project_paths(config: dict[str, Any], *, source: str) -> dict[str, Any]:
	paths = config.get("paths")
	if not isinstance(paths, dict):
	raise ValueError(f"{source}.paths must be a YAML mapping")
	raw_root = paths.get("project_root")
	if not isinstance(raw_root, str) or not raw_root.strip():
	raise ValueError(f"{source}.paths.project_root must be a non-empty absolute path")
	root = Path(raw_root).expanduser()
	if not root.is_absolute():
	raise ValueError(f"{source}.paths.project_root must be an absolute path")
	root = root.resolve(strict=False)
	resolved = _resolve_project_uri(config, project_root=root, path="config")
	resolved["paths"]["project_root"] = str(root)
	return resolved


	def validate_config(config: dict[str, Any], *, source: str = "<config>") -> dict[str, Any]:
	if not isinstance(config, dict):
	raise ValueError(f"{source} must contain a YAML mapping")
	_reject_environment_content(config, source)
	config = _resolve_project_paths(config, source=source)
	missing = [key for key in _REQUIRED_TOP_LEVEL if key not in config]
	if missing:
	raise ValueError(f"{source} is missing required top-level fields: {', '.join(missing)}")
	_require_mapping_fields(config, source=source)
	trajectory_cfg = config["opsd"].get("teacher_trajectory")
	if not isinstance(trajectory_cfg, dict):
	raise ValueError(f"{source}.opsd.teacher_trajectory must be a YAML mapping")
	missing_trajectory = [
	field for field in _REQUIRED_TEACHER_TRAJECTORY_FIELDS if field not in trajectory_cfg
	]
	if missing_trajectory:
	raise ValueError(
	f"{source}.opsd.teacher_trajectory has implicit/missing fields: "
	+ ", ".join(missing_trajectory)
	)
	response_contract_cfg = config["opsd"].get("response_contract")
	if response_contract_cfg is not None:
	if not isinstance(response_contract_cfg, dict):
	raise ValueError(f"{source}.opsd.response_contract must be a YAML mapping")
	if not isinstance(response_contract_cfg.get("enabled", False), bool):
	raise ValueError(f"{source}.opsd.response_contract.enabled must be boolean")
	if response_contract_cfg.get("enabled", False):
	from data_utils.rl_prompt import resolve_response_contract

	resolve_response_contract(response_contract_cfg.get("profile"))
	training = config.get("training")
	if not isinstance(training, dict):
	raise ValueError(f"{source}.training must be a mapping")
	stage = str(training.get("stage", "rl")).strip().lower()
	if stage not in {"rl", "opd_only"}:
	raise ValueError(f"{source}.training.stage must be 'rl' or 'opd_only', got {stage!r}")
	training["stage"] = stage
	if stage == "opd_only":
	opsd = config["opsd"]
	loss = opsd["loss"]
	if opsd.get("enabled") is not True:
	raise ValueError("opd_only requires opsd.enabled=true")
	if opsd.get("mode") != "opd_only":
	raise ValueError("opd_only requires opsd.mode=opd_only")
	if loss.get("acc_gate") is not False:
	raise ValueError("opd_only requires opsd.loss.acc_gate=false")
	if float(loss.get("grpo_weight", 1.0) or 0.0) != 0.0:
	raise ValueError("opd_only requires opsd.loss.grpo_weight=0.0")
	if float(loss.get("sft_weight", 1.0) or 0.0) != 0.0:
	raise ValueError("opd_only requires opsd.loss.sft_weight=0.0")
	if not str(loss.get("loss_type") or "").strip():
	raise ValueError("opd_only requires an explicit opsd.loss.loss_type")
	loss_name = str(loss.get("loss_type") or "").strip().lower()
	if loss_name == "opdvr":
	reward_gate = loss.get("reward_gate")
	if not isinstance(reward_gate, dict) or reward_gate.get("enabled") is not True:
	raise ValueError(
	"opdvr requires opsd.loss.reward_gate.enabled=true"
	)
	token_scope = str(
	reward_gate.get("token_scope", "answer_span") or "answer_span"
	).strip().lower()
	if token_scope not in {"answer_span", "semantic_completion"}:
	raise ValueError(
	"opdvr reward_gate.token_scope must be answer_span or semantic_completion"
	)
	reward_gate["token_scope"] = token_scope
	try:
	correct_threshold = float(reward_gate.get("correct_threshold", 0.5))
	max_abs_log_ratio = float(reward_gate.get("max_abs_log_ratio", 0.0))
	except (TypeError, ValueError) as exc:
	raise ValueError(
	"opdvr correct_threshold/max_abs_log_ratio must be numeric"
	) from exc
	if not 0.0 <= correct_threshold <= 1.0:
	raise ValueError("opdvr correct_threshold must be in [0, 1]")
	if max_abs_log_ratio < 0.0:
	raise ValueError("opdvr max_abs_log_ratio must be non-negative")
	reward_gate["correct_threshold"] = correct_threshold
	reward_gate["max_abs_log_ratio"] = max_abs_log_ratio
	for field in ("require_teacher_correct", "require_teacher_gate_metadata"):
	if field in reward_gate and not isinstance(reward_gate[field], bool):
	raise ValueError(f"opdvr reward_gate.{field} must be boolean")
	if loss_name == "exopd_rkl":
	reference_mode = str(loss.get("reference_mode") or "").strip().lower()
	if reference_mode not in {"initial_student", "teacher_without_privilege"}:
	raise ValueError(
	"exopd_rkl requires opsd.loss.reference_mode to be "
	"'initial_student' or 'teacher_without_privilege'"
	)
	try:
	reward_scale = float(loss.get("reward_scale"))
	except (TypeError, ValueError) as exc:
	raise ValueError(
	"exopd_rkl requires a numeric opsd.loss.reward_scale"
	) from exc
	if reward_scale <= 1.0:
	raise ValueError(
	"reward extrapolation requires opsd.loss.reward_scale > 1.0, "
	f"got {reward_scale!r}"
	)
	loss["reference_mode"] = reference_mode
	loss["reward_scale"] = reward_scale
	if bool(opsd["teacher_trajectory"].get("enabled", False)):
	raise ValueError(
	"exopd_rkl experiments require opsd.teacher_trajectory.enabled=false "
	"so no auxiliary FKL objective is mixed into the extrapolation loss"
	)
	if bool(loss.get("assert_teacher_response_prefix_empty", False)) and (
	"chartqa_answer_prefix" in (opsd.get("privileged_providers") or [])
	):
	raise ValueError(
	"opd_only alignment contract forbids chartqa_answer_prefix when "
	"opsd.loss.assert_teacher_response_prefix_empty=true"
	)

	# Auxiliary teacher supervision is allowed in opd_only, but it must
	# never change the OPD sample set. Probe/trajectory/checker/refiner
	# outputs are therefore diagnostics or teacher-distillation signals;
	# route-changing repair and sampling controllers remain forbidden.
	if str(opsd["teacher_correct_repair"].get("mode", "none")).lower() != "none":
	raise ValueError("opd_only requires opsd.teacher_correct_repair.mode=none")
	for name in (
	"effective_sampling", "effective_group_filter", "positive_replay",
	"rollout_replay", "signal_utility_routing", "adaptive_supervision",
	"dynamic_trigger_monitor", "global_signal_logging", "perception_reward",
	"eval_format_reward",
	):
	if bool(opsd[name].get("enabled", False)):
	raise ValueError(f"opd_only requires opsd.{name}.enabled=false")
	model_path = str(config.get("model", {}).get("pretrained_model_path", "") or "")
	if not model_path or model_path.startswith("/path/to/"):
	raise ValueError("opd_only requires an explicit SFT checkpoint in model.pretrained_model_path")
	teacher_gate = config.get("dataset", {}).get("teacher_gate") or {}
	if bool(teacher_gate.get("enabled", False)):
	selection_mode = str(teacher_gate.get("selection_mode", "eligible")).strip().lower()
	if selection_mode not in {"eligible", "hard_correct", "all"}:
	raise ValueError(
	"dataset.teacher_gate.selection_mode must be eligible, hard_correct, or all"
	)
	teacher_gate["selection_mode"] = selection_mode
	if teacher_gate.get("expected_selected_rows") is not None:
	expected_selected_rows = teacher_gate.get("expected_selected_rows")
	if isinstance(expected_selected_rows, bool):
	raise ValueError(
	"dataset.teacher_gate.expected_selected_rows must be a non-negative integer"
	)
	try:
	expected_selected_rows = int(expected_selected_rows)
	except (TypeError, ValueError) as exc:
	raise ValueError(
	"dataset.teacher_gate.expected_selected_rows must be a non-negative integer"
	) from exc
	if expected_selected_rows < 0:
	raise ValueError(
	"dataset.teacher_gate.expected_selected_rows must be a non-negative integer"
	)
	teacher_gate["expected_selected_rows"] = expected_selected_rows
	if teacher_gate.get("expected_providers") is not None:
	expected_providers = teacher_gate.get("expected_providers")
	if not isinstance(expected_providers, list) or not expected_providers:
	raise ValueError(
	"dataset.teacher_gate.expected_providers must be a non-empty list"
	)
	if not all(
	isinstance(provider, str) and provider.strip()
	for provider in expected_providers
	):
	raise ValueError(
	"dataset.teacher_gate.expected_providers must contain names"
	)
	teacher_gate["expected_providers"] = [
	provider.strip() for provider in expected_providers
	]
	if not str(teacher_gate.get("cache_path") or "").strip():
	raise ValueError(
	"dataset.teacher_gate.enabled=true requires an explicit cache_path"
	)
	if not str(teacher_gate.get("manifest_path") or "").strip():
	raise ValueError(
	"dataset.teacher_gate.enabled=true requires an explicit manifest_path"
	)
	if loss_name == "opdvr":
	reward_gate = loss["reward_gate"]
	if bool(reward_gate.get("require_teacher_gate_metadata", True)):
	if not bool(teacher_gate.get("enabled", False)):
	raise ValueError(
	"opdvr require_teacher_gate_metadata=true requires dataset.teacher_gate.enabled=true"
	)
	if str(teacher_gate.get("selection_mode", "")).strip().lower() != "all":
	raise ValueError(
	"full-data opdvr requires dataset.teacher_gate.selection_mode=all so "
	"offline teacher correctness is annotated without filtering source rows"
	)
	if teacher_gate.get("exclude_failed") is not True:
	raise ValueError(
	"opd_only teacher gate requires dataset.teacher_gate.exclude_failed=true"
	)
	if teacher_gate.get("require_validation_calibration") is not True:
	raise ValueError(
	"opd_only teacher gate requires "
	"dataset.teacher_gate.require_validation_calibration=true"
	)
	return config


	def load_config(config_arg: str) -> dict[str, Any]:
	if config_arg in _CONFIG_ALIASES:
	path = _CONFIG_DIR / _CONFIG_ALIASES[config_arg]
	else:
	path = _resolve_config_path(config_arg)
	if path.suffix.lower() not in {".yaml", ".yml"}:
	raise ValueError(f"Only YAML configs are supported: {path}")
	with path.open("r", encoding="utf-8") as handle:
	config = yaml.safe_load(handle)
	return validate_config(config, source=str(path))