src/agent_harness/specs.py

"""Immutable experiment specifications and configuration validation.

The project intentionally uses the Python standard library for its initial
schema layer. TOML inputs are converted into frozen dataclasses, validated, and
hashed using a canonical JSON representation.
"""

from __future__ import annotations

from dataclasses import asdict, dataclass
from hashlib import sha256
import json
from pathlib import Path
import re
import tomllib
from typing import Any, Iterable


class SpecError(ValueError):
    """Raised when a configuration cannot represent a valid treatment."""


def project_root() -> Path:
    return Path(__file__).resolve().parents[2]


def _read_toml(path: Path) -> dict[str, Any]:
    try:
        with path.open("rb") as handle:
            return tomllib.load(handle)
    except (OSError, tomllib.TOMLDecodeError) as exc:
        raise SpecError(f"Cannot read TOML configuration {path}: {exc}") from exc


def _canonical_hash(value: Any) -> str:
    payload = json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True)
    return sha256(payload.encode("utf-8")).hexdigest()


@dataclass(frozen=True, slots=True)
class HarnessSpec:
    schema_version: int
    harness_id: str
    name: str
    description: str
    exact_search: bool
    lexical: bool
    syntax: str
    dense: bool
    graph_hops: int
    query_policy: str
    interface: str
    packing: str
    fusion: str
    control: str
    adaptive: bool

    @classmethod
    def from_mapping(cls, value: dict[str, Any], source: Path | None = None) -> "HarnessSpec":
        try:
            spec = cls(
                schema_version=int(value["schema_version"]),
                harness_id=str(value["harness_id"]),
                name=str(value["name"]),
                description=str(value.get("description", "")),
                exact_search=bool(value["exact_search"]),
                lexical=bool(value["lexical"]),
                syntax=str(value["syntax"]),
                dense=bool(value["dense"]),
                graph_hops=int(value["graph_hops"]),
                query_policy=str(value["query_policy"]),
                interface=str(value["interface"]),
                packing=str(value["packing"]),
                fusion=str(value["fusion"]),
                control=str(value.get("control", "none")),
                adaptive=bool(value.get("adaptive", False)),
            )
        except (KeyError, TypeError, ValueError) as exc:
            location = f" in {source}" if source else ""
            raise SpecError(f"Malformed harness specification{location}: {exc}") from exc
        spec.validate(source)
        return spec

    @classmethod
    def load(cls, path: Path) -> "HarnessSpec":
        return cls.from_mapping(_read_toml(path), path)

    def validate(self, source: Path | None = None) -> None:
        errors: list[str] = []
        if self.schema_version != 1:
            errors.append("schema_version must be 1")
        if not re.fullmatch(r"H\d{3}", self.harness_id):
            errors.append("harness_id must match H000-style identifiers")
        if not re.fullmatch(r"[a-z][a-z0-9_]*", self.name):
            errors.append("name must be a lowercase semantic slug")
        if self.syntax not in {"raw", "tree_sitter"}:
            errors.append("syntax must be raw or tree_sitter")
        if self.graph_hops not in {0, 1, 2}:
            errors.append("graph_hops must be 0, 1, or 2")
        if self.graph_hops and self.syntax != "tree_sitter":
            errors.append("graph expansion requires the tree_sitter structural index")
        if self.query_policy not in {"one_shot", "iterative"}:
            errors.append("query_policy must be one_shot or iterative")
        if self.interface not in {"unified", "specialized"}:
            errors.append("interface must be unified or specialized")
        if self.packing not in {"ranked_snippets", "skeletons", "whole_files", "role_summaries"}:
            errors.append("unsupported context packing strategy")
        if self.packing == "skeletons" and self.syntax != "tree_sitter":
            errors.append("skeleton packing requires tree_sitter syntax")
        if self.fusion not in {"none", "rrf"}:
            errors.append("fusion must be none or rrf")

        active_advanced_sources = int(self.lexical) + int(self.syntax == "tree_sitter") + int(self.dense)
        expected_fusion = "rrf" if active_advanced_sources >= 2 else "none"
        if self.control == "none" and self.fusion != expected_fusion:
            errors.append(
                f"fusion must be {expected_fusion} for {active_advanced_sources} advanced retrieval sources"
            )

        allowed_controls = {"none", "no_search", "random_context", "oracle_file", "oracle_function"}
        if self.control not in allowed_controls:
            errors.append(f"control must be one of {sorted(allowed_controls)}")
        if self.control != "none":
            if self.exact_search or self.lexical or self.dense or self.syntax != "raw" or self.graph_hops:
                errors.append("control harnesses cannot enable repository retrieval capabilities")
            if self.fusion != "none" or self.adaptive:
                errors.append("control harnesses cannot enable fusion or adaptive routing")
        elif not self.exact_search:
            errors.append("non-control harnesses must retain the exact/regex baseline")

        if self.adaptive:
            if not (self.lexical and self.dense and self.syntax == "tree_sitter"):
                errors.append("adaptive routing requires lexical, syntax, and dense retrieval")
            if self.graph_hops < 1 or self.query_policy != "iterative":
                errors.append("adaptive routing requires graph expansion and iterative queries")

        if errors:
            location = f" ({source})" if source else ""
            raise SpecError(f"Invalid harness {self.harness_id}{location}: " + "; ".join(errors))

    @property
    def config_hash(self) -> str:
        return _canonical_hash(asdict(self))

    @property
    def treatment_hash(self) -> str:
        """Hash only causal treatment fields, excluding labels and prose."""

        value = asdict(self)
        for field in ("harness_id", "name", "description"):
            value.pop(field)
        return _canonical_hash(value)

    @property
    def uses_embedding(self) -> bool:
        return self.dense


