from __future__ import annotations

import json
import os
import tomllib
from dataclasses import asdict, dataclass, field, is_dataclass
from pathlib import Path
from typing import Any, Mapping


class SpecError(ValueError):
    """Raised when a structured spec is invalid."""


def _require_mapping(value: Any, *, context: str) -> Mapping[str, Any]:
    if not isinstance(value, Mapping):
        raise SpecError(f"{context} must be an object.")
    return value


def _require_str(value: Any, *, context: str) -> str:
    if not isinstance(value, str) or not value.strip():
        raise SpecError(f"{context} must be a non-empty string.")
    return value.strip()


def _require_float(value: Any, *, context: str) -> float:
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        raise SpecError(f"{context} must be numeric.")
    return float(value)


def _require_int(value: Any, *, context: str) -> int:
    if isinstance(value, bool) or not isinstance(value, int):
        raise SpecError(f"{context} must be an integer.")
    return value


def _require_bool(value: Any, *, context: str) -> bool:
    if not isinstance(value, bool):
        raise SpecError(f"{context} must be a boolean.")
    return value


def _coerce_str_map(value: Any, *, context: str) -> dict[str, str]:
    mapping = _require_mapping(value or {}, context=context)
    return {str(key): str(item) for key, item in mapping.items()}


def _coerce_any_map(value: Any, *, context: str) -> dict[str, Any]:
    mapping = _require_mapping(value or {}, context=context)
    return {str(key): item for key, item in mapping.items()}


def _coerce_str_list(value: Any, *, context: str) -> list[str]:
    if value is None:
        return []
    if not isinstance(value, list):
        raise SpecError(f"{context} must be a list.")
    result: list[str] = []
    for item in value:
        result.append(_require_str(item, context=context))
    return result


def _coerce_text(value: Any, *, context: str) -> str:
    if isinstance(value, str) and value.strip():
        return value.strip()
    if isinstance(value, Mapping) or isinstance(value, list):
        text = json.dumps(value, ensure_ascii=False, sort_keys=True)
        if text:
            return text
    raise SpecError(f"{context} must be a non-empty string.")


def _coerce_int_list(value: Any, *, context: str) -> list[int]:
    if value is None:
        return []
    if not isinstance(value, list):
        raise SpecError(f"{context} must be a list.")
    result: list[int] = []
    for item in value:
        result.append(_require_int(item, context=context))
    return result


def _resolve_codex_endpoint() -> tuple[str, str, str | None]:
    override = os.environ.get("AITUNER_CODEX_BASE_URL", "").strip()
    if override:
        return override, "chat.completions", None
    config_path = Path.home() / ".codex" / "config.toml"
    try:
        payload = tomllib.loads(config_path.read_text(encoding="utf-8"))
    except FileNotFoundError as exc:
        raise SpecError(
            "Unable to resolve llm.endpoint.base_url for provider=codex. "
            "Set llm.endpoint.base_url explicitly, set AITUNER_CODEX_BASE_URL, "
            f"or provide {config_path}."
        ) from exc
    except tomllib.TOMLDecodeError as exc:
        raise SpecError(f"Invalid TOML in {config_path}: {exc}") from exc

    raw_providers = payload.get("model_providers")
    providers = raw_providers if isinstance(raw_providers, Mapping) else {}
    provider_name = str(payload.get("model_provider") or "").strip()
    if provider_name:
        selected = providers.get(provider_name)
        if isinstance(selected, Mapping):
            base_url = str(selected.get("base_url") or "").strip()
            if base_url:
                wire_api = str(selected.get("wire_api") or "chat.completions").strip()
                reasoning_effort = str(payload.get("model_reasoning_effort") or "").strip()
                return base_url, wire_api, reasoning_effort or None
    if len(providers) == 1:
        only_provider = next(iter(providers.values()))
        if isinstance(only_provider, Mapping):
            base_url = str(only_provider.get("base_url") or "").strip()
            if base_url:
                wire_api = str(only_provider.get("wire_api") or "chat.completions").strip()
                reasoning_effort = str(payload.get("model_reasoning_effort") or "").strip()
                return base_url, wire_api, reasoning_effort or None
    root_base_url = str(payload.get("base_url") or "").strip()
    if root_base_url:
        wire_api = str(payload.get("wire_api") or "chat.completions").strip()
        reasoning_effort = str(payload.get("model_reasoning_effort") or "").strip()
        return root_base_url, wire_api, reasoning_effort or None
    raise SpecError(
        "Unable to resolve llm.endpoint.base_url for provider=codex from "
        f"{config_path}. Set llm.endpoint.base_url explicitly or set "
        "AITUNER_CODEX_BASE_URL."
    )


