src/yaml_edit/models.py

"""Public, immutable edit artifacts. These are not authorization tokens."""

import json
from dataclasses import dataclass
from enum import StrEnum
from math import isfinite
from typing import Never, TypeGuard, cast

# Keep int and float distinct: serialization and semantic comparison depend on it.
type Scalar = str | int | float
type Path = tuple[str | int, ...]
type Tree = Scalar | bool | list[Tree] | dict[str, Tree] | None

MAX_BATCH_OPERATIONS = 32
MAX_BATCH_ARTIFACT_BYTES = 65_536
MAX_BATCH_CANDIDATE_BYTES = 1_024
MAX_BATCH_SCALAR_BYTES = 4_096
MAX_BATCH_PATH_PARTS = 8
_VERSION_LENGTH = 64
_PROPOSAL_KEYS = frozenset(
    {
        "source_version",
        "candidate_id",
        "path",
        "expected_old_value",
        "value",
        "format",
        "operation",
        "document_index",
    }
)


class ErrorCode(StrEnum):
    """Stable machine-readable failures for library and CLI consumers."""

    INVALID_YAML = "invalid_yaml"
    UNSUPPORTED_DOCUMENT = "unsupported_document"
    INVALID_VALUE = "invalid_value"
    UNKNOWN_TARGET = "unknown_target"
    INVALID_PROPOSAL = "invalid_proposal"
    STALE_PROPOSAL = "stale_proposal"
    INVARIANT_VIOLATION = "invariant_violation"


class YamlEditError(ValueError):
    """An expected validation failure with a stable classification."""

    def __init__(self, code: ErrorCode, message: str) -> None:
        """Associate a human-readable explanation with a domain error code."""
        super().__init__(message)
        self.code = code


class FieldKind(StrEnum):
    """The complete v1 editable field allowlist."""

    REPLICAS = "replicas"
    IMAGE = "image"
    CPU_REQUEST = "cpu-request"
    CPU_LIMIT = "cpu-limit"
    MEMORY_REQUEST = "memory-request"
    MEMORY_LIMIT = "memory-limit"


@dataclass(frozen=True, slots=True)
class Location:
    """One-based line/column; zero-based Python character offsets [start, end)."""

    line: int
    column: int
    start: int
    end: int

    def to_dict(self) -> dict[str, int]:
        """Return JSON-compatible source coordinates."""
        return {"line": self.line, "column": self.column, "start": self.start, "end": self.end}


@dataclass(frozen=True, slots=True)
class Candidate:
    """An existing, editable scalar discovered in a specific source snapshot."""

    id: str
    path: Path
    kind: FieldKind
    description: str
    current_value: Scalar
    location: Location
    container: str | None = None

    def to_dict(self) -> dict[str, object]:
        """Return a JSON-compatible description without traversing arbitrary objects."""
        return {
            "id": self.id,
            "path": list(self.path),
            "kind": self.kind.value,
            "description": self.description,
            "current_value": self.current_value,
            "location": self.location.to_dict(),
            "container": self.container,
        }


@dataclass(frozen=True, slots=True)
class Inspection:
    """A source fingerprint and its complete, ordered candidate list."""

    source_version: str
    candidates: tuple[Candidate, ...]

    def to_dict(self) -> dict[str, object]:
        """Return JSON-compatible inspection results."""
        return {
            "source_version": self.source_version,
            "candidates": [candidate.to_dict() for candidate in self.candidates],
        }


def is_scalar(value: object) -> TypeGuard[Scalar]:
    """Recognize exact built-in scalar types, excluding bool and nonfinite floats."""
    return type(value) in (str, int) or (type(value) is float and isfinite(value))


def _invalid() -> Never:
    raise YamlEditError(ErrorCode.INVALID_PROPOSAL, "Malformed or unsupported edit proposal.")


def _record(value: object) -> dict[str, object]:
    if not isinstance(value, dict) or set(value) != _PROPOSAL_KEYS:
        _invalid()
    # Exact key-set validation above proves the string-key boundary.
    return cast("dict[str, object]", value)


def _version(value: object) -> str:
    if (
        not isinstance(value, str)
        or len(value) != _VERSION_LENGTH
        or any(character not in "0123456789abcdef" for character in value)
    ):
        _invalid()
    return value


def _path(value: object) -> Path:
    if not isinstance(value, list) or not value:
        _invalid()
    parts = cast("list[object]", value)
    if any(not (type(part) is str or (type(part) is int and part >= 0)) for part in parts):
        _invalid()
    # Each element has been checked; do not coerce bool to an index or strings to ints.
    return cast("Path", tuple(parts))