@dataclass(frozen=True, slots=True)
class EditInterfaceSpec:
    """Immutable action-interface treatment for protocol-normalized studies."""

    schema_version: int
    interface_id: str
    name: str
    description: str
    edit_tool: str
    prompt_contract: str

    @classmethod
    def load(cls, path: Path) -> "EditInterfaceSpec":
        value = _read_toml(path)
        try:
            spec = cls(
                schema_version=int(value["schema_version"]),
                interface_id=str(value["interface_id"]),
                name=str(value["name"]),
                description=str(value.get("description", "")),
                edit_tool=str(value["edit_tool"]),
                prompt_contract=str(value["prompt_contract"]),
            )
        except (KeyError, TypeError, ValueError) as exc:
            raise SpecError(f"Malformed edit-interface specification {path}: {exc}") from exc
        spec.validate(path)
        return spec

    def validate(self, source: Path | None = None) -> None:
        errors: list[str] = []
        if self.schema_version != 1:
            errors.append("schema_version must be 1")
        if not re.fullmatch(r"P\d{3}", self.interface_id):
            errors.append("interface_id must match P000-style identifiers")
        if not re.fullmatch(r"[a-z][a-z0-9_]*", self.name):
            errors.append("name must be a lowercase semantic slug")
        if self.edit_tool not in {"apply_patch", "replace_text", "write_file"}:
            errors.append("unsupported edit tool")
        if not self.prompt_contract.strip():
            errors.append("prompt_contract must be non-empty")
        if errors:
            location = f" ({source})" if source else ""
            raise SpecError(
                f"Invalid edit interface {self.interface_id}{location}: "
                + "; ".join(errors)
            )

    @property
    def config_hash(self) -> str:
        return _canonical_hash(asdict(self))

    @property
    def treatment_hash(self) -> str:
        value = asdict(self)
        for field in ("interface_id", "name", "description"):
            value.pop(field)
        return _canonical_hash(value)


@dataclass(frozen=True, slots=True)
class ModelSpec:
    schema_version: int
    model_id: str
    canonical_name: str
    expected_identity: str
    expected_inference_key: str
    expected_variant: str
    expected_format: str
    expected_quantization: str
    provider: str
    base_url: str
    api_token_env: str
    discovery_endpoint: str
    native_discovery_endpoint: str
    inference_endpoint: str
    temperature: float
    top_p: float
    max_tokens: int
    seed: int
    context_length: int
    reasoning_mode: str

    @classmethod
    def load(cls, path: Path) -> "ModelSpec":
        value = _read_toml(path)
        try:
            spec = cls(
                schema_version=int(value["schema_version"]),
                model_id=str(value["model_id"]),
                canonical_name=str(value["canonical_name"]),
                expected_identity=str(value["expected_identity"]),
                expected_inference_key=str(value["expected_inference_key"]),
                expected_variant=str(value["expected_variant"]),
                expected_format=str(value["expected_format"]),
                expected_quantization=str(value["expected_quantization"]),
                provider=str(value["provider"]),
                base_url=str(value["base_url"]).rstrip("/"),
                api_token_env=str(value.get("api_token_env", "LM_STUDIO_API_TOKEN")),
                discovery_endpoint=str(value.get("discovery_endpoint", "/v1/models")),
                native_discovery_endpoint=str(value.get("native_discovery_endpoint", "/api/v1/models")),
                inference_endpoint=str(value.get("inference_endpoint", "/v1/chat/completions")),
                temperature=float(value["temperature"]),
                top_p=float(value["top_p"]),
                max_tokens=int(value["max_tokens"]),
                seed=int(value["seed"]),
                context_length=int(value["context_length"]),
                reasoning_mode=str(value["reasoning_mode"]),
            )
        except (KeyError, TypeError, ValueError) as exc:
            raise SpecError(f"Malformed model specification {path}: {exc}") from exc
        spec.validate(path)
        return spec

    def validate(self, source: Path | None = None) -> None:
        errors: list[str] = []
        if self.schema_version != 1:
            errors.append("schema_version must be 1")
        if not re.fullmatch(r"M\d{3}", self.model_id):
            errors.append("model_id must match M000-style identifiers")
        if not self.canonical_name.strip() or not self.expected_identity.strip():
            errors.append("canonical_name and expected_identity must be pinned")
        if not self.expected_inference_key or not self.expected_variant:
            errors.append("the LM Studio model key and selected variant must be pinned")
        if not self.expected_format or not self.expected_quantization:
            errors.append("model format and quantization must be pinned")
        if self.provider != "lm_studio_local":
            errors.append("provider must be lm_studio_local")
        if self.base_url not in {"http://127.0.0.1:1234", "http://localhost:1234"}:
            errors.append("LM Studio must be configured on local port 1234")
        if not (0.0 <= self.temperature <= 2.0 and 0.0 < self.top_p <= 1.0):
            errors.append("invalid sampling parameters")
        if self.max_tokens <= 0:
            errors.append("max_tokens must be positive")
        if self.context_length <= 0:
            errors.append("context_length must be positive")
        if self.reasoning_mode not in {"none", "off", "on", "low", "medium", "high"}:
            errors.append("unsupported reasoning_mode")
        if errors:
            location = f" ({source})" if source else ""
            raise SpecError("Invalid model specification" + location + ": " + "; ".join(errors))

    @property
    def config_hash(self) -> str:
        return _canonical_hash(asdict(self))


