run.py

#!/usr/bin/env python
"""Lightweight entrypoint for running gage-eval-main without the full CLI stack."""

from __future__ import annotations

import argparse
import os
import sys
import signal
import platform
import time
import re
import copy
from textwrap import indent
from pathlib import Path
from typing import Optional

import yaml

REPO_ROOT = Path(__file__).resolve().parent
SRC_ROOT = REPO_ROOT / "src"

if not SRC_ROOT.exists():
    raise RuntimeError(f"Expected gage-eval sources under {SRC_ROOT}, but the directory is missing.")

if str(SRC_ROOT) not in sys.path:
    sys.path.insert(0, str(SRC_ROOT))

import gage_eval  # noqa: F401 - auto-discovery happens on import
from gage_eval.config import build_default_registry
from gage_eval.config.loader import (
    expand_env as _loader_expand_env,
    load_pipeline_config_payload,
    load_pre_smart_defaults_payload,
    load_yaml_mapping,
)
from gage_eval.config.loader_cli import CLIIntent, apply_cli_final_overrides, parse_metric_ids_csv
from gage_eval.tools.distill import DistillError, analyze_tasks_for_distill
from gage_eval.config.pipeline_config import PipelineConfig
from gage_eval.evaluation.runtime_builder import build_runtime
from gage_eval.observability.trace import ObservabilityTrace
from gage_eval.role.resource_profile import NodeResource, ResourceProfile
from gage_eval.agent_eval_kits.common import build_agentkit_v2_init_payload


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Run a gage-eval PipelineConfig.")
    parser.add_argument(
        "command",
        nargs="?",
        help="Optional positional command. Use 'init <kit>' to print an Agent Eval v2 template.",
    )
    parser.add_argument(
        "command_arg",
        nargs="?",
        help="Argument for the positional command.",
    )
    parser.add_argument(
        "--config",
        "-c",
        required=False,
        help="Path to the PipelineConfig YAML file.",
    )
    parser.add_argument(
        "--dataset-id",
        help="Optional dataset identifier override when multiple datasets are declared in the config.",
    )
    parser.add_argument(
        "--gpus",
        type=int,
        default=1,
        help="GPU count for the single-node ResourceProfile (default: 1).",
    )
    parser.add_argument(
        "--cpus",
        type=int,
        default=8,
        help="CPU count for the single-node ResourceProfile (default: 8).",
    )
    parser.add_argument(
        "--run-id",
        help="Optional run identifier; if omitted a random id will be generated by ObservabilityTrace.",
    )
    parser.add_argument(
        "--concurrency",
        type=int,
        help="Override SampleLoop concurrency (otherwise auto-detected from GPU/CPU resources).",
    )
    parser.add_argument(
        "--model-path",
        help="Override VLLM native backend model path (sets VLLM_NATIVE_MODEL_PATH).",
    )
    parser.add_argument(
        "--output-dir",
        help="Optional directory for run artifacts; defaults to ./runs/<run_id>.",
    )
    report_pack_group = parser.add_mutually_exclusive_group()
    report_pack_group.add_argument(
        "--report-pack",
        action="store_true",
        help="Write the static report pack under <run_dir>/report_pack.",
    )
    report_pack_group.add_argument(
        "--no-report-pack",
        action="store_true",
        help="Disable static report pack generation for this run.",
    )
    parser.add_argument(
        "--max-samples",
        type=int,
        help="Override task max_samples (>=0) and sets env GAGE_EVAL_MAX_SAMPLES.",
    )
    parser.add_argument(
        "--metric-ids",
        help=(
            "Comma-separated metric_id allowlist. When set, only these metrics will be kept in the config "
            "(useful for toggling judge vs non-judge scoring in a single PipelineConfig)."
        ),
    )
    parser.add_argument(
        "--skip-judge",
        action="store_true",
        help="Skip the judge step by removing it from custom.steps at runtime.",
    )
    parser.add_argument(
        "--show-expanded-config",
        action="store_true",
        help="Print the smart-default-expanded PipelineConfig payload and exit.",
    )
    parser.add_argument(
        "--no-smart-defaults",
        action="store_true",
        help="Disable smart defaults when loading PipelineConfig YAML.",
    )
    parser.add_argument(
        "--backend-id",
        help="Override the static DUT backend used for inference binding.",
    )
    parser.add_argument(
        "--env-provider",
        choices=("local_process", "docker"),
        help="Override an AgentKit v2 environment provider.",
    )
    parser.add_argument(
        "--dut-id",
        help="Target --env-provider to the AgentKit v2 dut_agents entry with this dut_id.",
    )
    parser.add_argument(
        "--env-id",
        help="Target --env-provider to the AgentKit v2 environment referenced by this env_id.",
    )
    parser.add_argument(
        "--distill",
        "-d",
        action="store_true",
        help="Distill mode: validate config for builtin template generation (single-task first).",
    )
    parser.add_argument(
        "--force-merge",
        action="store_true",
        help="Allow multi-task configs in distill mode by creating a monolithic template.",
    )
    parser.add_argument(
        "--init",
        "-i",
        help="Init mode: generate RunConfig or PipelineConfig from a builtin template name or template path.",
    )
    parser.add_argument(
        "--init-mode",
        choices=("run-config", "pipeline-config"),
        default="run-config",
        help="Init output type (default: run-config).",
    )
    parser.add_argument(
        "--builtin-name",
        help="Builtin suite name when running distill/init flows (required for distill).",
    )
    parser.add_argument(
        "--version",
        help="Template version when distilling; defaults to V1.",
    )
    parser.add_argument(
        "--distill-output",
        help="Output directory for distill templates (default: config/builtin_templates/<name>/vN.yaml).",
    )
    args = parser.parse_args()

    if args.command and args.command != "init":
        parser.error(f"unrecognized command: {args.command}")
    if args.command == "init" and not args.command_arg:
        parser.error("init requires a benchmark kit id")
    if args.command == "init" and args.init:
        parser.error("positional init and --init cannot be used together.")
    if not args.command and not args.init and not args.config and not args.distill:
        parser.error("--config is required unless running in --init or --distill mode.")
    if args.distill and not args.config:
        parser.error("--distill requires --config to point to a PipelineConfig.")
    if args.init and args.distill:
        parser.error("--init and --distill cannot be used together.")
    return args


def _apply_cli_metric_filter(payload: dict, metric_ids_csv: str) -> None:
    """Filter payload metrics by metric_id (in-place)."""
    apply_cli_final_overrides(payload, CLIIntent(metric_ids=parse_metric_ids_csv(metric_ids_csv)))