@dataclass(frozen=True)
class HardwareSpec:
    gpu_count: int
    gpu_model: str | None = None
    host_candidates: list[str] = field(default_factory=list)

    @classmethod
    def from_dict(cls, data: Mapping[str, Any]) -> "HardwareSpec":
        return cls(
            gpu_count=_require_int(data.get("gpu_count"), context="hardware.gpu_count"),
            gpu_model=str(data["gpu_model"]).strip() if data.get("gpu_model") else None,
            host_candidates=_coerce_str_list(
                data.get("host_candidates"), context="hardware.host_candidates"
            ),
        )


@dataclass(frozen=True)
class ModelSpec:
    model_id: str
    served_model_name: str

    @classmethod
    def from_dict(cls, data: Mapping[str, Any]) -> "ModelSpec":
        return cls(
            model_id=_require_str(data.get("model_id"), context="model.model_id"),
            served_model_name=_require_str(
                data.get("served_model_name"), context="model.served_model_name"
            ),
        )


@dataclass(frozen=True)
class EngineLaunchSpec:
    engine_name: str
    engine_version: str | None
    exec_path: str
    cwd: str | None
    host: str
    port: int
    ready_timeout_s: float
    request_timeout_s: float
    healthcheck_path: str
    launch_args: list[str]
    base_envs: dict[str, str]
    base_flags: dict[str, Any]
    tunable_envs: list[str]
    tunable_flags: list[str]
    topology_constraints: "TopologyConstraintSpec | None" = None
    python_executable: str = "python3"

    @property
    def base_url(self) -> str:
        return f"http://{self.host}:{self.port}"

    @classmethod
    def from_dict(cls, data: Mapping[str, Any]) -> "EngineLaunchSpec":
        return cls(
            engine_name=_require_str(data.get("engine_name"), context="engine.engine_name"),
            engine_version=str(data["engine_version"]).strip()
            if data.get("engine_version")
            else None,
            exec_path=_require_str(data.get("exec_path"), context="engine.exec_path"),
            cwd=str(data["cwd"]).strip() if data.get("cwd") else None,
            host=str(data.get("host") or "127.0.0.1").strip(),
            port=_require_int(data.get("port", 8000), context="engine.port"),
            ready_timeout_s=_require_float(
                data.get("ready_timeout_s", 600.0), context="engine.ready_timeout_s"
            ),
            request_timeout_s=_require_float(
                data.get("request_timeout_s", 600.0),
                context="engine.request_timeout_s",
            ),
            healthcheck_path=str(data.get("healthcheck_path") or "/v1/models").strip(),
            launch_args=_coerce_str_list(data.get("launch_args"), context="engine.launch_args"),
            base_envs=_coerce_str_map(data.get("base_envs"), context="engine.base_envs"),
            base_flags=_coerce_any_map(data.get("base_flags"), context="engine.base_flags"),
            tunable_envs=_coerce_str_list(
                data.get("tunable_envs"), context="engine.tunable_envs"
            ),
            tunable_flags=_coerce_str_list(
                data.get("tunable_flags"), context="engine.tunable_flags"
            ),
            topology_constraints=(
                TopologyConstraintSpec.from_dict(
                    _require_mapping(
                        data.get("topology_constraints"),
                        context="engine.topology_constraints",
                    )
                )
                if data.get("topology_constraints") is not None
                else None
            ),
            python_executable=str(data.get("python_executable") or "python3").strip(),
        )


@dataclass(frozen=True)
class TopologyConstraintSpec:
    require_tp_dp_product_equals_gpu_count: bool = False
    require_ep_size_leq_tp_dp_product: bool = False
    require_ep_size_divides_tp_dp_product: bool = False
    require_enable_expert_parallel_when_ep_gt_one: bool = True
    validate_cuda_graph_sizes_divisible_by_tp_when_tp_ep_reduce_scatter: bool = True
    allowed_tp_dp_products: list[int] = field(default_factory=list)
    allowed_tensor_parallel_sizes: list[int] = field(default_factory=list)
    allowed_data_parallel_sizes: list[int] = field(default_factory=list)
    allowed_expert_parallel_sizes: list[int] = field(default_factory=list)

    @classmethod
    def from_dict(cls, data: Mapping[str, Any]) -> "TopologyConstraintSpec":
        return cls(
            require_tp_dp_product_equals_gpu_count=_require_bool(
                data.get("require_tp_dp_product_equals_gpu_count", False),
                context="engine.topology_constraints.require_tp_dp_product_equals_gpu_count",
            ),
            require_ep_size_leq_tp_dp_product=_require_bool(
                data.get("require_ep_size_leq_tp_dp_product", False),
                context="engine.topology_constraints.require_ep_size_leq_tp_dp_product",
            ),
            require_ep_size_divides_tp_dp_product=_require_bool(
                data.get("require_ep_size_divides_tp_dp_product", False),
                context="engine.topology_constraints.require_ep_size_divides_tp_dp_product",
            ),
            require_enable_expert_parallel_when_ep_gt_one=_require_bool(
                data.get("require_enable_expert_parallel_when_ep_gt_one", True),
                context="engine.topology_constraints.require_enable_expert_parallel_when_ep_gt_one",
            ),
            validate_cuda_graph_sizes_divisible_by_tp_when_tp_ep_reduce_scatter=_require_bool(
                data.get(
                    "validate_cuda_graph_sizes_divisible_by_tp_when_tp_ep_reduce_scatter",
                    True,
                ),
                context="engine.topology_constraints.validate_cuda_graph_sizes_divisible_by_tp_when_tp_ep_reduce_scatter",
            ),
            allowed_tp_dp_products=_coerce_int_list(
                data.get("allowed_tp_dp_products"),
                context="engine.topology_constraints.allowed_tp_dp_products",
            ),
            allowed_tensor_parallel_sizes=_coerce_int_list(
                data.get("allowed_tensor_parallel_sizes"),
                context="engine.topology_constraints.allowed_tensor_parallel_sizes",
            ),
            allowed_data_parallel_sizes=_coerce_int_list(
                data.get("allowed_data_parallel_sizes"),
                context="engine.topology_constraints.allowed_data_parallel_sizes",
            ),
            allowed_expert_parallel_sizes=_coerce_int_list(
                data.get("allowed_expert_parallel_sizes"),
                context="engine.topology_constraints.allowed_expert_parallel_sizes",
            ),
        )


