cultcache-py: the cache holds one record per key and carries a store's unregistered records as foreign

One key-identity judge in _write, an unheld-removal refusal at foreign keys, a load that carries unregistered records
instead of refusing, and a load that refuses one key in two stores. The single-file store's commit guard stays the disk judge.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

docs/write-ownership-map.md

@@ -33,8 +33,8 @@ Probes: scratch tests in this worktree, removed. C# filtered runs on Starfire (o
 
 | Probe | C# | TypeScript | Rust | Python |
 |---|---|---|---|---|
-| P1 cache {A,B}: B at k, then A at k | **C**: B's record replaced, k now A's id | C, and **both** records live at k | C, both live at k | C, both live at k |
-| P2 cache {A} only, store holds B at k, write A at k | R `CultSchemaConflictException` (B is foreign, carried) | R at load: "No schema is registered for persisted schema id" | R at load, same | R at load: "Unknown persisted document type" |
+| P1 cache {A,B}: B at k, then A at k | **C**: B's record replaced, k now A's id | C, and **both** records live at k | C, both live at k | R `SchemaConflictError`, A's write refused, B kept (was C, both live) |
+| P2 cache {A} only, store holds B at k, write A at k | R `CultSchemaConflictException` (B is foreign, carried) | R at load: "No schema is registered for persisted schema id" | R at load, same | R `SchemaConflictError` (B is foreign, carried; the load admits it; was R at load) |
 | P3 cacheless `WriteDocument<A>` over B, registry null / {A} / {A,B} | R / R / R | no cacheless write | no cacheless write | no cacheless write |
 | P4 cache {A,B}: remove k and upsert A at k in one batch | C (k is A's) | n/a | n/a | n/a |
 | P5 cache {A,B}: remove, then add in a second commit | C | n/a | n/a | n/a |

packages/cultcache-py/CHANGELOG.md

@@ -9,6 +9,9 @@
 
 ### Breaking
 
+- 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.
+- 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.
+
 - A single-file store holding two records under one key is unreadable: the v1 reader and the legacy envelope-array reader refuse it with `StoreUnreadableError` (a key is compared by its exact code points, no case folding or normalisation). A catalog slot whose canonical schema is not a string is refused the same way. A store that loaded before can now be refused: repair it with the runtime that wrote the duplicate.
 - A `put` or `push_all` that would land a second type under a key the store holds is refused with `SchemaConflictError`, and the file is left as it was. Retype a key by deleting it, then putting the new type. A write replaces the record at a key only when the written type owns it: its catalog entry declares the stored schema id (its own, or one it lists compatible), so a renamed type that lists the old id compatible may replace the record, or it has the stored record's schema name. This runtime does not compare shapes, so a same-name type is never refused for dropping persisted slots as C# refuses it.
 - Writes of the same type replace by key: the last record wins, in a batch as in the file.

packages/cultcache-py/README.md

@@ -19,7 +19,11 @@ discovery, sessions, local server) build on it, mirroring `cultcache-rs` /
   `[format, schemaCatalog, records]`
 - records store raw MessagePack payload bytes under schema-catalog identity
 - payloads decode only through the registered document definition for that type