@dataclass(frozen=True, slots=True)
class EmbeddingSpec:
    schema_version: int
    embedding_id: str
    status: str
    provider: str
    base_url: str
    api_token_env: str
    discovery_endpoint: str
    inference_endpoint: str
    model_key: str
    expected_display_name: str
    expected_format: str
    expected_quantization: str
    expected_size_bytes: int
    max_context_length: int
    loaded_context_length: int
    vector_dimension: int
    normalized: bool
    query_instruction: str
    document_prefix_template: str
    chunk_lines: int
    chunk_overlap_lines: int
    chunk_char_limit: int
    batch_size: int
    notes: str

    @classmethod
    def load(cls, path: Path) -> "EmbeddingSpec":
        value = _read_toml(path)
        try:
            spec = cls(
                schema_version=int(value["schema_version"]),
                embedding_id=str(value["embedding_id"]),
                status=str(value["status"]),
                provider=str(value["provider"]),
                base_url=str(value["base_url"]).rstrip("/"),
                api_token_env=str(value.get("api_token_env", "LM_STUDIO_API_TOKEN")),
                discovery_endpoint=str(value.get("discovery_endpoint", "/api/v1/models")),
                inference_endpoint=str(value.get("inference_endpoint", "/v1/embeddings")),
                model_key=str(value["model_key"]),
                expected_display_name=str(value["expected_display_name"]),
                expected_format=str(value["expected_format"]),
                expected_quantization=str(value["expected_quantization"]),
                expected_size_bytes=int(value["expected_size_bytes"]),
                max_context_length=int(value["max_context_length"]),
                loaded_context_length=int(value["loaded_context_length"]),
                vector_dimension=int(value["vector_dimension"]),
                normalized=bool(value["normalized"]),
                query_instruction=str(value["query_instruction"]),
                document_prefix_template=str(value["document_prefix_template"]),
                chunk_lines=int(value["chunk_lines"]),
                chunk_overlap_lines=int(value["chunk_overlap_lines"]),
                chunk_char_limit=int(value["chunk_char_limit"]),
                batch_size=int(value["batch_size"]),
                notes=str(value.get("notes", "")),
            )
        except (KeyError, TypeError, ValueError) as exc:
            raise SpecError(f"Malformed embedding specification {path}: {exc}") from exc
        if spec.schema_version != 1 or not re.fullmatch(r"EMB\d{3}", spec.embedding_id):
            raise SpecError(f"Invalid embedding specification {path}")
        errors: list[str] = []
        if spec.status != "ready":
            errors.append("the pinned embedding profile must have status ready")
        if spec.provider != "lm_studio_local":
            errors.append("embedding provider must be lm_studio_local")
        if spec.base_url not in {"http://127.0.0.1:1234", "http://localhost:1234"}:
            errors.append("embedding service must use local LM Studio on port 1234")
        if not all(
            (
                spec.model_key,
                spec.expected_display_name,
                spec.expected_format,
                spec.expected_quantization,
            )
        ):
            errors.append("embedding identity, format, and quantization must be pinned")
        if min(
            spec.expected_size_bytes,
            spec.max_context_length,
            spec.loaded_context_length,
            spec.vector_dimension,
        ) <= 0:
            errors.append("embedding size, context length, and vector dimension must be positive")
        if spec.loaded_context_length > spec.max_context_length:
            errors.append("loaded embedding context cannot exceed the model maximum")
        if not spec.normalized:
            errors.append("the initial cosine-similarity protocol requires normalized embeddings")
        if not spec.query_instruction or "{path}" not in spec.document_prefix_template:
            errors.append("embedding query instruction and path-aware document prefix must be frozen")
        if min(spec.chunk_lines, spec.chunk_char_limit, spec.batch_size) <= 0:
            errors.append("embedding chunk size and batch size must be positive")
        if not 0 <= spec.chunk_overlap_lines < spec.chunk_lines:
            errors.append("embedding chunk overlap must be non-negative and smaller than the chunk")
        if errors:
            raise SpecError(f"Invalid embedding specification {path}: " + "; ".join(errors))
        return spec

    @property
    def config_hash(self) -> str:
        return _canonical_hash(asdict(self))