def _apply_cli_skip_judge(payload: dict) -> None:
    """Remove judge steps from payload custom and task steps (in-place)."""
    apply_cli_final_overrides(payload, CLIIntent(skip_judge=True))


def _normalize_version_tag(version: str) -> str:
    tag = version.strip()
    tag = tag.upper()
    return tag if tag.startswith("V") else f"V{tag}"


def _ensure_spawn_start_method() -> None:
    """Use 'spawn' to avoid CUDA re-init issues in forked processes."""

    try:
        import torch.multiprocessing as mp  # type: ignore
    except Exception:  # pragma: no cover - torch not installed or unavailable
        import multiprocessing as mp  # type: ignore

    current = mp.get_start_method(allow_none=True)
    if current == "spawn":
        return
    try:
        mp.set_start_method("spawn", force=True)
    except RuntimeError as exc:  # pragma: no cover - defensive guard
        raise RuntimeError("gage-eval requires multiprocessing start method 'spawn' for CUDA workloads") from exc


def load_config(path: Path) -> dict:
    return _loader_expand_env(load_yaml_mapping(path))


def _expand_env(value):
    return _loader_expand_env(value)


def _build_cli_intent(args: argparse.Namespace) -> CLIIntent:
    return CLIIntent(
        backend_id=args.backend_id,
        max_samples=args.max_samples,
        skip_judge=args.skip_judge,
        metric_ids=parse_metric_ids_csv(args.metric_ids),
        env_provider=args.env_provider,
        dut_id=args.dut_id,
        env_id=args.env_id,
    )


class _LiteralDumper(yaml.SafeDumper):
    """YAML dumper that renders multi-line strings in literal style for readability.

    Anchors (&id001/*id001) are disabled to keep generated configs simple.
    """

    # NOTE: Disable YAML anchors to avoid &id001/*id001 in generated configs.
    def ignore_aliases(self, data):
        return True


def _str_representer(dumper: yaml.SafeDumper, data: str):
    if "\n" in data:
        return dumper.represent_scalar("tag:yaml.org,2002:str", data, style="|")
    return dumper.represent_scalar("tag:yaml.org,2002:str", data)


_LiteralDumper.add_representer(str, _str_representer)


def _yaml_dump(data: dict, *, sort_keys: bool = False) -> str:
    content = yaml.dump(
        data,
        Dumper=_LiteralDumper,
        sort_keys=sort_keys,
        allow_unicode=True,
        default_flow_style=False,
    )
    return _insert_top_level_spacing(content)


_EMPTY_EXPANDED_CONFIG_DISPLAY_SECTIONS = {
    "models",
    "agent_backends",
    "sandbox_profiles",
    "mcp_clients",
    "prompts",
    "summary_generators",
}


def _hide_empty_expanded_config_sections(payload: dict) -> dict:
    display_payload = dict(payload)
    for key in _EMPTY_EXPANDED_CONFIG_DISPLAY_SECTIONS:
        if display_payload.get(key) == []:
            display_payload.pop(key)
    return display_payload


def _write_yaml(data: dict, path: Path) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(_yaml_dump(data, sort_keys=False), encoding="utf-8")


def _dedupe_path(path: Path) -> Path:
    if not path.exists():
        return path
    stem = path.stem
    suffix = path.suffix
    counter = 1
    while True:
        candidate = path.with_name(f"{stem}_{counter}{suffix}")
        if not candidate.exists():
            return candidate
        counter += 1


def _insert_top_level_spacing(content: str) -> str:
    lines = content.splitlines()
    formatted: list[str] = []
    for idx, line in enumerate(lines):
        stripped = line.lstrip()
        is_top_level_key = bool(line) and not line[0].isspace() and not line.startswith("-")
        if is_top_level_key and formatted and formatted[-1] != "":
            formatted.append("")
        formatted.append(line)
    return "\n".join(formatted) + ("\n" if content.endswith("\n") else "")


def _inject_runtime_hints(yaml_text: str, run_cfg: dict) -> str:
    runtime = run_cfg.get("runtime") or {}
    params: list[str] = []
    dataset_ids = list((runtime.get("datasets") or {}).keys())
    backend_ids = list((runtime.get("backends") or {}).keys())
    task_ids = list((runtime.get("tasks") or {}).keys())
    if dataset_ids:
        params.append(f"runtime.datasets.{dataset_ids[0]}: {{...}}  # override the whole dataset block")
    if backend_ids:
        params.append(f"runtime.backends.{backend_ids[0]}: {{...}}  # switch endpoint/model/params")
    if task_ids:
        params.append(f"runtime.tasks.{task_ids[0]}.max_samples: 50  # limit samples for this run")
    if not params:
        return yaml_text
    lines = yaml_text.splitlines()
    output: list[str] = []
    for line in lines:
        if line.startswith("runtime:"):
            output.append("# runtime override examples (edit keys below as needed):")
            for hint in params:
                output.append(f"#   {hint}")
        output.append(line)
    return "\n".join(output)


def _allocate_numbered_path(prefix: str, directory: Path, *, start: int = 1, suffix: str = ".yaml") -> Path:
    directory.mkdir(parents=True, exist_ok=True)
    pattern = re.compile(rf"^{re.escape(prefix)}(?:_(\d+))?{re.escape(suffix)}$")
    max_seen = start - 1
    for path in directory.glob(f"{prefix}*{suffix}"):
        match = pattern.match(path.name)
        if not match:
            continue
        num_str = match.group(1)
        num = int(num_str) if num_str is not None else 0
        max_seen = max(max_seen, num)
    next_num = max_seen + 1
    filename = f"{prefix}_{next_num}{suffix}"
    return directory / filename


def _find_latest_version_path(template_dir: Path) -> Path:
    best_num = -1
    best_path: Optional[Path] = None
    for path in template_dir.glob("v*.yaml"):
        name = path.name.lower()
        if not name.startswith("v") or not name.endswith(".yaml"):
            continue
        num = name[1:-5]
        if num.isdigit():
            value = int(num)
            if value > best_num:
                best_num = value
                best_path = path
    if best_path is None:
        raise FileNotFoundError(f"No template versions found under {template_dir}")
    return best_path