-- unknown persisted types fail closed
+- a cache holds one record per key whatever its type: a write of a second type at a held key is refused with
+  `SchemaConflictError` (in memory too), and two stores holding one key refuse the load
+- a store record no registered type claims loads and is carried as foreign: `cache.foreign_records()` lists it, the store
+  keeps its bytes through every write, a write or removal at its key is refused, and registering a type that declares its
+  schema id claims it at the next load
 - global documents are singleton-style per type
 - one home store per document type: a store added with `add_backing_store`
   owns its types, otherwise the generic store is the home; a type claimed by a

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
+from .cache import CultCache, CultCacheBuilder, ForeignRecord
 from .documents import (
     MAX_PAYLOAD_DEPTH,
     MAX_SLOT_KEY,
@@ -29,6 +29,7 @@ __all__ = [
     "CultCacheSchemaCatalogMember",
     "DatabaseEntryField",
     "DocumentDefinition",
+    "ForeignRecord",
     "JsonLinesBackingStore",
     "MAX_PAYLOAD_DEPTH",
     "MAX_SLOT_KEY",

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

@@ -9,7 +9,7 @@ from typing import Any, Callable, Generic, Iterable, TypeVar
 
 from .backing_store import BackingStore, CultCacheEnvelope
 from .documents import DocumentDefinition, extract_value
-from .stores import SchemaConflictError
+from .stores import SchemaConflictError, retyped, written_owns
 
 T = TypeVar("T")
 F = TypeVar("F", bound=Callable[..., Any])
@@ -120,6 +120,20 @@ class _State:
     # of that global removes it from the home store. Remove once no store holds a global under a key
     # other than GLOBAL_KEY.
     legacy_global_keys: dict[str, str] = field(default_factory=dict)
+    # 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)
+
+
+@dataclass(frozen=True)
+class ForeignRecord:
+    """A record a store holds that no registered type claims: the cache does not serve it, refuses a write or a removal at its
+    key, and a store keeps its bytes. Registering a type that declares its schema id claims it at the next load."""
+
+    key: str
+    schema_id: str
+    schema_name: str
+    stored_at: str
 
 
 @dataclass(frozen=True)
@@ -322,15 +336,23 @@ class CultCache:
         loaded_legacy_global_keys: dict[str, str] = {}
         routes, generic = self._state.stores_by_type, self._state.generic_store
         seen_globals: set[str] = set()
+        foreign: list[ForeignRecord] = []
+        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]:
             if candidate is not None and all(candidate is not store for store in stores):
                 stores.append(candidate)
-        for store in stores:
+        for position, store in enumerate(stores):
             for envelope in store.pull_all():
                 document = self._resolve_document_for_envelope(envelope)
                 if document is None:
-                    raise CultCacheError(f"Unknown persisted document type: {envelope.type}")
+                    entry = envelope.catalog_entry
+                    held_name = entry.schema_name if entry is not None else envelope.type
+                    self._claim_key(type_at_key, envelope.key, held_name, position)
+                    foreign.append(
+                        ForeignRecord(envelope.key, envelope.schema_id or envelope.type, held_name, envelope.stored_at)
+                    )
+                    continue
                 if envelope.type != document.type:
                     envelope = CultCacheEnvelope(
                         key=envelope.key,
@@ -353,6 +375,7 @@ class CultCache:
                     if envelope.key != GLOBAL_KEY:
                         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)
                 loaded_values.setdefault(envelope.type, {})[envelope.key] = value
@@ -369,7 +392,28 @@ class CultCache:
             self._state.indexes,
             self._state.lookup_keys,
             self._state.legacy_global_keys,
-        ) = (loaded_values, loaded_envelopes, loaded_names, loaded_indexes, loaded_lookup_keys, loaded_legacy_global_keys)
+            self._state.foreign,
+        ) = (
+            loaded_values,
+            loaded_envelopes,
+            loaded_names,
+            loaded_indexes,
+            loaded_lookup_keys,
+            loaded_legacy_global_keys,
+            sorted(foreign, key=lambda record: _utf16_order(record.key)),
+        )
+
+    @staticmethod
+    def _claim_key(type_at_key: dict[str, tuple[str, int]], key: str, type: str, store: int) -> None:
+        """A key holds one record whatever its type, across every store a load reads. The refusal names the key and both types,
+        never a payload."""
+        held, held_store = type_at_key.setdefault(key, (type, store))
+        if held != type or held_store != store:
+            raise CultCacheError(f"Record {key!r} is held by both {held} and {type}: a key holds one record; the load admitted nothing.")
+
+    def foreign_records(self) -> list[ForeignRecord]:
+        """The records the last load read that no registered type claims, sorted by key (UTF-16 code-unit order)."""
+        return list(self._state.foreign)
 
     def get(self, document: DocumentDefinition[T], key: str) -> T | None:
         self._assert_registered(document)
@@ -497,6 +541,7 @@ class CultCache:
             (envelope, value, self._lookup_keys_for(value, name_extractor, index_extractors))
             for envelope, value in records
         ]
