DataOps-platform/app/core/edge_gateway/contracts.py

"""Immutable, fail-closed contracts for the enterprise edge boundary."""

from __future__ import annotations

import hashlib
import json
import math
import re
from collections.abc import Mapping
from dataclasses import dataclass
from dataclasses import field as dataclass_field
from datetime import UTC, datetime, timedelta
from decimal import Decimal, InvalidOperation
from types import MappingProxyType
from typing import ClassVar

from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey


class EdgeContractError(ValueError):
    """Raised when an edge contract is malformed or ambiguous."""


TASK_OPERATIONS = frozenset(
    {"collect", "profile", "quality", "lineage", "controlled_query"}
)
TASK_CLASSIFICATIONS = frozenset(
    {
        "raw",
        "recent_detail",
        "restricted",
        "desensitized_metadata",
        "statistics",
        "lineage",
        "evidence",
    }
)
EVENT_CLASSIFICATIONS = frozenset(
    {
        "desensitized_metadata",
        "statistics",
        "lineage",
        "evidence",
        "health_summary",
        "diagnostic_summary",
    }
)
CONTRACT_VERSION = 1
MAX_ATTEMPT = 5
PREFLIGHT_MAX_DEPTH = 64
PREFLIGHT_MAX_NODES = 10_000
PREFLIGHT_MAX_STRING_BYTES = 131_072
PREFLIGHT_MAX_TOTAL_BYTES = 1_048_576

_IDENTIFIER = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,254}$")
_SHA256 = re.compile(r"^[0-9a-f]{64}$")
_ED25519_SIGNATURE = re.compile(r"^[0-9a-f]{128}$")
_RFC3339 = re.compile(
    r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d{1,6})?(?:Z|[+-]\d{2}:\d{2})$"
)