@dataclass(frozen=True)
class InputLengthFilterSpec:
    min_input_tokens: int | None = None
    max_input_tokens: int | None = None

    @classmethod
    def from_dict(cls, data: Mapping[str, Any], *, context: str) -> "InputLengthFilterSpec":
        min_input_tokens = data.get("min_input_tokens")
        max_input_tokens = data.get("max_input_tokens")
        spec = cls(
            min_input_tokens=(
                _require_int(min_input_tokens, context=f"{context}.min_input_tokens")
                if min_input_tokens is not None
                else None
            ),
            max_input_tokens=(
                _require_int(max_input_tokens, context=f"{context}.max_input_tokens")
                if max_input_tokens is not None
                else None
            ),
        )
        if spec.min_input_tokens is None and spec.max_input_tokens is None:
            raise SpecError(
                f"{context} must define at least one of min_input_tokens/max_input_tokens."
            )
        if (
            spec.min_input_tokens is not None
            and spec.max_input_tokens is not None
            and spec.min_input_tokens > spec.max_input_tokens
        ):
            raise SpecError(
                f"{context}.min_input_tokens must be <= {context}.max_input_tokens."
            )
        return spec


@dataclass(frozen=True)
class AdaptiveStopSpec:
    """Stop-A: truncate per-probe replay once the offered L-C-A converges.

    Disabled by default; the thresholds are calibrated per workload (Phase 3)
    before being switched on, so existing studies are unaffected.
    """

    enabled: bool = False
    tau: float = 0.9
    tau_c: float = 0.92
    stable_checks: int = 3
    max_checks: int = 20
    min_fraction: float = 0.1
    boundary_delta: float = 0.02

    @classmethod
    def from_dict(cls, data: Any) -> "AdaptiveStopSpec":
        if data is None:
            return cls()
        m = _require_mapping(data, context="trace.adaptive_stop")
        enabled = (
            _require_bool(m.get("enabled"), context="trace.adaptive_stop.enabled")
            if m.get("enabled") is not None
            else False
        )
        tau = _require_float(m.get("tau", 0.9), context="trace.adaptive_stop.tau")
        tau_c = _require_float(m.get("tau_c", 0.92), context="trace.adaptive_stop.tau_c")
        stable_checks = _require_int(
            m.get("stable_checks", 3), context="trace.adaptive_stop.stable_checks"
        )
        max_checks = _require_int(
            m.get("max_checks", 20), context="trace.adaptive_stop.max_checks"
        )
        min_fraction = _require_float(
            m.get("min_fraction", 0.1), context="trace.adaptive_stop.min_fraction"
        )
        boundary_delta = _require_float(
            m.get("boundary_delta", 0.02), context="trace.adaptive_stop.boundary_delta"
        )
        for name, value in (("tau", tau), ("tau_c", tau_c), ("min_fraction", min_fraction)):
            if not 0.0 < value <= 1.0:
                raise SpecError(f"trace.adaptive_stop.{name} must be in (0, 1].")
        if not 0.0 <= boundary_delta < 1.0:
            raise SpecError("trace.adaptive_stop.boundary_delta must be in [0, 1).")
        if stable_checks <= 0 or max_checks <= 0:
            raise SpecError(
                "trace.adaptive_stop.stable_checks and max_checks must be > 0."
            )
        if stable_checks > max_checks:
            raise SpecError(
                "trace.adaptive_stop.stable_checks must be <= max_checks, "
                "otherwise convergence can never be detected."
            )
        return cls(
            enabled=enabled,
            tau=tau,
            tau_c=tau_c,
            stable_checks=stable_checks,
            max_checks=max_checks,
            min_fraction=min_fraction,
            boundary_delta=boundary_delta,
        )


