"""Subject-kind plugins: the "any VCS can join" structural layer.

Each file in config/subject_kinds/<kind>.json declares, per subject kind,

what the harness needs to measure it: the argv template per operation

(null = capability gap, the caller emits a skip row

"capability_gap:<kind>:<operation>" — never a substituted command),

snapshot call-count semantics, lock-hint patterns, integrity commands,

workspace model, server-error patterns, and the remote env var.

This layer is a PARALLEL READ of existing configuration, not a rewrite:

existing lanes keep loading config/command_semantics.json through

oakbench.command_semantics, byte-for-byte unchanged. parity_report() is the

proof mechanism: it compares a plugin JSON against command_semantics.json

for that kind and returns the list of mismatches. An empty list means the

plugin layer is measurement-identical to the source of truth, so adding a

new VCS (jujutsu, sapling, pijul) is one JSON file plus one subjects.toml

entry plus golden output samples — zero lane-code changes.

"""

from __future__ import annotations

import json

from dataclasses import dataclass, field

from pathlib import Path

from typing import Any

ROOT = Path(__file__).resolve().parents[2]

DEFAULT_DIR = ROOT / "config" / "subject_kinds"

DEFAULT_COMMAND_SEMANTICS = ROOT / "config" / "command_semantics.json"

SCHEMA_VERSION = 1

# Operations whose value is a per-track mapping rather than a flat argv

# template, mirroring config/command_semantics.json harness_tracks shape.

TRACKED_OPERATIONS = ("status", "diff")

DEFAULT_TRACK = "agent-default"

@dataclass(frozen=True)

class SubjectKind:

    """One subject kind's measurement declaration, loaded from JSON."""

    kind: str

    schema_version: int

    commands: dict[str, Any]

    snapshot_call_semantics: dict[str, Any]

    lock_hint_patterns: tuple[str, ...]

    integrity_command: tuple[tuple[str, ...], ...] | None

    workspace_model: str

    server_error_patterns: tuple[str, ...]

    remote_env_var: str | None

    source_path: Path = field(compare=False, default=Path())

    def command_for(self, operation: str, track: str = DEFAULT_TRACK) -> list[str] | None:

        """Argv template for an operation, or None for a capability gap.

        None covers both an explicit null and an absent operation; callers

        emit a skip row "capability_gap:<kind>:<operation>" for either.

        """

        value = self.commands.get(operation)

        if isinstance(value, dict):

            value = value.get(track)

        if value is None:

            return None

        return [str(part) for part in value]

    def skip_reason(self, operation: str) -> str:

        return f"capability_gap:{self.kind}:{operation}"

def _fail(path: Path, message: str) -> ValueError:

    return ValueError(f"{path}: {message}")

def _validate_argv(path: Path, operation: str, value: Any) -> None:

    if value is None:

        return

    if not isinstance(value, list) or not value or not all(isinstance(part, str) for part in value):

        raise _fail(path, f"commands.{operation} must be null or a non-empty list of strings, got {value!r}")

def _validate(path: Path, data: dict[str, Any]) -> None:

    if data.get("schema_version") != SCHEMA_VERSION:

        raise _fail(path, f"schema_version must be {SCHEMA_VERSION}, got {data.get('schema_version')!r}")

    kind = data.get("kind")

    if not isinstance(kind, str) or not kind:

        raise _fail(path, "kind must be a non-empty string")

    if kind != path.stem:

        raise _fail(path, f"kind {kind!r} must match filename stem {path.stem!r}")

    commands = data.get("commands")

    if not isinstance(commands, dict) or not commands:

        raise _fail(path, "commands must be a non-empty object of operation -> argv template | null")

    for operation, value in commands.items():

        if isinstance(value, dict):

            for track, tracked in value.items():

                _validate_argv(path, f"{operation}.{track}", tracked)

        else:

            _validate_argv(path, operation, value)

    snapshot = data.get("snapshot_call_semantics")

    if not isinstance(snapshot, dict):

        raise _fail(path, "snapshot_call_semantics must be an object")

    if not isinstance(snapshot.get("stage_required"), bool):

        raise _fail(path, "snapshot_call_semantics.stage_required must be a boolean")

    if not isinstance(snapshot.get("calls_per_snapshot"), int):

        raise _fail(path, "snapshot_call_semantics.calls_per_snapshot must be an integer")

    for key in ("lock_hint_patterns", "server_error_patterns"):

        value = data.get(key)

        if not isinstance(value, list) or not all(isinstance(item, str) for item in value):

            raise _fail(path, f"{key} must be a list of strings")

    integrity = data.get("integrity_command")

    if integrity is not None:

        ok = isinstance(integrity, list) and integrity and all(

            isinstance(cmd, list) and cmd and all(isinstance(part, str) for part in cmd) for cmd in integrity

        )

        if not ok:

            raise _fail(path, "integrity_command must be null or a non-empty list of argv lists")

    model = data.get("workspace_model")

    if not isinstance(model, str) or not model:

        raise _fail(path, "workspace_model must be a non-empty string")

    remote = data.get("remote_env_var")

    if remote is not None and (not isinstance(remote, str) or not remote):

        raise _fail(path, "remote_env_var must be null or a non-empty string")