+        displaced = self._refuse_retype(document, [envelope for envelope, _ in records])
         self._refuse_index_sharing(
             [(type, envelope.key, keys) for envelope, _, keys in admitted],
             self._state.indexes,
@@ -515,6 +560,8 @@ class CultCache:
         values = self._state.values.setdefault(type, {})
         envelopes = self._state.envelopes.setdefault(type, {})
         lookup_keys = self._state.lookup_keys.setdefault(type, {})
+        for other, key in displaced:
+            self._drop(other, key)
         for envelope, value, keys in admitted:
             self._remove_lookups(type, envelope.key)
             values[envelope.key] = value
@@ -525,6 +572,39 @@ class CultCache:
             for index, index_value in keys.indexes.items():
                 self._state.indexes[(type, index, index_value)] = envelope.key
 
+    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
+        may not land at a key this cache holds another type's record at, unless the written type owns that record's schema id
+        (the single-file store's commit guard asks the same question of the file), and may not land at a key a store carries
+        foreign. Returns the (type, key) records an owning write replaces; refuses, changing nothing, otherwise."""
+        replaced: list[tuple[str, str]] = []
+        foreign = {record.key: record for record in self._state.foreign}
+        for envelope in written:
+            carried = foreign.get(envelope.key)
+            if carried is not None:
+                raise self._foreign_refusal(carried)
+            for other, held_envelopes in self._state.envelopes.items():
+                held = held_envelopes.get(envelope.key)
+                if other == document.type or held is None:
+                    continue
+                if not written_owns(held, envelope) or self._store_for_type(other) is not self._store_for_type(document.type):
+                    raise retyped(held, envelope)
+                replaced.append((other, envelope.key))
+        return replaced
+
+    @staticmethod
+    def _foreign_refusal(record: ForeignRecord) -> SchemaConflictError:
+        return SchemaConflictError(
+            record.schema_id,
+            [record.schema_name],
+            record.key,
+            message=(
+                f"Record {record.key!r} is held by the store as a {record.schema_name} (schema id {record.schema_id!r}) that no "
+                "registered type claims; a write or removal there would replace what this cache cannot read. Register the type "
+                "that owns the schema id, or remove the record with the runtime that wrote it."
+            ),
+        )
+
     @staticmethod
     def _retire_legacy_global(store: BackingStore, type: str, legacy_key: str) -> None:
         """The first write of a global loaded from a legacy key removes that record in the same write. If
@@ -564,19 +644,27 @@ 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)
         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)
         if legacy_key is not None:
             del self._state.legacy_global_keys[document.type]
-        self._remove_lookups(document.type, key)
+        self._drop(document.type, key)
+
+    def _drop(self, type: str, key: str) -> None:
+        """Removes the held record and its lookups from the cache's maps; the store is the caller's."""
+        self._remove_lookups(type, key)
         for by_type in (self._state.values, self._state.envelopes, self._state.lookup_keys):
-            held = by_type.get(document.type)
+            held = by_type.get(type)
             if held is not None:
                 held.pop(key, None)
                 if not held:
-                    by_type.pop(document.type, None)
+                    by_type.pop(type, None)
 
     @_mutating
     def delete_global(self, document: DocumentDefinition[Any]) -> None:

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

