docs/start/README.md

@@ -19,7 +19,7 @@ this page.
 | **C#** | [first call, tested](#c-first-call); [update, tested](#c-update-with-retry) | [package](../../packages/cultmath/README.md); snippet not yet | [package](../../src/GameCult.Geometry/README.md); snippet not yet | [first call, tested](#c-cultnet-first-call); [package](../../src/GameCult.Networking) | [quickstart](../../src/GameCult.Mesh/docs/durable-node-quickstart.md); run by `scripts/verify-eve-getting-started.ps1` |
 | **TypeScript** | [first call, tested](#typescript-first-call); [update, tested](#typescript-update-with-retry) | not yet | not yet | [first call, tested](#typescript-cultnet-first-call); [package](../../packages/cultnet-ts/README.md) | [package](../../packages/cultmesh-ts/README.md); snippet not yet; [recipe: two-client .io game](io-game.md), tested with two live clients, browser page not yet run |
 | **Rust** | [first call, tested](#rust-first-call); [update, tested](#rust-update-with-retry) | [native core](../../packages/cultmath/native/cultmath-core); snippet not yet | not yet | [first call, tested](#rust-cultnet-first-call); [package](../../packages/cultnet-rs/README.md), [examples](../../packages/cultnet-rs/examples) | [package](../../packages/cultmesh-rs); snippet not yet |
-| **Python** | [first call, tested](#python-first-call) | not yet | not yet | [first call, tested](#python-cultnet-first-call); [package](../../packages/cultnet-py/README.md) | [package](../../packages/cultmesh-py/README.md); snippet not yet |
+| **Python** | [first call, tested](#python-first-call); [update, tested](#python-update-with-retry) | not yet | not yet | [first call, tested](#python-cultnet-first-call); [package](../../packages/cultnet-py/README.md) | [package](../../packages/cultmesh-py/README.md); snippet not yet |
 
 For the whole
 picture of what each piece is for, see the
@@ -305,6 +305,29 @@ def first_call() -> Item:
     return reopened.get_required(item_doc, "item:potion")
 ```
 
+## Python update with retry
+
+`update` changes a record from what the store holds now and commits only if it is still that record. The source is the
+`update` region of
+[`packages/cultcache-py/tests/test_first_call.py`](../../packages/cultcache-py/tests/test_first_call.py), which uses the
+definitions of the first call above; a test in that file fails if the block stops matching the file.
+
+```python
+def restock(path: Path):
+    cache = (
+        CultCache.builder()
+        .register_document_type(item_doc)
+        .add_generic_store(SingleFileMessagePackBackingStore(path))
+        .build()
+    )
+    cache.pull_all_backing_stores()
+    return cache.update(item_doc, "item:potion", lambda item: Item("Potion", "Consumable", (item.value if item else 0) + 10))
+```
+
+The write is durable when `update` returns. A lost race re-runs the function, so keep it free of side effects; when the
+attempts run out the result's `outcome` is `UpdateOutcome.LOST_RACE` instead of a raise. `put` is the blind opt-out: last
+writer wins.
+
 ## CultNet first calls
 
 CultNet is schema-v0 messages in length-prefixed frames. The first call in each runtime is the same

packages/cultcache-py/CHANGELOG.md

@@ -4,10 +4,12 @@
 
 ### Added
 
+- `CultCache.update` and `update_global` return `UpdateResult(outcome, document)`; `UpdateOutcome` is `UPDATED`, `DECLINED` or `LOST_RACE`. The single-file store gains `read(key)` (the record and its content version, the SHA-256 C# computes) and `commit_if(envelope, expected)`, the version is the SHA-256 of the record re-encoded as `[key, schemaId, storedAt, payload]` with minimal MessagePack encodings, which is the SHA-256 of the record's stored bytes exactly when the writer encoded it minimally (every writer in this library does; a record another writer encoded with a wider-than-needed length or integer has the version of its minimal form, not of its bytes). Both looked up by the cache and not `BackingStore` protocol members (variants:cut_spec:cut-update-with-retry-py).
 - `read_payload_tree(type_name, payload, decoder=None)`: the exported owner of the one-value, depth-first payload read that `decode_payload` uses, for payloads read outside a definition (`cultnet-py`'s witness bundle decode needs it, so `cultnet-py` now requires `cultcache-py>=0.4.0`). The manifest version is 0.4.0, the minor these Unreleased breaking changes imply, set in this change so the workspace install resolves the new bound; the `## Unreleased` heading becomes `## [0.4.0]` when it is tagged.
 
 ### Changed
 
+- A write that replaces a record is stored at a later `stored_at` than the record it replaces, in the single-file store, the JSONL store and a store-less cache, with no exception for identical content (`variants:ruling:py-rs-update-follow-csharp-restamp`); the cache holds the stamp the store wrote. A stored_at is read only in the shape the runtimes write (an RFC 3339 date-time with T, a fraction of 1 to 9 digits, Z or a numeric offset), the same on every Python; any other text, or an instant outside `datetime`'s range, is not a stamp and a write over it is stamped at now. A write over a record stamped at the last instant `datetime` can hold (9999-12-31T23:59:59.999999Z) is refused with `ValueError` and writes nothing.
 - `store_write_lock` looks for a link at the store path again once the lock is held, so a link put at the path while a writer waited on the lock is refused with `StoreLinkedError` and the lock released, instead of being replaced. `JsonLinesBackingStore` takes no cross-process lock (only its thread lock, under which it already refuses a link), so it has no lock wait to re-check across. Pinned by a test that puts the link while a foreign process holds the lock.
 - `CultCache.is_writable` does not look at store paths: a link at a store path is not predicted, and the write refuses it with `StoreLinkedError`. Pinned by a test through a cache attached to a linked store.
 - `JsonLinesBackingStore` writes LF line endings on Windows. It wrote CRLF there, through `write_text` newline translation, before the commit_store_file change (#154, adbbb780).
@@ -17,6 +19,8 @@
 
 ### Breaking
 
+- `update(document, key, apply, *, max_attempts=5)` and `update_global(document, apply, *, max_attempts=5)` replace the blind pair. Each attempt reads the record from the home store, runs `apply` on a copy (None when absent) outside every lock, and commits only if the record is unchanged, running `apply` again on a lost race; it returns `UpdateResult`, not the value. `apply` returning None writes nothing; a lost race that outlasts `max_attempts` returns `LOST_RACE`; a store without `read` and `commit_if` (JSON lines among them) refuses before `apply`. `put` is the blind write.
+- `push` returns the `stored_at` it wrote and `push_all` a dict of them by key (`{}` when a soft push skipped); a store returning None keeps the cache's own stamp.
 - `push_all(envelopes, soft=False)` on the single-file and JSONL stores and on the `BackingStore` protocol takes the soft flag, as TypeScript, Rust and C# do: a soft push writes only where nothing is at the path; over a store the reader reads, empty included, it returns without writing, and over a file the reader cannot read it refuses (`StoreUnreadableError` for the single-file store). The `BackingStore` protocol has `pull_or_absent()`: the envelopes the reader reads, `None` when nothing is at the path, a raise when the path cannot be read; its default is `pull_all()`, which cannot say absent, so a store that inherits it is always read. The default `push_all` decides soft by it (writes where it says absent, skips over a store it read, refuses where it raises), and the single-file and JSONL stores expose their pull through it; an inheriting store that can tell nothing-at-the-path overrides `pull_or_absent`, not `push_all`. Before, a soft push through the default always skipped. A store kind that overrides `push_all(envelopes)` without the keyword fails when a caller passes `soft`.
 - A write at a store path that is a link, live or dangling, is refused with the new `StoreLinkedError` (an `OSError`, exported from `cultcache_py`; `.path` names the path) before the lock and the replace, for the single-file store and for `JsonLinesBackingStore` (`push`, `push_all`, `delete`), and nothing is written. Before, a write at a live link replaced the link with a regular file, and at a dangling link both stores raised `FileNotFoundError`. A caller catching `FileNotFoundError` there now catches `StoreLinkedError` (`OSError`); configure the link's target as the store path. On Windows a junction or any reparse point counts as a link. Reads through a live link work.
 - 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`.

packages/cultcache-py/README.md

@@ -94,8 +94,7 @@ store.
 - `put(...)`
 - `put_envelopes(...)`
 - `put_global(...)`
-- `update(...)`
-- `update_global(...)`
+- `update(document, key, apply, *, max_attempts=5)` and `update_global(document, apply, *, max_attempts=5)`: the conditional golden path; reads the record as its home store holds it now, runs `apply` on a fresh copy, writes only if the record is unchanged and runs `apply` again on a lost race; returns `UpdateResult(outcome, document)` with `UpdateOutcome.UPDATED`, `DECLINED` (`apply` returned `None`) or `LOST_RACE` (see below)
 - `delete(...)`
 - `delete_global(...)`
 - `is_writable(document, key)` (advisory: would a write or removal of the key be refused; see below)
@@ -143,6 +142,8 @@ used by the TypeScript, Rust, and C# runtimes:
   canonicalSchemaJson, compatibleSchemaIds, members]`
 - records: `[key, schemaId, storedAt, payload]`
 
+`update` needs a home store with the optional `read(key)` (the record at the key, of any type, and its content version, from one read) and `commit_if(envelope, expected)` (write only if the record at the key is still that version, under the store's write lock). The content version is the lowercase hex SHA-256 of the record re-encoded as the store writes it, `[key, schemaId, storedAt, payload]` with minimal MessagePack encodings, the same value C# computes (`tests/vectors/content-version`). It equals the SHA-256 of the record's bytes in the file exactly when the writer encoded the record minimally, as every writer in this library does; a record encoded with a wider-than-needed length or integer has the version of its minimal form. The single-file store has both; `JsonLinesBackingStore` and a store without them refuse `update` before `apply` runs. A cache with no store copies its held instance. Every write that replaces a record is stored at a later `stored_at` than the one it replaces, in each store kind and in a store-less cache; `push` and `push_all` return the stamp written, and the cache holds that stamp.
+
 Large corpora should use a sharded store or a real database. A single snapshot file is a scalpel, not a forklift.
 
 ## Store format and rollback

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

@@ -1,7 +1,7 @@
 from .backing_store import BackingStore, CultCacheEnvelope
 from .backing_store import CultCacheSchemaCatalogEntry, CultCacheSchemaCatalogMember
 from . import leaves
-from .cache import CultCache, CultCacheBuilder, ForeignRecord
+from .cache import CultCache, CultCacheBuilder, ForeignRecord, UpdateOutcome, UpdateResult
 from .documents import (
     MAX_PAYLOAD_DEPTH,
     MAX_SLOT_KEY,
@@ -42,6 +42,8 @@ __all__ = [
     "StoreUnreadableError",
     "UnionArm",
     "UnreadSlots",
+    "UpdateOutcome",
+    "UpdateResult",
     "read_payload_tree",
     "leaves",
     "database_entry_field",

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

@@ -58,10 +58,17 @@ class CultCacheEnvelope:
 
 
 class BackingStore(Protocol):
+    """The members every store has. A store may also offer `read(key) -> tuple[CultCacheEnvelope, str] | None` (the record at the
+    key, of any type, and its content version, from one read) and `commit_if(envelope, expected: str | None) -> str | None` (write
+    only if the record at the key is still the version `expected`, None meaning no record; returns the stored_at written, or None
+    with nothing written). They are looked up with getattr, not Protocol members, and `CultCache.update` needs both. `push` and
+    `push_all` may report the stored_at they wrote (`str` and `dict[str, str]` by key); None keeps the cache's own stamp. A store
+    stamps a record that replaces another strictly later than the one it replaces."""
+
     def pull_all(self) -> list[CultCacheEnvelope]:
         ...
 
-    def push(self, envelope: CultCacheEnvelope) -> None:
+    def push(self, envelope: CultCacheEnvelope) -> str | None:
         ...
 
     def delete(self, type: str, key: str) -> None:
@@ -74,7 +81,7 @@ class BackingStore(Protocol):
         that can tell nothing-at-the-path overrides this, and the default push_all then writes where nothing is on a soft push."""
         return self.pull_all()
 
-    def push_all(self, envelopes: list[CultCacheEnvelope], soft: bool = False) -> None:
+    def push_all(self, envelopes: list[CultCacheEnvelope], soft: bool = False) -> dict[str, str] | None:
         # Soft: write only where pull_or_absent says nothing is; skip over a store it read, empty included; its raise refuses.
         pulled = self.pull_or_absent()
         if soft and pulled is not None:

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

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import enum
 import functools
 import math
 import threading
@@ -9,13 +10,39 @@ from typing import Any, Callable, Generic, Iterable, TypeVar
 
 from .backing_store import BackingStore, CultCacheEnvelope
 from .documents import DocumentDefinition, UnreadSlots, extract_value
-from .stores import SchemaConflictError, retyped, written_owns
+from .stores import SchemaConflictError, _parse_stored_at, retyped, stored_at_over, written_owns
 
 T = TypeVar("T")
 F = TypeVar("F", bound=Callable[..., Any])
 GLOBAL_KEY = "__global__"
 
 
+class UpdateOutcome(enum.Enum):
+    UPDATED = "updated"
+    DECLINED = "declined"
+    LOST_RACE = "lost_race"
+
+
+@dataclass(frozen=True)
+class UpdateResult(Generic[T]):
+    """What `CultCache.update` did and the cache's record at the key after it (None when it holds nothing). The document decides
+    nothing: it is a read after the outcome."""
+
+    outcome: UpdateOutcome
+    document: T | None
+
+
+@dataclass(frozen=True)
+class _Expected:
+    """What an update's commit requires to still hold. A cache with no store requires the held envelope object at the key (None:
+    nothing held); a store-backed one requires the type's home to be `store` and the record there to have content version
+    `version` (None: no record)."""
+
+    held: CultCacheEnvelope | None = None
+    store: BackingStore | None = None
+    version: str | None = None
+
+
 def _locked(method: F) -> F:
     """Runs a read holding the cache's lock. Reads are reentrant and may run inside a mutation."""
 
@@ -482,7 +509,7 @@ class CultCache:
     def put(self, document: DocumentDefinition[T], key: str, value: T) -> None:
         self._put(document, key, value)
 
-    def _put(self, document: DocumentDefinition[T], key: str, value: T) -> None:
+    def _put(self, document: DocumentDefinition[T], key: str, value: T, expected: _Expected | None = None) -> str:
         self._assert_registered(document)
         if document.global_document and key != GLOBAL_KEY:
             raise CultCacheError(f"Global document {document.type} must use key {GLOBAL_KEY}")
@@ -494,7 +521,7 @@ class CultCache:
             schema_id=catalog_entry.schema_id,
             catalog_entry=catalog_entry,
         )
-        self._write(document, [(envelope, value)], batch=False)
+        return self._write(document, [(envelope, value)], batch=False, expected=expected)
 
     @_mutating
     def put_envelope(self, document: DocumentDefinition[T], envelope: CultCacheEnvelope) -> T:
@@ -562,9 +589,12 @@ class CultCache:
         *,
         batch: bool,
         unread: dict[str, UnreadSlots | None] | None = None,
-    ) -> None:
+        expected: _Expected | None = None,
+    ) -> str:
         """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."""
+        touched; once the store accepts, applying to the cache has no fallible step. With `expected` (an update's commit) the write
+        also requires the record to be unchanged since it was read and returns 'mismatch', having written and installed nothing,
+        when it is not; otherwise 'committed'. A record is held under the stored_at its store reports writing."""
         type = document.type
         # 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.
@@ -584,13 +614,34 @@ class CultCache:
         )
         store = self._store_for_type(type)
         legacy_key = self._state.legacy_global_keys.get(type)
-        if store is not None:
-            if batch:
-                store.push_all([envelope for envelope, _ in records])
+        stamps: dict[str, Any] = {}
+        if expected is not None:
+            if legacy_key is not None:
+                raise self._legacy_global_refusal(document, legacy_key)
+            if expected.store is None:
+                held = self._state.envelopes.get(type, {}).get(records[0][0].key)
+                if store is not None or held is not expected.held:
+                    return "mismatch"
+            elif store is not expected.store:
+                return "mismatch"
             else:
-                store.push(records[0][0])
+                stamp = store.commit_if(records[0][0], expected.version)  # type: ignore[attr-defined]
+                if stamp is None:
+                    return "mismatch"
+                stamps[records[0][0].key] = stamp
+        elif store is not None:
+            if batch:
+                reported = store.push_all([envelope for envelope, _ in records])
+                stamps = reported if isinstance(reported, dict) else {}
+            else:
+                stamps[records[0][0].key] = store.push(records[0][0])
             if legacy_key is not None:
                 self._retire_legacy_global(store, type, legacy_key)
+        else:
+            for envelope, _ in records:
+                held_at = next((held for held in (by_type.get(envelope.key) for by_type in self._state.envelopes.values()) if held is not None), None)
+                if held_at is not None:
+                    stamps[envelope.key] = stored_at_over(envelope.stored_at, held_at.stored_at)
         self._state.legacy_global_keys.pop(type, None)
         values = self._state.values.setdefault(type, {})
         envelopes = self._state.envelopes.setdefault(type, {})
@@ -598,6 +649,9 @@ class CultCache:
         for other, key in displaced:
             self._drop(other, key)
         for envelope, value, keys in admitted:
+            stamp = stamps.get(envelope.key)
+            if isinstance(stamp, str) and _parse_stored_at(stamp) is not None:
+                envelope = replace(envelope, stored_at=stamp)
             self._remove_lookups(type, envelope.key)
             values[envelope.key] = value
             envelopes[envelope.key] = envelope
@@ -609,6 +663,7 @@ class CultCache:
                 self._state.names[(type, keys.name)] = envelope.key
             for index, index_value in keys.indexes.items():
                 self._state.indexes[(type, index, index_value)] = envelope.key
+        return "committed"
 
     def _refuse_retype(self, document: DocumentDefinition[Any], written: list[CultCacheEnvelope]) -> list[tuple[str, str]]:
         """The one key-identity judge (variants:ruling:cache-refuses-retype): a key holds one record whatever its type, so a write
@@ -663,18 +718,34 @@ class CultCache:
             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,
-            )
+            raise self._read_only_refusal(self._state.documents[type], key, self._state.envelopes[type][key].schema_id, slots)
+
+    @staticmethod
+    def _read_only_refusal(
+        document: DocumentDefinition[Any], key: str, schema_id: str | None, slots: UnreadSlots
+    ) -> SchemaConflictError:
+        """The one refusal of a write to a record that carries slots this build would drop: a record the cache holds read-only,
+        or one an update reads from the store."""
+        name = document.catalog_entry().schema_name
+        return SchemaConflictError(
+            schema_id or document.type,
+            [name],
+            key,
+            message=(
+                f"{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 _legacy_global_refusal(document: DocumentDefinition[Any], legacy_key: str) -> CultCacheError:
+        return CultCacheError(
+            f"Global {document.type} is held under the legacy key {legacy_key!r}; update cannot copy it. "
+            "put_global writes the global and retires the legacy key."
+        )
 
     @staticmethod
     def _foreign_refusal(record: ForeignRecord) -> SchemaConflictError:
@@ -707,20 +778,81 @@ class CultCache:
         self._assert_global(document)
         self._put(document, GLOBAL_KEY, value)
 
-    @_mutating
-    def update(self, document: DocumentDefinition[T], key: str, updater: Any) -> T:
-        return self._update(document, key, updater)
+    def update(
+        self, document: DocumentDefinition[T], key: str, apply: Callable[[T | None], T | None], *, max_attempts: int = 5
+    ) -> UpdateResult[T]:
+        """Reads the record at the key as its home store holds it, applies `apply` to a fresh copy of it, and commits the result
+        only if the record is still the one that was read; when another writer got there first it reads again and runs `apply`
+        again, up to `max_attempts` times.
 
-    def _update(self, document: DocumentDefinition[T], key: str, updater: Any) -> T:
-        current = self.get_required(document, key)
-        updated = updater(current)
-        self._put(document, key, updated)
-        return updated
+        `apply` may run up to `max_attempts` times, on a fresh copy of the record as the store holds it now (None when the key
+        holds nothing), which it may change and return. It must not write the same key through this cache or have side effects.
+        Returning None writes nothing (DECLINED). A lost race that outlasts `max_attempts` returns LOST_RACE and does not raise;
+        a refusal, or a raise from `apply`, propagates and is not retried. The write is durable when update returns. The
+        result's document is the cache's record after the outcome. `put` is the blind write. A cache with no store copies its
+        held instance; a store without `read` and `commit_if` (the JSON lines store among them) refuses before `apply` runs."""
+        if not callable(apply):
+            raise TypeError("CultCache update needs a function to apply.")
+        if not isinstance(max_attempts, int) or isinstance(max_attempts, bool):
+            raise TypeError(f"CultCache update max_attempts must be an int, not {type(max_attempts).__name__}.")
+        if max_attempts < 1:
+            raise ValueError(f"CultCache update max_attempts must be at least 1, not {max_attempts}.")
+        for attempt in range(1, max_attempts + 1):
+            copy, expected = self._read_for_update(document, key)
+            # Outside the cache's lock and the store's: apply may take its time and may write other keys through this cache.
+            next_value = apply(copy)
+            if next_value is None:
+                return UpdateResult(UpdateOutcome.DECLINED, self.get(document, key))
+            if self._commit_update(document, key, next_value, expected) == "committed":
+                return UpdateResult(UpdateOutcome.UPDATED, self.get(document, key))
+        # Display only: the cache shows what the winner left.
+        if self._state.stores_by_type or self._state.generic_store is not None:
+            self.pull_all_backing_stores()
+        return UpdateResult(UpdateOutcome.LOST_RACE, self.get(document, key))
 
     @_mutating
-    def update_global(self, document: DocumentDefinition[T], updater: Any) -> T:
+    def _read_for_update(self, document: DocumentDefinition[T], key: str) -> tuple[T | None, _Expected]:
+        """One attempt's read: the copy to hand `apply` and the condition its commit holds. Every refusal that does not depend on
+        what `apply` returns is raised here, so `apply` never runs for a write that cannot land. Installs nothing."""
+        self._assert_registered(document)
+        self._refuse_at(document, key)
+        legacy_key = self._state.legacy_global_keys.get(document.type)
+        if legacy_key is not None:
+            raise self._legacy_global_refusal(document, legacy_key)
+        store = self._store_for_type(document.type)
+        if store is None:
+            held = self._state.envelopes.get(document.type, {}).get(key)
+            return (None if held is None else document.decode_payload(held.payload)), _Expected(held=held)
+        read, commit_if = getattr(store, "read", None), getattr(store, "commit_if", None)
+        if read is None or commit_if is None:
+            raise CultCacheError(
+                f"Backing store {type(store).__name__} for {document.type} cannot write conditionally; "
+                "update needs a store with read and commit_if."
+            )
+        if document.global_document:
+            for envelope in store.pull_all():
+                if envelope.key != GLOBAL_KEY and self._resolve_document_for_envelope(envelope) is document:
+                    raise self._legacy_global_refusal(document, envelope.key)
+        found = read(key)
+        if found is None:
+            return None, _Expected(store=store)
+        envelope, version = found
+        if self._resolve_document_for_envelope(envelope) is not document:
+            raise CultCacheError(f"Record {key} is not a {document.type}; update cannot copy it.")
+        value, unread = self._decode_marked(document, envelope)
+        if unread is not None:
+            raise self._read_only_refusal(document, key, envelope.schema_id, unread)
+        return value, _Expected(store=store, version=version)
+
+    @_mutating
+    def _commit_update(self, document: DocumentDefinition[T], key: str, value: T, expected: _Expected) -> str:
+        return self._put(document, key, value, expected)
+
+    def update_global(
+        self, document: DocumentDefinition[T], apply: Callable[[T | None], T | None], *, max_attempts: int = 5
+    ) -> UpdateResult[T]:
         self._assert_global(document)
-        return self._update(document, GLOBAL_KEY, updater)
+        return self.update(document, GLOBAL_KEY, apply, max_attempts=max_attempts)
 
     @_mutating
     def delete(self, document: DocumentDefinition[Any], key: str) -> None:

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

@@ -1,13 +1,17 @@
 from __future__ import annotations
 
 import base64
+import hashlib
 import json
 import os
+import re
 import sys
 import threading
 import time
 import uuid
 from contextlib import contextmanager
+from dataclasses import replace
+from datetime import datetime, timedelta, timezone
 from pathlib import Path
 from typing import Any, Callable, Iterator, TypeVar
 
@@ -107,14 +111,23 @@ class JsonLinesBackingStore:
                     raise ValueError(f"Malformed CultCache JSONL envelope at {self.path}:{line_number}: missing {exc}") from exc
             return envelopes
 
-    def push(self, envelope: CultCacheEnvelope) -> None:
-        self.push_all([envelope])
+    def push(self, envelope: CultCacheEnvelope) -> str | None:
+        return self.push_all([envelope])[envelope.key]
+
+    def push_all(self, envelopes: list[CultCacheEnvelope], soft: bool = False) -> dict[str, str]:
+        """Merges the envelopes, each stored later than any record it replaces, and returns the stamps written by key
+        ({} when a soft push skipped)."""
+        stamps: dict[str, str] = {}
 
-    def push_all(self, envelopes: list[CultCacheEnvelope], soft: bool = False) -> None:
         def merge(existing: dict[tuple[str, str], CultCacheEnvelope]) -> None:
             for envelope in envelopes:
+                replaced = existing.get((envelope.type, envelope.key))
+                if replaced is not None:
+                    envelope = replace(envelope, stored_at=stored_at_over(envelope.stored_at, replaced.stored_at))
                 existing[(envelope.type, envelope.key)] = envelope
+                stamps[envelope.key] = envelope.stored_at
         self._rewrite(merge, soft)
+        return stamps
 
     def delete(self, type: str, key: str) -> None:
         self._rewrite(lambda existing: existing.pop((type, key), None))
@@ -177,23 +190,40 @@ class SingleFileMessagePackBackingStore:
             except Exception as exc:
                 raise StoreUnreadableError(self.path, exc) from exc
 
-    def push(self, envelope: CultCacheEnvelope) -> None:
+    def push(self, envelope: CultCacheEnvelope) -> str | None:
         with self._write_transaction():
-            self._replace_all([envelope], self.pull_all())
+            return self._replace_all([envelope], self.pull_all())[envelope.key]
 
-    def push_all(self, envelopes: list[CultCacheEnvelope], soft: bool = False) -> None:
+    def push_all(self, envelopes: list[CultCacheEnvelope], soft: bool = False) -> dict[str, str]:
         """A soft push writes only where nothing is at the path: a store the reader reads is left byte for byte, and a file it
-        cannot read is refused by the same read, as in TS and Rust."""
+        cannot read is refused by the same read, as in TS and Rust. Returns the stamps written by key ({} when skipped)."""
         with self._write_transaction():
             arrived = self._pull()
             if soft and arrived is not None:
-                return
-            self._replace_all(list(envelopes), arrived or [])
+                return {}
+            return self._replace_all(list(envelopes), arrived or [])
 
     def delete(self, type: str, key: str) -> None:
         with self._write_transaction():
             self._replace_all([], [item for item in self.pull_all() if not (item.type == type and item.key == key)])
 
+    def read(self, key: str) -> tuple[CultCacheEnvelope, str] | None:
+        """The record at the key, of any type, and its content version, from one read; None when the key holds nothing."""
+        for envelope in self._pull() or []:
+            if envelope.key == key:
+                return envelope, record_version(envelope)
+        return None
+
+    def commit_if(self, envelope: CultCacheEnvelope, expected: str | None) -> str | None:
+        """Writes the envelope only if the record at its key (any type) is still the one whose version is `expected`
+        (None: no record), judged under the write lock; returns the stored_at written, or None with nothing written."""
+        with self._write_transaction():
+            arrived = self._pull() or []
+            current = next((item for item in arrived if item.key == envelope.key), None)
+            if (None if current is None else record_version(current)) != expected:
+                return None
+            return self._replace_all([envelope], arrived)[envelope.key]
+
     @contextmanager
     def _write_transaction(self) -> Iterator[None]:
         """The thread lock, then the cross-process lock on `<path>.lock`, held across the read and the replace so two
@@ -203,8 +233,10 @@ class SingleFileMessagePackBackingStore:
             with store_write_lock(self.path, self.lock_timeout):
                 yield
 
-    def _replace_all(self, supplied: list[CultCacheEnvelope], arrived: list[CultCacheEnvelope]) -> None:
-        commit_store_file(self.path, msgpack.packb(_encode_snapshot(supplied, arrived, self._format), use_bin_type=True))
+    def _replace_all(self, supplied: list[CultCacheEnvelope], arrived: list[CultCacheEnvelope]) -> dict[str, str]:
+        snapshot, stamps = _encode_snapshot(supplied, arrived, self._format)
+        commit_store_file(self.path, msgpack.packb(snapshot, use_bin_type=True))
+        return stamps
 
 
 
@@ -305,7 +337,7 @@ def written_owns(held: CultCacheEnvelope, written: CultCacheEnvelope) -> bool:
 
 def _encode_snapshot(
     supplied: list[CultCacheEnvelope], arrived: list[CultCacheEnvelope], format_version: str
-) -> list[Any]:
+) -> tuple[list[Any], dict[str, str]]:
     # A key names one record whatever its type, so a supplied record replaces the arrived one by key alone, and only of the
     # type that owns it: retyping a key is a removal and an add, so a record another type owns under the key refuses the write.
     held_by_key = {envelope.key: envelope for envelope in arrived}
@@ -314,8 +346,12 @@ def _encode_snapshot(
         for held in (landing.get(envelope.key), held_by_key.get(envelope.key)):
             if held is not None and not written_owns(held, envelope):
                 raise retyped(held, envelope)
+        held = held_by_key.get(envelope.key)
+        if held is not None:
+            envelope = replace(envelope, stored_at=stored_at_over(envelope.stored_at, held.stored_at))
         landing[envelope.key] = envelope
     supplied = list(landing.values())
+    stamps = {envelope.key: envelope.stored_at for envelope in supplied}
     replaced = set(landing)
     kept = [envelope for envelope in arrived if envelope.key not in replaced]
     envelopes = [*kept, *supplied]
@@ -330,7 +366,7 @@ def _encode_snapshot(
         [envelope.key, _schema_id_for(envelope), envelope.stored_at, envelope.payload]
         for envelope in sorted(envelopes, key=lambda item: item.key)
     ]
-    return [format_version, catalog, records]
+    return [format_version, catalog, records], stamps
 
 def commit_store_file(path: Path, data: bytes) -> None:
     """Writes `data` to a unique staged file beside the store, fsyncs it, replaces the store with it and, off Windows,
@@ -739,6 +775,65 @@ def _schema_id_for(envelope: CultCacheEnvelope) -> str:
     return envelope.schema_id or envelope.type
 
 
+_STORED_AT_SHAPE = re.compile(
+    r"([0-9]{4})-([0-9]{2})-([0-9]{2})T([0-9]{2}):([0-9]{2}):([0-9]{2})(?:\.([0-9]{1,9}))?(?:(Z)|([+-])([01][0-9]|2[0-3]):([0-5][0-9]))"
+)
+
+
+def _parse_stored_at(text: Any) -> datetime | None:
+    """The UTC instant a stored_at text names, or None. Only the shape the runtimes write is a stamp: an RFC 3339 date-time with T,
+    an optional fraction of 1 to 9 digits (cut to the microsecond) and Z or a numeric offset of hours 00-23 and minutes 00-59. Checked here, not left to
+    fromisoformat, whose accepted set differs by interpreter (3.10 and 3.13) and from Rust and TypeScript. An instant outside
+    datetime's range is not a stamp."""
+    if not isinstance(text, str):
+        return None
+    shape = _STORED_AT_SHAPE.fullmatch(text)
+    if shape is None:
+        return None
+    year, month, day, hour, minute, second, fraction, zulu, sign, offset_hours, offset_minutes = shape.groups()
+    try:
+        offset = timezone.utc if zulu else timezone(
+            (-1 if sign == "-" else 1) * timedelta(hours=int(offset_hours), minutes=int(offset_minutes))
+        )
+        local = datetime(
+            int(year), int(month), int(day), int(hour), int(minute), int(second), int(((fraction or "") + "000000")[:6]), offset
+        )
+        return local.astimezone(timezone.utc)
+    except (ValueError, OverflowError):
+        return None
+
+
+def mint_stored_at(previous: str | None, now: datetime) -> str:
+    """A stored_at later than `previous` (C# MintStoredAt): `now`, or the microsecond after `previous` when `now` is not later.
+    A previous that does not parse leaves `now`. Raises ValueError, nothing minted, when `previous` is the last instant a stamp
+    can name, so a write over it is refused and writes nothing."""
+    now = now.astimezone(timezone.utc)
+    last = _parse_stored_at(previous)
+    if last is not None and now <= last:
+        try:
+            now = last + timedelta(microseconds=1)
+        except OverflowError:
+            raise ValueError(f"stored_at {previous!r} is the last instant a stamp can name; a write over it cannot be stored later") from None
+    return now.isoformat(timespec="microseconds")
+
+
+def stored_at_over(staged: str, replaced: str) -> str:
+    """The stored_at a record replacing `replaced` is written with (C# StoredAtOver): `staged` when both parse and it is strictly
+    later, else one minted over `replaced`. No content is compared: every replacing write is stamped later."""
+    later, earlier = _parse_stored_at(staged), _parse_stored_at(replaced)
+    if later is not None and earlier is not None and later > earlier:
+        return staged
+    return mint_stored_at(replaced, datetime.now(timezone.utc))
+
+
+def record_version(envelope: CultCacheEnvelope) -> str:
+    """The content version of a stored record: the lowercase hex SHA-256 of its persisted encoding, the array
+    [key, schema id, stored_at, payload] exactly as the single-file store writes it
+    (variants:ruling:write-set-content-version-declared-ownership)."""
+    record = [envelope.key, _schema_id_for(envelope), envelope.stored_at, envelope.payload]
+    return hashlib.sha256(msgpack.packb(record, use_bin_type=True)).hexdigest()
+
+
 def _default_catalog_entry(envelope: CultCacheEnvelope) -> CultCacheSchemaCatalogEntry:
     schema_id = _schema_id_for(envelope)
     return CultCacheSchemaCatalogEntry(

packages/cultcache-py/tests/test_content_version_vector.py

@@ -0,0 +1,38 @@
+from __future__ import annotations
+
+import shutil
+import tempfile
+import unittest
+from pathlib import Path
+
+import msgpack
+
+from cultcache_py import SingleFileMessagePackBackingStore
+from cultcache_py.stores import record_version
+
+# variants:ruling:write-set-content-version-declared-ownership. The vector in tests/vectors/content-version is written only by
+# the C# reference; every row of it is computed here from the store file, by the store's read and by record_version.
+VECTORS = Path(__file__).resolve().parents[3] / "tests" / "vectors" / "content-version"
+
+
+class ContentVersionVectorTests(unittest.TestCase):
+    def test_every_row_of_the_vector_equals_this_runtimes_version(self) -> None:
+        rows = msgpack.unpackb((VECTORS / "versions.msgpack").read_bytes(), raw=False)
+        self.assertEqual([key for key, _ in rows], ["alpha", "beta"])
+        directory = Path(tempfile.mkdtemp())
+        self.addCleanup(shutil.rmtree, directory, ignore_errors=True)
+        copy = directory / "store.cc"
+        shutil.copyfile(VECTORS / "store.cc", copy)
+        store = SingleFileMessagePackBackingStore(copy)
+        by_key = {envelope.key: envelope for envelope in store.pull_all()}
+        for key, version in rows:
+            with self.subTest(key=key):
+                found = store.read(key)
+                self.assertIsNotNone(found)
+                self.assertEqual(found[1], version)
+                self.assertEqual(record_version(by_key[key]), version)
+        self.assertIsNone(store.read("gamma"))
+
+
+if __name__ == "__main__":
+    unittest.main()

packages/cultcache-py/tests/test_cultcache.py

@@ -86,7 +86,7 @@ class CultCacheTests(unittest.TestCase):
             )
             cache.pull_all_backing_stores()
             cache.put(document, "item:potion", Item(name="Potion", category="Consumable", value=50))
-            cache.update(document, "item:potion", lambda item: Item(item.name, item.category, item.value + 5))
+            cache.put(document, "item:potion", Item(name="Potion", category="Consumable", value=55))
             self.assertEqual(cache.get_required(document, "item:potion").value, 55)
             cache.delete(document, "item:potion")
             self.assertIsNone(cache.get(document, "item:potion"))

packages/cultcache-py/tests/test_first_call.py

@@ -4,6 +4,8 @@
 import unittest
 from pathlib import Path
 
+from cultcache_py import UpdateOutcome
+
 # snippet:begin
 import tempfile
 from dataclasses import asdict, dataclass
@@ -51,15 +53,44 @@ def first_call() -> Item:
 # snippet:end
 
 
+# snippet:update:begin
+def restock(path: Path):
+    cache = (
+        CultCache.builder()
+        .register_document_type(item_doc)
+        .add_generic_store(SingleFileMessagePackBackingStore(path))
+        .build()
+    )
+    cache.pull_all_backing_stores()
+    return cache.update(item_doc, "item:potion", lambda item: Item("Potion", "Consumable", (item.value if item else 0) + 10))
+# snippet:update:end
+
+
 class FirstCallTests(unittest.TestCase):
     def test_first_call_round_trips_through_a_second_cache(self) -> None:
         self.assertEqual(first_call(), Item("Potion", "Consumable", 50))
 
-    def test_docs_show_the_tested_snippet(self) -> None:
+    def test_update_with_retry_through_a_second_cache(self) -> None:
+        path = Path(tempfile.mkdtemp()) / "restock.cc"
+        restock(path)
+        result = restock(path)
+        self.assertIs(result.outcome, UpdateOutcome.UPDATED)
+        reopened = (
+            CultCache.builder()
+            .register_document_type(item_doc)
+            .add_generic_store(SingleFileMessagePackBackingStore(path))
+            .build()
+        )
+        reopened.pull_all_backing_stores()
+        self.assertEqual(reopened.get_required(item_doc, "item:potion"), Item("Potion", "Consumable", 20))
+
+    def test_docs_show_the_tested_snippets(self) -> None:
         source = Path(__file__).read_text(encoding="utf-8")
-        snippet = source.split("# snippet:begin\n", 1)[1].split("# snippet:end\n", 1)[0]
         docs = (Path(__file__).parents[3] / "docs" / "start" / "README.md").read_text(encoding="utf-8")
-        self.assertIn("```python\n" + snippet + "```", docs)
+        for region in ("snippet", "snippet:update"):
+            with self.subTest(region=region):
+                snippet = source.split(f"# {region}:begin\n", 1)[1].split(f"# {region}:end\n", 1)[0]
+                self.assertIn("```python\n" + snippet + "```", docs)
 
 
 if __name__ == "__main__":

packages/cultcache-py/tests/test_unique_index.py

@@ -148,7 +148,6 @@ class UniqueIndexTests(unittest.TestCase):
                 "put": lambda cache, d, k, i: cache.put(d, k, i),
                 "put_envelope": lambda cache, d, k, i: cache.put_envelope(d, envelope(d, k, i)),
                 "put_envelopes": lambda cache, d, k, i: cache.put_envelopes(d, [envelope(d, k, i)]),
-                "update": lambda cache, d, k, i: cache.update(d, k, lambda _held: i),
             }
             for name, write in paths.items():
                 with self.subTest(alias=alias, path=name):

packages/cultcache-py/tests/test_update.py

@@ -0,0 +1,731 @@
+"""CultCache.update (variants:cut_spec:cut-update-with-retry-py): read the record as its home store holds it, apply a function,
+commit only if the record is unchanged, run the function again on a lost race. Mirrors C# UpdateTests and the TS update tests."""
+
+from __future__ import annotations
+
+import shutil
+import tempfile
+import threading
+import time
+import unittest
+from dataclasses import asdict, dataclass
+from datetime import datetime, timezone
+from pathlib import Path
+
+from cultcache_py import (
+    CultCache,
+    CultCacheEnvelope,
+    JsonLinesBackingStore,
+    SchemaConflictError,
+    SingleFileMessagePackBackingStore,
+    UpdateOutcome,
+    UpdateResult,
+    database_entry_field,
+    define_database_entry_type,
+    define_document_type,
+    leaves as t,
+)
+from cultcache_py.cache import GLOBAL_KEY, CultCacheError
+from cultcache_py.stores import _parse_stored_at, mint_stored_at, record_version, stored_at_over
+
+
+@dataclass
+class Item:
+    name: str = "n"
+    value: int = 0
+    sku: str | None = None
+
+
+def _doc(type: str, **extra):
+    return define_document_type(type, encode=asdict, decode=lambda raw: Item(**raw), **extra)
+
+
+item_doc = _doc("item")
+other_doc = _doc("other")
+sku_doc = _doc("skuitem", indexes={"sku": "sku"})
+global_doc = _doc("gitem", global_document=True)
+
+@dataclass
+class Wide:
+    name: str = ""
+    extra: str = ""
+
+
+@dataclass
+class Narrow:
+    name: str = ""
+
+
+ALPHA = "2099-01-01T00:00:00.0000000Z"
+
+
+class Fixture(unittest.TestCase):
+    def setUp(self) -> None:
+        self.dir = Path(tempfile.mkdtemp())
+        self.addCleanup(shutil.rmtree, self.dir, ignore_errors=True)
+        self.path = self.dir / "state.cc"
+
+    def cache(self, *documents, path: Path | None = None, store=True) -> CultCache:
+        builder = CultCache.builder()
+        for document in documents or (item_doc,):
+            builder.register_document_type(document)
+        if store:
+            builder.add_generic_store(SingleFileMessagePackBackingStore(path or self.path))
+        cache = builder.build()
+        if store:
+            cache.pull_all_backing_stores()
+        return cache
+
+    def stored(self, key: str = "k", document=item_doc):
+        """What a fresh cache reads at the key."""
+        return self.cache(document).get(document, key)
+
+    def bytes(self) -> bytes | None:
+        return self.path.read_bytes() if self.path.exists() else None
+
+
+class Counting:
+    """An apply that records what it was given."""
+
+    def __init__(self, fn=None) -> None:
+        self.seen: list = []
+        self.fn = fn or (lambda item, calls: Item("n", (item.value if item else 0) + 1))
+
+    def __call__(self, item):
+        self.seen.append(None if item is None else item.value)
+        # An always-losing apply must fail by assertion when the attempt bound is gone, not hang.
+        assert len(self.seen) <= 10, "apply ran past the attempt bound"
+        return self.fn(item, len(self.seen))
+
+    @property
+    def calls(self) -> int:
+        return len(self.seen)
+
+
+class UpdateTests(Fixture):
+    def test_two_writers_updating_lose_no_increment(self) -> None:
+        results: list[UpdateResult] = []
+        calls = [0]
+        lock = threading.Lock()
+        a, b = self.cache(), self.cache()
+        a.put(item_doc, "k", Item("n", 0))
+
+        def bump(item):
+            with lock:
+                calls[0] += 1
+            time.sleep(0)
+            return Item("n", item.value + 1)
+
+        def run(cache) -> None:
+            for _ in range(50):
+                results.append(cache.update(item_doc, "k", bump, max_attempts=200))
+
+        threads = [threading.Thread(target=run, args=(cache,)) for cache in (a, b)]
+        for thread in threads:
+            thread.start()
+        for thread in threads:
+            thread.join()
+        self.assertTrue(all(result.outcome is UpdateOutcome.UPDATED for result in results))
+        self.assertEqual(self.stored().value, 100)
+        self.assertGreater(calls[0], 100)
+
+    def test_a_lost_attempt_reruns_apply_on_the_fresh_value(self) -> None:
+        cache, rival = self.cache(), self.cache()
+        cache.put(item_doc, "k", Item("n", 1))
+        rival.pull_all_backing_stores()
+        apply = Counting(lambda item, calls: (rival.put(item_doc, "k", Item("n", 100)) if calls == 1 else None) or Item("n", item.value + 1))
+        result = cache.update(item_doc, "k", apply)
+        self.assertEqual(apply.seen, [1, 100])
+        self.assertIs(result.outcome, UpdateOutcome.UPDATED)
+        self.assertEqual(result.document.value, 101)
+        self.assertEqual(self.stored().value, 101)
+
+    def test_exhausted_attempts_return_lost_race(self) -> None:
+        cache, rival = self.cache(), self.cache()
+        cache.put(item_doc, "k", Item("n", 1))
+        rival.pull_all_backing_stores()
+        apply = Counting(lambda item, calls: rival.put(item_doc, "k", Item("n", 100 * calls)) or Item("n", -1))
+        result = cache.update(item_doc, "k", apply, max_attempts=3)
+        self.assertEqual(apply.calls, 3)
+        self.assertIs(result.outcome, UpdateOutcome.LOST_RACE)
+        self.assertEqual(result.document.value, 300)  # the pull before LOST_RACE shows the winner
+        self.assertEqual(self.stored().value, 300)
+
+    def test_max_attempts_bounds(self) -> None:
+        cache, rival = self.cache(), self.cache()
+        cache.put(item_doc, "k", Item("n", 1))
+        rival.pull_all_backing_stores()
+
+        def always_losing(item, calls):
+            rival.put(item_doc, "k", Item("n", 100 + calls))
+            return Item("n", -1)
+
+        apply = Counting(always_losing)
+        result = cache.update(item_doc, "k", apply, max_attempts=1)
+        self.assertEqual((apply.calls, result.outcome), (1, UpdateOutcome.LOST_RACE))
+        for bad, error in ((0, ValueError), (-1, ValueError), (True, TypeError), (1.5, TypeError), (None, TypeError)):
+            with self.subTest(max_attempts=bad):
+                untouched = Counting()
+                with self.assertRaises(error):
+                    cache.update(item_doc, "k", untouched, max_attempts=bad)
+                self.assertEqual(untouched.calls, 0)
+        with self.assertRaises(TypeError):
+            cache.update(item_doc, "k", "not callable")
+        with self.assertRaises(TypeError):
+            cache.update(item_doc, "k", None)
+
+    def test_apply_gets_a_fresh_copy(self) -> None:
+        cache, rival = self.cache(), self.cache()
+        cache.put(item_doc, "k", Item("n", 1))
+        rival.pull_all_backing_stores()
+        arguments: list[Item] = []
+
+        def apply(item):
+            arguments.append(item)
+            self.assertIsNot(item, cache.get(item_doc, "k"))
+            if len(arguments) == 1:
+                item.value = 777  # changing the copy in place changes nothing the cache or the next attempt sees
+                rival.put(item_doc, "k", Item("n", 5))
+                self.assertEqual(cache.get(item_doc, "k").value, 1)
+            return Item("n", item.value + 1)
+
+        result = cache.update(item_doc, "k", apply)
+        self.assertEqual([item.value for item in arguments], [777, 5])
+        self.assertIsNot(arguments[0], arguments[1])
+        self.assertEqual(result.document.value, 6)
+
+    def test_an_in_place_change_to_the_held_value_is_neither_applied_nor_persisted(self) -> None:
+        cache = self.cache()
+        cache.put(item_doc, "k", Item("n", 1))
+        cache.get(item_doc, "k").value = 999
+        apply = Counting(lambda item, calls: Item("n", item.value + 1))
+        result = cache.update(item_doc, "k", apply)
+        self.assertEqual(apply.seen, [1])
+        self.assertEqual(self.stored().value, 2)
+        result.document.value = 999
+        cache.update(item_doc, "k", apply)
+        self.assertEqual(apply.seen, [1, 2])
+        self.assertEqual(self.stored().value, 3)
+
+        bare = self.cache(store=False)  # a cache with no store copies its held instance
+        bare.put(item_doc, "k", Item("n", 1))
+        bare.get(item_doc, "k").value = 999
+        seen = Counting(lambda item, calls: Item("n", item.value + 1))
+        self.assertEqual(bare.update(item_doc, "k", seen).document.value, 2)
+        self.assertEqual(seen.seen, [1])
+
+    def test_none_declines(self) -> None:
+        cache = self.cache()
+        cache.put(item_doc, "k", Item("n", 1))
+        before = self.bytes()
+        result = cache.update(item_doc, "k", lambda item: None)
+        self.assertIs(result.outcome, UpdateOutcome.DECLINED)
+        self.assertEqual(result.document, cache.get(item_doc, "k"))
+        self.assertEqual(self.bytes(), before)
+        absent = cache.update(item_doc, "missing", lambda item: None)
+        self.assertEqual((absent.outcome, absent.document), (UpdateOutcome.DECLINED, None))
+        self.assertEqual(self.bytes(), before)
+        empty = Fixture.cache(self, path=self.dir / "empty.cc")
+        self.assertEqual(empty.update(item_doc, "k", lambda item: None).outcome, UpdateOutcome.DECLINED)
+        self.assertFalse((self.dir / "empty.cc").exists())
+
+    def test_a_declined_or_lost_attempt_installs_nothing(self) -> None:
+        cache, rival = self.cache(), self.cache()
+        cache.put(item_doc, "k", Item("n", 1))
+        rival.pull_all_backing_stores()
+        rival.put(item_doc, "k", Item("n", 5))
+        self.assertIs(cache.update(item_doc, "k", lambda item: None).outcome, UpdateOutcome.DECLINED)
+        self.assertEqual(cache.get(item_doc, "k").value, 1)
+        cache.pull_all_backing_stores()
+        self.assertEqual(cache.get(item_doc, "k").value, 5)
+
+        during: list[int] = []
+
+        def apply(item, calls):
+            during.append(cache.get(item_doc, "k").value)
+            if calls == 1:
+                rival.put(item_doc, "k", Item("n", 9))
+            return Item("n", item.value + 1)
+
+        cache.update(item_doc, "k", Counting(apply))
+        self.assertEqual(during, [5, 5])  # the lost attempt published nothing mid-update
+
+    def test_absent_key_rival_removal_and_a_store_that_holds_nothing(self) -> None:
+        a, b = self.cache(), self.cache()
+        bump = lambda item: Item("n", (item.value if item else 0) + 1)  # noqa: E731
+        self.assertIs(a.update(item_doc, "k", bump).outcome, UpdateOutcome.UPDATED)
+        self.assertIs(b.update(item_doc, "k", bump).outcome, UpdateOutcome.UPDATED)
+        self.assertEqual(self.stored().value, 2)
+
+        cache, rival = self.cache(), self.cache()
+        cache.put(item_doc, "d", Item("n", 1))
+        rival.pull_all_backing_stores()
+        apply = Counting(lambda item, calls: (rival.delete(item_doc, "d") if calls == 1 else None) or Item("n", (item.value if item else 0) + 1))
+        cache.update(item_doc, "d", apply)
+        self.assertEqual(apply.seen, [1, None])
+        self.assertEqual(self.stored("d").value, 1)
+
+        holder = self.cache()
+        holder.put(item_doc, "h", Item("n", 3))
+        rival.pull_all_backing_stores()
+        rival.delete(item_doc, "h")
+        nothing = Counting(lambda item, calls: Item("n", 1))
+        holder.update(item_doc, "h", nothing)
+        self.assertEqual(nothing.seen, [None])
+
+    def test_a_rival_writing_and_restoring_the_value_still_reruns_apply(self) -> None:
+        cache, rival = self.cache(), self.cache()
+        cache.put(item_doc, "k", Item("n", 1))
+        rival.pull_all_backing_stores()
+
+        def apply(item, calls):
+            if calls == 1:
+                rival.put(item_doc, "k", Item("n", 5))
+                rival.put(item_doc, "k", Item("n", 1))
+            return Item("n", item.value + 1)
+
+        counting = Counting(apply)
+        cache.update(item_doc, "k", counting)
+        self.assertEqual(counting.seen, [1, 1])
+        self.assertEqual(self.stored().value, 2)
+
+    def test_a_stale_caches_first_attempt_reads_the_store(self) -> None:
+        a, b = self.cache(), self.cache()
+        a.put(item_doc, "k", Item("n", 1))
+        b.pull_all_backing_stores()
+        a.put(item_doc, "k", Item("n", 7))
+        apply = Counting(lambda item, calls: Item("n", item.value + 1))
+        b.update(item_doc, "k", apply)
+        self.assertEqual(apply.seen, [7])
+        self.assertEqual(self.stored().value, 8)
+
+
+class UpdateRefusalTests(Fixture):
+    def refused(self, cache: CultCache, key: str = "k", document=item_doc, *errors) -> None:
+        before = self.bytes()
+        apply = Counting()
+        with self.assertRaises(errors or (CultCacheError, SchemaConflictError)):
+            cache.update(document, key, apply)
+        self.assertEqual(apply.calls, 0)
+        self.assertEqual(self.bytes(), before)
+
+    def test_refusals_before_apply(self) -> None:
+        writer = self.cache(item_doc, other_doc)
+        writer.put(other_doc, "k", Item("n", 1))
+        self.refused(writer)  # the record at the key is another document's, held
+        self.refused(self.fresh(item_doc, other_doc))  # unpulled: the store's record is another document's
+        self.refused(self.cache(item_doc))  # foreign: the store holds a type this cache never registered
+        self.refused(self.fresh(item_doc))
+        storeless = self.cache(item_doc, other_doc, store=False)
+        storeless.put(other_doc, "k", Item("n", 1))
+        apply = Counting()
+        with self.assertRaises((CultCacheError, SchemaConflictError)):
+            storeless.update(item_doc, "k", apply)
+        self.assertEqual(apply.calls, 0)
+
+    def fresh(self, *documents) -> CultCache:
+        """A cache that has not pulled: whatever it reads, it reads from the store at update."""
+        builder = CultCache.builder()
+        for document in documents:
+            builder.register_document_type(document)
+        return builder.add_generic_store(SingleFileMessagePackBackingStore(self.path)).build()
+
+    def test_a_home_without_read_and_commit_if_refuses(self) -> None:
+        jsonl = JsonLinesBackingStore(self.dir / "state.jsonl")
+        cache = CultCache.builder().register_document_type(item_doc).add_generic_store(jsonl).build()
+        cache.pull_all_backing_stores()
+        cache.put(item_doc, "k", Item("n", 1))
+        before = (self.dir / "state.jsonl").read_bytes()
+        apply = Counting()
+        with self.assertRaisesRegex(CultCacheError, "cannot write conditionally"):
+            cache.update(item_doc, "k", apply)
+        self.assertEqual(apply.calls, 0)
+        self.assertEqual((self.dir / "state.jsonl").read_bytes(), before)
+        self.assertFalse(hasattr(JsonLinesBackingStore, "read") or hasattr(JsonLinesBackingStore, "commit_if"))
+
+        class Bare:
+            def pull_all(self):
+                return []
+
+            def push(self, envelope):
+                raise AssertionError("update wrote through a store that cannot write conditionally")
+
+            def push_all(self, envelopes, soft=False):
+                raise AssertionError("push_all")
+
+            def delete(self, type, key):
+                raise AssertionError("delete")
+
+        bare = CultCache.builder().register_document_type(item_doc).add_generic_store(Bare()).build()
+        with self.assertRaisesRegex(CultCacheError, "Bare"):
+            bare.update(item_doc, "k", apply)
+        self.assertEqual(apply.calls, 0)
+
+    def test_read_only_records_refuse_before_apply(self) -> None:
+        wide = define_database_entry_type(
+            "tests.rw",
+            [database_entry_field("name", 0, codec=t.string), database_entry_field("extra", 1, codec=t.string)],
+            cls=Wide,
+            schema_id="tests.rw.v1",
+        )
+        narrow = define_database_entry_type("tests.rw", [database_entry_field("name", 0, codec=t.string)], cls=Narrow, schema_id="tests.rw.v1")
+        self.cache(wide).put(wide, "k", Wide("n", "x"))
+        before = self.bytes()
+        for pulled in (True, False):  # held read-only, then a store record the cache never loaded
+            with self.subTest(pulled=pulled):
+                cache = self.cache(narrow) if pulled else self.fresh(narrow)
+                apply = Counting()
+                with self.assertRaises(SchemaConflictError) as caught:
+                    cache.update(narrow, "k", apply)
+                self.assertEqual(caught.exception.record_key, "k")
+                self.assertEqual(apply.calls, 0)
+                self.assertEqual(self.bytes(), before)
+
+    def test_a_non_global_key_for_a_global_document_refuses(self) -> None:
+        cache = self.cache(global_doc)
+        apply = Counting()
+        with self.assertRaises(CultCacheError):
+            cache.update(global_doc, "not-global", apply)
+        self.assertEqual(apply.calls, 0)
+
+    def test_refusals_after_apply_are_not_retried(self) -> None:
+        cache = self.cache()
+        cache.put(item_doc, "k", Item("n", 1))
+        before = self.bytes()
+        boom = Counting(lambda item, calls: (_ for _ in ()).throw(RuntimeError("boom")))
+        with self.assertRaisesRegex(RuntimeError, "boom"):
+            cache.update(item_doc, "k", boom)
+        self.assertEqual(boom.calls, 1)
+        self.assertEqual(self.bytes(), before)
+
+        indexed = self.cache(sku_doc, path=self.dir / "sku.cc")
+        indexed.put(sku_doc, "a", Item("n", 1, "X"))
+        indexed.put(sku_doc, "b", Item("n", 1, "Y"))
+        before = (self.dir / "sku.cc").read_bytes()
+        clash = Counting(lambda item, calls: Item("n", 2, "X"))
+        with self.assertRaisesRegex(CultCacheError, "unique index"):
+            indexed.update(sku_doc, "b", clash)
+        self.assertEqual(clash.calls, 1)
+        self.assertEqual((self.dir / "sku.cc").read_bytes(), before)
+
+
+class UpdateOwnWritesTests(Fixture):
+    def test_own_writes_and_the_held_instance(self) -> None:
+        cache = self.cache()
+        cache.put(item_doc, "k", Item("n", 1))
+        apply = Counting(lambda item, calls: Item("n", item.value + 1))
+        cache.update(item_doc, "k", apply)
+        second = cache.update(item_doc, "k", apply)
+        self.assertEqual(apply.calls, 2)
+        self.assertIs(second.document, cache.get(item_doc, "k"))
+        self.assertEqual(second.document.value, 3)
+
+        def sideways(item, calls):
+            cache.put(item_doc, "side", Item("n", 5))  # apply runs outside the cache's hold
+            return Item("n", item.value + 1)
+
+        sideways_apply = Counting(sideways)
+        result = cache.update(item_doc, "k", sideways_apply)
+        self.assertEqual((result.outcome, sideways_apply.calls), (UpdateOutcome.UPDATED, 1))
+        self.assertEqual(self.stored("side").value, 5)
+
+    def test_store_less_cache(self) -> None:
+        cache = self.cache(store=False)
+        cache.put(item_doc, "k", Item("n", 0))
+        results: list[UpdateResult] = []
+
+        def bump(item):
+            time.sleep(0)
+            return Item("n", item.value + 1)
+
+        def run() -> None:
+            for _ in range(50):
+                results.append(cache.update(item_doc, "k", bump, max_attempts=500))
+
+        threads = [threading.Thread(target=run) for _ in range(2)]
+        for thread in threads:
+            thread.start()
+        for thread in threads:
+            thread.join()
+        self.assertTrue(all(result.outcome is UpdateOutcome.UPDATED for result in results))
+        self.assertEqual(cache.get(item_doc, "k").value, 100)
+
+        # A store that becomes the type's home while apply runs holds a record the cache never read: the commit loses.
+        cache = self.cache(store=False)  # holds no record of the type, so a store may become its home
+        other = self.cache(path=self.dir / "late.cc")
+        other.put(item_doc, "fresh", Item("n", 40))
+        late = SingleFileMessagePackBackingStore(self.dir / "late.cc")
+
+        def attach(item, calls):
+            if calls == 1:
+                cache.add_generic_store(late)
+            return Item("n", (item.value if item else 0) + 1)
+
+        apply = Counting(attach)
+        result = cache.update(item_doc, "fresh", apply)
+        self.assertEqual(apply.seen, [None, 40])
+        self.assertIs(result.outcome, UpdateOutcome.UPDATED)
+        self.assertEqual(self.cache(path=self.dir / "late.cc").get(item_doc, "fresh").value, 41)
+
+    def test_a_home_attached_while_apply_runs_is_reread_or_refused(self) -> None:
+        """The cache's home for the type changes between the read and the commit (attaching is allowed while it holds no record
+        of the type): the version read from the old home is no condition on the new one."""
+        cache = self.cache(path=self.dir / "generic.cc")
+        late = SingleFileMessagePackBackingStore(self.dir / "late.cc")
+
+        def attach(item, calls):
+            if calls == 1:
+                cache.add_backing_store(late, ["item"])
+            return Item("n", 1)
+
+        apply = Counting(attach)
+        self.assertIs(cache.update(item_doc, "k", apply).outcome, UpdateOutcome.UPDATED)
+        self.assertEqual(apply.seen, [None, None])  # reread from the new home, not committed with the old home's version
+        self.assertEqual(self.cache(path=self.dir / "late.cc").get(item_doc, "k").value, 1)
+
+        cache = self.cache(path=self.dir / "generic2.cc")
+        jsonl = JsonLinesBackingStore(self.dir / "late.jsonl")
+
+        def attach_jsonl(item, calls):
+            cache.add_backing_store(jsonl, ["item"])
+            return Item("n", 1)
+
+        apply = Counting(attach_jsonl)
+        with self.assertRaisesRegex(CultCacheError, "cannot write conditionally"):
+            cache.update(item_doc, "k", apply)
+        self.assertEqual(apply.calls, 1)
+        self.assertFalse((self.dir / "late.jsonl").exists())
+
+    def test_ledger_under_contention(self) -> None:
+        caches = [self.cache() for _ in range(3)]
+        outcomes: list[UpdateOutcome] = []
+
+        def run(cache) -> None:
+            for _ in range(25):
+                outcomes.append(cache.update(item_doc, "k", lambda item: Item("n", (item.value if item else 0) + 1)).outcome)
+
+        threads = [threading.Thread(target=run, args=(cache,)) for cache in caches]
+        for thread in threads:
+            thread.start()
+        for thread in threads:
+            thread.join()
+        self.assertEqual(self.stored().value, outcomes.count(UpdateOutcome.UPDATED))
+        self.assertEqual(len(outcomes), 75)
+
+
+class UpdateLegacyGlobalTests(Fixture):
+    def plant(self, key: str = "legacy") -> None:
+        entry = global_doc.catalog_entry()
+        SingleFileMessagePackBackingStore(self.path).push(
+            CultCacheEnvelope.create(key=key, type=global_doc.type, payload=global_doc.encode_payload(Item("n", 1)), schema_id=entry.schema_id, catalog_entry=entry)
+        )
+
+    def test_update_global_updates_the_global_record(self) -> None:
+        cache = self.cache(global_doc)
+        result = cache.update_global(global_doc, lambda item: Item("n", (item.value if item else 0) + 1))
+        self.assertIs(result.outcome, UpdateOutcome.UPDATED)
+        self.assertEqual(self.cache(global_doc).get(global_doc, GLOBAL_KEY).value, 1)
+
+    def test_a_legacy_global_is_refused_until_put_global_retires_it(self) -> None:
+        self.plant()
+        cache = self.cache(global_doc)  # loaded from the legacy key
+        before = self.bytes()
+        for call in (lambda a: cache.update_global(global_doc, a), lambda a: cache.update(global_doc, GLOBAL_KEY, a)):
+            apply = Counting()
+            with self.assertRaisesRegex(CultCacheError, "put_global"):
+                call(apply)
+            self.assertEqual(apply.calls, 0)
+        self.assertEqual(self.bytes(), before)
+        # Judged from the cache's load, not only from the store: the legacy record is gone from the store, the cache still holds it.
+        SingleFileMessagePackBackingStore(self.path).delete(global_doc.type, "legacy")
+        apply = Counting()
+        with self.assertRaisesRegex(CultCacheError, "put_global"):
+            cache.update_global(global_doc, apply)
+        self.assertEqual(apply.calls, 0)
+        cache.put_global(global_doc, Item("n", 5))
+        self.assertEqual(cache.update_global(global_doc, lambda item: Item("n", item.value + 1)).document.value, 6)
+
+    def test_a_pull_inside_apply_that_loads_a_legacy_global_fails_the_commit(self) -> None:
+        cache = self.cache(global_doc)
+
+        def apply(item):
+            self.plant()
+            cache.pull_all_backing_stores()
+            return Item("n", 9)
+
+        with self.assertRaisesRegex(CultCacheError, "put_global"):
+            cache.update_global(global_doc, apply)
+        self.assertEqual([envelope.key for envelope in SingleFileMessagePackBackingStore(self.path).pull_all()], ["legacy"])
+
+    def test_an_unpulled_cache_judges_the_legacy_global_from_the_store(self) -> None:
+        self.plant()
+        before = self.bytes()
+        cache = SingleFileMessagePackBackingStore(self.path)
+        unpulled = CultCache.builder().register_document_type(global_doc).add_generic_store(cache).build()
+        apply = Counting()
+        with self.assertRaisesRegex(CultCacheError, "put_global"):
+            unpulled.update_global(global_doc, apply)
+        self.assertEqual(apply.calls, 0)
+        self.assertEqual(self.bytes(), before)
+
+
+class StoreRestampTests(Fixture):
+    def envelope(self, value: int, stored_at: str = ALPHA, key: str = "k") -> CultCacheEnvelope:
+        entry = item_doc.catalog_entry()
+        return CultCacheEnvelope(key, "item", item_doc.encode_payload(Item("n", value)), stored_at, entry.schema_id, entry)
+
+    def stamp_on_disk(self, store, key: str = "k") -> str:
+        return next(envelope.stored_at for envelope in store.pull_all() if envelope.key == key)
+
+    def later_than_alpha(self, text: str) -> bool:
+        return _parse_stored_at(text) > _parse_stored_at(ALPHA)
+
+    def test_commit_if_and_the_restamp_at_the_store(self) -> None:
+        store = SingleFileMessagePackBackingStore(self.path)
+        self.assertIsNone(store.read("k"))
+        stamp = store.commit_if(self.envelope(1), None)  # None against an absent key commits
+        self.assertEqual(stamp, ALPHA)
+        envelope, version = store.read("k")
+        self.assertEqual(version, record_version(envelope))
+        before = self.bytes()
+        self.assertIsNone(store.commit_if(self.envelope(2, "2100-01-01T00:00:00Z"), None))  # a record is there: not absent
+        self.assertIsNone(store.commit_if(self.envelope(2, "2100-01-01T00:00:00Z"), "0" * 64))
+        self.assertEqual(self.bytes(), before)
+        written = store.commit_if(self.envelope(2), version)  # a staged stamp that is no later is minted over
+        self.assertTrue(self.later_than_alpha(written))
+        self.assertEqual(self.stamp_on_disk(store), written)
+        self.assertEqual(stamp_of(store, "k"), written)
+
+    def test_every_replacing_write_is_stored_strictly_later(self) -> None:
+        for make in (SingleFileMessagePackBackingStore, JsonLinesBackingStore):
+            for name in ("push", "push_all", "commit_if"):
+                if name == "commit_if" and make is JsonLinesBackingStore:
+                    continue
+                with self.subTest(store=make.__name__, write=name):
+                    path = self.dir / f"{make.__name__}-{name}"
+                    store = make(path)
+                    store.push(self.envelope(1))
+                    if name == "push":
+                        reported = store.push(self.envelope(2))
+                    elif name == "push_all":
+                        reported = store.push_all([self.envelope(2)])["k"]
+                    else:
+                        reported = store.commit_if(self.envelope(2), store.read("k")[1])
+                    self.assertTrue(self.later_than_alpha(reported))
+                    self.assertEqual(self.stamp_on_disk(store), reported)
+
+    def test_the_cache_holds_the_stamp_the_store_wrote(self) -> None:
+        for name in ("put", "put_envelope", "put_envelopes", "update"):
+            with self.subTest(write=name):
+                path = self.dir / name
+                SingleFileMessagePackBackingStore(path).push(self.envelope(1))
+                cache = self.cache(path=path)
+                self.assertEqual(cache.get_envelope(item_doc, "k").stored_at, ALPHA)
+                if name == "put":
+                    cache.put(item_doc, "k", Item("n", 2))
+                elif name == "put_envelope":
+                    cache.put_envelope(item_doc, self.envelope(2))
+                elif name == "put_envelopes":
+                    cache.put_envelopes(item_doc, [self.envelope(2)])
+                else:
+                    cache.update(item_doc, "k", lambda item: Item("n", 2))
+                held = cache.get_envelope(item_doc, "k").stored_at
+                self.assertTrue(self.later_than_alpha(held))
+                self.assertEqual(held, self.stamp_on_disk(SingleFileMessagePackBackingStore(path)))
+
+    def test_a_store_less_cache_stamps_a_replacing_write_later(self) -> None:
+        cache = self.cache(store=False)
+        cache.put_envelope(item_doc, self.envelope(1))
+        cache.put_envelope(item_doc, self.envelope(2, "2000-01-01T00:00:00Z"))
+        self.assertTrue(self.later_than_alpha(cache.get_envelope(item_doc, "k").stored_at))
+
+    def test_only_the_stamp_shape_the_runtimes_write_parses(self) -> None:
+        """An RFC 3339 date-time with T, a fraction of 1 to 9 digits and Z or a numeric offset, on every interpreter (the shape is
+        checked here, not by fromisoformat), so a stamp reads the same on 3.10 and 3.13. The edges follow RFC 3339: offset hours 00-23,
+        minutes 00-59."""
+        utc = timezone.utc
+        accepted = {
+            "2026-01-01T00:00:00Z": datetime(2026, 1, 1, tzinfo=utc),
+            "2026-01-01T00:00:00.1Z": datetime(2026, 1, 1, 0, 0, 0, 100000, utc),
+            "2026-01-01T00:00:00.123456789Z": datetime(2026, 1, 1, 0, 0, 0, 123456, utc),
+            "2026-01-01T00:00:00.1234567+00:00": datetime(2026, 1, 1, 0, 0, 0, 123456, utc),
+            "2026-01-01T05:30:00+05:30": datetime(2026, 1, 1, tzinfo=utc),
+            "2025-12-31T19:00:00-05:00": datetime(2026, 1, 1, tzinfo=utc),
+            "2026-01-01T23:59:00+23:59": datetime(2026, 1, 1, 0, 0, tzinfo=utc),
+        }
+        for text, instant in accepted.items():
+            with self.subTest(text=text):
+                self.assertEqual(_parse_stored_at(text), instant)
+        for text in (
+            "2026-01-01", "20260101T000000Z", "2026-W01-1T00:00:00Z", "2026-01-01 00:00:00Z", "2026-01-01T00:00:00,5Z",
+            "2026-01-01T00Z", "2026-01-01T00:00:00", "2026-01-01T00:00:00.Z", "2026-01-01T00:00:00.1234567890Z",
+            "2026-01-01T00:00:00z", " 2026-01-01T00:00:00Z", "2026-01-01T00:00:00Z ", "2026-01-01T24:00:00Z",
+            "2026-13-01T00:00:00Z", "2026-01-01T00:00:00+24:00", "2026-01-01T00:00:00+05:60", "2026-01-01T00:00:00+05:75", "2026-01-01T00:00Z", "2026-01-01T00:00+00:00",
+            "026-01-01T00:00:00Z", "12026-01-01T00:00:00Z", "2026-01-01T00:00:00+5:00", "2026-01-01T00:00:00+0000", "\u0662026-01-01T00:00:00Z",
+            "", "garbage",
+        ):
+            with self.subTest(text=text):
+                self.assertIsNone(_parse_stored_at(text))
+
+    def test_a_stamp_outside_the_range_is_unparseable_and_one_at_the_edge_refuses_a_later_stamp(self) -> None:
+        for text in ("9999-12-31T23:30:00-05:00", "0001-01-01T00:00:00+05:00"):
+            with self.subTest(text=text):
+                self.assertIsNone(_parse_stored_at(text))
+        now = datetime(2030, 1, 1, tzinfo=timezone.utc)
+        self.assertEqual(mint_stored_at("9999-12-31T23:30:00-05:00", now), "2030-01-01T00:00:00.000000+00:00")
+        edge = "9999-12-31T23:59:59.9999999Z"
+        with self.assertRaisesRegex(ValueError, "last instant"):
+            mint_stored_at(edge, now)
+        with self.assertRaisesRegex(ValueError, "last instant"):
+            stored_at_over("2000-01-01T00:00:00Z", edge)
+        # A write over a record stamped at the edge is refused by the store and writes nothing, from push, commit_if and update.
+        store = SingleFileMessagePackBackingStore(self.path)
+        entry = item_doc.catalog_entry()
+        record = lambda value, at: CultCacheEnvelope("k", "item", item_doc.encode_payload(Item("n", value)), at, entry.schema_id, entry)  # noqa: E731
+        store.push(record(1, edge))
+        before = self.bytes()
+        version = store.read("k")[1]
+        for write in (lambda: store.push(record(2, ALPHA)), lambda: store.push_all([record(2, ALPHA)]), lambda: store.commit_if(record(2, ALPHA), version)):
+            with self.assertRaises(ValueError):
+                write()
+        cache = self.cache()
+        apply = Counting()
+        with self.assertRaises(ValueError):
+            cache.update(item_doc, "k", apply)
+        self.assertEqual(apply.calls, 1)
+        self.assertEqual(self.bytes(), before)
+        self.assertEqual(cache.get_envelope(item_doc, "k").stored_at, edge)
+
+    def test_stamp_parsing_and_install(self) -> None:
+        for text in ("2026-01-01T00:00:00.1234567+00:00", "2026-01-01T00:00:00Z", "2026-01-01T00:00:00.123456+00:00"):
+            with self.subTest(text=text):
+                self.assertIsNotNone(_parse_stored_at(text))
+                self.assertEqual(_parse_stored_at(text).tzinfo, timezone.utc)
+        self.assertIsNone(_parse_stored_at("garbage"))
+        self.assertIsNone(_parse_stored_at(None))
+        now = datetime(2030, 1, 1, tzinfo=timezone.utc)
+        self.assertEqual(mint_stored_at("garbage", now), "2030-01-01T00:00:00.000000+00:00")
+        self.assertEqual(stored_at_over("2000-01-01T00:00:00Z", "2000-01-01T00:00:00Z")[:4], str(datetime.now(timezone.utc).year))  # no later: minted at now
+        self.assertEqual(mint_stored_at("2030-01-01T00:00:00Z", now), "2030-01-01T00:00:00.000001+00:00")  # equal is not later
+        self.assertEqual(mint_stored_at("2099-01-01T00:00:00.1234567Z", now), "2099-01-01T00:00:00.123457+00:00")
+        self.assertEqual(stored_at_over("2000-01-01T00:00:00Z", "2099-01-01T00:00:00.1234567Z"), "2099-01-01T00:00:00.123457+00:00")
+
+        for reported in (True, 1, "garbage"):
+            with self.subTest(reported=reported):
+                class Lying(SingleFileMessagePackBackingStore):
+                    def push(self, envelope):
+                        super().push(envelope)
+                        return reported
+
+                cache = CultCache.builder().register_document_type(item_doc).add_generic_store(Lying(self.dir / f"lying-{reported}")).build()
+                cache.pull_all_backing_stores()
+                cache.put_envelope(item_doc, self.envelope(1, "2026-02-02T00:00:00.000000+00:00"))
+                self.assertEqual(cache.get_envelope(item_doc, "k").stored_at, "2026-02-02T00:00:00.000000+00:00")
+
+
+def stamp_of(store, key: str) -> str:
+    return store.read(key)[0].stored_at
+
+
+if __name__ == "__main__":
+    unittest.main()

packages/cultmesh-py/src/cultmesh_py/node.py

@@ -570,12 +570,14 @@ class CultMeshDatabase:
             catalog_entry=catalog_entry,
         )
         value = self.cache.put_envelope(registered, envelope)
+        # The stamp the write was stored at, which a replacing write mints later than the record it replaces: the cache holds it.
+        stored_at = self.cache.get_required_envelope(registered, key).stored_at
         self._publish_local_change(registered, key, "added" if previous is None else "updated", value, previous)
         message = document_put_raw(
             message_id=message_id,
             key=key,
             schema_id=envelope.schema_id or catalog_entry.schema_id,
-            stored_at=envelope.stored_at,
+            stored_at=stored_at,
             payload=envelope.payload,
             source_runtime_id=self.runtime_id,
             shard_id=shard_id,
@@ -683,10 +685,11 @@ class CultMeshDatabase:
         value = self.cache.put_envelope(document, envelope)
         change_kind = "added" if previous is None else "updated"
         change = self._publish_local_change(document, record_key, change_kind, value, previous)
-        self._append_shard_log_put(CultNetMessage(
-            "cultnet.document_put_raw.v0",
-            {key: value for key, value in message.items() if key != "schemaVersion"},
-        ), change_kind)
+        # The log carries the stamp the cache holds, which a replacing put mints later than the record it replaces, not the peer's.
+        stored_at = self.cache.get_required_envelope(document, record_key).stored_at
+        logged = {key: value for key, value in message.items() if key != "schemaVersion"}
+        logged["document"] = {**message["document"], "storedAt": stored_at}
+        self._append_shard_log_put(CultNetMessage("cultnet.document_put_raw.v0", logged), change_kind)
         return change
 
     def raw_put_served_alone(self, message: dict[str, Any], *, message_id: str) -> dict[str, Any] | None:

packages/cultmesh-py/src/cultmesh_py/server.py

@@ -686,9 +686,11 @@ class CultMeshLocalServer:
                 message_id=str(message.get("messageId") or ""),
                 code="unregistered_document_put",
             )]
+        # Subscribers are sent the stamp the store holds, which a replacing put mints later than the record it replaces, not the peer's.
+        stored = self.node.database.cache.get_required_envelope(change.document, change.record_key).stored_at
         return self._database_change_notifications(
             message,
-            document_record,
+            {**document_record, "storedAt": stored},
             change.change_kind,
             subscriptions,
             resolved_schema_id=change.schema_id,

packages/cultmesh-py/tests/test_cultmesh.py

@@ -19,6 +19,7 @@ from typing import Any, Callable
 import msgpack
 from cultcache_py import (
     CultCache,
+    CultCacheEnvelope,
     SingleFileMessagePackBackingStore,
     define_database_entry_type,
     define_document_type,
@@ -796,6 +797,51 @@ class CultMeshTests(unittest.TestCase):
         self.assertEqual(node.get_required(document, "note:second").body, "second catalog publication")
         self.assertIsInstance(node.get_required(alias, "note:second"), UiNote)
 
+    def test_raw_put_publishes_and_logs_the_stamp_the_store_wrote(self) -> None:
+        document = define_database_entry_type("mesh.stamp_item", [("name", 0), ("category", 1), ("value", 2, 0)], cls=Item)
+        node = create_node(runtime_id="python-node")
+        node.register_document(document)
+        cache = node.database.cache
+        entry = document.catalog_entry()
+        future = "2099-01-01T00:00:00.000000+00:00"
+        cache.put_envelope(
+            document,
+            CultCacheEnvelope(
+                key="item:stamp",
+                type=document.type,
+                payload=document.encode_payload(Item("old", "gear", 1)),
+                stored_at=future,
+                schema_id=entry.schema_id,
+                catalog_entry=entry,
+            ),
+        )
+        # A replacing write is stored later than the record it replaces, so the stamp is not the one the write was created with.
+        wire = node.put_raw_message(document, "item:stamp", Item("new", "gear", 2), shard_id="mesh", shard_epoch=1).to_wire()
+        held = cache.get_required_envelope(document, "item:stamp").stored_at
+        self.assertGreater(held, future)
+        self.assertEqual(wire["document"]["storedAt"], held)
+        self.assertEqual(node.database._shard_logs["mesh"][-1]["put"]["document"]["storedAt"], held)
+
+    def test_applied_raw_put_logs_the_stamp_the_cache_holds(self) -> None:
+        document = define_database_entry_type("mesh.applied_stamp", [("name", 0), ("category", 1), ("value", 2, 0)], cls=Item)
+        node = create_node(runtime_id="python-node")
+        node.register_document(document)
+        cache = node.database.cache
+        entry = document.catalog_entry()
+
+        def envelope(value: int, stored_at: str) -> CultCacheEnvelope:
+            return CultCacheEnvelope("item:applied", document.type, document.encode_payload(Item("n", "gear", value)), stored_at, entry.schema_id, entry)
+
+        cache.put_envelope(document, envelope(1, "2099-01-01T00:00:00.000000+00:00"))
+        peer = node.put_raw_message(document, "item:peer", Item("peer", "gear", 2)).to_wire()  # a well-formed put to replay
+        peer["document"] = {**peer["document"], "recordKey": "item:applied", "storedAt": "2000-01-01T00:00:00Z",
+                            "payload": envelope(2, "x").payload}
+        peer["shardId"], peer["shardEpoch"] = "mesh", 1
+        node.database.apply_raw_put_message(peer)
+        held = cache.get_required_envelope(document, "item:applied").stored_at
+        self.assertGreater(held, "2099-01-01T00:00:00.000000+00:00")
+        self.assertEqual(node.database._shard_logs["mesh"][-1]["put"]["document"]["storedAt"], held)
+
     def test_cultmesh_node_emits_raw_put_and_delete_messages_for_local_writes(self) -> None:
         document = define_database_entry_type(
             "mesh.emit_item",

packages/cultmesh-py/tests/test_remote_write_authority.py

@@ -8,6 +8,7 @@ the peer the connection loop would pass, and one test drives the same path over
 from __future__ import annotations
 
 import json
+import tempfile
 import unittest
 from pathlib import Path
 from typing import Any
@@ -23,7 +24,7 @@ from cultmesh_py import (
     CultMeshServerPeer,
     create_node,
 )
-from cultmesh_py.server import CultMeshLocalServer
+from cultmesh_py.server import CultMeshLocalServer, _DatabaseSubscription
 from cultnet_py import CultNetPeerError, create_rudp_schema_transport, document_delete, document_put_raw, error_message
 
 SCHEMA_ID = "sha256:authority-note"
@@ -55,12 +56,12 @@ class Recording:
         return self._decide(request)
 
 
-def put_of(key: str, text: str, *, schema_id: str = SCHEMA_ID) -> dict[str, Any]:
+def put_of(key: str, text: str, *, schema_id: str = SCHEMA_ID, stored_at: str = "2026-11-07T00:00:00Z") -> dict[str, Any]:
     return document_put_raw(
         message_id="wire",
         key=key,
         schema_id=schema_id,
-        stored_at="2026-11-07T00:00:00Z",
+        stored_at=stored_at,
         payload=NOTE.encode_payload({"schema_version": "tests.authority_note.v1", "text": text}),
         shard_id=SHARD_ID,
         shard_epoch=1,
@@ -73,6 +74,26 @@ def delete_of(key: str, *, schema_id: str = SCHEMA_ID) -> dict[str, Any]:
     ).to_wire()
 
 
+class NotificationStampTests(unittest.TestCase):
+    """A subscriber is sent the stamp the store holds after a peer's put, as the shard log is, not the peer's stamp."""
+
+    def test_the_notification_carries_the_stamp_the_cache_holds(self) -> None:
+        local, peer_stamp = "2099-01-01T00:00:00.000000+00:00", "2000-01-01T00:00:00Z"
+        with tempfile.TemporaryDirectory() as directory:
+            for backing in (None, Path(directory) / "node.cc"):  # store-less, then store-backed
+                with self.subTest(store=backing is not None):
+                    node = create_node(backing, runtime_id="stamp-server")
+                    node.database.register_document(NOTE)
+                    server = CultMeshLocalServer(node=node, port=0, write_authority=CultMeshGrantAllWrites())
+                    node.database.apply_raw_put_message(put_of(KEY_A, "server", stored_at=local))
+                    subscriptions = {"s": _DatabaseSubscription("s", set(), set())}
+                    replies = server._handle_connection_message(put_of(KEY_A, "peer", stored_at=peer_stamp), subscriptions, peer=PEER)
+                    held = node.cache.get_required_envelope(NOTE, KEY_A).stored_at
+                    self.assertGreater(held, local)
+                    self.assertEqual([reply["document"]["storedAt"] for reply in replies], [held])
+                    self.assertEqual(node.database._shard_logs[SHARD_ID][-1]["put"]["document"]["storedAt"], held)
+
+
 class RemoteWriteAuthorityTests(unittest.TestCase):
     def build(self, authority: Any = None) -> CultMeshLocalServer:
         node = create_node(runtime_id="authority-server")