def _resolve_template_path(init_value: str, version: Optional[str]) -> Path:
    candidate = Path(init_value).expanduser()
    if candidate.exists():
        return candidate

    name = init_value
    template_dir = REPO_ROOT / "config" / "builtin_templates" / name
    if not template_dir.exists():
        raise FileNotFoundError(f"Builtin template '{name}' not found under {template_dir}")
    if version:
        normalized = _normalize_version_tag(version)
        filename = f"{normalized.lower()}.yaml"
        target = template_dir / filename
        if not target.exists():
            raise FileNotFoundError(f"Template version '{normalized}' not found at {target}")
        return target
    return _find_latest_version_path(template_dir)


def _load_template(path: Path) -> dict:
    data = load_config(path)
    if (data.get("kind") or "").lower() != "builtintemplate":
        raise ValueError(f"Template at {path} must have kind=BuiltInTemplate/BuiltinTemplate")
    return data


def _build_run_config_payload(template: dict, *, name_hint: Optional[str] = None) -> dict:
    meta = template.get("metadata") or {}
    template_name = meta.get("name") or name_hint or "builtin_task"
    version = meta.get("version") or "V1"
    parameters = template.get("parameters") or []
    runtime_defaults = _parameters_to_runtime_defaults(parameters) if parameters else _extract_runtime_defaults(template.get("definition") or {})
    payload = {
        "api_version": template.get("api_version") or "gage/v1alpha1",
        "kind": "RunConfig",
        "metadata": {
            "name": f"{template_name}_run",
        },
        "base_task": f"builtin/{template_name}",
        "template_version": _normalize_version_tag(version),
        "template_digest": meta.get("digest"),
        "runtime": runtime_defaults,
    }
    return payload


def _lookup_runtime_value(runtime_params: dict, dotted_path: str):
    node: object = runtime_params
    for token in dotted_path.split("."):
        if not isinstance(node, dict) or token not in node:
            raise KeyError(f"runtime parameter '{dotted_path}' is missing")
        node = node[token]
    return node


def _apply_runtime_params(definition: object, runtime_params: dict) -> object:
    pattern = re.compile(r"\$\{runtime\.([A-Za-z0-9_\.]+)\}")

    def _apply(value: object) -> object:
        if isinstance(value, dict):
            return {k: _apply(v) for k, v in value.items()}
        if isinstance(value, list):
            return [_apply(v) for v in value]
        if isinstance(value, str):
            full_match = pattern.fullmatch(value)
            if full_match:
                key = full_match.group(1)
                return _lookup_runtime_value(runtime_params, key)

            def _repl(match: re.Match[str]) -> str:
                key = match.group(1)
                resolved = _lookup_runtime_value(runtime_params, key)
                return str(resolved)

            return pattern.sub(_repl, value)
        return value

    return _apply(definition)


def _apply_runtime_overrides(definition: dict, runtime_params: dict) -> dict:
    """Apply common runtime overrides even when templates lack placeholders."""

    rendered = copy.deepcopy(definition)

    runtime_datasets = runtime_params.get("datasets") or {}
    for ds in rendered.get("datasets") or []:
        ds_id = ds.get("dataset_id")
        if not ds_id:
            continue
        overrides = runtime_datasets.get(ds_id) or {}
        if not isinstance(overrides, dict) or not overrides:
            continue

        # NOTE: Backward compatibility: runtime.datasets.<id>.hub_limit only overrides hub_params.limit.
        hub_limit = None
        if isinstance(overrides.get("hub_params"), dict):
            hub_limit = overrides["hub_params"].get("limit")
        if hub_limit is None:
            hub_limit = overrides.get("hub_limit")
        if hub_limit is not None:
            hub_params = dict(ds.get("hub_params") or ds.get("hub_args") or {})
            hub_params["limit"] = hub_limit
            ds["hub_params"] = hub_params
            ds.pop("hub_args", None)

        # NOTE: New semantics: allow a full dataset config object in RunConfig and merge by keys.
        for key, value in overrides.items():
            if key in ("hub_params", "hub_args", "params"):
                base = ds.get(key) or {}
                if isinstance(base, dict) and isinstance(value, dict):
                    merged = dict(base)
                    merged.update(copy.deepcopy(value))
                    ds[key] = merged
                else:
                    ds[key] = copy.deepcopy(value)
            elif key in ("hub_limit",):
                # hub_limit -> hub_params.limit mapping handled above; skip here.
                continue
            else:
                if value is not None:
                    ds[key] = copy.deepcopy(value)

    runtime_backends = runtime_params.get("backends") or {}
    for backend in rendered.get("backends") or []:
        backend_id = backend.get("backend_id")
        if not backend_id:
            continue
        overrides = runtime_backends.get(backend_id) or {}
        if not isinstance(overrides, dict) or not overrides:
            continue

        # NOTE: Allow overriding backend.type.
        if "type" in overrides and overrides["type"] is not None:
            backend["type"] = overrides["type"]

        cfg = backend.setdefault("config", {})

        # NOTE: New semantics: if runtime.backends.<id> provides a config object, deep-merge it into backend.config.
        override_cfg = overrides.get("config") or {}
        if isinstance(override_cfg, dict):
            for key, value in override_cfg.items():
                if isinstance(value, dict) and isinstance(cfg.get(key), dict):
                    merged = dict(cfg[key])
                    merged.update(copy.deepcopy(value))
                    cfg[key] = merged
                else:
                    cfg[key] = copy.deepcopy(value)

        # NOTE: Backward compatibility: still support base_url/model/async_max_concurrency/pool_size at top level.
        for key in ("base_url", "model", "async_max_concurrency"):
            if key in overrides and overrides[key] is not None:
                cfg[key] = overrides[key]
        if "pool_size" in overrides and overrides["pool_size"] is not None:
            resource = cfg.setdefault("resource_requirement", {})
            resource["pool_size"] = overrides["pool_size"]

    runtime_tasks = runtime_params.get("tasks") or {}
    for task in rendered.get("tasks") or []:
        task_id = task.get("task_id")
        if not task_id:
            continue
        overrides = runtime_tasks.get(task_id) or {}
        for key in ("max_samples", "concurrency", "shuffle", "shuffle_seed", "prefetch_factor", "max_inflight"):
            if key in overrides and overrides[key] is not None:
                task[key] = overrides[key]

    output_path = (runtime_params.get("global") or {}).get("output_path")
    if output_path:
        for task in rendered.get("tasks") or []:
            reporting = task.get("reporting") or {}
            sinks = reporting.get("sinks") or []
            for sink in sinks:
                if sink.get("type") != "file":
                    continue
                params = sink.get("params") or {}
                params["output_path"] = output_path
                sink["params"] = params
            if sinks:
                reporting["sinks"] = sinks
                task["reporting"] = reporting

    return rendered


