def canonical_payload(
value: Any,
*,
path: str,
capability: str,
_depth: int = 0,
_seen: frozenset[int] = frozenset(),
) -> Any:
"""Return the payload in canonical immutable form, or refuse it.
Fail-closed, per review of PR #94. The first fix froze containers and left
unrecognised objects by reference with the limit written in a docstring --
which is a note, not a boundary. A resolver handing over a nested mutable
object still let one compilation stage mutate what the next one read.
Frozen dataclasses are admitted, traversed, **and rebuilt when a field's
canonical form differs**. Returning them unchanged was the standing residual
for three rounds, and it was not small: ``@dataclass(frozen=True)`` freezes
the *bindings*, so a list field stays mutable and an enum field stays the
process-wide singleton, both shared with whoever handed the value over.
Rebuilding was tried twice before and reverted both times, because it
rewrote ``MemoryLayer`` to ``str`` inside a retrieval result that still
expected an enum and the recall stopped rendering. What changed is upstream:
memory now renders in its resolution stage and hands over strings, so the
live object graph that made rebuilding destructive is no longer in any
payload. The earlier fix is what made this one available -- the reverse of
how these rounds usually go.
A field that needs canonicalising but is ``init=False`` cannot be rebuilt,
and is refused rather than silently left alone.
It *returns* rather than only validating, because round 5 found that
admitting an enum member was not enough: the member is a process-wide
singleton, so a compilation stage setting an attribute on one reaches every
turn in the process, not merely the next stage. Canonicalising to ``.value``
is what makes the payload a value instead of a handle on shared state.
"""
if _depth > MAX_PAYLOAD_DEPTH:
raise ConfigurationError(
f"the resolved input from {capability!r} nests past "
f"{MAX_PAYLOAD_DEPTH} levels at {path}. The snapshot walks a payload "
"to freeze it and to check what it holds, and a structure this deep "
"is either a mistake or a value that should have been projected into "
"something flatter before being handed over."
)
if not _is_exact_leaf(value):
# Cycle detection by identity, over the *ancestors* of this value rather
# than everything visited: a payload may legitimately hold the same
# frozen value twice in two branches, and treating that as a cycle would
# refuse ordinary sharing. Only a value that contains itself is a cycle.
#
# Needed because a container can reach itself even when every link is
# nominally immutable -- a dict that holds itself, a frozen dataclass
# whose field was set to a tuple containing it. Without this the walk
# spins to a RecursionError, which is the same unusable failure the
# depth limit exists to avoid, arriving by a different route.
if id(value) in _seen:
raise ConfigurationError(
f"the resolved input from {capability!r} contains itself at "
f"{path}. A payload that reaches itself cannot be frozen or "
"walked to a conclusion; hand over a value with no cycle."
)
_seen = _seen | {id(value)}
if _is_exact_leaf(value):
return value
if isinstance(value, enum.Enum):
# Canonicalised to its ``.value``, not admitted as the member. An enum
# member is a process-wide singleton and Python lets you set attributes
# on one, so keeping the member would hand every compilation stage a
# handle on state shared by every turn in the process -- the global
# mutable configuration Guide 22 is about, arriving through a payload.
# ``MemoryLayer`` inside a frozen retrieval result is the case that
# occurs, and ``contribution_spec`` already emits ``.value`` for exactly
# this reason.
return canonical_payload(
value.value,
path=f"{path}.value",
capability=capability,
_depth=_depth + 1,
_seen=_seen,
)
if isinstance(value, _IMMUTABLE_LEAVES):
raise ConfigurationError(
f"the resolved input from {capability!r} carries "
f"{type(value).__name__} at {path}, a subclass of "
f"{type(value).__mro__[1].__name__} rather than the built-in itself. "
"A subclass passes every immutability check Python offers and can "
"still carry mutable attributes, so it would stay shared between "
"compilation stages. Pass the built-in value -- str(x), int(x) -- "
"which is what an enum member's .value already is."
)
if isinstance(value, Mapping):
rebuilt: dict[Any, Any] = {}
for key, item in value.items():
# The key as well as the value. ``deep_freeze`` rebuilds a mapping
# with its keys kept by reference, and a hashable-but-mutable key
# survives that intact: an object whose ``__hash__`` does not depend
# on the field being mutated stays a valid key while its state
# changes underneath. One compilation stage rewrites the label, the
# next reads the rewritten one, and the mapping never noticed.
# Named by type when the key is not a scalar: the default repr of
# a custom object carries a memory address, which would make this
# message differ between runs and useless in a diff.
shown = repr(key) if _is_exact_leaf(key) else f"<{type(key).__name__}>"
rebuilt[
canonical_payload(
key,
path=f"{path} key {shown}",
capability=capability,
_depth=_depth + 1,
_seen=_seen,
)
] = canonical_payload(
item,
path=f"{path}[{key!r}]",
capability=capability,
_depth=_depth + 1,
_seen=_seen,
)
return MappingProxyType(rebuilt)
if isinstance(value, (tuple, list, frozenset, set)):
items = tuple(
canonical_payload(
item,
path=f"{path}[{index}]",
capability=capability,
_depth=_depth + 1,
_seen=_seen,
)
for index, item in enumerate(value)
)
return frozenset(items) if isinstance(value, (frozenset, set)) else items
if dataclasses.is_dataclass(value) and not isinstance(value, type):
params = getattr(type(value), "__dataclass_params__", None)
if params is None or not params.frozen:
raise ConfigurationError(
f"the resolved input from {capability!r} carries a mutable dataclass "
f"{type(value).__name__} at {path}. A compilation stage could rewrite "
"its fields and the next stage would compile the rewritten value. "
"Declare it frozen, or project it into one."
)
# Traversed to refuse what it must not hold, and returned **unchanged**.
# Rebuilding it was tried and reverted: canonicalising an enum field of
# a nested value object rewrote ``MemoryLayer`` to ``str`` inside a
# retrieval result, and the recall stopped rendering. A frozen dataclass
# is the producing capability's own value, and its field types are part
# of what it means; the snapshot canonicalises the structure it owns,
# not the insides of a value somebody else defined.
#
# The residual, stated at its real width rather than the narrow version
# this comment used to give. ``dataclasses.fields()`` sees *declared
# fields*, so anything hanging off the class outside them is untouched:
# a mutable ``ClassVar`` (process-wide shared, reproduced), a
# ``@property`` returning module-level state, and -- on a dataclass
# without ``slots=True`` -- a writable ``__dict__`` that rewrites a
# declared field with no ``object.__setattr__`` needed. An enum member
# reachable only through a field is one instance of this, not the whole
# of it.
#
# Not enforced, and the choice is deliberate: refusing it means refusing
# frozen dataclasses that declare a ClassVar or omit slots, which is a
# real narrowing of what a capability may hand over, for a hazard that
# needs a capability to attack the process it is running in. Recorded so
# the next reader inherits the boundary rather than the impression that
# a frozen dataclass is airtight.
changed: dict[str, Any] = {}
for field_ in dataclasses.fields(value):
current = getattr(value, field_.name)
canon = canonical_payload(
current,
path=f"{path}.{field_.name}",
capability=capability,
_depth=_depth + 1,
_seen=_seen,
)
if canon is not current:
if not field_.init:
raise ConfigurationError(
f"the resolved input from {capability!r} holds "
f"{type(value).__name__} at {path}, whose field "
f"{field_.name!r} needs canonicalising but is init=False, "
"so it cannot be rebuilt. Project it into a value whose "
"fields are all constructor arguments."
)
changed[field_.name] = canon
return dataclasses.replace(value, **changed) if changed else value
raise ConfigurationError(
f"the resolved input from {capability!r} carries {type(value).__name__} at "
f"{path}, which has no immutable representation the snapshot can hold. "
"STG-7 makes compilation a pure function of the snapshot, and an object "
"one compilation stage can mutate makes the next stage's compile depend "
"on execution order. Hand over a frozen dataclass, a mapping, a sequence, "
"or a scalar -- project the object into one rather than passing it."
)