def available_kinds(root: Path = DEFAULT_DIR) -> list[str]:

    """Sorted kind names with a plugin JSON under config/subject_kinds/."""

    if not root.is_dir():

        return []

    return sorted(path.stem for path in root.glob("*.json"))

def load_subject_kind(kind: str, root: Path = DEFAULT_DIR) -> SubjectKind:

    path = root / f"{kind}.json"

    if not path.is_file():

        known = ", ".join(available_kinds(root)) or "none"

        raise ValueError(f"unknown subject kind {kind!r}: no plugin at {path} (available: {known})")

    try:

        data = json.loads(path.read_text())

    except json.JSONDecodeError as exc:

        raise _fail(path, f"invalid JSON: {exc}") from exc

    if not isinstance(data, dict):

        raise _fail(path, "plugin root must be a JSON object")

    _validate(path, data)

    integrity = data["integrity_command"]

    return SubjectKind(

        kind=data["kind"],

        schema_version=data["schema_version"],

        commands=data["commands"],

        snapshot_call_semantics=data["snapshot_call_semantics"],

        lock_hint_patterns=tuple(data["lock_hint_patterns"]),

        integrity_command=tuple(tuple(cmd) for cmd in integrity) if integrity is not None else None,

        workspace_model=data["workspace_model"],

        server_error_patterns=tuple(data["server_error_patterns"]),

        remote_env_var=data["remote_env_var"],

        source_path=path,

    )

def command_for(kind: str, operation: str, track: str = DEFAULT_TRACK, root: Path = DEFAULT_DIR) -> list[str] | None:

    """Argv template for kind+operation, or None = capability gap (skip row)."""

    return load_subject_kind(kind, root).command_for(operation, track)

def parity_report(

    kind: str,

    command_semantics_path: Path = DEFAULT_COMMAND_SEMANTICS,

    root: Path = DEFAULT_DIR,

) -> list[str]:

    """Mismatches between the plugin JSON and command_semantics.json for kind.

    Empty list = the plugin layer is measurement-identical to the existing

    source of truth for every operation the harness runs. This is the

    row_parity proof for the config layer: identical commands in, identical

    rows out, so lanes can later read either source without drift.

    """

    plugin = load_subject_kind(kind, root)

    data = json.loads(Path(command_semantics_path).read_text())

    tracks = data.get("harness_tracks", {})

    mismatches: list[str] = []

    if kind not in tracks:

        return [f"kind_not_in_command_semantics:{kind} ({command_semantics_path})"]

    semantics = tracks[kind]

    for operation in sorted(semantics):

        expected = semantics[operation]

        if operation not in plugin.commands:

            mismatches.append(f"missing_operation:{kind}:{operation} (in command_semantics.json, not in plugin)")

            continue

        actual = plugin.commands[operation]

        if actual != expected:

            mismatches.append(

                f"command_mismatch:{kind}:{operation} plugin={actual!r} command_semantics={expected!r}"

            )

    for operation in sorted(set(plugin.commands) - set(semantics)):

        mismatches.append(

            f"extra_operation:{kind}:{operation} (in plugin, not in command_semantics.json harness_tracks)"

        )

    # Snapshot call-count semantics derive from the stage entry: a non-null

    # stage means the kind pays stage+commit as two recorded calls

    # (oakbench.command_semantics.snapshot_commands).

    stage = semantics.get("stage")

    expected_stage_required = stage is not None

    expected_calls = 2 if stage is not None else 1

    snapshot = plugin.snapshot_call_semantics

    if snapshot.get("stage_required") != expected_stage_required:

        mismatches.append(

            f"snapshot_mismatch:{kind}:stage_required plugin={snapshot.get('stage_required')!r} "

            f"derived_from_command_semantics={expected_stage_required!r}"

        )

    if snapshot.get("calls_per_snapshot") != expected_calls:

        mismatches.append(

            f"snapshot_mismatch:{kind}:calls_per_snapshot plugin={snapshot.get('calls_per_snapshot')!r} "

            f"derived_from_command_semantics={expected_calls!r}"

        )

    return mismatches