@dataclass(frozen=True, slots=True)
class TaskSpec:
    schema_version: int
    task_id: str
    repository_url: str
    base_commit: str
    gold_commit: str
    language: str
    statement: str
    gold_patch: str
    test_patch: str
    gold_files: tuple[str, ...]
    gold_symbols: tuple[str, ...]
    fail_to_pass_tests: tuple[str, ...]
    pass_to_pass_tests: tuple[str, ...]
    difficulty: str
    provenance: str
    validation_status: str

    @classmethod
    def load(cls, path: Path) -> "TaskSpec":
        value = _read_toml(path)
        try:
            spec = cls(
                schema_version=int(value["schema_version"]),
                task_id=str(value["task_id"]),
                repository_url=str(value["repository_url"]),
                base_commit=str(value["base_commit"]),
                gold_commit=str(value["gold_commit"]),
                language=str(value["language"]),
                statement=str(value["statement"]),
                gold_patch=str(value["gold_patch"]),
                test_patch=str(value.get("test_patch", "")),
                gold_files=tuple(str(item) for item in value["gold_files"]),
                gold_symbols=tuple(str(item) for item in value["gold_symbols"]),
                fail_to_pass_tests=tuple(str(item) for item in value["fail_to_pass_tests"]),
                pass_to_pass_tests=tuple(str(item) for item in value["pass_to_pass_tests"]),
                difficulty=str(value["difficulty"]),
                provenance=str(value["provenance"]),
                validation_status=str(value["validation_status"]),
            )
        except (KeyError, TypeError, ValueError) as exc:
            raise SpecError(f"Malformed task specification {path}: {exc}") from exc
        spec.validate(path)
        return spec

    def validate(self, source: Path | None = None) -> None:
        errors: list[str] = []
        if self.schema_version != 1:
            errors.append("schema_version must be 1")
        if not re.fullmatch(r"TASK_[A-Z0-9_]+", self.task_id):
            errors.append("task_id must match TASK_UPPERCASE_STYLE")
        if not self.repository_url.startswith("https://gitlab.com/"):
            errors.append("initial benchmark tasks must use a version-pinned GitLab repository")
        if not re.fullmatch(r"[0-9a-fA-F]{40}", self.base_commit):
            errors.append("base_commit must be a full 40-character Git commit SHA")
        if not re.fullmatch(r"[0-9a-fA-F]{40}", self.gold_commit):
            errors.append("gold_commit must be a full 40-character Git commit SHA")
        if not self.language or not self.statement.strip():
            errors.append("language and statement must be non-empty")
        if not self.gold_files or not self.gold_symbols:
            errors.append("gold_files and gold_symbols must be non-empty")
        allowed_statuses = {"template_only", "retrieval_ready", "end_to_end_ready"}
        if self.validation_status not in allowed_statuses:
            errors.append(f"validation_status must be one of {sorted(allowed_statuses)}")
        if self.validation_status == "end_to_end_ready":
            if not self.gold_patch:
                errors.append("end-to-end-ready tasks require a gold patch")
            if not self.test_patch:
                errors.append("end-to-end-ready tasks require a hidden test patch")
            if not self.fail_to_pass_tests or not self.pass_to_pass_tests:
                errors.append("end-to-end-ready tasks require both test sets")
        if not self.provenance.strip() or not self.difficulty.strip():
            errors.append("difficulty and provenance must be documented")
        for repository_path in (
            *self.gold_files,
            *((self.gold_patch,) if self.gold_patch else ()),
            *((self.test_patch,) if self.test_patch else ()),
        ):
            path = Path(repository_path)
            if path.is_absolute() or ".." in path.parts:
                errors.append(f"repository path must be safe and relative: {repository_path}")
        if errors:
            location = f" ({source})" if source else ""
            raise SpecError(f"Invalid task {self.task_id}{location}: " + "; ".join(errors))

    @property
    def config_hash(self) -> str:
        return _canonical_hash(asdict(self))