def preflight_json(
    value: object,
    *,
    max_depth: int = PREFLIGHT_MAX_DEPTH,
    max_nodes: int = PREFLIGHT_MAX_NODES,
    max_string_bytes: int = PREFLIGHT_MAX_STRING_BYTES,
    max_total_bytes: int = PREFLIGHT_MAX_TOTAL_BYTES,
) -> None:
    """Bound a JSON-like graph iteratively before copying or encoding it."""

    nodes = 0
    estimated_bytes = 0
    active: set[int] = set()
    stack: list[tuple[bool, object, int]] = [(False, value, 0)]
    try:
        while stack:
            exiting, current, depth = stack.pop()
            if exiting:
                active.remove(id(current))
                continue
            if depth > max_depth:
                raise EdgeContractError("contract exceeds the maximum depth")
            nodes += 1
            if nodes > max_nodes:
                raise EdgeContractError("contract exceeds the maximum node count")
            if isinstance(current, str):
                size = len(current.encode("utf-8"))
                estimated_bytes += size + 16
                if estimated_bytes > max_total_bytes:
                    raise EdgeContractError(
                        "contract exceeds the conservative byte limit"
                    )
                if size > max_string_bytes:
                    raise EdgeContractError("contract string exceeds the byte limit")
            elif current is None or isinstance(current, bool):
                estimated_bytes += 8
            elif isinstance(current, int):
                if current.bit_length() > 4096:
                    raise EdgeContractError("contract integer exceeds the byte limit")
                estimated_bytes += max(8, current.bit_length() // 3 + 4)
            elif isinstance(current, float):
                if not math.isfinite(current):
                    raise EdgeContractError("contract numbers must be finite")
                estimated_bytes += 32
            elif isinstance(current, Mapping):
                identity = id(current)
                if identity in active:
                    raise EdgeContractError("contract contains a cycle")
                active.add(identity)
                stack.append((True, current, depth))
                estimated_bytes += 16
                children: list[object] = []
                for key, item in current.items():
                    if not isinstance(key, str):
                        raise EdgeContractError("contract property names must be strings")
                    key_size = len(key.encode("utf-8"))
                    if key_size > max_string_bytes:
                        raise EdgeContractError("contract key exceeds the byte limit")
                    estimated_bytes += key_size + 8
                    children.append(item)
                stack.extend((False, item, depth + 1) for item in reversed(children))
            elif isinstance(current, (list, tuple)):
                identity = id(current)
                if identity in active:
                    raise EdgeContractError("contract contains a cycle")
                active.add(identity)
                stack.append((True, current, depth))
                estimated_bytes += 16
                stack.extend((False, item, depth + 1) for item in reversed(current))
            else:
                raise EdgeContractError("contract contains a non-JSON value")
            if estimated_bytes > max_total_bytes:
                raise EdgeContractError("contract exceeds the conservative byte limit")
    except (MemoryError, RecursionError) as exc:
        raise EdgeContractError("contract exceeds safe structural limits") from exc


def _length_prefixed(tag: bytes, content: bytes) -> bytes:
    return tag + str(len(content)).encode("ascii") + b":" + content


def _number_text(value: int | float) -> str:
    try:
        number = Decimal(value if isinstance(value, int) else str(value))
    except (InvalidOperation, ValueError) as exc:
        raise EdgeContractError("contract number is invalid") from exc
    if not number.is_finite():
        raise EdgeContractError("contract numbers must be finite")
    if number == 0:
        return "0"
    normalized = format(number.normalize(), "f")
    if "." in normalized:
        normalized = normalized.rstrip("0").rstrip(".")
    return normalized


def _canonical_encode(value: object) -> bytes:
    if value is None:
        return b"n"
    if isinstance(value, bool):
        return b"b1" if value else b"b0"
    if isinstance(value, (int, float)):
        return _length_prefixed(b"d", _number_text(value).encode("ascii"))
    if isinstance(value, str):
        return _length_prefixed(b"s", value.encode("utf-8"))
    if isinstance(value, Mapping):
        encoded = bytearray(b"m" + str(len(value)).encode("ascii") + b":")
        for key in sorted(value, key=lambda item: item.encode("utf-8")):
            key_bytes = key.encode("utf-8")
            encoded.extend(_length_prefixed(b"k", key_bytes))
            encoded.extend(_canonical_encode(value[key]))
        return bytes(encoded)
    if isinstance(value, (list, tuple)):
        encoded = bytearray(b"l" + str(len(value)).encode("ascii") + b":")
        for item in value:
            encoded.extend(_canonical_encode(item))
        return bytes(encoded)
    raise EdgeContractError("contract contains a non-JSON value")


def canonical_json_bytes(value: object) -> bytes:
    """Encode the explicit type-tagged, length-prefixed canonical protocol."""

    preflight_json(value)
    try:
        return _canonical_encode(value)
    except (MemoryError, RecursionError) as exc:
        raise EdgeContractError("contract exceeds safe encoding limits") from exc


def canonical_sha256(value: object) -> str:
    return hashlib.sha256(canonical_json_bytes(value)).hexdigest()


def strict_json_bytes(value: object) -> bytes:
    """Serialize validated data as strict transport/storage JSON."""

    preflight_json(value)
    try:
        return json.dumps(
            value,
            sort_keys=True,
            separators=(",", ":"),
            ensure_ascii=False,
            allow_nan=False,
        ).encode("utf-8")
    except (MemoryError, RecursionError, TypeError, ValueError) as exc:
        raise EdgeContractError("contract cannot be serialized as strict JSON") from exc


def stable_event_id(identity: Mapping[str, object]) -> str:
    if not isinstance(identity, Mapping):
        raise EdgeContractError("event identity must be a mapping")
    normalized = dict(identity)
    if "occurred_at" in normalized:
        normalized["occurred_at"] = _timestamp(
            normalized["occurred_at"], "occurred_at"
        )
    return f"evt_{canonical_sha256(normalized)}"


def _strict_mapping(value: object, *, label: str) -> dict[str, object]:
    if not isinstance(value, Mapping):
        raise EdgeContractError(f"{label} must be a mapping")
    preflight_json(value)
    return dict(value)


def _require_exact_properties(
    value: Mapping[str, object],
    *,
    required: frozenset[str],
    optional: frozenset[str] = frozenset(),
) -> None:
    keys = frozenset(value)
    unknown = keys - required - optional
    missing = required - keys
    if unknown:
        raise EdgeContractError(
            f"contract has unknown properties: {', '.join(sorted(unknown))}"
        )
    if missing:
        raise EdgeContractError(
            f"contract is missing properties: {', '.join(sorted(missing))}"
        )


def _text(value: object, label: str, *, maximum: int = 255) -> str:
    if (
        not isinstance(value, str)
        or not value
        or value.strip() != value
        or len(value.encode("utf-8")) > maximum
        or "\x00" in value
    ):
        raise EdgeContractError(f"{label} is invalid")
    return value


def _identifier(value: object, label: str) -> str:
    candidate = _text(value, label)
    if not _IDENTIFIER.fullmatch(candidate):
        raise EdgeContractError(f"{label} is invalid")
    return candidate


def _attempt(value: object) -> int:
    if isinstance(value, bool) or not isinstance(value, int) or not 1 <= value <= 5:
        raise EdgeContractError("attempt must be between 1 and 5")
    return value


def _version(value: object) -> int:
    if isinstance(value, bool) or value != CONTRACT_VERSION:
        raise EdgeContractError("contract_version is not supported")
    return CONTRACT_VERSION


def _timestamp(value: object, label: str) -> str:
    candidate = _text(value, label, maximum=64)
    if not _RFC3339.fullmatch(candidate):
        raise EdgeContractError(f"{label} must be an RFC3339 timestamp")
    try:
        parsed = datetime.fromisoformat(candidate.replace("Z", "+00:00"))
    except ValueError as exc:
        raise EdgeContractError(f"{label} must be an RFC3339 timestamp") from exc
    if parsed.tzinfo is None or parsed.utcoffset() is None:
        raise EdgeContractError(f"{label} must be an RFC3339 timestamp")
    utc = parsed.astimezone(UTC)
    canonical = utc.strftime("%Y-%m-%dT%H:%M:%S")
    if utc.microsecond:
        canonical += f".{utc.microsecond:06d}".rstrip("0")
    return f"{canonical}Z"


def canonical_timestamp(value: object, label: str = "timestamp") -> str:
    return _timestamp(value, label)


def _digest(value: object, label: str) -> str:
    candidate = _text(value, label, maximum=64)
    if not _SHA256.fullmatch(candidate):
        raise EdgeContractError(f"{label} must be a lowercase SHA256 digest")
    return candidate


def _json_copy(value: object) -> object:
    try:
        encoded = strict_json_bytes(value)
        return json.loads(encoded.decode("utf-8"))
    except (MemoryError, RecursionError, TypeError, ValueError) as exc:
        raise EdgeContractError("contract cannot be copied as strict JSON") from exc


def _deep_freeze(value: object) -> object:
    if isinstance(value, dict):
        return MappingProxyType({key: _deep_freeze(item) for key, item in value.items()})
    if isinstance(value, list):
        return tuple(_deep_freeze(item) for item in value)
    return value


def _deep_thaw(value: object) -> object:
    if isinstance(value, Mapping):
        return {key: _deep_thaw(item) for key, item in value.items()}
    if isinstance(value, tuple):
        return [_deep_thaw(item) for item in value]
    return value


@dataclass(frozen=True, slots=True)
class EdgeTaskContract:
    task_id: str
    gateway_id: str
    environment: str
    network_zone: str
    purpose: str
    classification: str
    operation: str
    contract_version: int
    deadline_at: str
    attempt: int
    idempotency_key: str
    policy_digest: str

    def __post_init__(self) -> None:
        values = {
            "task_id": _identifier(self.task_id, "task_id"),
            "gateway_id": _identifier(self.gateway_id, "gateway_id"),
            "environment": _identifier(self.environment, "environment"),
            "network_zone": _identifier(self.network_zone, "network_zone"),
            "purpose": _identifier(self.purpose, "purpose"),
            "contract_version": _version(self.contract_version),
            "deadline_at": _timestamp(self.deadline_at, "deadline_at"),
            "attempt": _attempt(self.attempt),
            "idempotency_key": _identifier(self.idempotency_key, "idempotency_key"),
            "policy_digest": _digest(self.policy_digest, "policy_digest"),
        }
        if self.operation not in TASK_OPERATIONS:
            raise EdgeContractError("task operation is not approved")
        if self.classification not in TASK_CLASSIFICATIONS:
            raise EdgeContractError("task classification is invalid")
        for field, value in values.items():
            object.__setattr__(self, field, value)

    @property
    def task_type(self) -> str:
        return self.operation

    @classmethod
    def from_mapping(cls, value: Mapping[str, object]) -> EdgeTaskContract:
        data = _strict_mapping(value, label="task contract")
        required = frozenset(
            {
                "task_id",
                "gateway_id",
                "environment",
                "network_zone",
                "purpose",
                "classification",
                "contract_version",
                "deadline_at",
                "attempt",
                "idempotency_key",
                "policy_digest",
            }
        )
        _require_exact_properties(
            data,
            required=required,
            optional=frozenset({"task_type", "operation"}),
        )
        if ("task_type" in data) == ("operation" in data):
            raise EdgeContractError(
                "task contract requires exactly one operation property"
            )
        return cls(
            task_id=data["task_id"],  # type: ignore[arg-type]
            gateway_id=data["gateway_id"],  # type: ignore[arg-type]
            environment=data["environment"],  # type: ignore[arg-type]
            network_zone=data["network_zone"],  # type: ignore[arg-type]
            purpose=data["purpose"],  # type: ignore[arg-type]
            classification=data["classification"],  # type: ignore[arg-type]
            operation=data.get("task_type", data.get("operation")),  # type: ignore[arg-type]
            contract_version=data["contract_version"],  # type: ignore[arg-type]
            deadline_at=data["deadline_at"],  # type: ignore[arg-type]
            attempt=data["attempt"],  # type: ignore[arg-type]
            idempotency_key=data["idempotency_key"],  # type: ignore[arg-type]
            policy_digest=data["policy_digest"],  # type: ignore[arg-type]
        )

    def to_mapping(self) -> dict[str, object]:
        return {
            "task_id": self.task_id,
            "gateway_id": self.gateway_id,
            "environment": self.environment,
            "network_zone": self.network_zone,
            "purpose": self.purpose,
            "classification": self.classification,
            "task_type": self.operation,
            "contract_version": self.contract_version,
            "deadline_at": self.deadline_at,
            "attempt": self.attempt,
            "idempotency_key": self.idempotency_key,
            "policy_digest": self.policy_digest,
        }

    @property
    def digest(self) -> str:
        return canonical_sha256(self.to_mapping())


@dataclass(frozen=True, slots=True)
class SignedTaskEnvelope:
    """Control-plane authority over an exact, context-bound task contract."""

    task: EdgeTaskContract
    authority_key_id: str
    signature_algorithm: str
    contract_digest: str
    gateway_id: str
    environment: str
    network_zone: str
    policy_digest: str
    purpose: str
    issued_at: str
    expires_at: str
    signature: str
    _verified: bool = dataclass_field(default=False, init=False, repr=False, compare=False)

    _UNSIGNED_FIELDS: ClassVar[frozenset[str]] = frozenset(
        {
            "task", "authority_key_id", "signature_algorithm", "contract_digest",
            "gateway_id", "environment", "network_zone", "policy_digest", "purpose",
            "issued_at", "expires_at",
        }
    )

    def __post_init__(self) -> None:
        if not isinstance(self.task, EdgeTaskContract):
            raise EdgeContractError("signed task envelope requires a task contract")
        normalized_task = EdgeTaskContract.from_mapping(self.task.to_mapping())
        unsigned = {
            "task": normalized_task.to_mapping(),
            "authority_key_id": self.authority_key_id,
            "signature_algorithm": self.signature_algorithm,
            "contract_digest": self.contract_digest,
            "gateway_id": self.gateway_id,
            "environment": self.environment,
            "network_zone": self.network_zone,
            "policy_digest": self.policy_digest,
            "purpose": self.purpose,
            "issued_at": self.issued_at,
            "expires_at": self.expires_at,
        }
        self.canonical_unsigned_bytes(unsigned)
        if not isinstance(self.signature, str) or not _ED25519_SIGNATURE.fullmatch(self.signature):
            raise EdgeContractError("signature must be a lowercase Ed25519 signature")
        object.__setattr__(self, "task", normalized_task)

    @classmethod
    def canonical_unsigned_bytes(cls, value: Mapping[str, object]) -> bytes:
        data = _strict_mapping(value, label="signed task envelope")
        _require_exact_properties(data, required=cls._UNSIGNED_FIELDS)
        task_raw = data["task"]
        if isinstance(task_raw, EdgeTaskContract):
            task_raw = task_raw.to_mapping()
        task = EdgeTaskContract.from_mapping(task_raw)  # type: ignore[arg-type]
        normalized = {
            "task": task.to_mapping(),
            "authority_key_id": _identifier(data["authority_key_id"], "authority_key_id"),
            "signature_algorithm": data["signature_algorithm"],
            "contract_digest": _digest(data["contract_digest"], "contract_digest"),
            "gateway_id": _identifier(data["gateway_id"], "gateway_id"),
            "environment": _identifier(data["environment"], "environment"),
            "network_zone": _identifier(data["network_zone"], "network_zone"),
            "policy_digest": _digest(data["policy_digest"], "policy_digest"),
            "purpose": _identifier(data["purpose"], "purpose"),
            "issued_at": _timestamp(data["issued_at"], "issued_at"),
            "expires_at": _timestamp(data["expires_at"], "expires_at"),
        }
        if normalized["signature_algorithm"] != "Ed25519":
            raise EdgeContractError("signature_algorithm is not supported")
        for field in ("gateway_id", "environment", "network_zone", "policy_digest", "purpose"):
            if normalized[field] != task.to_mapping()[field]:
                raise EdgeContractError(f"signed authority {field} does not bind the task")
        if normalized["contract_digest"] != task.digest:
            raise EdgeContractError("contract_digest does not bind the task")
        issued = datetime.fromisoformat(str(normalized["issued_at"]).replace("Z", "+00:00"))
        expires = datetime.fromisoformat(str(normalized["expires_at"]).replace("Z", "+00:00"))
        deadline = datetime.fromisoformat(task.deadline_at.replace("Z", "+00:00"))
        if expires <= issued or expires > deadline:
            raise EdgeContractError("signed authority validity window is invalid")
        return canonical_json_bytes(normalized)

    @classmethod
    def verify_mapping(
        cls,
        value: Mapping[str, object],
        *,
        authority_keys: Mapping[str, bytes | Ed25519PublicKey],
        now: datetime,
        allowed_future_skew_seconds: int = 0,
    ) -> SignedTaskEnvelope:
        data = _strict_mapping(value, label="signed task envelope")
        _require_exact_properties(data, required=cls._UNSIGNED_FIELDS | {"signature"})
        if not isinstance(now, datetime) or now.tzinfo is None or now.utcoffset() is None:
            raise EdgeContractError("verification time must be timezone-aware")
        if (
            isinstance(allowed_future_skew_seconds, bool)
            or not isinstance(allowed_future_skew_seconds, int)
            or not 0 <= allowed_future_skew_seconds <= 300
        ):
            raise EdgeContractError("task authority clock skew is invalid")
        signature = _text(data["signature"], "signature", maximum=128)
        if not _ED25519_SIGNATURE.fullmatch(signature):
            raise EdgeContractError("signature must be a lowercase Ed25519 signature")
        unsigned = {key: data[key] for key in cls._UNSIGNED_FIELDS}
        signed_bytes = cls.canonical_unsigned_bytes(unsigned)
        key_id = _identifier(data["authority_key_id"], "authority_key_id")
        key = authority_keys.get(key_id)
        if key is None:
            raise EdgeContractError("task authority key is not trusted")
        try:
            public_key = (
                key if isinstance(key, Ed25519PublicKey)
                else Ed25519PublicKey.from_public_bytes(bytes(key))
            )
            public_key.verify(bytes.fromhex(signature), signed_bytes)
        except (InvalidSignature, TypeError, ValueError) as exc:
            raise EdgeContractError("task authority signature is invalid") from exc
        issued_at = _timestamp(data["issued_at"], "issued_at")
        expires_at = _timestamp(data["expires_at"], "expires_at")
        instant = now.astimezone(UTC)
        issued = datetime.fromisoformat(issued_at.replace("Z", "+00:00"))
        expires = datetime.fromisoformat(expires_at.replace("Z", "+00:00"))
        if instant + timedelta(seconds=allowed_future_skew_seconds) < issued:
            raise EdgeContractError("signed task is not yet valid")
        if instant >= expires:
            raise EdgeContractError("signed task authority has expired")
        task = EdgeTaskContract.from_mapping(data["task"])  # type: ignore[arg-type]
        envelope = cls(
            task=task,
            authority_key_id=key_id,
            signature_algorithm="Ed25519",
            contract_digest=task.digest,
            gateway_id=task.gateway_id,
            environment=task.environment,
            network_zone=task.network_zone,
            policy_digest=task.policy_digest,
            purpose=task.purpose,
            issued_at=issued_at,
            expires_at=expires_at,
            signature=signature,
        )
        object.__setattr__(envelope, "_verified", True)
        return envelope

    @property
    def is_verified(self) -> bool:
        return self._verified

    def to_mapping(self) -> dict[str, object]:
        return {
            "task": self.task.to_mapping(),
            "authority_key_id": self.authority_key_id,
            "signature_algorithm": self.signature_algorithm,
            "contract_digest": self.contract_digest,
            "gateway_id": self.gateway_id,
            "environment": self.environment,
            "network_zone": self.network_zone,
            "policy_digest": self.policy_digest,
            "purpose": self.purpose,
            "issued_at": self.issued_at,
            "expires_at": self.expires_at,
            "signature": self.signature,
        }

    @property
    def digest(self) -> str:
        return canonical_sha256(self.to_mapping())


@dataclass(frozen=True, slots=True)
class EdgeEventContract:
    event_id: str
    task_id: str
    gateway_id: str
    environment: str
    network_zone: str
    purpose: str
    classification: str
    contract_version: int
    occurred_at: str
    attempt: int
    idempotency_key: str
    policy_digest: str
    payload: Mapping[str, object]

    def __post_init__(self) -> None:
        payload = _strict_mapping(self.payload, label="event payload")
        payload_copy = _json_copy(payload)
        if not isinstance(payload_copy, dict):
            raise EdgeContractError("event payload must be a mapping")
        values = {
            "task_id": _identifier(self.task_id, "task_id"),
            "gateway_id": _identifier(self.gateway_id, "gateway_id"),
            "environment": _identifier(self.environment, "environment"),
            "network_zone": _identifier(self.network_zone, "network_zone"),
            "purpose": _identifier(self.purpose, "purpose"),
            "contract_version": _version(self.contract_version),
            "occurred_at": _timestamp(self.occurred_at, "occurred_at"),
            "attempt": _attempt(self.attempt),
            "idempotency_key": _identifier(self.idempotency_key, "idempotency_key"),
            "policy_digest": _digest(self.policy_digest, "policy_digest"),
        }
        if self.classification not in EVENT_CLASSIFICATIONS:
            raise EdgeContractError("event classification is not approved")
        identity = {
            "task_id": values["task_id"],
            "gateway_id": values["gateway_id"],
            "environment": values["environment"],
            "network_zone": values["network_zone"],
            "purpose": values["purpose"],
            "classification": self.classification,
            "contract_version": values["contract_version"],
            "occurred_at": values["occurred_at"],
            "attempt": values["attempt"],
            "idempotency_key": values["idempotency_key"],
            "policy_digest": values["policy_digest"],
            "payload": payload_copy,
        }
        expected_id = stable_event_id(identity)
        if self.event_id != expected_id:
            raise EdgeContractError("event_id is not the stable event identifier")
        object.__setattr__(self, "event_id", expected_id)
        object.__setattr__(self, "payload", _deep_freeze(payload_copy))
        for field, value in values.items():
            object.__setattr__(self, field, value)

    @classmethod
    def from_mapping(cls, value: Mapping[str, object]) -> EdgeEventContract:
        data = _strict_mapping(value, label="event contract")
        required = frozenset(
            {
                "event_id",
                "task_id",
                "gateway_id",
                "environment",
                "network_zone",
                "purpose",
                "classification",
                "contract_version",
                "occurred_at",
                "attempt",
                "idempotency_key",
                "policy_digest",
                "payload",
            }
        )
        _require_exact_properties(data, required=required)
        return cls(**data)  # type: ignore[arg-type]

    def to_mapping(self) -> dict[str, object]:
        return {
            "event_id": self.event_id,
            "task_id": self.task_id,
            "gateway_id": self.gateway_id,
            "environment": self.environment,
            "network_zone": self.network_zone,
            "purpose": self.purpose,
            "classification": self.classification,
            "contract_version": self.contract_version,
            "occurred_at": self.occurred_at,
            "attempt": self.attempt,
            "idempotency_key": self.idempotency_key,
            "policy_digest": self.policy_digest,
            "payload": _deep_thaw(self.payload),
        }

    @property
    def digest(self) -> str:
        return canonical_sha256(self.to_mapping())