def _serialized_path(value: object) -> object:
    # Preserve malformed dataclass fields for from_dict() to reject, without coercion.
    return list(value) if isinstance(value, tuple) else value


@dataclass(frozen=True, slots=True)
class EditProposal:
    """A snapshot-bound, single-field review artifact, not an authorization token."""

    source_version: str
    candidate_id: str
    path: Path
    expected_old_value: Scalar
    value: Scalar
    format: str = "yaml-edit/v1"
    operation: str = "set"
    document_index: int = 0

    def to_dict(self) -> dict[str, object]:
        """Project fields shallowly so malformed values reach validation safely."""
        return {
            "source_version": self.source_version,
            "candidate_id": self.candidate_id,
            "path": _serialized_path(self.path),
            "expected_old_value": self.expected_old_value,
            "value": self.value,
            "format": self.format,
            "operation": self.operation,
            "document_index": self.document_index,
        }

    @classmethod
    def from_dict(cls, data: object) -> "EditProposal":
        """Validate untrusted decoded JSON without recursively copying its values."""
        record = _record(data)
        if (
            record["format"] != "yaml-edit/v1"
            or record["operation"] != "set"
            or type(record["document_index"]) is not int
            or record["document_index"] != 0
        ):
            _invalid()
        candidate = record["candidate_id"]
        if not isinstance(candidate, str) or not candidate.startswith("/"):
            _invalid()
        old_value, new_value = record["expected_old_value"], record["value"]
        if not is_scalar(old_value) or not is_scalar(new_value):
            _invalid()
        return cls(
            source_version=_version(record["source_version"]),
            candidate_id=candidate,
            path=_path(record["path"]),
            expected_old_value=old_value,
            value=new_value,
        )


@dataclass(frozen=True, slots=True)
class EditPreview:
    """The verified replacement source and a human-readable review diff."""

    source: str
    diff: str
    changed: bool
    candidate: Candidate


_BATCH_PROPOSAL_KEYS = frozenset(
    {"format", "operation", "source_version", "document_index", "entries"}
)
_BATCH_ENTRY_KEYS = frozenset({"candidate_id", "path", "expected_old_value", "value"})
_MAX_ARTIFACT_INTEGER = 2**63 - 1
_MAX_ARTIFACT_INTEGER_BITS = 63
_MISSING = object()


@dataclass(frozen=True, slots=True)
class BatchRequest:
    """One explicit requested set, before snapshot binding."""

    candidate_id: str
    value: Scalar


@dataclass(frozen=True, slots=True)
class BatchEntry:
    """One source-bound set inside a batch proposal."""

    candidate_id: str
    path: Path
    expected_old_value: Scalar
    value: Scalar


@dataclass(frozen=True, slots=True)
class BatchProposal:
    """A snapshot-bound atomic batch review artifact."""

    source_version: str
    entries: tuple[BatchEntry, ...]
    format: str = "yaml-edit/batch-v1"
    operation: str = "set-existing-scalars"
    document_index: int = 0

    def to_dict(self) -> dict[str, object]:
        """Return a bounded JSON-compatible artifact after runtime validation."""
        data = _batch_dataclass_dict(self)
        _check_batch_size(data)
        return data

    @classmethod
    def from_dict(cls, data: object) -> "BatchProposal":
        """Validate one untrusted, decoded batch artifact without recursive copying."""
        record = _batch_record(data, _BATCH_PROPOSAL_KEYS)
        if (
            record["format"] != "yaml-edit/batch-v1"
            or record["operation"] != "set-existing-scalars"
            or type(record["document_index"]) is not int
            or record["document_index"] != 0
        ):
            _invalid()
        entries_value = record["entries"]
        if type(entries_value) is not list or not 1 <= len(entries_value) <= MAX_BATCH_OPERATIONS:
            _invalid()
        raw_entries = cast("list[object]", entries_value)
        entries = tuple(_batch_entry_from_dict(entry) for entry in raw_entries)
        proposal = cls(source_version=_batch_version(record["source_version"]), entries=entries)
        _check_batch_size(_batch_dataclass_dict(proposal))
        return proposal


@dataclass(frozen=True, slots=True)
class BatchPreview:
    """The verified complete source and one unified review diff."""

    source: str
    diff: str
    changed: bool
    candidates: tuple[Candidate, ...]


def _batch_record(value: object, keys: frozenset[str]) -> dict[str, object]:
    if type(value) is not dict or len(value) != len(keys) or set(value) != keys:
        _invalid()
    return cast("dict[str, object]", value)