@dataclass(frozen=True, slots=True)
class ExperimentSpec:
    schema_version: int
    experiment_id: str
    name: str
    mode: str
    description: str
    harness_ids: tuple[str, ...]
    edit_interface_ids: tuple[str, ...]
    agent_system_ids: tuple[str, ...]
    repository_ids: tuple[str, ...]
    backend_ids: tuple[str, ...]
    model_ids: tuple[str, ...]
    embedding_id: str
    task_split: str
    context_budgets: tuple[int, ...]
    seeds: tuple[int, ...]
    repetitions: int
    max_tool_calls: int
    max_test_runs: int
    timeout_seconds: int

    @classmethod
    def load(cls, path: Path) -> "ExperimentSpec":
        value = _read_toml(path)
        try:
            spec = cls(
                schema_version=int(value["schema_version"]),
                experiment_id=str(value["experiment_id"]),
                name=str(value["name"]),
                mode=str(value["mode"]),
                description=str(value.get("description", "")),
                harness_ids=tuple(str(item) for item in value["harness_ids"]),
                edit_interface_ids=tuple(
                    str(item) for item in value.get("edit_interface_ids", [])
                ),
                agent_system_ids=tuple(str(item) for item in value.get("agent_system_ids", [])),
                repository_ids=tuple(str(item) for item in value.get("repository_ids", [])),
                backend_ids=tuple(str(item) for item in value.get("backend_ids", [])),
                model_ids=tuple(str(item) for item in value["model_ids"]),
                embedding_id=str(value["embedding_id"]),
                task_split=str(value["task_split"]),
                context_budgets=tuple(int(item) for item in value["context_budgets"]),
                seeds=tuple(int(item) for item in value["seeds"]),
                repetitions=int(value["repetitions"]),
                max_tool_calls=int(value["max_tool_calls"]),
                max_test_runs=int(value["max_test_runs"]),
                timeout_seconds=int(value["timeout_seconds"]),
            )
        except (KeyError, TypeError, ValueError) as exc:
            raise SpecError(f"Malformed experiment specification {path}: {exc}") from exc
        spec.validate(path)
        return spec

    def validate(self, source: Path | None = None) -> None:
        errors: list[str] = []
        if self.schema_version != 1:
            errors.append("schema_version must be 1")
        if not re.fullmatch(r"E\d{2}", self.experiment_id):
            errors.append("experiment_id must match E00-style identifiers")
        if not self.harness_ids or len(set(self.harness_ids)) != len(self.harness_ids):
            errors.append("harness_ids must be non-empty and unique")
        if len(set(self.edit_interface_ids)) != len(self.edit_interface_ids):
            errors.append("edit_interface_ids must be unique")
        if len(set(self.agent_system_ids)) != len(self.agent_system_ids):
            errors.append("agent_system_ids must be unique")
        if len(set(self.repository_ids)) != len(self.repository_ids):
            errors.append("repository_ids must be unique")
        is_study2 = self.mode in {"study2_live_agent", "study2_reliability"}
        is_protocol = self.mode == "protocol_interface"
        if is_study2 and (not self.agent_system_ids or not self.repository_ids):
            errors.append("Study 2 experiments must enumerate agent systems and repositories")
        if is_protocol and (not self.edit_interface_ids or not self.repository_ids):
            errors.append("protocol experiments must enumerate edit interfaces and repositories")
        if not is_protocol and self.edit_interface_ids:
            errors.append("only protocol experiments may enumerate edit interfaces")
        if not (is_study2 or is_protocol) and (self.agent_system_ids or self.repository_ids):
            errors.append(
                "only repository-scale Study 2/protocol experiments may enumerate "
                "agent systems or repositories"
            )
        if is_protocol and self.agent_system_ids:
            errors.append("protocol experiments cannot enumerate agent systems")
        if len(set(self.backend_ids)) != len(self.backend_ids):
            errors.append("backend_ids must be unique")
        if self.mode == "index_backend" and not self.backend_ids:
            errors.append("index_backend experiments must enumerate backend_ids")
        if self.mode != "index_backend" and self.backend_ids:
            errors.append("only index_backend experiments may enumerate backend_ids")
        if not self.model_ids or len(set(self.model_ids)) != len(self.model_ids):
            errors.append("model_ids must be non-empty and unique")
        if not self.context_budgets or any(value <= 0 for value in self.context_budgets):
            errors.append("context budgets must be positive")
        if not self.seeds or self.repetitions <= 0:
            errors.append("seeds and repetitions must be positive/non-empty")
        if min(self.max_tool_calls, self.max_test_runs, self.timeout_seconds) <= 0:
            errors.append("execution limits must be positive")
        if errors:
            location = f" ({source})" if source else ""
            raise SpecError("Invalid experiment specification" + location + ": " + "; ".join(errors))

    def cells_per_task(self) -> int:
        return (
            (len(self.harness_ids) + len(self.agent_system_ids))
            * max(len(self.edit_interface_ids), 1)
            * max(len(self.backend_ids), 1)
            * len(self.model_ids)
            * len(self.context_budgets)
            * len(self.seeds)
            * self.repetitions
        )