@@ -245,7 +245,7 @@ def _derive_catalog(records: list[tuple[CultCacheEnvelope, bool]]) -> list[CultC
     return [entry for entry, _ in chosen.values()]
 
 
-def _retyped(held: CultCacheEnvelope, written: CultCacheEnvelope) -> SchemaConflictError:
+def retyped(held: CultCacheEnvelope, written: CultCacheEnvelope) -> SchemaConflictError:
     """The refusal of a write that would put a second type under a key the store already holds (C# CacheBackingStore.Retyped)."""
     return SchemaConflictError(
         _schema_id_for(held),
@@ -258,7 +258,7 @@ def _retyped(held: CultCacheEnvelope, written: CultCacheEnvelope) -> SchemaConfl
     )
 
 
-def _written_owns(held: CultCacheEnvelope, written: CultCacheEnvelope) -> bool:
+def written_owns(held: CultCacheEnvelope, written: CultCacheEnvelope) -> bool:
     """Whether a write of `written` may replace `held` at its key (C# WrittenOwns): the written type's entry declares the held
     record's schema id (its own id or one it lists compatible), or it is the type of the same schema name. The same-name half
     stands in for C#'s "reads the published entry whole": this runtime does not compare shapes, so it never refuses a
@@ -277,8 +277,8 @@ def _encode_snapshot(
     landing: dict[str, CultCacheEnvelope] = {}
     for envelope in supplied:
         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)
+            if held is not None and not written_owns(held, envelope):
+                raise retyped(held, envelope)
         landing[envelope.key] = envelope
     supplied = list(landing.values())
     replaced = set(landing)

packages/cultcache-py/tests/test_key_identity_vector.py

@@ -137,3 +137,86 @@ class KeyIdentityVectorTests(unittest.TestCase):
             for key in kept:
                 self.assertIsNotNone(stored(vector, key))
                 self.assertEqual(stored(path, key), stored(vector, key))
+
+
+def alpha_beta(*names: str, store=None) -> CultCache:
+    made = CultCache.builder()
+    for name in names:
+        made.register_document_type(DOCUMENTS[name])
+    if store is not None:
+        made.add_generic_store(store)
+    return made.build()
+
+
+class KeyIdentityTests(unittest.TestCase):
+    def test_every_write_path_refuses_a_second_type_and_never_echoes_the_payload(self) -> None:
+        alpha, beta = DOCUMENTS["alpha"], DOCUMENTS["beta"]
+        for name in ("put", "put_envelope", "put_envelopes"):
+            with self.subTest(path=name):
+                cache = alpha_beta("alpha", "beta")
+                cache.put(alpha, "k", {"text": "held"})
+                envelope = cache.get_envelope(alpha, "k")
+                other = type(envelope).create(
+                    key="k", type=beta.type, payload=beta.encode_payload({"text": "CANARY"}), schema_id=IDS["beta"]
+                )
+                writes = {
+                    "put": lambda: cache.put(beta, "k", {"text": "CANARY"}),
+                    "put_envelope": lambda: cache.put_envelope(beta, other),
+                    "put_envelopes": lambda: cache.put_envelopes(beta, [other]),
+                }
+                with self.assertRaises(SchemaConflictError) as caught:
+                    writes[name]()
+                self.assertEqual(caught.exception.record_key, "k")
+                self.assertNotIn("CANARY", str(caught.exception))
+                self.assertEqual(cache.get(alpha, "k"), {"text": "held"})
+                self.assertIsNone(cache.get(beta, "k"))
+
+    def test_two_routed_stores_holding_one_key_refuse_the_pull_and_keep_the_earlier_state(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            one, two = Path(tmp) / "one.cc", Path(tmp) / "two.cc"
+            shutil.copyfile(VECTORS / "catalog-store.cc", one)
+            shutil.copyfile(VECTORS / "run-store.cc", two)
+            cache = alpha_beta("alpha", "beta")
+            cache.add_backing_store(SingleFileMessagePackBackingStore(one), [DOCUMENTS["alpha"].type])
+            cache.pull_all_backing_stores()
+            cache.add_backing_store(SingleFileMessagePackBackingStore(two), [DOCUMENTS["beta"].type])
+            with self.assertRaises(Exception) as caught:
+                cache.pull_all_backing_stores()
+            self.assertIn("'k'", str(caught.exception))
+            self.assertEqual(cache.get(DOCUMENTS["alpha"], "k"), {"text": "a"})
+            self.assertIsNone(cache.get(DOCUMENTS["beta"], "k"))
+
+    def test_a_foreign_record_is_listed_and_its_bytes_survive_writes_and_removals_elsewhere(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            path = Path(tmp) / "store.cc"
+            shutil.copyfile(VECTORS / "foreign-store.cc", path)
+            before = stored(path, "k")
+            cache = alpha_beta("alpha", store=SingleFileMessagePackBackingStore(path))
+            cache.pull_all_backing_stores()
+            [record] = cache.foreign_records()
+            self.assertEqual((record.key, record.schema_id, record.schema_name), ("k", IDS["gamma"], "tests.ki_gamma"))
+            self.assertEqual(record.stored_at, before[1])
+            cache.put(DOCUMENTS["alpha"], "m", {"text": "m"})
+            cache.delete(DOCUMENTS["alpha"], "j")
+            self.assertEqual(stored(path, "k"), before)
+            self.assertEqual(file_records(path), [["k", "tests.ki_gamma"], ["m", "tests.ki_alpha"]])
+            claimed = alpha_beta("alpha", "gamma", store=SingleFileMessagePackBackingStore(path))
+            claimed.pull_all_backing_stores()
+            self.assertEqual(claimed.foreign_records(), [])
+            self.assertEqual(claimed.get(DOCUMENTS["gamma"], "k"), {"text": "g"})
+
+    def test_the_foreign_list_is_sorted_by_utf16_code_unit_and_replaced_by_the_next_pull(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            path = Path(tmp) / "store.cc"
+            store = SingleFileMessagePackBackingStore(path)
+            gamma = DOCUMENTS["gamma"]
+            writer = alpha_beta("gamma", store=store)
+            for key in ("\U00010400", "Ａ", "b", "B"):
+                writer.put(gamma, key, {"text": "x"})
+            cache = alpha_beta("alpha", store=SingleFileMessagePackBackingStore(path))
+            cache.pull_all_backing_stores()
+            self.assertEqual([record.key for record in cache.foreign_records()], ["B", "b", "\U00010400", "Ａ"])
+            for key in ("B", "b", "\U00010400", "Ａ"):
+                writer.delete(gamma, key)
+            cache.pull_all_backing_stores()
+            self.assertEqual(cache.foreign_records(), [])