feat: release experimental YAML edit model under MIT

8d20d7d verified 8 days ago

14 kB

"""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)

	"""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)