@dataclass(frozen=True)
class TraceSpec:
    windows_path: str
    window_id: str
    trace_file_override: str | None
    request_mode: str
    completion_tokens_override: int | None
    u_field: str
    timestamp_field: str
    max_concurrency: int
    input_length_filter: InputLengthFilterSpec | None = None
    max_requests_per_probe: int | None = None
    synthetic_prompt_cap_tokens: int | None = None
    replay_time_scale: float = 1.0
    early_stop_max_lag_s: float | None = None
    early_stop_max_elapsed_s: float | None = None
    restart_engine_after_early_stop: bool = False
    adaptive_stop: AdaptiveStopSpec = AdaptiveStopSpec()

    @classmethod
    def from_dict(cls, data: Mapping[str, Any]) -> "TraceSpec":
        max_requests = data.get("max_requests_per_probe")
        synthetic_prompt_cap = data.get("synthetic_prompt_cap_tokens")
        completion_tokens_override = data.get("completion_tokens_override")
        request_mode = str(data.get("request_mode") or "chat").strip().lower()
        if request_mode not in {"chat", "decode_only", "raw_completion"}:
            raise SpecError(
                "trace.request_mode must be one of: chat, decode_only, raw_completion."
            )
        if completion_tokens_override is not None:
            completion_tokens_override = _require_int(
                completion_tokens_override,
                context="trace.completion_tokens_override",
            )
            if completion_tokens_override < 0:
                raise SpecError("trace.completion_tokens_override must be >= 0.")
        max_requests_value = (
            _require_int(max_requests, context="trace.max_requests_per_probe")
            if max_requests is not None
            else None
        )
        if max_requests_value is not None and max_requests_value <= 0:
            raise SpecError("trace.max_requests_per_probe must be > 0.")
        synthetic_prompt_cap_value = (
            _require_int(
                synthetic_prompt_cap,
                context="trace.synthetic_prompt_cap_tokens",
            )
            if synthetic_prompt_cap is not None
            else None
        )
        if synthetic_prompt_cap_value is not None and synthetic_prompt_cap_value < 0:
            raise SpecError("trace.synthetic_prompt_cap_tokens must be >= 0.")
        replay_time_scale = _require_float(
            data.get("replay_time_scale", 1.0), context="trace.replay_time_scale"
        )
        if replay_time_scale <= 0:
            raise SpecError("trace.replay_time_scale must be > 0.")
        max_concurrency = _require_int(
            data.get("max_concurrency", 64), context="trace.max_concurrency"
        )
        if max_concurrency <= 0:
            raise SpecError("trace.max_concurrency must be > 0.")
        return cls(
            windows_path=_require_str(data.get("windows_path"), context="trace.windows_path"),
            window_id=_require_str(data.get("window_id"), context="trace.window_id"),
            trace_file_override=str(data["trace_file_override"]).strip()
            if data.get("trace_file_override")
            else None,
            request_mode=request_mode,
            completion_tokens_override=completion_tokens_override,
            u_field=str(data.get("u_field") or "sampling_u").strip(),
            timestamp_field=str(data.get("timestamp_field") or "timestamp").strip(),
            max_concurrency=max_concurrency,
            input_length_filter=(
                InputLengthFilterSpec.from_dict(
                    _require_mapping(
                        data.get("input_length_filter"),
                        context="trace.input_length_filter",
                    ),
                    context="trace.input_length_filter",
                )
                if data.get("input_length_filter") is not None
                else None
            ),
            max_requests_per_probe=max_requests_value,
            synthetic_prompt_cap_tokens=synthetic_prompt_cap_value,
            replay_time_scale=replay_time_scale,
            early_stop_max_lag_s=(
                _require_float(
                    data.get("early_stop_max_lag_s"), context="trace.early_stop_max_lag_s"
                )
                if data.get("early_stop_max_lag_s") is not None
                else None
            ),
            early_stop_max_elapsed_s=(
                _require_float(
                    data.get("early_stop_max_elapsed_s"),
                    context="trace.early_stop_max_elapsed_s",
                )
                if data.get("early_stop_max_elapsed_s") is not None
                else None
            ),
            restart_engine_after_early_stop=(
                _require_bool(
                    data.get("restart_engine_after_early_stop"),
                    context="trace.restart_engine_after_early_stop",
                )
                if data.get("restart_engine_after_early_stop") is not None
                else request_mode == "decode_only"
            ),
            adaptive_stop=AdaptiveStopSpec.from_dict(data.get("adaptive_stop")),
        )