def _validate_run_config_payload(payload: dict) -> None:
    allowed_keys = {"api_version", "kind", "metadata", "base_task", "template_version", "template_digest", "runtime"}
    logic_keys = {
        "datasets",
        "backends",
        "role_adapters",
        "metrics",
        "tasks",
        "custom",
        "builtin",
        "models",
        "prompts",
        "observability",
    }
    unexpected = set(payload.keys()) - allowed_keys
    logic_conflict = unexpected & logic_keys
    if logic_conflict:
        raise ValueError(f"RunConfig must not contain pipeline logic fields: {', '.join(sorted(logic_conflict))}")
    if unexpected:
        raise ValueError(f"RunConfig contains unsupported fields: {', '.join(sorted(unexpected))}")
    if not payload.get("base_task"):
        raise ValueError("RunConfig missing required field: base_task")
    if not payload.get("template_version"):
        raise ValueError("RunConfig missing required field: template_version")


def _compile_run_config(run_cfg: dict, *, template_payload: Optional[dict] = None) -> tuple[dict, Path]:
    _validate_run_config_payload(run_cfg)
    base_task: str = run_cfg["base_task"]
    version = _normalize_version_tag(str(run_cfg["template_version"]))
    template_path: Path
    if template_payload is not None:
        template_path = Path(template_payload.get("_source_path") or "")
    else:
        if base_task.startswith("builtin/"):
            template_path = _resolve_template_path(base_task.replace("builtin/", "", 1), version)
        else:
            template_path = Path(base_task).expanduser()
        template_payload = _load_template(template_path)

    meta = template_payload.get("metadata") or {}
    template_version = _normalize_version_tag(str(meta.get("version") or version))
    if template_version != version:
        raise ValueError(
            f"RunConfig template_version={version} does not match template version={template_version} "
            f"from {template_path}"
        )
    expected_digest = meta.get("digest")
    run_digest = run_cfg.get("template_digest")
    if run_digest and expected_digest and run_digest != expected_digest:
        raise ValueError(
            f"RunConfig template_digest mismatch: expected {expected_digest}, got {run_digest} (template: {template_path})"
        )

    definition = copy.deepcopy(template_payload.get("definition") or {})
    runtime_params = run_cfg.get("runtime") or {}
    rendered = _apply_runtime_params(definition, runtime_params)
    rendered = _apply_runtime_overrides(rendered, runtime_params)
    pipeline_payload = {"api_version": "gage/v1alpha1", "kind": "PipelineConfig"}
    pipeline_payload.update(rendered)
    return pipeline_payload, template_path


def _prune_empty_sections(payload: dict, *, keys: set[str]) -> dict:
    trimmed = dict(payload)
    for key in keys:
        val = trimmed.get(key)
        if val in (None, {}, []):
            trimmed.pop(key, None)
    return trimmed


def _extract_runtime_defaults(definition: dict) -> dict:
    task_by_dataset: dict[str, dict] = {}
    for task in definition.get("tasks") or []:
        ds_id = task.get("dataset_id")
        if ds_id:
            task_by_dataset.setdefault(ds_id, {})
            if "max_samples" in task:
                task_by_dataset[ds_id]["max_samples"] = task.get("max_samples")

    runtime = {
        "datasets": {},
        "backends": {},
        "tasks": {},
        "global": {},
    }
    for ds in definition.get("datasets") or []:
        ds_id = ds.get("dataset_id")
        if not ds_id:
            continue
        ds_runtime: dict = {}
        ds_runtime_hub_params: dict = {}
        hub_params = ds.get("hub_params") or {}
        if "limit" in hub_params:
            ds_runtime_hub_params["limit"] = hub_params.get("limit")
        elif task_by_dataset.get(ds_id, {}).get("max_samples") is not None:
            ds_runtime_hub_params["limit"] = task_by_dataset[ds_id]["max_samples"]
        if ds_runtime_hub_params:
            ds_runtime["hub_params"] = ds_runtime_hub_params
        runtime["datasets"][ds_id] = ds_runtime
    for backend in definition.get("backends") or []:
        backend_id = backend.get("backend_id")
        if not backend_id:
            continue
        cfg = backend.get("config") or {}
        b_runtime: dict = {}
        for key in ("base_url", "model", "async_max_concurrency"):
            if key in cfg:
                b_runtime[key] = cfg.get(key)
        resource = cfg.get("resource_requirement") or {}
        if "pool_size" in resource:
            b_runtime["pool_size"] = resource.get("pool_size")
        runtime["backends"][backend_id] = b_runtime
    for task in definition.get("tasks") or []:
        task_id = task.get("task_id")
        if not task_id:
            continue
        t_runtime: dict = {}
        for key in ("concurrency", "max_samples"):
            if key in task:
                t_runtime[key] = task.get(key)
        runtime["tasks"][task_id] = t_runtime
        reporting = task.get("reporting") or {}
        sinks = reporting.get("sinks") or []
        for sink in sinks:
            if sink.get("type") == "file":
                params = sink.get("params") or {}
                if "output_path" in params:
                    runtime["global"]["output_path"] = params.get("output_path")
                    break
    # Ensure keys exist even if empty
    for key in ("datasets", "backends", "tasks", "global"):
        runtime.setdefault(key, {})
    return runtime


def _parameters_to_runtime_defaults(parameters: list[dict]) -> dict:
    runtime: dict = {"datasets": {}, "backends": {}, "tasks": {}, "global": {}}
    for param in parameters:
        name = param.get("name")
        default = param.get("default")
        if not isinstance(name, str) or not name.startswith("runtime."):
            continue
        path = name[len("runtime.") :].split(".")
        cursor = runtime
        for token in path[:-1]:
            if token not in cursor or not isinstance(cursor[token], dict):
                cursor[token] = {}
            cursor = cursor[token]
        cursor[path[-1]] = default
    for key in ("datasets", "backends", "tasks", "global"):
        runtime.setdefault(key, {})
    return runtime