def _batch_text(value: object, maximum: int, *, pointer: bool = False) -> str:
    if type(value) is not str or len(value) > maximum or (pointer and not value.startswith("/")):
        _invalid()
    try:
        raw = value.encode("utf-8")
    except UnicodeEncodeError as error:
        raise YamlEditError(
            ErrorCode.INVALID_PROPOSAL, "Malformed or unsupported edit proposal."
        ) from error
    if len(raw) > maximum:
        _invalid()
    return value


def _batch_version(value: object) -> str:
    if (
        type(value) is not str
        or len(value) != _VERSION_LENGTH
        or any(character not in "0123456789abcdef" for character in value)
    ):
        _invalid()
    return value


def _batch_scalar(value: object) -> Scalar:
    if not is_scalar(value):
        _invalid()
    if type(value) is str:
        _batch_text(value, MAX_BATCH_SCALAR_BYTES)
    elif type(value) is int and value.bit_length() > _MAX_ARTIFACT_INTEGER.bit_length():
        _invalid()
    return value


def _batch_path(value: object, *, serialized: bool) -> Path:
    expected_type = list if serialized else tuple
    if type(value) is not expected_type or not 1 <= len(value) <= MAX_BATCH_PATH_PARTS:
        _invalid()
    parts = cast("list[object] | tuple[object, ...]", value)
    for part in parts:
        if type(part) is str:
            _batch_text(part, MAX_BATCH_CANDIDATE_BYTES)
        elif type(part) is not int or part < 0 or part.bit_length() > _MAX_ARTIFACT_INTEGER_BITS:
            _invalid()
    return cast("Path", tuple(parts))


def _batch_entry_from_dict(value: object) -> BatchEntry:
    record = _batch_record(value, _BATCH_ENTRY_KEYS)
    return BatchEntry(
        candidate_id=_batch_text(record["candidate_id"], MAX_BATCH_CANDIDATE_BYTES, pointer=True),
        path=_batch_path(record["path"], serialized=True),
        expected_old_value=_batch_scalar(record["expected_old_value"]),
        value=_batch_scalar(record["value"]),
    )


def _batch_entry_dict(value: object) -> dict[str, object]:
    if type(value) is not BatchEntry:
        _invalid()
    candidate_id = getattr(value, "candidate_id", _MISSING)
    path = getattr(value, "path", _MISSING)
    old_value = getattr(value, "expected_old_value", _MISSING)
    new_value = getattr(value, "value", _MISSING)
    return {
        "candidate_id": _batch_text(candidate_id, MAX_BATCH_CANDIDATE_BYTES, pointer=True),
        "path": list(_batch_path(path, serialized=False)),
        "expected_old_value": _batch_scalar(old_value),
        "value": _batch_scalar(new_value),
    }


def _batch_dataclass_dict(value: object) -> dict[str, object]:
    if type(value) is not BatchProposal:
        _invalid()
    format_value = getattr(value, "format", _MISSING)
    operation = getattr(value, "operation", _MISSING)
    document_index = getattr(value, "document_index", _MISSING)
    entries = getattr(value, "entries", _MISSING)
    if (
        format_value != "yaml-edit/batch-v1"
        or operation != "set-existing-scalars"
        or type(document_index) is not int
        or document_index != 0
        or type(entries) is not tuple
        or not 1 <= len(entries) <= MAX_BATCH_OPERATIONS
    ):
        _invalid()
    checked_entries = cast("tuple[object, ...]", entries)
    return {
        "format": "yaml-edit/batch-v1",
        "operation": "set-existing-scalars",
        "source_version": _batch_version(getattr(value, "source_version", _MISSING)),
        "document_index": 0,
        "entries": [_batch_entry_dict(entry) for entry in checked_entries],
    }


def _batch_json(data: dict[str, object]) -> str:
    return json.dumps(data, indent=2, ensure_ascii=False, allow_nan=False) + "\n"


def _check_batch_size(data: dict[str, object]) -> None:
    try:
        size = len(_batch_json(data).encode("utf-8"))
    except (TypeError, ValueError, UnicodeEncodeError) as error:
        raise YamlEditError(
            ErrorCode.INVALID_PROPOSAL, "Malformed or unsupported edit proposal."
        ) from error
    if size > MAX_BATCH_ARTIFACT_BYTES:
        _invalid()


def validate_batch_proposal(value: object) -> BatchProposal:
    """Validate a decoded or typed batch proposal before source processing."""
    if type(value) is BatchProposal:
        return BatchProposal.from_dict(_batch_dataclass_dict(value))
    return BatchProposal.from_dict(value)


def serialize_batch_proposal(value: object) -> str:
    """Serialize one validated batch proposal within the reader's byte limit."""
    checked = validate_batch_proposal(value)
    data = _batch_dataclass_dict(checked)
    _check_batch_size(data)
    return _batch_json(data)