@dataclass(frozen=True)
class ThresholdRule:
    kind: str
    threshold_ms: float | None = None
    buckets: list[dict[str, float]] = field(default_factory=list)
    intercept_ms: float | None = None
    per_token_ms: float | None = None

    @classmethod
    def from_dict(cls, data: Mapping[str, Any], *, context: str) -> "ThresholdRule":
        kind = _require_str(data.get("kind"), context=f"{context}.kind")
        if kind == "fixed_ms":
            return cls(
                kind=kind,
                threshold_ms=_require_float(
                    data.get("threshold_ms"), context=f"{context}.threshold_ms"
                ),
            )
        if kind == "linear_ms":
            # threshold = intercept_ms + per_token_ms * input_tokens
            # e.g. "4s + L_in/8k" -> intercept_ms=4000, per_token_ms=0.125
            intercept_ms = _require_float(
                data.get("intercept_ms"), context=f"{context}.intercept_ms"
            )
            per_token_ms = _require_float(
                data.get("per_token_ms"), context=f"{context}.per_token_ms"
            )
            if intercept_ms < 0 or per_token_ms < 0:
                raise SpecError(f"{context}.intercept_ms/per_token_ms must be >= 0.")
            return cls(kind=kind, intercept_ms=intercept_ms, per_token_ms=per_token_ms)
        if kind == "step_ms":
            raw = data.get("buckets")
            if not isinstance(raw, list) or not raw:
                raise SpecError(f"{context}.buckets must be a non-empty list.")
            buckets: list[dict[str, float]] = []
            for idx, item in enumerate(raw):
                mapping = _require_mapping(item, context=f"{context}.buckets[{idx}]")
                bucket: dict[str, float] = {
                    "threshold_ms": _require_float(
                        mapping.get("threshold_ms"),
                        context=f"{context}.buckets[{idx}].threshold_ms",
                    )
                }
                if "max_input_tokens" in mapping and mapping["max_input_tokens"] is not None:
                    bucket["max_input_tokens"] = _require_float(
                        mapping["max_input_tokens"],
                        context=f"{context}.buckets[{idx}].max_input_tokens",
                    )
                buckets.append(bucket)
            return cls(kind=kind, buckets=buckets)
        raise SpecError(f"Unsupported threshold rule kind: {kind}")


@dataclass(frozen=True)
class SloSpec:
    target_pass_rate: float
    ttft_rule: ThresholdRule | None
    tpot_rule: ThresholdRule | None

    @classmethod
    def from_dict(cls, data: Mapping[str, Any]) -> "SloSpec":
        ttft_rule = (
            ThresholdRule.from_dict(
                _require_mapping(data["ttft_rule"], context="slo.ttft_rule"),
                context="slo.ttft_rule",
            )
            if data.get("ttft_rule")
            else None
        )
        tpot_rule = (
            ThresholdRule.from_dict(
                _require_mapping(data["tpot_rule"], context="slo.tpot_rule"),
                context="slo.tpot_rule",
            )
            if data.get("tpot_rule")
            else None
        )
        return cls(
            target_pass_rate=_require_float(
                data.get("target_pass_rate", 0.95), context="slo.target_pass_rate"
            ),
            ttft_rule=ttft_rule,
            tpot_rule=tpot_rule,
        )


@dataclass(frozen=True)
class SearchAutoHighSpec:
    enabled: bool = False
    max_sampling_u: float = 1.0
    require_human_confirmation_beyond_trace: bool = True

    @classmethod
    def from_dict(cls, data: Any) -> "SearchAutoHighSpec":
        if data is None:
            return cls()
        m = _require_mapping(data, context="search.auto_high")
        enabled = (
            _require_bool(m.get("enabled"), context="search.auto_high.enabled")
            if m.get("enabled") is not None
            else False
        )
        max_sampling_u = _require_float(
            m.get("max_sampling_u", 1.0), context="search.auto_high.max_sampling_u"
        )
        if not 0.0 < max_sampling_u <= 1.0:
            raise SpecError("search.auto_high.max_sampling_u must be in (0, 1].")
        require_confirmation = (
            _require_bool(
                m.get("require_human_confirmation_beyond_trace"),
                context="search.auto_high.require_human_confirmation_beyond_trace",
            )
            if m.get("require_human_confirmation_beyond_trace") is not None
            else True
        )
        return cls(
            enabled=enabled,
            max_sampling_u=max_sampling_u,
            require_human_confirmation_beyond_trace=require_confirmation,
        )


@dataclass(frozen=True)
class SamplingSearchSpec:
    low: float
    high: float
    tolerance: float
    max_probes: int
    sample_seed: int
    inherit_incumbent_floor: bool = False
    auto_high: SearchAutoHighSpec = field(default_factory=SearchAutoHighSpec)

    @classmethod
    def from_dict(cls, data: Mapping[str, Any]) -> "SamplingSearchSpec":
        low = _require_float(data.get("low", 0.0), context="search.low")
        high = _require_float(data.get("high", 1.0), context="search.high")
        tolerance = _require_float(data.get("tolerance", 0.01), context="search.tolerance")
        max_probes = _require_int(data.get("max_probes", 8), context="search.max_probes")
        if low < 0:
            raise SpecError("search.low must be >= 0.")
        if high < low:
            raise SpecError("search.high must be >= search.low.")
        if tolerance <= 0:
            raise SpecError("search.tolerance must be > 0.")
        if max_probes <= 0:
            raise SpecError("search.max_probes must be > 0.")
        return cls(
            low=low,
            high=high,
            tolerance=tolerance,
            max_probes=max_probes,
            sample_seed=_require_int(
                data.get("sample_seed", 20260325), context="search.sample_seed"
            ),
            inherit_incumbent_floor=_require_bool(
                data.get("inherit_incumbent_floor", False),
                context="search.inherit_incumbent_floor",
            ),
            auto_high=SearchAutoHighSpec.from_dict(data.get("auto_high")),
        )


