# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/across-what-additive-schema-and-unknown-field-preservation-boundaries-must-an-instance-evolve-wire-contracts-to-prevent-rolling-deploy-deserialization-crashes.md
# Remnant Warrant:    https://corpusfetch.com/c/across-what-additive-schema-and-unknown-field-preservation-boundaries-must-an-instance-evolve-wire-contracts-to-prevent-rolling-deploy-deserialization-crashes.warrant.json
# Markdown SHA-256:   22fcdfe62dcb074ce8c36a5310256a556e01af101c819c67f7d1da513d3961aa
# Code AST SHA-256:   1fdfc586d9baf3da01dde7dd9e7164e61264e36a793368dd939486fbec00c742
# Verified Classes:   WireContractEvolutionError, AdditiveWirePayload, AdditiveWireContractGate
# Defensive Guards:   3 fail-closed invariants (zero external deps)

from dataclasses import dataclass, field
from typing import Any, Final


class WireContractEvolutionError(RuntimeError):
    """Raised when a wire payload violates additive-only schema evolution invariants."""
    pass


@dataclass(frozen=True)
class AdditiveWirePayload:
    schema_version: int
    task_id: str
    known_fields: dict[str, Any] = field(default_factory=dict)
    unknown_extensions: dict[str, Any] = field(default_factory=dict)


class AdditiveWireContractGate:
    """
    Enforces forward- and backward-compatible wire serialization:
    1. Preserves unrecognized fields in `unknown_extensions` during read-modify-write cycles.
    2. Forbids destructive type narrowing or required-field additions without defaults.
    """
    REQUIRED_CORE_KEYS: Final[frozenset[str]] = frozenset({"schema_version", "task_id"})
    KNOWN_OPTIONAL_KEYS: Final[frozenset[str]] = frozenset({"status", "priority", "metadata"})

    @classmethod
    def deserialize_preserving_unknowns(cls, raw_payload: dict[str, Any]) -> AdditiveWirePayload:
        missing = cls.REQUIRED_CORE_KEYS - raw_payload.keys()
        if missing:
            raise WireContractEvolutionError(f"Missing required core wire keys: {sorted(missing)}")

        version = raw_payload["schema_version"]
        task_id = raw_payload["task_id"]
        if not isinstance(version, int) or version < 1:
            raise WireContractEvolutionError("schema_version must be a positive integer.")
        if not isinstance(task_id, str) or not task_id.strip():
            raise WireContractEvolutionError("task_id must be a non-empty string.")

        known: dict[str, Any] = {}
        unknown: dict[str, Any] = {}
        for k, v in raw_payload.items():
            if k in cls.REQUIRED_CORE_KEYS:
                continue
            if k in cls.KNOWN_OPTIONAL_KEYS:
                known[k] = v
            else:
                unknown[k] = v

        return AdditiveWirePayload(
            schema_version=version,
            task_id=task_id,
            known_fields=known,
            unknown_extensions=unknown,
        )

    @staticmethod
    def serialize_roundtrip(envelope: AdditiveWirePayload) -> dict[str, Any]:
        out: dict[str, Any] = {
            "schema_version": envelope.schema_version,
            "task_id": envelope.task_id,
        }
        out.update(envelope.unknown_extensions)
        out.update(envelope.known_fields)
        return out