def _summarize_template_comment(template: dict) -> str:
    meta = template.get("metadata") or {}
    definition = template.get("definition") or {}
    lines = [
        "# =================================================================",
        "# Read-only template snapshot (editing this block does not affect the template definition)",
        f"# Template: {meta.get('name', '<unknown>')} version={meta.get('version', '<unknown>')} monolithic={meta.get('monolithic', False)}",
        "# -----------------------------------------------------------------",
    ]
    datasets = definition.get("datasets") or []
    if datasets:
        lines.append("# Datasets:")
        hint_added = False
        for ds in datasets:
            ds_id = ds.get("dataset_id")
            loader = ds.get("loader") or ds.get("hub") or "<loader>"
            hub_params = ds.get("hub_params") or {}
            params = ds.get("params") or {}
            preprocess = params.get("preprocess")
            doc_to_visual = params.get("doc_to_visual")
            lines.append(f"#   - dataset_id: {ds_id}")
            lines.append(f"#     loader: {loader}")
            if hub_params:
                hub_id = hub_params.get("hub_id")
                split = hub_params.get("split")
                subset = hub_params.get("subset")
                limit = hub_params.get("limit")
                lines.append(f"#     hub_params:")
                if hub_id:
                    lines.append(f"#       hub_id: {hub_id}")
                if split:
                    lines.append(f"#       split: {split}")
                if subset:
                    lines.append(f"#       subset: {subset}")
                if limit is not None:
                    lines.append(f"#       limit: {limit}")
            if preprocess:
                lines.append(f"#     preprocess: {preprocess}")
            if doc_to_visual:
                lines.append(f"#     doc_to_visual: {doc_to_visual}")
            if not hint_added:
                lines.append(
                    "#     Tip: override runtime.datasets.<id> to set hub_params.limit/local_path, etc. for local test data."
                )
                hint_added = True
    backends = definition.get("backends") or []
    if backends:
        lines.append("# Backends:")
        for backend in backends:
            backend_id = backend.get("backend_id")
            b_type = backend.get("type")
            cfg = backend.get("config") or {}
            base_url = cfg.get("base_url") or cfg.get("endpoint")
            model = cfg.get("model")
            async_max = cfg.get("async_max_concurrency")
            pool_size = (cfg.get("resource_requirement") or {}).get("pool_size")
            lines.append(f"#   - backend_id: {backend_id}")
            lines.append(f"#     type: {b_type}")
            if model:
                lines.append(f"#     model: {model}")
            if base_url:
                lines.append(f"#     base_url: {base_url}")
            if async_max is not None:
                lines.append(f"#     async_max_concurrency: {async_max}")
            if pool_size is not None:
                lines.append(f"#     pool_size: {pool_size}")
    roles = {r.get("adapter_id"): r for r in (definition.get("role_adapters") or []) if r.get("adapter_id")}
    steps = (definition.get("custom") or {}).get("steps") or []
    if steps:
        lines.append("# Steps:")
        for step in steps:
            lines.append(f"#   - step: {step.get('step')}")
            lines.append(f"#     adapter_id: {step.get('adapter_id') or '<auto>'}")
    prompts = definition.get("prompts") or []
    if prompts:
        # Build a prompt_id -> adapter_id list mapping.
        prompt_usage: dict[str, list[str]] = {}
        for adapter_id, role in roles.items():
            prompt_id = role.get("prompt_id")
            if not prompt_id:
                continue
            prompt_usage.setdefault(str(prompt_id), []).append(str(adapter_id))

        lines.append("# Prompts (read-only snapshot; edit via PipelineConfig):")
        lines.append(
            "#   Tip: Prompts are part of evaluation logic. To edit them, use "
            "run.py --init <name> --init-mode pipeline-config to generate a PipelineConfig, then edit the prompts section."
        )
        for prompt in prompts:
            prompt_id = prompt.get("prompt_id") or "<unknown>"
            renderer = prompt.get("renderer") or "<renderer>"
            params = prompt.get("params") or {}
            used_by = ", ".join(sorted(prompt_usage.get(prompt_id, []))) or "<unbound>"
            lines.append(f"#   - prompt_id: {prompt_id}")
            lines.append(f"#     renderer: {renderer}")
            lines.append(f"#     used_by_adapters: {used_by}")
            if isinstance(params, dict) and params:
                lines.append("#     params:")
                for key, value in params.items():
                    text = str(value)
                    if "\n" in text:
                        lines.append(f"#       {key}: |")
                        for sub in text.splitlines():
                            lines.append(f"#         {sub}")
                    else:
                        lines.append(f"#       {key}: {text}")
            template_text = prompt.get("template")
            if isinstance(template_text, str) and template_text.strip():
                lines.append("#     template: |")
                for line in template_text.splitlines():
                    lines.append(f"#       {line}")

    metrics = definition.get("metrics") or []
    if metrics:
        lines.append("# Metrics:")
        for metric in metrics:
            lines.append(f"#   - metric_id: {metric.get('metric_id')}")
            lines.append(f"#     implementation: {metric.get('implementation')}")
    lines.append("# =================================================================")
    return "\n".join(lines)


def _write_run_config_with_comments(run_cfg: dict, template_payload: dict, path: Path) -> None:
    comment_block = _summarize_template_comment(template_payload)
    yaml_block = _insert_top_level_spacing(yaml.safe_dump(run_cfg, sort_keys=False, allow_unicode=True))
    yaml_block = _inject_runtime_hints(yaml_block, run_cfg)
    content = f"{comment_block}\n{yaml_block}"
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(content, encoding="utf-8")


def _handle_init_mode(args: argparse.Namespace) -> None:
    template_path = _resolve_template_path(args.init, args.version)
    template_payload = _load_template(template_path)
    template_payload["_source_path"] = str(template_path)
    template_name = template_payload.get("metadata", {}).get("name") or Path(args.init).stem
    version = _normalize_version_tag(template_payload.get("metadata", {}).get("version") or (args.version or "V1"))
    run_cfg = _build_run_config_payload(template_payload, name_hint=template_name)

    if args.init_mode == "pipeline-config":
        compiled, _ = _compile_run_config(run_cfg, template_payload=template_payload)
        output_dir = REPO_ROOT / "config" / "custom"
        compiled = _prune_empty_sections(
            compiled,
            keys={"models", "prompts", "parameters", "role_adapters", "observability", "builtin"},
        )
        output = _allocate_numbered_path(f"{template_name}_from_{version}", output_dir, start=1)
        _write_yaml(compiled, output)
        print(f"[gage-eval][init] mode=pipeline-config base_task=builtin/{template_name} template_version={version}")
        print(f"[gage-eval][init] pipeline config written to {output}")
        return

    output_dir = REPO_ROOT / "config" / "run_configs"
    output = _allocate_numbered_path(f"{template_name}_run", output_dir, start=1)
    _write_run_config_with_comments(run_cfg, template_payload, output)
    print(f"[gage-eval][init] mode=run-config base_task=builtin/{template_name} template_version={version}")
    print(f"[gage-eval][init] run config written to {output}")