@dataclass(frozen=True)
class LLMEndpointSpec:
    base_url: str
    model: str
    provider: str = "custom"
    wire_api: str = "chat.completions"
    stream: bool = False
    reasoning_effort: str | None = None
    api_key_env: str = "OPENAI_API_KEY"
    timeout_s: float = 120.0

    @classmethod
    def from_dict(cls, data: Mapping[str, Any]) -> "LLMEndpointSpec":
        provider = str(data.get("provider") or "custom").strip().lower()
        base_url = str(data.get("base_url") or "").strip()
        wire_api = str(data.get("wire_api") or "").strip()
        stream = data.get("stream")
        stream_value = _require_bool(stream, context="llm.endpoint.stream") if stream is not None else False
        reasoning_effort = str(data.get("reasoning_effort") or "").strip()
        api_key_env = str(data.get("api_key_env") or "").strip()
        if provider == "codex":
            if not base_url or not wire_api or not reasoning_effort:
                resolved_base_url, resolved_wire_api, resolved_reasoning_effort = (
                    _resolve_codex_endpoint()
                )
                if not base_url:
                    base_url = resolved_base_url
                if not wire_api:
                    wire_api = resolved_wire_api
                if not reasoning_effort and resolved_reasoning_effort:
                    reasoning_effort = resolved_reasoning_effort
            if stream_value:
                wire_api = "chat.completions"
            if not api_key_env:
                api_key_env = "OPENAI_API_KEY"
        elif provider == "bailian":
            if not base_url:
                base_url = "https://dashscope.aliyuncs.com/compatible-mode/v1"
            if not wire_api:
                wire_api = "chat.completions"
            if not api_key_env:
                api_key_env = "DASHSCOPE_API_KEY"
        elif provider == "custom":
            if not base_url:
                raise SpecError("llm.endpoint.base_url must be a non-empty string.")
            if not wire_api:
                wire_api = "chat.completions"
            if not api_key_env:
                api_key_env = "OPENAI_API_KEY"
        else:
            raise SpecError(f"Unsupported llm.endpoint.provider: {provider}")
        return cls(
            base_url=_require_str(base_url, context="llm.endpoint.base_url"),
            model=_require_str(data.get("model"), context="llm.endpoint.model"),
            provider=provider,
            wire_api=_require_str(wire_api, context="llm.endpoint.wire_api"),
            stream=stream_value,
            reasoning_effort=reasoning_effort or None,
            api_key_env=_require_str(api_key_env, context="llm.endpoint.api_key_env"),
            timeout_s=_require_float(
                data.get("timeout_s", 120.0), context="llm.endpoint.timeout_s"
            ),
        )


@dataclass(frozen=True)
class InitialConfigReviewSpec:
    mode: str = "off"

    @classmethod
    def from_dict(cls, data: Any) -> "InitialConfigReviewSpec":
        if data is None:
            return cls()
        payload = _require_mapping(data, context="llm.initial_config_review")
        mode = str(payload.get("mode") or "off").strip().lower()
        if mode not in {"off", "warn"}:
            raise SpecError("llm.initial_config_review.mode must be one of: off, warn.")
        return cls(mode=mode)


@dataclass(frozen=True)
class LLMPolicySpec:
    endpoint: LLMEndpointSpec | None
    system_prompt: str
    max_history_trials: int
    use_harness: bool = True
    harness_candidate_policy: str = "advisory"
    initial_config_review: InitialConfigReviewSpec = field(
        default_factory=InitialConfigReviewSpec
    )

    @classmethod
    def from_dict(cls, data: Mapping[str, Any] | None) -> "LLMPolicySpec":
        payload = _require_mapping(data or {}, context="llm")
        endpoint = (
            LLMEndpointSpec.from_dict(
                _require_mapping(payload["endpoint"], context="llm.endpoint")
            )
            if payload.get("endpoint")
            else None
        )
        harness_candidate_policy = str(
            payload.get("harness_candidate_policy") or "advisory"
        ).strip()
        if harness_candidate_policy not in {"advisory", "strict"}:
            raise SpecError(
                "llm.harness_candidate_policy must be one of: advisory, strict."
            )
        return cls(
            endpoint=endpoint,
            system_prompt=str(payload.get("system_prompt") or "").strip(),
            max_history_trials=_require_int(
                payload.get("max_history_trials", 8), context="llm.max_history_trials"
            ),
            use_harness=(
                _require_bool(payload.get("use_harness"), context="llm.use_harness")
                if payload.get("use_harness") is not None
                else True
            ),
            harness_candidate_policy=harness_candidate_policy,
            initial_config_review=InitialConfigReviewSpec.from_dict(
                payload.get("initial_config_review")
            ),
        )