@dataclass(frozen=True, slots=True)
class BackendSpec:
    schema_version: int
    backend_id: str
    name: str
    kind: str
    distance: str
    description: str
    library: str
    version: str
    index_type: str
    query_repetitions: int
    neighbors: int | None = None
    ef_construction: int | None = None
    ef_search: int | None = None

    @classmethod
    def load(cls, path: Path) -> "BackendSpec":
        value = _read_toml(path)
        try:
            spec = cls(
                schema_version=int(value["schema_version"]),
                backend_id=str(value["backend_id"]),
                name=str(value["name"]),
                kind=str(value["kind"]),
                distance=str(value["distance"]),
                description=str(value.get("description", "")),
                library=str(value["library"]),
                version=str(value["version"]),
                index_type=str(value["index_type"]),
                query_repetitions=int(value["query_repetitions"]),
                neighbors=int(value["neighbors"]) if "neighbors" in value else None,
                ef_construction=int(value["ef_construction"]) if "ef_construction" in value else None,
                ef_search=int(value["ef_search"]) if "ef_search" in value else None,
            )
        except (KeyError, TypeError, ValueError) as exc:
            raise SpecError(f"Malformed backend specification {path}: {exc}") from exc
        errors: list[str] = []
        if spec.schema_version != 1 or not re.fullmatch(r"B\d{3}", spec.backend_id):
            errors.append("invalid backend schema or identifier")
        if not all((spec.library, spec.version, spec.index_type)) or spec.query_repetitions < 2:
            errors.append("backend version, index type, and at least two repetitions are required")
        if spec.backend_id == "B002" and None in (
            spec.neighbors, spec.ef_construction, spec.ef_search
        ):
            errors.append("B002 must freeze HNSW parameters")
        if errors:
            raise SpecError(f"Invalid backend specification {path}: " + "; ".join(errors))
        return spec

    @property
    def config_hash(self) -> str:
        return _canonical_hash(asdict(self))


@dataclass(frozen=True, slots=True)
class AgentSystemSpec:
    schema_version: int
    system_id: str
    name: str
    family: str
    description: str
    implementation: str
    model_calls: int
    max_tool_calls: int
    max_test_runs: int
    interactive: bool
    same_model_required: bool
    reference: str

    @classmethod
    def load(cls, path: Path) -> "AgentSystemSpec":
        value = _read_toml(path)
        try:
            spec = cls(
                schema_version=int(value["schema_version"]),
                system_id=str(value["system_id"]),
                name=str(value["name"]),
                family=str(value["family"]),
                description=str(value["description"]),
                implementation=str(value["implementation"]),
                model_calls=int(value["model_calls"]),
                max_tool_calls=int(value["max_tool_calls"]),
                max_test_runs=int(value["max_test_runs"]),
                interactive=bool(value["interactive"]),
                same_model_required=bool(value["same_model_required"]),
                reference=str(value["reference"]),
            )
        except (KeyError, TypeError, ValueError) as exc:
            raise SpecError(f"Malformed agent-system specification {path}: {exc}") from exc
        errors: list[str] = []
        if spec.schema_version != 1 or not re.fullmatch(r"A\d{3}", spec.system_id):
            errors.append("invalid agent-system schema or identifier")
        if not re.fullmatch(r"[a-z][a-z0-9_]*", spec.name):
            errors.append("agent-system name must be a lowercase semantic slug")
        if spec.implementation != "local_reimplementation":
            errors.append("system baselines must be explicitly labeled local_reimplementation")
        if spec.model_calls <= 0 or min(spec.max_tool_calls, spec.max_test_runs) < 0:
            errors.append("agent-system budgets must be nonnegative and model_calls positive")
        if not spec.same_model_required or not spec.reference.startswith("https://"):
            errors.append("same-model control and a public reference are required")
        if errors:
            raise SpecError(f"Invalid agent-system specification {path}: " + "; ".join(errors))
        return spec

    @property
    def config_hash(self) -> str:
        return _canonical_hash(asdict(self))