def _handle_agentkit_v2_init(kit_id: str) -> None:
    payload = build_agentkit_v2_init_payload(kit_id)
    print(_yaml_dump(payload, sort_keys=False), end="")


def ensure_save_dir(directory: Optional[str]) -> None:
    if not directory:
        return
    resolved = Path(directory).expanduser().resolve()
    resolved.mkdir(parents=True, exist_ok=True)
    os.environ["GAGE_EVAL_SAVE_DIR"] = str(resolved)


def _detect_gpu_count() -> Optional[int]:
    """Best-effort GPU count detection covering torch/nvidia-smi/env hints."""

    torch = sys.modules.get("torch")
    if torch is not None:
        if torch.cuda.is_available():
            count = torch.cuda.device_count()
            if count:
                return count

    env_visible = os.environ.get("CUDA_VISIBLE_DEVICES")
    if env_visible:
        tokens = [token.strip() for token in env_visible.split(",") if token.strip() not in {"", "-1"}]
        if tokens:
            return len(tokens)

    env_world = os.environ.get("WORLD_SIZE")
    if env_world and env_world.isdigit():
        return int(env_world)

    for env_name in ("SLURM_GPUS_ON_NODE", "SLURM_JOB_GPUS"):
        env_value = os.environ.get(env_name)
        if env_value and env_value.isdigit():
            return int(env_value)

    try:
        import subprocess

        result = subprocess.run(
            ["nvidia-smi", "--query-gpu=count", "--format=csv,noheader"],
            check=False,
            capture_output=True,
            text=True,
        )
        if result.returncode == 0:
            stripped = result.stdout.strip()
            if stripped.isdigit():
                return int(stripped)
    except FileNotFoundError:
        pass
    return None


def _ensure_default_concurrency(args: argparse.Namespace) -> None:
    if args.concurrency and args.concurrency > 0:
        os.environ["GAGE_EVAL_THREADS"] = str(args.concurrency)
        print(f"[gage-eval] concurrency set from CLI: {args.concurrency}")
        return
    if os.environ.get("GAGE_EVAL_THREADS"):
        return
    auto = _detect_gpu_count()
    if auto:
        os.environ["GAGE_EVAL_THREADS"] = str(max(1, auto))
        print(f"[gage-eval] auto-detected concurrency from GPU count: {auto}")
        return
    fallback = min(os.cpu_count() or 1, 4)
    os.environ["GAGE_EVAL_THREADS"] = str(fallback)
    print(f"[gage-eval] concurrency fallback to CPU-based default: {fallback}")


def _preflight_checks() -> None:
    """Runs best-effort environment health checks.

    The goal is to reduce avoidable startup failures caused by dirty GPU memory
    or stale background processes.
    """

    try:
        import pynvml  # type: ignore

        pynvml.nvmlInit()
        device_count = pynvml.nvmlDeviceGetCount()
        for idx in range(device_count):
            handle = pynvml.nvmlDeviceGetHandleByIndex(idx)
            mem = pynvml.nvmlDeviceGetMemoryInfo(handle)
            used_ratio = mem.used / mem.total if mem.total else 0.0
            if used_ratio > 0.1:
                procs = pynvml.nvmlDeviceGetComputeRunningProcesses(handle)
                if not procs:
                    print(
                        f"[gage-eval][CRITICAL] GPU{idx} mem used {mem.used/1024/1024:.0f}MiB "
                        f"({used_ratio:.0%}) but no compute processes detected. "
                        "Environment may be dirty; consider reboot/kill stale drivers."
                    )
        pynvml.nvmlShutdown()
    except Exception:
        # NOTE: Ignore if NVML is unavailable.
        pass

    try:
        import psutil  # type: ignore

        me = psutil.Process()
        conflicts = []
        for proc in psutil.process_iter(["pid", "name", "cmdline"]):
            if proc.pid == me.pid:
                continue
            cmd = " ".join(proc.info.get("cmdline") or [])
            name = proc.info.get("name") or ""
            if "vllm" in cmd.lower() or "gage_eval" in cmd.lower() or "python -m gage_eval" in cmd.lower():
                conflicts.append((proc.pid, name or cmd))
        if conflicts:
            print("[gage-eval][WARNING] Potential conflicting processes detected:")
            for pid, cmd in conflicts:
                print(f"  - PID {pid}: {cmd}")
    except Exception:
        pass


_RUNTIME_REF: Optional[object] = None
_SHUTDOWN_SIGNAL_RECEIVED = False


def _install_signal_handlers():
    global _SHUTDOWN_SIGNAL_RECEIVED
    _SHUTDOWN_SIGNAL_RECEIVED = False

    def _handler(signum, frame):
        del frame
        global _RUNTIME_REF, _SHUTDOWN_SIGNAL_RECEIVED
        exit_code = 128 + int(signum)
        if _SHUTDOWN_SIGNAL_RECEIVED:
            print(f"[gage-eval] Caught signal {signum} again, forcing exit…", file=sys.stderr)
            _kill_child_processes()
            os._exit(exit_code)
        _SHUTDOWN_SIGNAL_RECEIVED = True
        print(f"[gage-eval] Caught signal {signum}, shutting down…", file=sys.stderr)
        try:
            if _RUNTIME_REF and hasattr(_RUNTIME_REF, "shutdown"):
                _RUNTIME_REF.shutdown()
        except Exception as exc:
            print(f"[gage-eval] shutdown hook failed after signal {signum}: {exc}", file=sys.stderr)
        raise SystemExit(exit_code)

    for sig in (signal.SIGINT, signal.SIGTERM):
        try:
            signal.signal(sig, _handler)
        except Exception:
            pass


def _kill_child_processes():
    try:
        import psutil  # type: ignore

        me = psutil.Process()
        for child in me.children(recursive=True):
            try:
                child.kill()
            except Exception:
                pass
    except Exception:
        pass


def _detect_hardware_profile() -> Optional[str]:
    """Detects a best-effort hardware profile.

    Returns:
        One of: `'h100'`, `'apple'`, or `None` if no known profile is detected.
    """

    try:
        import subprocess

        result = subprocess.run(
            ["nvidia-smi", "--query-gpu=name", "--format=csv,noheader"],
            check=False,
            capture_output=True,
            text=True,
        )
        if result.returncode == 0:
            names = [line.strip().lower() for line in result.stdout.splitlines() if line.strip()]
            if any("h100" in n for n in names):
                return "h100"
    except FileNotFoundError:
        pass
    sys_platform = platform.system().lower()
    machine = platform.machine().lower()
    if sys_platform == "darwin" and "arm" in machine:
        return "apple"
    return None