@dataclass(frozen=True)
class StudySpec:
    study_id: str
    hardware: HardwareSpec
    model: ModelSpec
    engine: EngineLaunchSpec
    trace: TraceSpec
    slo: SloSpec
    search: SamplingSearchSpec
    llm: LLMPolicySpec
    capability_profile_path: str | None = None

    @classmethod
    def from_dict(cls, data: Mapping[str, Any]) -> "StudySpec":
        study = cls(
            study_id=_require_str(data.get("study_id"), context="study_id"),
            hardware=HardwareSpec.from_dict(
                _require_mapping(data.get("hardware"), context="hardware")
            ),
            model=ModelSpec.from_dict(_require_mapping(data.get("model"), context="model")),
            engine=EngineLaunchSpec.from_dict(
                _require_mapping(data.get("engine"), context="engine")
            ),
            trace=TraceSpec.from_dict(_require_mapping(data.get("trace"), context="trace")),
            slo=SloSpec.from_dict(_require_mapping(data.get("slo"), context="slo")),
            search=SamplingSearchSpec.from_dict(
                _require_mapping(data.get("search"), context="search")
            ),
            llm=LLMPolicySpec.from_dict(data.get("llm")),
            capability_profile_path=str(data["capability_profile_path"]).strip()
            if data.get("capability_profile_path")
            else None,
        )
        served_model_name = str(study.engine.base_flags.get("served-model-name") or "").strip()
        if served_model_name and served_model_name != study.model.served_model_name:
            raise SpecError(
                "model.served_model_name must match engine.base_flags['served-model-name'] "
                f"when both are set. Got model.served_model_name={study.model.served_model_name!r} "
                f"and engine.base_flags['served-model-name']={served_model_name!r}."
            )
        return study


@dataclass(frozen=True)
class ConfigPatch:
    env_patch: dict[str, str] = field(default_factory=dict)
    flag_patch: dict[str, Any] = field(default_factory=dict)

    @classmethod
    def from_dict(cls, data: Mapping[str, Any]) -> "ConfigPatch":
        return cls(
            env_patch=_coerce_str_map(data.get("env_patch"), context="config_patch.env_patch"),
            flag_patch=_coerce_any_map(
                data.get("flag_patch"), context="config_patch.flag_patch"
            ),
        )


@dataclass(frozen=True)
class Proposal:
    observation: str
    diagnosis: str
    config_patch: ConfigPatch
    expected_effects: list[str]
    why_not_previous_failures: str = ""
    should_stop: bool = False

    @classmethod
    def from_dict(cls, data: Mapping[str, Any]) -> "Proposal":
        expected_effects = data.get("expected_effects")
        if isinstance(expected_effects, str):
            expected_effects_value = [expected_effects.strip()] if expected_effects.strip() else []
        elif isinstance(expected_effects, Mapping):
            expected_effects_value = []
            for key, value in expected_effects.items():
                key_text = str(key).strip()
                value_text = str(value).strip()
                if key_text and value_text:
                    expected_effects_value.append(f"{key_text}: {value_text}")
                elif key_text:
                    expected_effects_value.append(key_text)
                elif value_text:
                    expected_effects_value.append(value_text)
        else:
            expected_effects_value = _coerce_str_list(
                expected_effects, context="proposal.expected_effects"
            )
        return cls(
            observation=_coerce_text(data.get("observation"), context="proposal.observation"),
            diagnosis=_coerce_text(data.get("diagnosis"), context="proposal.diagnosis"),
            config_patch=ConfigPatch.from_dict(
                _require_mapping(data.get("config_patch"), context="proposal.config_patch")
            ),
            expected_effects=expected_effects_value,
            why_not_previous_failures=str(data.get("why_not_previous_failures") or "").strip(),
            should_stop=(
                _require_bool(data.get("should_stop"), context="proposal.should_stop")
                if data.get("should_stop") is not None
                else False
            ),
        )


@dataclass(frozen=True)
class TrialSpec:
    study_id: str
    trial_id: str
    config_patch: ConfigPatch
    search: SamplingSearchSpec
    study_spec_path: str
    artifact_dir: str
    probe_log_path: str
    engine_log_path: str
    result_path: str
    search_evidence: dict[str, Any] = field(default_factory=dict)


@dataclass
class TrialSummary:
    trial_id: str
    status: str
    parallel_size: int | None = None
    best_sampling_u: float | None = None
    best_request_rate: float | None = None
    best_request_rate_per_gpu: float | None = None
    best_pass_rate: float | None = None
    result_path: str | None = None
    diagnosis: str = ""
    config_patch: dict[str, Any] | None = None
    failure_reason: str = ""
    failure_stage: str = ""


