packages/cultcache-py/CHANGELOG.md

@@ -9,6 +9,8 @@
 
 ### Breaking
 
+- A record whose payload carries a slot past its reader's declared slots, at any depth (the document, a nested record, a list element, a union arm, a unit arm), is read, held read-only, and every write or removal of its key (`put`, `put_global`, `update`, `put_envelope`, `put_envelopes`, `delete`, `delete_global`) is refused with `SchemaConflictError` before any store is touched (`cultlib-gaps:ruling:extra-trailing-slots`). Reading dropped those slots silently and the next write shed them. `SchemaConflictError.unread_slots` is an `UnreadSlots(type_name, slot_count, declared_count, occurrences, depth0_only)` and the message names the type and both counts, never a slot value. A depth-0 mark is consented, and the record writable, only when the record is stored under an id the document lists in `compatible_schema_ids` and that id's catalog entry has slots past the document's, covering the mark's slot count (`cultlib-gaps:ruling:slot-mark-consent`); a nested mark, or a mark under the document's own id, keeps the record read-only whatever is declared. Missing trailing slots keep their declared defaults. An untouched marked record keeps its stored bytes through every other write. Added `CultCache.is_writable(document, key)` (advisory; the write still decides: it also answers False for a foreign-carried key, for a key held as a type the document does not own, and for a global document at a non-global key; `delete` asks the same judges, so removing a key held as another type is refused with `SchemaConflictError` unless the document owns that record's schema id (its own or one it lists compatible), in which case the removal removes that record, its file row and its lookups, as a write of the key would replace it; it returned silently and removed nothing), `DocumentDefinition.decode_payload_marked`, and `UnreadSlots`. A document with a custom `decode` (no slot codec) never marks. A store holding a variant record is still unreadable here. Implies a minor bump while the version is 0.x (0.3.0 to 0.4.0): code that wrote back a record from a newer writer now gets `SchemaConflictError`.
+
 - A cache holds one record per key whatever its type (`variants:ruling:cache-refuses-retype`). A `put`, `put_envelope`, `put_envelopes` or `put_global` of a second type under a key the cache holds, or under a key a store carries foreign, is refused with `SchemaConflictError` naming the key, even with no store, and changes nothing; retype a key by deleting it, then putting the new type. A write may replace another registered type's record only when the written type owns its schema id (declares it or lists it compatible) and both types share one home store. A load that reads one key from two stores, whatever the types, is refused with `CultCacheError` naming the key and both types, and admits nothing. A store that holds a record no registered type claims now loads and is carried: the cache does not serve it, the store keeps its bytes through every write, `delete` or `put` at its key is refused with `SchemaConflictError`, and registering a type that declares its schema id claims it at the next load. It raised `CultCacheError` "Unknown persisted document type". Two global types share the key `__global__`, so a second global type in one cache is now refused, and a store that already holds two global types at `__global__` (a JSONL store written by caches that never pulled could produce one) no longer loads: the pull raises `CultCacheError` and admits nothing. C# refuses a second persisted global on pull the same way (`cultcache-store-composition.md`, "Globals never invented"); repair such a file by deleting one of the two records with the runtime that wrote it.
 - Added `foreign_records()` and the frozen `ForeignRecord(key, schema_id, schema_name, stored_at)`, sorted by key in UTF-16 code-unit order. `stores.retyped` and `stores.written_owns` lose their underscore: the cache judge and the single-file store's commit guard ask one predicate.
 

packages/cultcache-py/README.md

@@ -97,6 +97,7 @@ store.
 - `update_global(...)`
 - `delete(...)`
 - `delete_global(...)`
+- `is_writable(document, key)` (advisory: would a write or removal of the key be refused; see below)
 - `snapshot()`
 
 ## Indexes And Roles
@@ -180,6 +181,16 @@ The payload is a MessagePack array. Field keys are durable slot indexes:
 - newly added fields should use new keys
 - fields with defaults tolerate missing or nil slots when older payloads are read
 
+A payload carrying a slot past the slots a reader declares, at any depth (the document, a nested record, a list element, a
+union arm, a unit arm), was written by a newer build. It is read, and the record is held read-only: `put`, `update`,
+`put_envelope(s)`, `delete` and their global forms refuse its key with `SchemaConflictError` before any store is touched, and
+`SchemaConflictError.unread_slots` is an `UnreadSlots(type_name, slot_count, declared_count, occurrences, depth0_only)` naming the
+first type read with extra slots (never a slot value). `CultCache.is_writable(document, key)` answers the same question before
+the write: False for such a record, for a key a store carries foreign, and for a key held as another type's record; the write
+still decides. `DocumentDefinition.decode_payload_marked(payload)` returns the value and its `UnreadSlots` (or None). Declaring the
+stored id in `compatible_schema_ids` consents to the loss of depth-0 slots that id's catalog entry reports; a nested mark never is.
+A store keeps a read-only record's bytes through every other write. A document with a custom `decode` never marks.
+
 That matches the Rust `#[derive(DatabaseEntry)]` formatter shape and the C#
 `[Key(n)]` intent: schema evolution comes from stable field slots, not source
 member order.

packages/cultcache-py/src/cultcache_py/__init__.py