def _apply_hardware_profile_env(profile: Optional[str]) -> None:
    """Applies conservative runtime defaults based on hardware profile.

    The environment is only modified when the user did not explicitly set the
    corresponding knobs (for example: `GAGE_EVAL_THREADS`).
    """

    if profile == "h100":
        if not os.environ.get("GAGE_EVAL_THREADS"):
            os.environ["GAGE_EVAL_THREADS"] = "32"
            print("[gage-eval] hardware profile=h100 -> set GAGE_EVAL_THREADS=32 (override with --concurrency)")
    elif profile == "apple":
        if not os.environ.get("GAGE_EVAL_THREADS"):
            os.environ["GAGE_EVAL_THREADS"] = "4"
            print("[gage-eval] hardware profile=apple -> set GAGE_EVAL_THREADS=4 (override with --concurrency)")


def _apply_cli_max_samples_override(payload: dict, max_samples: int) -> None:
    """Override task/dataset sample limits in a PipelineConfig payload (in-place)."""

    for task in payload.get("tasks") or []:
        if not isinstance(task, dict):
            continue
        task["max_samples"] = max_samples
    # NOTE: Only push the limit down into the dataset loader when the dataset has
    # no preprocess/bundle stage that might filter records. Otherwise a tiny
    # loader-side limit (for example: 1) can be exhausted by filtered records
    # before the first valid sample is reached.
    for dataset in payload.get("datasets") or []:
        if not isinstance(dataset, dict):
            continue
        params = dataset.get("params")
        has_record_filter_stage = False
        if isinstance(params, dict):
            has_record_filter_stage = bool(params.get("preprocess") or params.get("bundle"))
        if isinstance(params, dict) and not has_record_filter_stage:
            params["limit"] = max_samples
        hub_params = dataset.get("hub_params")
        if isinstance(hub_params, dict) and not has_record_filter_stage:
            hub_params["limit"] = max_samples


def _validate_config_wiring(
    config: PipelineConfig,
    *,
    registry=None,
    run_id: str = "validate",
) -> list[str]:
    """Validate config references without loading data or running the pipeline."""

    from gage_eval.registry import registry as global_asset_registry

    runtime_registry_context = None
    asset_registry = global_asset_registry
    if registry is not None:
        try:
            runtime_registry_context = registry.prepare_runtime_registry_context(
                config,
                run_id=run_id,
            )
            asset_registry = runtime_registry_context.view
        except Exception as exc:
            return [f"Registry discovery failed: {exc}"]

    try:
        return _validate_config_wiring_with_registry(config, asset_registry)
    finally:
        if runtime_registry_context is not None:
            runtime_registry_context.close()


def _validate_config_wiring_with_registry(config: PipelineConfig, asset_registry) -> list[str]:
    import importlib

    from gage_eval.metrics.registry import MetricRegistry

    errors: list[str] = []
    dataset_ids = {spec.dataset_id for spec in config.datasets}
    backend_ids = {spec.backend_id for spec in config.backends}
    prompt_ids = {spec.prompt_id for spec in config.prompts}

    for spec in config.datasets:
        try:
            asset_registry.get("dataset_loaders", spec.loader)
        except KeyError:
            errors.append(f"Dataset '{spec.dataset_id}' references unknown loader '{spec.loader}'")
        if spec.hub:
            try:
                asset_registry.get("dataset_hubs", spec.hub)
            except KeyError:
                errors.append(f"Dataset '{spec.dataset_id}' references unknown hub '{spec.hub}'")
        elif spec.loader in {"hf_hub", "modelscope"}:
            errors.append(f"Dataset '{spec.dataset_id}' using loader '{spec.loader}' requires field 'hub'")

        preprocess = spec.params.get("preprocess")
        if preprocess:
            try:
                asset_registry.get("dataset_preprocessors", str(preprocess))
            except KeyError:
                errors.append(f"Dataset '{spec.dataset_id}' references unknown preprocessor '{preprocess}'")

    for spec in config.backends:
        try:
            asset_registry.get("backends", spec.type)
        except KeyError:
            errors.append(f"Backend '{spec.backend_id}' references unknown type '{spec.type}'")

    for spec in config.role_adapters:
        if spec.class_path:
            try:
                module_name, class_name = _split_import_class_path(spec.class_path)
                module = importlib.import_module(module_name)
                getattr(module, class_name)
            except Exception as exc:
                errors.append(f"RoleAdapter '{spec.adapter_id}' invalid class_path '{spec.class_path}': {exc}")
        else:
            try:
                asset_registry.get("roles", spec.role_type)
            except KeyError:
                errors.append(f"RoleAdapter '{spec.adapter_id}' references unknown role_type '{spec.role_type}'")
        if spec.backend_id and spec.backend_id not in backend_ids:
            errors.append(f"RoleAdapter '{spec.adapter_id}' references missing backend_id '{spec.backend_id}'")
        if spec.prompt_id and spec.prompt_id not in prompt_ids:
            errors.append(f"RoleAdapter '{spec.adapter_id}' references missing prompt_id '{spec.prompt_id}'")

    metric_registry = MetricRegistry(registry_view=asset_registry)
    for spec in config.metrics:
        try:
            metric_registry.build_metric(spec)
        except Exception as exc:
            errors.append(f"Metric '{spec.metric_id}' failed to build implementation '{spec.implementation}': {exc}")

    for spec in config.tasks:
        if spec.dataset_id not in dataset_ids:
            errors.append(f"Task '{spec.task_id}' references missing dataset_id '{spec.dataset_id}'")

    return errors


def _split_import_class_path(class_path: str) -> tuple[str, str]:
    if ":" in class_path:
        module_name, class_name = class_path.split(":", 1)
    else:
        module_name, class_name = class_path.rsplit(".", 1)
    if not module_name or not class_name:
        raise ValueError(f"Invalid class_path '{class_path}'")
    return module_name, class_name