@dataclass
class StudyState:
    study_id: str
    best_trial_id: str | None = None
    best_parallel_size: int | None = None
    best_sampling_u: float | None = None
    best_request_rate: float | None = None
    best_request_rate_per_gpu: float | None = None
    next_trial_index: int = 1
    tuning_stop_reason: str = ""
    tuning_stop_diagnosis: str = ""
    tuning_stop_details: dict[str, Any] = field(default_factory=dict)
    best_by_parallel_size: dict[str, dict[str, Any]] = field(default_factory=dict)
    trials: list[TrialSummary] = field(default_factory=list)


@dataclass(frozen=True)
class CompareTrialRefSpec:
    study_root: str
    trial_id: str

    @classmethod
    def from_dict(cls, data: Mapping[str, Any]) -> "CompareTrialRefSpec":
        return cls(
            study_root=_require_str(data.get("study_root"), context="trial_ref.study_root"),
            trial_id=_require_str(data.get("trial_id"), context="trial_ref.trial_id"),
        )


@dataclass(frozen=True)
class CompareCandidateSpec:
    config_patch: ConfigPatch | None = None
    trial_ref: CompareTrialRefSpec | None = None

    @classmethod
    def from_dict(cls, data: Mapping[str, Any], *, context: str) -> "CompareCandidateSpec":
        config_patch = (
            ConfigPatch.from_dict(
                _require_mapping(data.get("config_patch"), context=f"{context}.config_patch")
            )
            if data.get("config_patch") is not None
            else None
        )
        trial_ref = (
            CompareTrialRefSpec.from_dict(
                _require_mapping(data.get("trial_ref"), context=f"{context}.trial_ref")
            )
            if data.get("trial_ref") is not None
            else None
        )
        if (config_patch is None) == (trial_ref is None):
            raise SpecError(
                f"{context} must define exactly one of config_patch or trial_ref."
            )
        return cls(config_patch=config_patch, trial_ref=trial_ref)


@dataclass(frozen=True)
class WindowSelectorSpec:
    trace_type: str | None = None
    date_prefix: str | None = None
    date_from: str | None = None
    date_to: str | None = None
    slot_token: str | None = None

    @classmethod
    def from_dict(cls, data: Mapping[str, Any], *, context: str) -> "WindowSelectorSpec":
        selector = cls(
            trace_type=str(data.get("trace_type") or "").strip() or None,
            date_prefix=str(data.get("date_prefix") or "").strip() or None,
            date_from=str(data.get("date_from") or "").strip() or None,
            date_to=str(data.get("date_to") or "").strip() or None,
            slot_token=str(data.get("slot_token") or "").strip() or None,
        )
        if (
            selector.trace_type is None
            and selector.date_prefix is None
            and selector.date_from is None
            and selector.date_to is None
            and selector.slot_token is None
        ):
            raise SpecError(f"{context} must define at least one selector field.")
        return selector


@dataclass(frozen=True)
class CompareSpec:
    compare_id: str
    study_spec_path: str
    window_ids: list[str]
    window_selector: WindowSelectorSpec | None
    baseline: CompareCandidateSpec
    tuned: CompareCandidateSpec

    @classmethod
    def from_dict(cls, data: Mapping[str, Any]) -> "CompareSpec":
        compare = cls(
            compare_id=_require_str(data.get("compare_id"), context="compare_id"),
            study_spec_path=_require_str(
                data.get("study_spec_path"), context="study_spec_path"
            ),
            window_ids=_coerce_str_list(data.get("window_ids"), context="window_ids"),
            window_selector=(
                WindowSelectorSpec.from_dict(
                    _require_mapping(data.get("window_selector"), context="window_selector"),
                    context="window_selector",
                )
                if data.get("window_selector") is not None
                else None
            ),
            baseline=CompareCandidateSpec.from_dict(
                _require_mapping(data.get("baseline"), context="baseline"),
                context="baseline",
            ),
            tuned=CompareCandidateSpec.from_dict(
                _require_mapping(data.get("tuned"), context="tuned"),
                context="tuned",
            ),
        )
        if not compare.window_ids and compare.window_selector is None:
            raise SpecError("Compare spec must define window_ids or window_selector.")
        return compare


def to_jsonable(value: Any) -> Any:
    if is_dataclass(value):
        return {key: to_jsonable(item) for key, item in asdict(value).items()}
    if isinstance(value, dict):
        return {str(key): to_jsonable(item) for key, item in value.items()}
    if isinstance(value, list):
        return [to_jsonable(item) for item in value]
    return value


def load_structured_file(path: Path) -> Mapping[str, Any]:
    suffix = path.suffix.lower()
    if suffix == ".json":
        payload = json.loads(path.read_text(encoding="utf-8"))
    elif suffix in {".toml", ".tml"}:
        payload = tomllib.loads(path.read_text(encoding="utf-8"))
    else:
        raise SpecError(f"Unsupported spec file type: {path}")
    return _require_mapping(payload, context=str(path))


def load_study_spec(path: Path) -> StudySpec:
    return StudySpec.from_dict(load_structured_file(path))