@@ -9,6 +9,7 @@ from .documents import (
     DatabaseEntryField,
     DocumentDefinition,
     UnionArm,
+    UnreadSlots,
     database_entry_field,
     define_database_entry_type,
     define_document_registry,
@@ -38,6 +39,7 @@ __all__ = [
     "SlotCodec",
     "StoreUnreadableError",
     "UnionArm",
+    "UnreadSlots",
     "leaves",
     "database_entry_field",
     "define_database_entry_type",

packages/cultcache-py/src/cultcache_py/cache.py

@@ -8,7 +8,7 @@ from dataclasses import dataclass, field, replace
 from typing import Any, Callable, Generic, Iterable, TypeVar
 
 from .backing_store import BackingStore, CultCacheEnvelope
-from .documents import DocumentDefinition, extract_value
+from .documents import DocumentDefinition, UnreadSlots, extract_value
 from .stores import SchemaConflictError, retyped, written_owns
 
 T = TypeVar("T")
@@ -123,6 +123,10 @@ class _State:
     # The records the last pull read that no registered type claims, sorted by key. Derived from the pull and swapped with the
     # rest of the state, written nowhere else; it decides only refusals at its keys. A store keeps their bytes.
     foreign: list["ForeignRecord"] = field(default_factory=list)
+    # The records held read-only, by key (a key holds one record) with the type that holds it: the payload carried slots past the
+    # declared ones and nothing consents to the loss. Derived from the load or the write that admitted the record and swapped with
+    # the rest of the state; it decides only refusals at its keys, and any other key is absent.
+    unread: dict[str, tuple[str, UnreadSlots]] = field(default_factory=dict)
 
 
 @dataclass(frozen=True)
@@ -337,6 +341,7 @@ class CultCache:
         routes, generic = self._state.stores_by_type, self._state.generic_store
         seen_globals: set[str] = set()
         foreign: list[ForeignRecord] = []
+        loaded_unread: dict[str, tuple[str, UnreadSlots]] = {}
         type_at_key: dict[str, tuple[str, int]] = {}  # one record per key across every store: its type (or schema name) and store
         stores: list[BackingStore] = []
         for candidate in [*self._state.stores_by_type.values(), self._state.generic_store]:
@@ -376,8 +381,10 @@ class CultCache:
                         loaded_legacy_global_keys[envelope.type] = envelope.key
                         envelope = replace(envelope, key=GLOBAL_KEY)
                 self._claim_key(type_at_key, envelope.key, envelope.type, position)
-                envelope = self._stamped(document, envelope)
-                value = document.decode_payload(envelope.payload)
+                arrived, envelope = envelope, self._stamped(document, envelope)
+                value, unread = self._decode_marked(document, arrived)
+                if unread is not None:
+                    loaded_unread[envelope.key] = (envelope.type, unread)
                 loaded_values.setdefault(envelope.type, {})[envelope.key] = value
                 read_values[(envelope.type, envelope.key)] = value
                 loaded_envelopes.setdefault(envelope.type, {})[envelope.key] = envelope
@@ -393,6 +400,7 @@ class CultCache:
             self._state.lookup_keys,
             self._state.legacy_global_keys,
             self._state.foreign,
+            self._state.unread,
         ) = (
             loaded_values,
             loaded_envelopes,
@@ -401,6 +409,7 @@ class CultCache:
             loaded_lookup_keys,
             loaded_legacy_global_keys,
             sorted(foreign, key=lambda record: _utf16_order(record.key)),
+            loaded_unread,
         )
 
     @staticmethod
@@ -491,9 +500,9 @@ class CultCache:
     def put_envelope(self, document: DocumentDefinition[T], envelope: CultCacheEnvelope) -> T:
         self._assert_registered(document)
         self._check_envelope(document, envelope)
-        envelope = self._stamped(document, envelope)
-        value = document.decode_payload(envelope.payload)
-        self._write(document, [(envelope, value)], batch=False)
+        arrived, envelope = envelope, self._stamped(document, envelope)
+        value, unread = self._decode_marked(document, envelope)
+        self._write(document, [(envelope, value)], batch=False, unread={envelope.key: unread})
         return value
 
     @_mutating
@@ -501,13 +510,33 @@ class CultCache:
         self._assert_registered(document)
         values: list[T] = []
         stamped: list[CultCacheEnvelope] = []
+        unread: dict[str, UnreadSlots | None] = {}
         for envelope in envelopes:
             self._check_envelope(document, envelope)
             stamped.append(self._stamped(document, envelope))
-            values.append(document.decode_payload(envelope.payload))
-        self._write(document, list(zip(stamped, values)), batch=True)
+            value, unread[envelope.key] = self._decode_marked(document, stamped[-1])
+            values.append(value)
+        self._write(document, list(zip(stamped, values)), batch=True, unread=unread)
         return values
 
+    @staticmethod
+    def _decode_marked(document: DocumentDefinition[Any], held_as: CultCacheEnvelope) -> tuple[Any, UnreadSlots | None]:
+        """The payload decoded under the document, with the mark of any slot it carried past the declared ones, or no mark when the
+        loss is consented. `held_as` is the envelope under the id the record is stored under: a load's arrival, a write's stamped
+        envelope, so a key is writable in memory exactly when it is after a reload. A depth-0 mark is consented only when that id is
+        one the document lists compatible and its catalog entry has slots past the document's that cover the mark's slot count (the
+        loss a compatible-id declaration consents to, cultlib-gaps:ruling:slot-mark-consent). Any other mark keeps the record read-only."""
+        value, slots = document.decode_payload_marked(held_as.payload)
+        if slots is None:
+            return value, None
+        entry, stored = document.catalog_entry(), held_as.catalog_entry
+        if slots.depth0_only and stored is not None and held_as.schema_id in entry.compatible_schema_ids and held_as.schema_id != entry.schema_id:
+            declared = {member.slot for member in entry.members}
+            reported = [member.slot for member in stored.members if member.slot not in declared]
+            if reported and slots.slot_count <= max(reported) + 1:
+                return value, None
+        return value, slots
+
     @staticmethod
     def _stamped(document: DocumentDefinition[Any], envelope: CultCacheEnvelope) -> CultCacheEnvelope:
         """The record under its registered schema's id, whatever id the envelope arrived under, so the catalog entry a store
@@ -527,7 +556,12 @@ class CultCache:
             raise CultCacheError(f"Global document {document.type} must use key {GLOBAL_KEY}")
 
     def _write(
-        self, document: DocumentDefinition[Any], records: list[tuple[CultCacheEnvelope, Any]], *, batch: bool
+        self,
+        document: DocumentDefinition[Any],
+        records: list[tuple[CultCacheEnvelope, Any]],
+        *,
+        batch: bool,
+        unread: dict[str, UnreadSlots | None] | None = None,
     ) -> None:
         """Validates every record (home and name and index extractors included) before the store is
         touched; once the store accepts, applying to the cache has no fallible step."""
@@ -535,6 +569,7 @@ class CultCache:
         # A batch naming a key twice lands the last record for it, as C# CultCacheBatch stages one operation per key. The
         # judge, the store and the apply all read this one list.
         records = list({envelope.key: (envelope, value) for envelope, value in records}.values())
+        self._refuse_unread([envelope.key for envelope, _ in records])
         name_extractor = self._state.name_extractors.get(type)
         index_extractors = self._state.index_extractors.get(type, {})
         admitted = [
@@ -566,6 +601,9 @@ class CultCache:
             self._remove_lookups(type, envelope.key)
             values[envelope.key] = value
             envelopes[envelope.key] = envelope
+            # A held mark refused this write above, so a key reaching here carries none to clear.
+            if (mark := (unread or {}).get(envelope.key)) is not None:
+                self._state.unread[envelope.key] = (type, mark)
             lookup_keys[envelope.key] = keys
             if keys.name is not None:
                 self._state.names[(type, keys.name)] = envelope.key
@@ -592,6 +630,51 @@ class CultCache:
                 replaced.append((other, envelope.key))
         return replaced
 
+    @_locked
+    def is_writable(self, document: DocumentDefinition[Any], key: str) -> bool:
+        """Whether a write or removal of the key by this document would pass the judges that run before any store is touched: False for
+        a record held read-only (its payload carried slots past the declared ones, at any depth, and nothing consents to the loss, so
+        writing it would drop data a newer writer stored), for a key a store carries foreign, and for a key held as another type's
+        record this document does not own. Advisory (the write still decides): it answers at call time, and a key nothing holds is
+        writable."""
+        self._assert_registered(document)
+        try:
+            self._refuse_at(document, key)
+        except (SchemaConflictError, CultCacheError):
+            return False
+        return True
+
+    def _refuse_at(self, document: DocumentDefinition[Any], key: str) -> list[tuple[str, str]]:
+        """The judges a write or removal of the key by this document passes before any store is touched, asked by `is_writable` and
+        by `delete`: a global document writes only its global key, a record held read-only is never replaced, and a key held as
+        another type's record or carried foreign is refused (`_refuse_retype`)."""
+        if document.global_document and key != GLOBAL_KEY:
+            raise CultCacheError(f"Global document {document.type} must use key {GLOBAL_KEY}")
+        self._refuse_unread([key])
+        entry = document.catalog_entry()
+        probe = CultCacheEnvelope.create(key=key, type=document.type, payload=b"", schema_id=entry.schema_id, catalog_entry=entry)
+        return self._refuse_retype(document, [probe])
+
+    def _refuse_unread(self, keys: list[str]) -> None:
+        """The one check every write and removal asks before any store is touched."""
+        for key in keys:
+            held = self._state.unread.get(key)
+            if held is None:
+                continue
+            type, slots = held
+            raise SchemaConflictError(
+                self._state.envelopes[type][key].schema_id or type,
+                [self._state.documents[type].catalog_entry().schema_name],
+                key,
+                message=(
+                    f"{self._state.documents[type].catalog_entry().schema_name} record {key} is read-only: {slots.type_name} was read with "
+                    f"{slots.slot_count} slots where this build declares {slots.declared_count} ({slots.occurrences} object(s) in the "
+                    "record). Writing it would drop data a newer writer stored; read it with types that declare those slots, or "
+                    "write it from that build."
+                ),
+                unread_slots=slots,
+            )
+
     @staticmethod
     def _foreign_refusal(record: ForeignRecord) -> SchemaConflictError:
         return SchemaConflictError(
@@ -644,16 +727,18 @@ class CultCache:
 
     def _delete(self, document: DocumentDefinition[Any], key: str) -> None:
         self._assert_registered(document)
-        if key not in self._state.envelopes.get(document.type, {}):
-            carried = next((record for record in self._state.foreign if record.key == key), None)
-            if carried is not None:
-                raise self._foreign_refusal(carried)
+        # A key held as another type's record this document owns is removed, as a write of the key would replace it (C# Remove is by key).
+        owned = self._refuse_at(document, key)
         store = self._store_for_type(document.type)
         legacy_key = self._state.legacy_global_keys.get(document.type) if key == GLOBAL_KEY else None
         if store is not None:
-            store.delete(document.type, key if legacy_key is None else legacy_key)
+            # One record per key: the owned record when this document does not hold the key, else its own, in one store call.
+            holder = owned[0][0] if owned else document.type
+            store.delete(holder, key if legacy_key is None else legacy_key)
         if legacy_key is not None:
             del self._state.legacy_global_keys[document.type]
+        for other, _ in owned:
+            self._drop(other, key)
         self._drop(document.type, key)
 
     def _drop(self, type: str, key: str) -> None:

packages/cultcache-py/src/cultcache_py/documents.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import json
+import threading
 from dataclasses import MISSING, dataclass, field
 from collections.abc import Mapping, Sequence, Set
 from itertools import chain
@@ -30,6 +31,49 @@ MAX_SLOT_KEY = 65535
 _STACK_FAULTS: tuple[type[BaseException], ...] = (_MsgpackStackError,)
 
 
+@dataclass(frozen=True)
+class UnreadSlots:
+    """What a payload decode found past the slots this runtime declares, at any depth: the first type read with extra slots, how
+    many slots it carried against how many it declares, and how many objects in the payload did. It holds no slot value, so a
+    refusal can name it. `depth0_only` is True when every mark sat in the document's own record (the half of consent the payload
+    decides). A record carrying it is read-only unless the schema comparison explains every mark (see `CultCache.is_writable`)."""
+
+    type_name: str
+    slot_count: int
+    declared_count: int
+    occurrences: int
+    depth0_only: bool
+
+
+class _SlotScope:
+    """Opened by `DocumentDefinition.decode_payload_marked`; the record codec and the union codec note into it. It is thread-local,
+    so no codec signature carries it, and a codec decoded outside a payload decode notes nothing."""
+
+    def __init__(self) -> None:
+        self.depth = 0  # records entered: 0 while the document's own record is read
+        self.first: tuple[str, int, int] | None = None
+        self.occurrences = 0
+        self.depth0_only = True
+
+    def note(self, type_name: str, slot_count: int, declared_count: int, depth0: bool) -> None:
+        if self.first is None:
+            self.first = (type_name, slot_count, declared_count)
+        self.occurrences += 1
+        self.depth0_only = self.depth0_only and depth0
+
+    def result(self) -> UnreadSlots | None:
+        if self.first is None:
+            return None
+        return UnreadSlots(*self.first, self.occurrences, self.depth0_only)
+
+
+_scope = threading.local()
+
+
+def _current_scope() -> _SlotScope | None:
+    return getattr(_scope, "open", None)
+
+
 _CONTAINERS = (Mapping, Sequence, Set)
 _SCALAR_SEQUENCES = (str, bytes, bytearray, memoryview)
 
@@ -185,15 +229,25 @@ class DocumentDefinition(Generic[T]):
         return self.payload_encoder(tree)
 
     def decode_payload(self, payload: bytes) -> T:
+        return self.decode_payload_marked(payload)[0]
+
+    def decode_payload_marked(self, payload: bytes) -> tuple[T, UnreadSlots | None]:
+        """The value and what the decode found past the declared slots, at any depth. The payload depth check runs first: a
+        payload past the limit is refused, never marked."""
         try:
             tree = self.payload_decoder(payload)
         except _STACK_FAULTS:
             raise CultCachePayloadDepthError(self.type, "read") from None
         _refuse_past_depth_limit(tree, self.type, "read")
-        value = self.decode(tree)
+        outer, opened = _current_scope(), _SlotScope()
+        _scope.open = opened
+        try:
+            value = self.decode(tree)
+        finally:
+            _scope.open = outer
         if self.validate:
             self.validate(value)
-        return value
+        return value, opened.result()
 
     def __post_init__(self) -> None:
         # Judged here so every constructor passes it: the factories, the dataclass constructor and dataclasses.replace.
@@ -418,6 +472,9 @@ def define_union(
             raise ValueError(f"Union {name!r} value is malformed: expected [armKey, [armSlots]]")
         arm_name, codec = by_key[key]
         if codec is None:
+            scope = _current_scope()
+            if scope is not None and raw[1]:
+                scope.note(arm_name, len(raw[1]), 0, False)
             return UnionArm(arm_name, None)
         try:
             return UnionArm(arm_name, codec.decode(raw[1]))
@@ -479,17 +536,26 @@ def _record_codec(
     def decode(raw: Any) -> Any:
         if not isinstance(raw, list):
             raise ValueError(f"DatabaseEntry document type {type!r} expected a MessagePack array")
+        scope = _current_scope()
+        if scope is not None and len(raw) > max_key + 1:
+            scope.note(type, len(raw), max_key + 1, scope.depth == 0)
         values: dict[str, Any] = {}
-        for field in normalized:
-            if field.key < len(raw):
-                try:
-                    values[field.name] = field.codec.decode(raw[field.key])
-                except ValueError as error:
-                    raise ValueError(f"{type} slot {field.key} ({field.name}): {error}") from None
-            elif field.default is MISSING:
-                values[field.name] = None
-            else:
-                values[field.name] = field.default() if callable(field.default) else field.default
+        if scope is not None:
+            scope.depth += 1
+        try:
+            for field in normalized:
+                if field.key < len(raw):
+                    try:
+                        values[field.name] = field.codec.decode(raw[field.key])
+                    except ValueError as error:
+                        raise ValueError(f"{type} slot {field.key} ({field.name}): {error}") from None
+                elif field.default is MISSING:
+                    values[field.name] = None
+                else:
+                    values[field.name] = field.default() if callable(field.default) else field.default
+        finally:
+            if scope is not None:
+                scope.depth -= 1
         if cls is None:
             return values
         return cls(**values)

packages/cultcache-py/src/cultcache_py/stores.py

@@ -13,6 +13,7 @@ from typing import Any, Callable, Iterator, TypeVar
 
 import msgpack
 
+from .documents import UnreadSlots
 from .backing_store import (
     CultCacheEnvelope,
     CultCacheSchemaCatalogEntry,
@@ -46,9 +47,19 @@ class SchemaConflictError(ValueError):
     """A write the catalog cannot describe: records of different types under one schema id, two arrived schemas that share an id
     and disagree on the schema name, or a record no chosen entry publishes. Nothing is written. It names the id, the names
     involved (`schema_names`) and a record key (`record_key`). A cache raises it too, with an empty record key, when a
-    document registers a schema id another registered document already carries."""
+    document registers a schema id another registered document already carries. `unread_slots` is None except for the refusal to
+    write or remove a record held read-only because its payload carries slots past the declared ones; its message names the type
+    and both slot counts and never holds a slot value."""
 
-    def __init__(self, schema_id: str, schema_names: list[str], record_key: str, *, message: str | None = None) -> None:
+    def __init__(
+        self,
+        schema_id: str,
+        schema_names: list[str],
+        record_key: str,
+        *,
+        message: str | None = None,
+        unread_slots: UnreadSlots | None = None,
+    ) -> None:
         super().__init__(
             message
             or f"Schema id {schema_id!r} cannot describe record {record_key!r}: {', '.join(repr(name) for name in schema_names)}"
@@ -56,6 +67,7 @@ class SchemaConflictError(ValueError):
         self.schema_id = schema_id
         self.schema_names = schema_names
         self.record_key = record_key
+        self.unread_slots = unread_slots
 
 
 class JsonLinesBackingStore:

packages/cultcache-py/tests/test_slot_readonly.py

@@ -0,0 +1,492 @@
+from __future__ import annotations
+
+import dataclasses
+import shutil
+import threading
+import tempfile
+import unittest
+from pathlib import Path
+
+import msgpack
+from cultcache_py.cache import CultCacheError
+
+from cultcache_py import (
+    CultCache,
+    CultCachePayloadDepthError,
+    SlotCodec,
+    SchemaConflictError,
+    SingleFileMessagePackBackingStore,
+    UnionArm,
+    UnreadSlots,
+    database_entry_field,
+    define_database_entry_type,
+    define_record_codec,
+    leaves as t,
+)
+import test_union_parity as _parity
+from test_union_parity import (
+    VECTORS,
+    _surface_value,
+    _vector,
+    surface_document,
+    unit_outcome,
+)
+
+SLOT = "extra-slot"  # the canary: the value of every extra slot in the vectors; no refusal may carry it
+
+# The vectors name the C# types; this runtime names a type by its codec's name, a document by its document type.
+_NAMES = {"UnionSurface": surface_document.type, "SurfaceDeclaring": surface_document.type, "Denied2": "Denied"}
+
+slot_unit_document = define_database_entry_type(
+    "tests.slot_unit",
+    [database_entry_field("name", 0, codec=t.string), database_entry_field("outcome", 1, codec=unit_outcome)],
+    name="name",
+)
+# The reader of extra-slot-consent.cc: the surface document's slots under a new schema id that lists the old one compatible.
+declaring_document = define_database_entry_type(
+    "tests.union_surface",
+    [
+        database_entry_field("id", 0, codec=t.string),
+        database_entry_field("version", 1, codec=t.int32),
+        database_entry_field("root", 2, codec=_parity._node),
+        database_entry_field("outcomes", 3, codec=t.list(_parity.outcome)),
+        database_entry_field("part", 4, codec=t.nullable(_parity.item)),
+        database_entry_field("crafted", 5, codec=t.nullable(_parity.crafted)),
+        database_entry_field("missing", 6, codec=_parity._node),
+    ],
+    name="id",
+    schema_id="tests.union_surface.v2",
+    schema_name="tests.union_surface",
+    compatible_schema_ids=("tests.union_surface.wide",),
+)
+
+# file -> (type, slots carried, slots declared, every mark at depth 0): the README table.
+_PAYLOADS = {
+    "extra-slot-record.msgpack": ("UnionSurface", 8, 7, True),
+    "extra-slot-nested-record.msgpack": ("Composition", 3, 2, False),
+    "extra-slot-arm.msgpack": ("Box", 3, 2, False),
+    "extra-slot-list-element.msgpack": ("Text", 3, 2, False),
+    "depth-skipped-at-limit.msgpack": ("UnionSurface", 8, 7, True),
+}
+
+
+def _copy_store_without_variants(source: Path, target: Path) -> None:
+    """This runtime holds no variant records (a five-slot record makes a store unreadable here), so the store vector is
+    walked with its variant row removed; every other byte is the C# writer's."""
+    [snapshot] = list(msgpack.Unpacker(open(source, "rb"), raw=False))
+    snapshot[2] = [record for record in snapshot[2] if len(record) <= 4]
+    target.write_bytes(msgpack.packb(snapshot, use_bin_type=True))
+
+
+def _open(path: Path, *documents) -> CultCache:
+    builder = CultCache.builder()
+    for document in documents:
+        builder.register_document_type(document)
+    cache = builder.add_generic_store(SingleFileMessagePackBackingStore(path)).build()
+    cache.pull_all_backing_stores()
+    return cache
+
+
+def _rows():
+    return msgpack.unpackb(_vector("slot-outcomes.msgpack"), raw=False)
+
+
+class PayloadMarkTests(unittest.TestCase):
+    def test_every_read_only_payload_decodes_and_names_its_mark(self) -> None:
+        for file, (type_name, carried, declared, depth0) in _PAYLOADS.items():
+            with self.subTest(file=file):
+                value, mark = surface_document.decode_payload_marked(_vector(file))
+                self.assertEqual(value, _surface_value())
+                self.assertEqual(mark, UnreadSlots(_NAMES.get(type_name, type_name), carried, declared, 1, depth0))
+                self.assertEqual(surface_document.decode_payload(_vector(file)), _surface_value())
+
+    def test_the_outcomes_manifest_still_counts_nine_rows_and_four_read_only(self) -> None:
+        rows = msgpack.unpackb(_vector("outcomes.msgpack"), raw=False)
+        self.assertEqual(len(rows), 9)
+        self.assertEqual([row[0] for row in rows if row[1] == "read-only"], [f for f in _PAYLOADS if f != "depth-skipped-at-limit.msgpack"])
+
+    def test_a_clean_payload_and_a_short_one_carry_no_mark(self) -> None:
+        self.assertEqual(surface_document.decode_payload_marked(_vector("union-surface.msgpack")), (_surface_value(), None))
+        slots = msgpack.unpackb(_vector("union-surface.msgpack"), raw=False)
+        slots[2] = [1, ["only-id"]]  # a Text arm with a missing trailing slot: defaults, no mark
+        _, mark = surface_document.decode_payload_marked(msgpack.packb(slots, use_bin_type=True))
+        self.assertIsNone(mark)
+
+    def test_a_retired_slot_below_the_highest_key_does_not_mark(self) -> None:
+        slots = msgpack.unpackb(_vector("union-surface.msgpack"), raw=False)
+        # Image's declared slots are 0, 1, 3 (2 retired): four carried, four declared.
+        image = slots[2][1][1][2]
+        self.assertEqual(image[0], 5)
+        image[1][2] = "retired-slot-value"
+        _, mark = surface_document.decode_payload_marked(msgpack.packb(slots, use_bin_type=True))
+        self.assertIsNone(mark)
+
+    def test_the_depth_check_runs_before_the_mark(self) -> None:
+        with self.assertRaises(CultCachePayloadDepthError):
+            surface_document.decode_payload_marked(_vector("depth-skipped-past-limit.msgpack"))
+
+    def test_the_unit_arm_with_extra_slots_is_read_past_and_marked(self) -> None:
+        payload = msgpack.packb(["unit-arm", [2, [SLOT]]], use_bin_type=True)
+        value, mark = slot_unit_document.decode_payload_marked(payload)
+        self.assertEqual(value, {"name": "unit-arm", "outcome": UnionArm("Denied")})
+        self.assertEqual(mark, UnreadSlots("Denied", 1, 0, 1, False))
+        self.assertEqual(slot_unit_document.decode_payload_marked(msgpack.packb(["ok", [2, []]]))[1], None)
+
+    def test_a_codec_decoded_outside_a_payload_decode_notes_nothing(self) -> None:
+        self.assertEqual(unit_outcome.decode([2, [SLOT]]), UnionArm("Denied"))
+
+    def test_the_scope_belongs_to_its_thread(self) -> None:
+        def decode_elsewhere(raw):
+            thread = threading.Thread(target=lambda: unit_outcome.decode([2, [SLOT]]))
+            thread.start()
+            thread.join()
+            return raw
+
+        outer = define_database_entry_type("tests.threaded", [database_entry_field("a", 0, codec=SlotCodec(lambda v: v, decode_elsewhere))])
+        self.assertIsNone(outer.decode_payload_marked(msgpack.packb(["x"]))[1])
+
+    def test_a_nested_payload_decode_restores_the_outer_scope(self) -> None:
+        def decode_clean_inside(raw):
+            surface_document.decode_payload_marked(_vector("union-surface.msgpack"))
+            return raw
+
+        inner = define_record_codec("Inner", [database_entry_field("x", 0, codec=t.string)])
+        outer = define_database_entry_type(
+            "tests.nested",
+            [
+                database_entry_field("a", 0, codec=SlotCodec(lambda v: v, decode_clean_inside)),
+                database_entry_field("b", 1, codec=inner),
+            ],
+        )
+        _, mark = outer.decode_payload_marked(msgpack.packb(["x", ["v", SLOT]]))
+        self.assertEqual(mark, UnreadSlots("Inner", 2, 1, 1, False))
+
+
+class StoreRowTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.tmp = Path(tempfile.mkdtemp())
+        self.addCleanup(shutil.rmtree, self.tmp, True)
+
+    def _store(self, name: str) -> tuple[Path, CultCache]:
+        path = self.tmp / name
+        _copy_store_without_variants(VECTORS / name, path)
+        documents = (declaring_document,) if name == "extra-slot-consent.cc" else (surface_document, slot_unit_document)
+        return path, _open(path, *documents)
+
+    def _document(self, store: str, key: str):
+        if key == "unit-arm":
+            return slot_unit_document
+        return declaring_document if store == "extra-slot-consent.cc" else surface_document
+
+    def test_every_row_is_writable_exactly_where_the_vector_says_and_refuses_with_its_mark(self) -> None:
+        walked = 0
+        for store, key, writable, explained, type_name, carried, declared, occurrences, flatten in _rows():
+            if flatten is not None:
+                continue  # a variant: this runtime holds none (deviation)
+            walked += 1
+            with self.subTest(store=store, key=key):
+                path, cache = self._store(store)
+                document = self._document(store, key)
+                before = path.read_bytes()
+                self.assertEqual(cache.is_writable(document, key), writable)
+                held = cache.get(document, key)
+                self.assertIsNotNone(held)
+                if writable:
+                    continue
+                self.assertEqual(explained, False)
+                for attempt in (
+                    lambda: cache.put(document, key, held),
+                    lambda: cache.delete(document, key),
+                    lambda: cache.put_envelope(document, cache.get_required_envelope(document, key)),
+                    lambda: cache.put_envelopes(document, [cache.get_required_envelope(document, key)]),
+                    lambda: cache.update(document, key, lambda value: value),
+                ):
+                    with self.assertRaises(SchemaConflictError) as caught:
+                        attempt()
+                    refusal = caught.exception
+                    self.assertEqual(
+                        refusal.unread_slots,
+                        UnreadSlots(_NAMES.get(type_name, type_name), carried, declared, occurrences, refusal.unread_slots.depth0_only),
+                    )
+                    self.assertEqual(refusal.record_key, key)
+                    message = str(refusal)
+                    self.assertNotIn(SLOT, message)
+                    self.assertIn(refusal.unread_slots.type_name, message)
+                    self.assertIn(f"{carried} slots", message)
+                    self.assertIn(f"declares {declared}", message)
+                    self.assertEqual(path.read_bytes(), before)
+                self.assertTrue(cache.get(document, key) is held)
+        self.assertEqual(walked, 7)  # eight rows, the variant not walked
+
+    def test_a_key_no_store_holds_and_another_key_are_writable_while_a_marked_one_is_held(self) -> None:
+        path, cache = self._store("extra-slot-store.cc")
+        self.assertFalse(cache.is_writable(surface_document, "marked"))
+        self.assertTrue(cache.is_writable(surface_document, "nothing-here"))
+        cache.put(surface_document, "other", cache.get_required(surface_document, "clean"))
+        self.assertEqual(cache.get(surface_document, "other"), _surface_value())
+
+    def test_consent_lands_an_explained_mark_and_refuses_the_others(self) -> None:
+        path, cache = self._store("extra-slot-consent.cc")
+        self.assertTrue(cache.is_writable(declaring_document, "consented"))
+        cache.put(declaring_document, "consented", cache.get_required(declaring_document, "consented"))
+        [stored] = [e for e in SingleFileMessagePackBackingStore(path).pull_all() if e.key == "consented"]
+        self.assertEqual(stored.schema_id, "tests.union_surface.v2")
+        self.assertEqual(len(msgpack.unpackb(stored.payload, raw=False)), 7)  # the consented slot is gone
+        self.assertTrue(cache.is_writable(declaring_document, "consented"))
+        for key in ("over-consent", "nested"):
+            self.assertFalse(cache.is_writable(declaring_document, key), key)
+            with self.assertRaises(SchemaConflictError):
+                cache.put(declaring_document, key, cache.get_required(declaring_document, key))
+
+    def test_a_reload_marks_again_and_a_write_of_a_clean_envelope_over_a_marked_key_is_refused(self) -> None:
+        path, cache = self._store("extra-slot-store.cc")
+        envelope = cache.get_required_envelope(surface_document, "clean")
+        with self.assertRaises(SchemaConflictError):
+            cache.put_envelope(surface_document, type(envelope)(**{**envelope.__dict__, "key": "marked"}))
+        cache.pull_all_backing_stores()
+        self.assertFalse(cache.is_writable(surface_document, "marked"))
+
+    def test_a_written_envelope_that_carries_extra_slots_is_held_read_only(self) -> None:
+        path, cache = self._store("extra-slot-store.cc")
+        envelope = cache.get_required_envelope(surface_document, "clean")
+        carrying = type(envelope)(**{**envelope.__dict__, "key": "arrived", "payload": _vector("extra-slot-arm.msgpack")})
+        cache.put_envelope(surface_document, carrying)
+        self.assertFalse(cache.is_writable(surface_document, "arrived"))
+        with self.assertRaises(SchemaConflictError):
+            cache.delete(surface_document, "arrived")
+        cache.pull_all_backing_stores()
+        self.assertFalse(cache.is_writable(surface_document, "arrived"))
+
+    def test_a_batch_of_envelopes_that_carry_extra_slots_is_held_read_only(self) -> None:
+        path, cache = self._store("extra-slot-store.cc")
+        envelope = cache.get_required_envelope(surface_document, "clean")
+        batch = [
+            type(envelope)(**{**envelope.__dict__, "key": key, "payload": _vector(file)})
+            for key, file in (("b1", "extra-slot-arm.msgpack"), ("b2", "union-surface.msgpack"))
+        ]
+        cache.put_envelopes(surface_document, batch)
+        self.assertEqual([cache.is_writable(surface_document, key) for key in ("b1", "b2")], [False, True])
+
+    def test_a_mark_under_an_id_the_reader_does_not_list_compatible_is_not_consented(self) -> None:
+        unlisted = define_database_entry_type(
+            "tests.union_surface",
+            [database_entry_field("id", 0, codec=t.string)] + [database_entry_field(n, i, codec=c) for n, i, c in (
+                ("version", 1, t.int32), ("root", 2, _parity._node), ("outcomes", 3, t.list(_parity.outcome)),
+                ("part", 4, t.nullable(_parity.item)), ("crafted", 5, t.nullable(_parity.crafted)), ("missing", 6, _parity._node))],
+            name="id",
+            schema_id="tests.union_surface.v2",
+            schema_name="tests.union_surface",
+        )
+        path = self.tmp / "unlisted.cc"
+        _copy_store_without_variants(VECTORS / "extra-slot-consent.cc", path)
+        cache = _open(path, unlisted)
+        self.assertFalse(cache.is_writable(unlisted, "consented"))
+
+    def test_a_written_envelope_is_judged_on_the_id_it_is_stored_under_so_a_reload_agrees(self) -> None:
+        path, cache = self._store("extra-slot-consent.cc")
+        arrived = next(e for e in SingleFileMessagePackBackingStore(path).pull_all() if e.key == "consented")
+        self.assertEqual(arrived.schema_id, "tests.union_surface.wide")
+        cache.put_envelope(declaring_document, dataclasses.replace(arrived, key="via-envelope"))
+        cache.put_envelopes(declaring_document, [dataclasses.replace(arrived, key="via-batch")])
+        in_memory = [cache.is_writable(declaring_document, k) for k in ("via-envelope", "via-batch")]
+        cache.pull_all_backing_stores()
+        self.assertEqual([cache.is_writable(declaring_document, k) for k in ("via-envelope", "via-batch")], in_memory)
+        self.assertEqual(in_memory, [False, False])
+
+    def test_is_writable_agrees_with_the_write_for_a_key_a_store_carries_foreign(self) -> None:
+        path = self.tmp / "foreign.cc"
+        _copy_store_without_variants(VECTORS / "extra-slot-store.cc", path)
+        cache = _open(path, surface_document)  # tests.slot_unit is not registered: its record is carried foreign
+        self.assertEqual([r.key for r in cache.foreign_records()], ["unit-arm"])
+        self.assertFalse(cache.is_writable(surface_document, "unit-arm"))
+        with self.assertRaises(SchemaConflictError):
+            cache.put(surface_document, "unit-arm", _surface_value())
+        with self.assertRaises(SchemaConflictError):
+            cache.delete(surface_document, "unit-arm")
+
+    def test_every_key_of_a_batch_is_checked_wherever_the_marked_one_stands(self) -> None:
+        path, cache = self._store("extra-slot-store.cc")
+        before = path.read_bytes()
+        envelope = cache.get_required_envelope(surface_document, "clean")
+        fresh = dataclasses.replace(envelope, key="fresh")
+        marked = dataclasses.replace(envelope, key="marked")
+        for batch in ([fresh, marked], [marked, fresh], [fresh, dataclasses.replace(envelope, key="f2"), marked]):
+            with self.assertRaises(SchemaConflictError) as caught:
+                cache.put_envelopes(surface_document, batch)
+            self.assertEqual(caught.exception.record_key, "marked")
+            self.assertEqual(path.read_bytes(), before)
+            self.assertIsNone(cache.get(surface_document, "fresh"))
+
+    def test_only_the_marks_that_hold_a_record_read_only_are_kept_by_key(self) -> None:
+        _, store = self._store("extra-slot-store.cc")
+        self.assertEqual({k: v[0] for k, v in store._state.unread.items()}, {"marked": surface_document.type, "unit-arm": "tests.slot_unit"})
+        _, consent = self._store("extra-slot-consent.cc")
+        self.assertEqual(sorted(consent._state.unread), ["nested", "over-consent"])  # the consented mark is no state
+
+    def test_an_arrival_with_no_catalog_entry_under_a_listed_id_is_not_consented(self) -> None:
+        wide = next(e for e in SingleFileMessagePackBackingStore(VECTORS / "extra-slot-consent.cc").pull_all() if e.key == "consented")
+        bare = dataclasses.replace(wide, catalog_entry=None)
+
+        class Held:
+            def pull_all(self):
+                return [bare]
+
+            def push(self, envelope): ...
+
+            def push_all(self, envelopes): ...
+
+            def delete(self, type, key): ...
+
+        cache = CultCache.builder().register_document_type(declaring_document).add_generic_store(Held()).build()
+        cache.pull_all_backing_stores()
+        self.assertFalse(cache.is_writable(declaring_document, "consented"))
+
+    def test_a_removal_is_refused_where_is_writable_says_no(self) -> None:
+        path, cache = self._store("extra-slot-store.cc")
+        before = path.read_bytes()
+        other = define_database_entry_type("tests.other", [database_entry_field("a", 0, codec=t.string)], name="a")
+        cache2 = _open(self.tmp / "extra-slot-store.cc", surface_document, slot_unit_document, other)
+        self.assertFalse(cache2.is_writable(other, "clean"))
+        with self.assertRaises(SchemaConflictError):
+            cache2.delete(other, "clean")
+        self.assertEqual(path.read_bytes(), before)
+        self.assertIsNotNone(cache2.get(surface_document, "clean"))
+        self.assertTrue(cache2.is_writable(other, "nothing-here"))
+
+    def test_a_global_document_is_not_writable_at_a_key_other_than_the_global_one(self) -> None:
+        solo = define_database_entry_type("tests.solo", [database_entry_field("a", 0, codec=t.string)], global_document=True)
+        cache = CultCache.builder().register_document_type(solo).build()
+        self.assertFalse(cache.is_writable(solo, "elsewhere"))
+        with self.assertRaises(CultCacheError):
+            cache.put(solo, "elsewhere", {"a": "x"})
+        with self.assertRaises(CultCacheError):
+            cache.delete(solo, "elsewhere")
+        self.assertTrue(cache.is_writable(solo, CultCache.GLOBAL_KEY))
+
+    def test_a_reload_replaces_the_marks_wholesale(self) -> None:
+        path, cache = self._store("extra-slot-store.cc")
+        self.assertFalse(cache.is_writable(surface_document, "marked"))
+        # The store is repaired by its owner: the marked record is rewritten clean. The reload must forget the old mark.
+        [snapshot] = list(msgpack.Unpacker(open(path, "rb"), raw=False))
+        clean = next(r for r in snapshot[2] if r[0] == "clean")
+        for record in snapshot[2]:
+            if record[0] in ("marked", "unit-arm"):
+                record[3] = clean[3] if record[0] == "marked" else record[3]
+        snapshot[2] = [r for r in snapshot[2] if r[0] != "unit-arm"]
+        path.write_bytes(msgpack.packb(snapshot, use_bin_type=True))
+        cache.pull_all_backing_stores()
+        self.assertTrue(cache.is_writable(surface_document, "marked"))
+        self.assertEqual(cache._state.unread, {})
+
+    def test_a_listed_id_whose_catalog_reports_no_extra_slot_does_not_consent(self) -> None:
+        wide = next(e for e in SingleFileMessagePackBackingStore(VECTORS / "extra-slot-consent.cc").pull_all() if e.key == "consented")
+        same = dataclasses.replace(wide, catalog_entry=declaring_document.catalog_entry())  # the stored entry has no slot past the reader's
+
+        class Held:
+            def pull_all(self):
+                return [same]
+
+            def push(self, envelope): ...
+
+            def push_all(self, envelopes): ...
+
+            def delete(self, type, key): ...
+
+        cache = CultCache.builder().register_document_type(declaring_document).add_generic_store(Held()).build()
+        cache.pull_all_backing_stores()
+        self.assertFalse(cache.is_writable(declaring_document, "consented"))
+
+    def test_a_record_stored_under_the_readers_own_id_is_never_consented(self) -> None:
+        own = dataclasses.replace(
+            next(e for e in SingleFileMessagePackBackingStore(VECTORS / "extra-slot-consent.cc").pull_all() if e.key == "consented"),
+            schema_id="tests.union_surface.v2",
+        )
+        wider = dataclasses.replace(own.catalog_entry, schema_id="tests.union_surface.v2", compatible_schema_ids=("tests.union_surface.v2",))
+        listing = dataclasses.replace(
+            declaring_document, schema_id="tests.union_surface.v2", compatible_schema_ids=("tests.union_surface.wide", "tests.union_surface.v2")
+        )
+
+        class Held:
+            def pull_all(self):
+                return [dataclasses.replace(own, catalog_entry=wider)]
+
+            def push(self, envelope): ...
+
+            def push_all(self, envelopes): ...
+
+            def delete(self, type, key): ...
+
+        cache = CultCache.builder().register_document_type(listing).add_generic_store(Held()).build()
+        cache.pull_all_backing_stores()
+        self.assertFalse(cache.is_writable(listing, "consented"))
+
+    def test_a_removal_by_the_owner_of_a_key_held_as_another_type_removes_it_as_a_write_would_replace_it(self) -> None:
+        a = define_database_entry_type(
+            "tests.owned_a", [database_entry_field("a", 0, codec=t.string)], name="a", indexes={"by_a": "a"}
+        )
+        b = define_database_entry_type(
+            "tests.owned_b",
+            [database_entry_field("a", 0, codec=t.string)],
+            compatible_schema_ids=(a.catalog_entry().schema_id,),
+        )
+        path = self.tmp / "owned.cc"
+        writer = _open(path, a)
+        writer.put(a, "k", {"a": "x"})
+        writer.put(a, "other", {"a": "y"})
+        cache = _open(path, a, b)
+        self.assertTrue(cache.is_writable(b, "k"))
+        cache.delete(b, "k")
+        self.assertIsNone(cache.get(a, "k"))
+        self.assertIsNone(cache.get_by_index(a, "by_a", "x"))
+        self.assertEqual([e.key for e in SingleFileMessagePackBackingStore(path).pull_all()], ["other"])
+        self.assertTrue(cache.is_writable(b, "k"))
+
+    def test_an_owners_removal_is_one_store_call_so_the_cache_and_the_store_agree_when_it_fails(self) -> None:
+        a = define_database_entry_type("tests.one_a", [database_entry_field("a", 0, codec=t.string)], name="a")
+        b = define_database_entry_type(
+            "tests.one_b", [database_entry_field("a", 0, codec=t.string)], compatible_schema_ids=(a.catalog_entry().schema_id,)
+        )
+        path = self.tmp / "one.cc"
+        _open(path, a).put(a, "k", {"a": "x"})
+
+        class SecondDeleteFails(SingleFileMessagePackBackingStore):
+            calls = 0
+
+            def delete(self, type, key):
+                self.calls += 1
+                if self.calls > 1:
+                    raise OSError("second delete")
+                super().delete(type, key)
+
+        store = SecondDeleteFails(path)
+        cache = CultCache.builder().register_document_type(a).register_document_type(b).add_generic_store(store).build()
+        cache.pull_all_backing_stores()
+        cache.delete(b, "k")
+        self.assertEqual(store.calls, 1)
+        self.assertIsNone(cache.get(a, "k"))
+        self.assertEqual(SingleFileMessagePackBackingStore(path).pull_all(), [])
+
+    def test_writing_another_record_and_flushing_carries_the_marked_one_unchanged(self) -> None:
+        path, cache = self._store("extra-slot-store.cc")
+        original = {e.key: e for e in SingleFileMessagePackBackingStore(path).pull_all()}
+        cache.put(surface_document, "clean", {**_surface_value(), "version": 9})
+        after = {e.key: e for e in SingleFileMessagePackBackingStore(path).pull_all()}
+        for key in ("marked", "unit-arm"):
+            self.assertEqual(after[key].payload, original[key].payload, key)
+            self.assertEqual(after[key].schema_id, original[key].schema_id, key)
+            self.assertEqual(after[key].stored_at, original[key].stored_at, key)
+        self.assertNotEqual(after["clean"].payload, original["clean"].payload)
+        self.assertEqual(after["marked"].catalog_entry, original["marked"].catalog_entry)
+
+    def test_the_vector_store_loads_under_the_reference_documents_and_marks_nothing_else(self) -> None:
+        path, cache = self._store("extra-slot-store.cc")
+        self.assertTrue(cache.is_writable(surface_document, "clean"))
+        self.assertFalse(cache.is_writable(surface_document, "marked"))
+        self.assertFalse(cache.is_writable(slot_unit_document, "unit-arm"))
+        self.assertEqual(
+            cache.get(slot_unit_document, "unit-arm"), {"name": "unit-arm", "outcome": UnionArm("Denied")}
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

packages/cultcache-py/tests/test_union_parity.py

@@ -261,7 +261,9 @@ class UnionSurfaceVectorTests(unittest.TestCase):
             else:
                 read_only += 1
                 self.assertEqual(outcome_kind, "read-only", file)
-                self.assertEqual(surface_document.decode_payload(payload), _surface_value(), file)
+                value, mark = surface_document.decode_payload_marked(payload)
+                self.assertEqual(value, _surface_value(), file)
+                self.assertIsNotNone(mark, file)  # read past and marked: tests/test_slot_readonly.py pins what the mark holds
         self.assertEqual((refused, read_only), (5, 4))
 
 
@@ -526,7 +528,9 @@ class BoundsVectorTests(unittest.TestCase):
                     surface_document.encode_payload(surface_document.decode_payload(payload)), _vector(detail), file
                 )
             elif outcome_kind == "read-only":
-                self.assertEqual(surface_document.decode_payload(payload), _surface_value(), file)
+                value, mark = surface_document.decode_payload_marked(payload)
+                self.assertEqual(value, _surface_value(), file)
+                self.assertIsNotNone(mark, file)
             else:
                 with self.assertRaises(ValueError, msg=file) as caught:
                     surface_document.decode_payload(payload)