def main() -> None:
    wall_clock_start = time.perf_counter()
    args = parse_args()
    cli_intent = _build_cli_intent(args)

    if args.init:
        try:
            _handle_init_mode(args)
        except Exception as exc:
            print(f"[gage-eval][init] {exc}", file=sys.stderr)
            sys.exit(1)
        sys.exit(0)

    if args.command == "init":
        try:
            _handle_agentkit_v2_init(args.command_arg)
        except Exception as exc:
            print(f"[gage-eval][init] {exc}", file=sys.stderr)
            sys.exit(1)
        sys.exit(0)

    if args.show_expanded_config:
        try:
            config_path = Path(args.config).expanduser()
            if args.no_smart_defaults:
                payload = load_pre_smart_defaults_payload(
                    config_path,
                    run_config_compiler=_compile_run_config,
                    cli_intent=cli_intent,
                )
            else:
                payload = load_pipeline_config_payload(
                    config_path,
                    cli_intent=cli_intent,
                    smart_defaults=True,
                    run_config_compiler=_compile_run_config,
                )
        except Exception as exc:
            print(f"[gage-eval] failed to expand config: {exc}", file=sys.stderr)
            sys.exit(1)
        print(_yaml_dump(_hide_empty_expanded_config_sections(payload), sort_keys=False))
        sys.exit(0)

    if args.distill:
        builtin_name = args.builtin_name
        try:
            config_payload = load_pipeline_config_payload(
                Path(args.config).expanduser(),
                cli_intent=cli_intent,
                smart_defaults=not args.no_smart_defaults,
                run_config_compiler=_compile_run_config,
            )
        except Exception as exc:
            print(f"[gage-eval][distill] {exc}", file=sys.stderr)
            sys.exit(1)
        if not builtin_name and args.distill_output:
            # NOTE: If distill_output is provided without builtin_name, use the last
            # path component as the template name.
            builtin_name = Path(args.distill_output).name
        try:
            analysis = analyze_tasks_for_distill(config_payload, force_merge=args.force_merge)
        except DistillError as exc:
            context = getattr(exc, "context", None)
            if context:
                task_info = ", ".join(context.task_ids) if context.task_ids else "<none>"
                print(
                    f"[gage-eval][distill] mode={context.mode} tasks={task_info}",
                    file=sys.stderr,
                )
            print(f"[gage-eval][distill] {exc}", file=sys.stderr)
            sys.exit(1)

        task_info = ", ".join(analysis.task_ids) if analysis.task_ids else "<none>"
        print(f"[gage-eval][distill] mode={analysis.mode} tasks={task_info}")
        if not builtin_name:
            print("[gage-eval][distill] please provide --builtin-name or set --distill-output to derive a name", file=sys.stderr)
            sys.exit(1)
        from gage_eval.tools.distill import distill_to_template

        version = args.version or None
        output_root = (
            Path(args.distill_output).expanduser()
            if args.distill_output
            else REPO_ROOT / "config" / "builtin_templates"
        )
        try:
            target = distill_to_template(
                config_payload,
                builtin_name=builtin_name,
                version=version,
                output_root=output_root,
                force_merge=args.force_merge,
            )
        except DistillError as exc:
            print(f"[gage-eval][distill] {exc}", file=sys.stderr)
            sys.exit(1)
        print(f"[gage-eval][distill] template written to {target}")
        sys.exit(0)

    if not args.config:
        print("[gage-eval] --config is required for run mode (omit only with --init/--distill)", file=sys.stderr)
        sys.exit(2)

    if args.max_samples is not None and args.max_samples < 0:
        print("[gage-eval] --max-samples must be >= 0", file=sys.stderr)
        sys.exit(2)

    config_path = Path(args.config).expanduser()
    try:
        raw_run_payload = load_yaml_mapping(config_path)
    except Exception as exc:
        print(f"[gage-eval] failed to load config: {exc}", file=sys.stderr)
        sys.exit(1)
    if sys.stdin.isatty() and os.environ.get("GAGE_EVAL_HUMAN_INPUT") is None:
        os.environ["GAGE_EVAL_HUMAN_INPUT"] = "stdin"

    _ensure_spawn_start_method()
    profile_hint = _detect_hardware_profile()
    _ensure_default_concurrency(args)
    _apply_hardware_profile_env(profile_hint)
    _preflight_checks()
    if args.max_samples is not None:
        os.environ["GAGE_EVAL_MAX_SAMPLES"] = str(args.max_samples)
    if args.model_path:
        os.environ["VLLM_NATIVE_MODEL_PATH"] = args.model_path
    if args.report_pack:
        os.environ["GAGE_EVAL_REPORT_PACK"] = "1"
        os.environ["GAGE_EVAL_REPORT_PACK_SOURCE"] = "cli"
    elif args.no_report_pack:
        os.environ["GAGE_EVAL_REPORT_PACK"] = "0"
        os.environ["GAGE_EVAL_REPORT_PACK_SOURCE"] = "cli"
    _install_signal_handlers()
    config_source_desc = args.config
    try:
        materialized = load_pipeline_config_payload(
            Path(args.config).expanduser(),
            cli_intent=cli_intent,
            smart_defaults=not args.no_smart_defaults,
            run_config_compiler=_compile_run_config,
        )
    except Exception as exc:
        print(f"[gage-eval] failed to load config: {exc}", file=sys.stderr)
        sys.exit(1)
    config = PipelineConfig.from_dict(materialized)

    registry = build_default_registry()

    if args.max_samples == 0:
        errors = _validate_config_wiring(
            config,
            registry=registry,
            run_id=args.run_id or f"validate-{int(time.time() * 1000)}",
        )
        if errors:
            for err in errors:
                print(f"[gage-eval][validate] {err}", file=sys.stderr)
            sys.exit(1)
        print("[gage-eval] --max-samples 0: config wiring validated, skip execution.")
        sys.exit(0)

    ensure_save_dir(args.output_dir)
    profile = ResourceProfile(nodes=[NodeResource(node_id="local", gpus=args.gpus, cpus=args.cpus)])
    trace = ObservabilityTrace(run_id=args.run_id)

    print(f"[gage-eval] building runtime from {config_source_desc}")
    runtime = build_runtime(
        config=config,
        registry=registry,
        resource_profile=profile,
        dataset_id=args.dataset_id,
        trace=trace,
    )
    global _RUNTIME_REF
    _RUNTIME_REF = runtime
    if hasattr(runtime, "set_wall_clock_start"):
        try:
            runtime.set_wall_clock_start(wall_clock_start)  # type: ignore[call-arg]
        except Exception:
            pass
    print("[gage-eval] runtime ready, starting execution…")
    runtime.run()
    print("[gage-eval] run completed successfully.")


if __name__ == "__main__":
    main()