@dataclass(frozen=True, slots=True)
class RepositorySpec:
    schema_version: int
    repository_id: str
    name: str
    repository_url: str
    local_path: str
    language: str
    source_suffixes: tuple[str, ...]
    test_pattern: str
    test_runner: str
    pinned_head: str

    @classmethod
    def load(cls, path: Path) -> "RepositorySpec":
        value = _read_toml(path)
        try:
            spec = cls(
                schema_version=int(value["schema_version"]),
                repository_id=str(value["repository_id"]),
                name=str(value["name"]),
                repository_url=str(value["repository_url"]),
                local_path=str(value["local_path"]),
                language=str(value["language"]),
                source_suffixes=tuple(str(item) for item in value["source_suffixes"]),
                test_pattern=str(value["test_pattern"]),
                test_runner=str(value["test_runner"]),
                pinned_head=str(value["pinned_head"]),
            )
        except (KeyError, TypeError, ValueError) as exc:
            raise SpecError(f"Malformed repository specification {path}: {exc}") from exc
        errors: list[str] = []
        if spec.schema_version != 1 or not re.fullmatch(r"R\d{3}", spec.repository_id):
            errors.append("invalid repository schema or identifier")
        if not spec.repository_url.startswith("https://gitlab.com/"):
            errors.append("Study 2 repositories must be public GitLab URLs")
        local = Path(spec.local_path)
        if local.is_absolute() or ".." in local.parts:
            errors.append("local_path must be a safe project-relative path")
        if spec.language not in {"go", "python"}:
            errors.append("unsupported Study 2 language")
        if not spec.source_suffixes or any(not item.startswith(".") for item in spec.source_suffixes):
            errors.append("source_suffixes must be non-empty file suffixes")
        if not re.fullmatch(r"[0-9a-f]{40}", spec.pinned_head):
            errors.append("pinned_head must be a lowercase full Git SHA")
        if errors:
            raise SpecError(f"Invalid repository specification {path}: " + "; ".join(errors))
        return spec

    @property
    def config_hash(self) -> str:
        return _canonical_hash(asdict(self))


def _load_specs(directory: Path, loader: Any, identifier_field: str) -> dict[str, Any]:
    result: dict[str, Any] = {}
    for path in sorted(directory.glob("*.toml")):
        spec = loader(path)
        identifier = getattr(spec, identifier_field)
        if identifier in result:
            raise SpecError(f"Duplicate specification identifier {identifier} in {directory}")
        result[identifier] = spec
    return result


def load_harnesses(root: Path | None = None) -> dict[str, HarnessSpec]:
    base = root or project_root()
    return _load_specs(base / "configs" / "harnesses", HarnessSpec.load, "harness_id")


def load_edit_interfaces(root: Path | None = None) -> dict[str, EditInterfaceSpec]:
    base = root or project_root()
    return _load_specs(
        base / "configs" / "edit_interfaces",
        EditInterfaceSpec.load,
        "interface_id",
    )


def load_models(root: Path | None = None) -> dict[str, ModelSpec]:
    base = root or project_root()
    return _load_specs(base / "configs" / "models", ModelSpec.load, "model_id")


def load_embeddings(root: Path | None = None) -> dict[str, EmbeddingSpec]:
    base = root or project_root()
    return _load_specs(base / "configs" / "embeddings", EmbeddingSpec.load, "embedding_id")


def load_experiments(root: Path | None = None) -> dict[str, ExperimentSpec]:
    base = root or project_root()
    return _load_specs(base / "configs" / "experiments", ExperimentSpec.load, "experiment_id")


def load_backends(root: Path | None = None) -> dict[str, BackendSpec]:
    base = root or project_root()
    return _load_specs(base / "configs" / "backends", BackendSpec.load, "backend_id")


def load_agent_systems(root: Path | None = None) -> dict[str, AgentSystemSpec]:
    base = root or project_root()
    return _load_specs(
        base / "configs" / "agent_systems", AgentSystemSpec.load, "system_id"
    )


def load_repositories(root: Path | None = None) -> dict[str, RepositorySpec]:
    base = root or project_root()
    return _load_specs(
        base / "configs" / "repositories", RepositorySpec.load, "repository_id"
    )


def load_tasks(root: Path | None = None) -> dict[str, TaskSpec]:
    base = root or project_root()
    return _load_specs(base / "tasks" / "manifests", TaskSpec.load, "task_id")


def load_task_split(path: Path) -> tuple[str, ...]:
    try:
        identifiers = tuple(
            line.strip()
            for line in path.read_text(encoding="utf-8").splitlines()
            if line.strip() and not line.lstrip().startswith("#")
        )
    except OSError as exc:
        raise SpecError(f"Cannot read task split {path}: {exc}") from exc
    if len(set(identifiers)) != len(identifiers):
        raise SpecError(f"Task split contains duplicate identifiers: {path}")
    return identifiers


def _validate_plain_toml(paths: Iterable[Path], errors: list[str]) -> None:
    for path in paths:
        try:
            _read_toml(path)
        except SpecError as exc:
            errors.append(str(exc))


def validate_configuration_tree(root: Path | None = None) -> tuple[list[str], list[str]]:
    base = root or project_root()
    errors: list[str] = []
    warnings: list[str] = []
    try:
        harnesses = load_harnesses(base)
        edit_interfaces = load_edit_interfaces(base)
        models = load_models(base)
        embeddings = load_embeddings(base)
        experiments = load_experiments(base)
        backends = load_backends(base)
        agent_systems = load_agent_systems(base)
        repositories = load_repositories(base)
        tasks = load_tasks(base)
    except SpecError as exc:
        return [str(exc)], []

    expected_harness_ids = {f"H{number:03d}" for number in range(21)}
    if set(harnesses) != expected_harness_ids:
        errors.append(
            "Initial catalog must contain exactly H000-H020; found "
            + ", ".join(sorted(harnesses))
        )
    if set(models) != {"M001", "M002", "M003", "M004"}:
        errors.append("The study must define exactly the pinned runtime profiles M001-M004")
    if set(edit_interfaces) != {"P001", "P002", "P003"}:
        errors.append("Study 3 must define exactly edit interfaces P001-P003")
    if set(embeddings) != {"EMB001", "EMB002"}:
        errors.append(
            "The study must define exactly EMB001 (Study 1) and EMB002 (Study 2)"
        )
    if set(backends) != {"B001", "B002", "B003"}:
        errors.append("The backend study must define exactly B001-B003")
    if set(agent_systems) != {"A001", "A002"}:
        errors.append("Study 2 must define exactly controlled agent systems A001-A002")
    if set(repositories) != {"R001", "R002", "R003"}:
        errors.append("Study 2 must define exactly repositories R001-R003")

    for experiment in experiments.values():
        missing_harnesses = set(experiment.harness_ids) - set(harnesses)
        missing_edit_interfaces = set(experiment.edit_interface_ids) - set(edit_interfaces)
        missing_models = set(experiment.model_ids) - set(models)
        missing_systems = set(experiment.agent_system_ids) - set(agent_systems)
        missing_repositories = set(experiment.repository_ids) - set(repositories)
        if missing_harnesses:
            errors.append(f"{experiment.experiment_id} references missing harnesses {sorted(missing_harnesses)}")
        if missing_edit_interfaces:
            errors.append(
                f"{experiment.experiment_id} references missing edit interfaces "
                f"{sorted(missing_edit_interfaces)}"
            )
        if missing_models:
            errors.append(f"{experiment.experiment_id} references missing models {sorted(missing_models)}")
        if missing_systems:
            errors.append(
                f"{experiment.experiment_id} references missing agent systems {sorted(missing_systems)}"
            )
        if missing_repositories:
            errors.append(
                f"{experiment.experiment_id} references missing repositories {sorted(missing_repositories)}"
            )
        missing_backends = set(experiment.backend_ids) - set(backends)
        if missing_backends:
            errors.append(
                f"{experiment.experiment_id} references missing backends {sorted(missing_backends)}"
            )
        if experiment.embedding_id not in embeddings:
            errors.append(
                f"{experiment.experiment_id} references missing embedding profile {experiment.embedding_id}"
            )
            continue
        uses_dense = any(harnesses[item].uses_embedding for item in experiment.harness_ids)
        if uses_dense and embeddings[experiment.embedding_id].status != "ready":
            warnings.append(
                f"{experiment.experiment_id} includes dense retrieval but {experiment.embedding_id} "
                "is not configured; execution must remain blocked"
            )

        split_path = base / "tasks" / "splits" / f"{experiment.task_split}.txt"
        try:
            split = load_task_split(split_path)
        except SpecError as exc:
            errors.append(str(exc))
            continue
        missing_tasks = set(split) - set(tasks)
        if missing_tasks:
            errors.append(
                f"{experiment.experiment_id} split references missing tasks {sorted(missing_tasks)}"
            )
        if "TASK_EXAMPLE" in split:
            errors.append(f"{experiment.experiment_id} cannot include the template TASK_EXAMPLE")
        if not split:
            warnings.append(
                f"{experiment.experiment_id} task split {experiment.task_split} is empty; "
                "execution must remain blocked"
            )

        if experiment.mode in {
            "study2_live_agent",
            "study2_reliability",
            "protocol_interface",
        }:
            allowed_urls = {
                repositories[item].repository_url for item in experiment.repository_ids
            }
            unexpected = {
                tasks[item].repository_url
                for item in split
                if item in tasks and tasks[item].repository_url not in allowed_urls
            }
            if unexpected:
                errors.append(
                    f"{experiment.experiment_id} split contains repositories outside its registry: "
                    f"{sorted(unexpected)}"
                )

    hashes: dict[str, str] = {}
    for harness in harnesses.values():
        if harness.treatment_hash in hashes:
            errors.append(
                f"{harness.harness_id} duplicates the causal treatment of "
                f"{hashes[harness.treatment_hash]}"
            )
        hashes[harness.treatment_hash] = harness.harness_id

    interface_hashes: dict[str, str] = {}
    for interface in edit_interfaces.values():
        if interface.treatment_hash in interface_hashes:
            errors.append(
                f"{interface.interface_id} duplicates the causal treatment of "
                f"{interface_hashes[interface.treatment_hash]}"
            )
        interface_hashes[interface.treatment_hash] = interface.interface_id

    _validate_plain_toml((base / "configs" / "scenarios").glob("*.toml"), errors)
    _validate_plain_toml((base / "configs" / "backends").glob("*.toml"), errors)
    return errors, warnings