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 | R `SchemaConflictError` (cultcache-ts: one record per key) | 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 `SchemaConflictError` at write (foreign loads, carried; cultcache-ts) | 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 | R `SchemaConflictError` (cultcache-ts: one record per key) | 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 `SchemaConflictError` at write (foreign loads, carried; cultcache-ts) | 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, and a store that already holds two global types at `__global__` (a JSONL store written by caches that never pulled could produce one) no longer loads: the pull raises `CultCacheError` and admits nothing. C# refuses a second persisted global on pull the same way (`cultcache-store-composition.md`, "Globals never invented"); repair such a file by deleting one of the two records with the runtime that wrote it.
+- Added `foreign_records()` and the frozen `ForeignRecord(key, schema_id, schema_name, stored_at)`, sorted by key in UTF-16 code-unit order. `stores.retyped` and `stores.written_owns` lose their underscore: the cache judge and the single-file store's commit guard ask one predicate.
+
 - 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

@@ -0,0 +1,316 @@
+from __future__ import annotations
+
+import re
+import shutil
+import tempfile
+import unittest
+from pathlib import Path
+
+import msgpack
+
+from cultcache_py.cache import CultCacheError
+from cultcache_py import (
+    CultCache,
+
+    JsonLinesBackingStore,
+    SchemaConflictError,
+    SingleFileMessagePackBackingStore,
+    database_entry_field,
+    define_database_entry_type,
+    leaves as t,
+)
+
+# variants:ruling:cache-refuses-retype, write-ownership-by-written-type, unheld-removal-refused-at-commit. The vector in
+# tests/vectors/key-identity is written only by the C# reference; this walks every row against the Python cache.
+VECTORS = Path(__file__).resolve().parents[3] / "tests" / "vectors" / "key-identity"
+
+IDS = {
+    "alpha": "sha256:b20703a98f49c253898404ab50ba2c630720f0b8a5e1a41614d5650be22799fe",
+    "beta": "sha256:dee4b727cf2ee04e61ab635ff0c5f3dca7345a411dc5f1018e583ab16009cf94",
+    "gamma": "sha256:fbc26e351e00c9e7185e8de171c7486aba707c229c949f0aae6b889288af5f6d",
+}
+DOCUMENTS = {
+    label: define_database_entry_type(
+        f"tests.ki_{label}",
+        [database_entry_field("text", 0, codec=t.string)],
+        schema_id=schema_id,
+        schema_name=f"tests.ki_{label}",
+    )
+    for label, schema_id in IDS.items()
+}
+LABEL_OF = {document.type: label for label, document in DOCUMENTS.items()}
+
+def file_records(path: Path) -> list[list[str]]:
+    return sorted([envelope.key, envelope.type] for envelope in SingleFileMessagePackBackingStore(path).pull_all())
+
+def stored(path: Path, key: str):
+    for envelope in SingleFileMessagePackBackingStore(path).pull_all():
+        if envelope.key == key:
+            return envelope.schema_id, envelope.stored_at, envelope.payload
+    return None
+
+class KeyIdentityVectorTests(unittest.TestCase):
+    def test_rows(self) -> None:
+        rows = msgpack.unpackb((VECTORS / "rows.msgpack").read_bytes(), raw=False)
+        walked = 0
+        for label, registered, ops, expect in rows:
+            with self.subTest(row=label), tempfile.TemporaryDirectory() as tmp:
+                self.walk(Path(tmp), registered, ops, expect)
+                walked += 1
+        self.assertEqual(walked, 12)
+
+    def walk(self, tmp: Path, registered: list[str], ops: list, expect: list) -> None:
+        refusal, held, file, foreign, kept = expect
+        state: dict = {"cache": None, "path": None, "before": None, "attached": []}
+
+        def builder() -> CultCache.builder:
+            made = CultCache.builder()
+            for name in registered:
+                made.register_document_type(DOCUMENTS[name])
+            return made
+
+        def open_cache(target) -> None:
+            if target == "memory":
+                state["cache"], state["path"] = builder().build(), None
+                return
+            path = tmp / "store.cc"
+            if target is not None and target != path:
+                shutil.copyfile(VECTORS / target, path)
+                state["before"] = path.read_bytes()
+            state["path"] = path
+            made = builder().add_generic_store(SingleFileMessagePackBackingStore(path))
+            state["cache"] = made.build()
+            state["cache"].pull_all_backing_stores()
+
+        def perform(op: list) -> None:
+            cache = state["cache"]
+            if op[0] == "open":
+                open_cache(op[1])
+            elif op[0] == "attach":
+                path = tmp / f"attached-{len(state['attached'])}.cc"
+                shutil.copyfile(VECTORS / op[1], path)
+                state["attached"].append((path, path.read_bytes()))
+                cache.add_backing_store(
+                    SingleFileMessagePackBackingStore(path), [DOCUMENTS[name].type for name in op[2]]
+                )
+                cache.pull_all_backing_stores()
+            elif op[0] == "reopen":
+                open_cache(state["path"])  # the same file, no fresh copy
+            elif op[0] == "put":
+                cache.put(DOCUMENTS[op[1]], op[2], {"text": op[3]})
+            elif op[0] == "delete":
+                cache.delete(DOCUMENTS[op[1]], op[2])
+            else:
+                self.fail(f"unknown op {op[0]}")
+
+        for op in ops[:-1]:
+            perform(op)
+        if refusal is None:
+            perform(ops[-1])
+        else:
+            key = refusal[0]
+            with self.assertRaises((SchemaConflictError, ValueError, RuntimeError)) as caught:
+                perform(ops[-1])
+            if ops[-1][0] == "attach":
+                # The key is read out of the refusal's own text (it opens "Record 'key' is held by") and occurs only as a
+                # delimited token, never inside a schema name or other word.
+                message = str(caught.exception)
+                named = re.match(r"Record '([^']*)' is held by", message)
+                self.assertIsNotNone(named, message)
+                self.assertEqual(named.group(1), key)
+                self.assertEqual(len(re.findall(rf"(?<![A-Za-z0-9_]){re.escape(key)}(?![A-Za-z0-9_])", message)), 1)
+                self.assertEqual(message.count(key), 1)
+                for path, before in state["attached"]:
+                    self.assertEqual(path.read_bytes(), before)
+            else:
+                self.assertIsInstance(caught.exception, SchemaConflictError)
+                self.assertEqual(caught.exception.record_key, key)
+                self.assertIn(key, str(caught.exception))
+        cache = state["cache"]
+        got_held = sorted(
+            [LABEL_OF[type], key, value["text"]]
+            for type, values in cache.snapshot().items()
+            for key, value in values.items()
+        )
+        self.assertEqual(got_held, held)
+        path = state["path"]
+        if file is None:
+            self.assertIsNone(path)
+        else:
+            self.assertEqual(file_records(path), file)
+        self.assertEqual([record.key for record in cache.foreign_records()], foreign)
+        if kept:
+            vector = tmp / "vector-before.cc"
+            vector.write_bytes(state["before"])
+            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_two_global_types_are_refused_in_one_cache_and_a_store_holding_both_does_not_load(self) -> None:
+        # C# refuses a second persisted global on pull (cultcache-store-composition.md, "Globals never invented"); the Python
+        # JSONL store and a cache with no store used to let two global types share __global__, so such a file can exist.
+        ga = define_database_entry_type(
+            "tests.ki_ga", [database_entry_field("text", 0, codec=t.string)], schema_id="sha256:" + "1" * 64, global_document=True
+        )
+        gb = define_database_entry_type(
+            "tests.ki_gb", [database_entry_field("text", 0, codec=t.string)], schema_id="sha256:" + "2" * 64, global_document=True
+        )
+        with tempfile.TemporaryDirectory() as tmp:
+            path = Path(tmp) / "store.jsonl"
+            for document in (ga, gb):  # one cache per type that never pulled: how a file got two globals
+                writer = CultCache.builder().register_document_type(document).add_generic_store(JsonLinesBackingStore(path)).build()
+                writer.put_global(document, {"text": document.type})
+            both = CultCache.builder().register_document_type(ga).register_document_type(gb).add_generic_store(JsonLinesBackingStore(path)).build()
+            with self.assertRaises(CultCacheError) as caught:
+                both.pull_all_backing_stores()
+            self.assertIn("'__global__' is held by both", str(caught.exception))
+            self.assertEqual(both.snapshot(), {})
+            memory = CultCache.builder().register_document_type(ga).register_document_type(gb).build()
+            memory.put_global(ga, {"text": "a"})
+            with self.assertRaises(SchemaConflictError):
+                memory.put_global(gb, {"text": "b"})
+            self.assertEqual(memory.get_required_global(ga), {"text": "a"})
+
+    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("'qx'", str(caught.exception))
+            self.assertEqual(cache.get(DOCUMENTS["alpha"], "qx"), {"text": "a"})
+            self.assertIsNone(cache.get(DOCUMENTS["beta"], "qx"))
+
+    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(), [])
+
+class OwnershipAndStoreKindTests(unittest.TestCase):
+    """The cache judge answers on its own where the store has no commit guard (JSONL) or the records live in two stores."""
+
+    def test_a_jsonl_store_gets_no_help_from_a_disk_judge(self) -> None:
+        from cultcache_py import JsonLinesBackingStore
+
+        with tempfile.TemporaryDirectory() as tmp:
+            path = Path(tmp) / "store.jsonl"
+            gamma = DOCUMENTS["gamma"]
+            alpha_beta("gamma", store=JsonLinesBackingStore(path)).put(gamma, "k", {"text": "g"})
+            before = path.read_bytes()
+            cache = alpha_beta("alpha", store=JsonLinesBackingStore(path))
+            cache.pull_all_backing_stores()
+            self.assertEqual([record.key for record in cache.foreign_records()], ["k"])
+            with self.assertRaises(SchemaConflictError):
+                cache.put(DOCUMENTS["alpha"], "k", {"text": "x"})
+            with self.assertRaises(SchemaConflictError):
+                cache.delete(DOCUMENTS["alpha"], "k")
+            self.assertEqual(path.read_bytes(), before)
+            cache.put(DOCUMENTS["alpha"], "j", {"text": "a"})  # the foreign line is carried through a write elsewhere
+            self.assertIn(b"tests.ki_gamma", path.read_bytes())
+
+    def test_a_foreign_key_in_two_stores_refuses_the_pull(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            one, two = Path(tmp) / "one.cc", Path(tmp) / "two.cc"
+            for path in (one, two):
+                alpha_beta("gamma", store=SingleFileMessagePackBackingStore(path)).put(DOCUMENTS["gamma"], "k", {"text": "g"})
+            cache = alpha_beta("alpha")
+            cache.add_backing_store(SingleFileMessagePackBackingStore(one), [DOCUMENTS["alpha"].type])
+            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))
+
+    def test_a_type_that_owns_the_held_schema_id_replaces_the_record_and_the_key_keeps_one_type(self) -> None:
+        successor = define_database_entry_type(
+            "tests.ki_alpha2",
+            [database_entry_field("text", 0, codec=t.string)],
+            schema_id="sha256:" + "a" * 64,
+            schema_name="tests.ki_alpha2",
+            compatible_schema_ids=[IDS["alpha"]],
+        )
+        for routed in (False, True):
+            with self.subTest(routed=routed), tempfile.TemporaryDirectory() as tmp:
+                one, two = Path(tmp) / "one.cc", Path(tmp) / "two.cc"
+                made = CultCache.builder().register_document_type(DOCUMENTS["alpha"]).register_document_type(successor)
+                if routed:
+                    made.add_backing_store(SingleFileMessagePackBackingStore(one), [DOCUMENTS["alpha"].type])
+                    made.add_backing_store(SingleFileMessagePackBackingStore(two), [successor.type])
+                else:
+                    made.add_generic_store(SingleFileMessagePackBackingStore(one))
+                cache = made.build()
+                cache.put(DOCUMENTS["alpha"], "k", {"text": "old"})
+                if routed:
+                    with self.assertRaises(SchemaConflictError):  # one key would live in two stores
+                        cache.put(successor, "k", {"text": "new"})
+                    self.assertEqual(cache.get(DOCUMENTS["alpha"], "k"), {"text": "old"})
+                else:
+                    cache.put(successor, "k", {"text": "new"})
+                    self.assertIsNone(cache.get(DOCUMENTS["alpha"], "k"))
+                    self.assertEqual(cache.get(successor, "k"), {"text": "new"})
+                    self.assertEqual(file_records(one), [["k", "tests.ki_alpha2"]])

packages/cultnet-py/CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ## Unreleased (breaking; version 0.x, so the next minor)
 
+- `apply_raw_document_record` and `apply_raw_snapshot` raise `SchemaConflictError` when a remote record is of another type than the one the cache holds at its key (`cultcache-py` holds one record per key whatever its type); the cache is unchanged, and the caller decides whether to drop the record or the peer. `cultmesh-py`'s server answers such a put with `message_invalid`; its client's `read_next_change` lets the error out.
 - `CultNetDatabaseChange` carries `commit_id` and `commit_count` (`commitId`/`commitCount`; None outside a multi-record commit, as C# writes nil): `from_wire` refuses an empty or non-string id and a count that is not an integer of at least 2, `to_wire` writes them when set, and the published `database_change_raw` schema lists both. Python applies no batch. The `simulation_fact` document's `shard`, `frame`, `subject` and `claim_kind` aliases are `roles` (many facts share each); the catalog bytes do not move.
 - Decodes the write receipt frames of `contracts/cultnet/interop/write-receipt.md`: `parse_message` validates `cultnet.write_admitted.v0` and `cultnet.error.v1` against their closed schemas (unknown keys, empty or non-string ids, an unknown operation or code are refused; `error.v1` requires `messageId`, string or null; `limit_reached` is `error.v1` only). Added `CULTNET_ERROR_V1_CODES`, `CULTNET_WRITE_RECEIPT_MESSAGE_VERSIONS`, `write_content_hash` and `hello_advertises_write_receipts`; `hello` takes `supported_message_versions`. `tests/test_write_receipt.py` replays the vectors' `contentHash`, `wire` and `schemaInvalid` sections. Not implemented here: the client write tracker (`scenarios`) and the server rules (`dedupe`, `blankFields`); this package has no receipt server and no write client.
 - Added `message_invalid` to `CULTNET_ERROR_CODES` and `message_invalid_error(field, reason)`, byte-identical to the C# `ForMessageInvalid`. `cultmesh-py` refuses a malformed put or delete with it instead of the out-of-schema `malformed_document_put` and `malformed_document_delete`, which are removed.

packages/cultnet-py/tests/test_cultnet.py

@@ -14,7 +14,7 @@ from pathlib import Path
 from typing import Callable
 from uuid import uuid4
 import msgpack
-from cultcache_py import CultCache, define_database_entry_type
+from cultcache_py import CultCache, SchemaConflictError, define_database_entry_type
 from cultnet_py.benchmark import run_benchmark
 from cultnet_py.compare_csharp import (
     DEFAULT_PARITY_THRESHOLD,
@@ -1942,6 +1942,35 @@ class CultNetTests(unittest.TestCase):
         self.assertEqual(raised.exception.response["messageId"], "subscription-error")
         thread.join(2.0)
 
+    def test_replication_refuses_a_remote_record_of_another_type_at_a_held_key_and_changes_nothing(self) -> None:
+        # variants:ruling:cache-refuses-retype: a cache holds one record per key whatever its type, so a peer that reuses a
+        # key across types is refused where the record is applied; replication does not catch it.
+        first = define_database_entry_type("replica.first", [("name", 0), ("category", 1), ("value", 2, 0)], cls=Item)
+        second = define_database_entry_type("replica.second", [("name", 0), ("category", 1), ("value", 2, 0)], cls=Item)
+        cache = CultCache()
+        cache.register_document_type(first)
+        cache.register_document_type(second)
+        cache.put(first, "shared", Item("held", "", 0))
+
+        def raw(document):
+            return {
+                "schemaId": document.catalog_entry().schema_id,
+                "recordKey": "shared",
+                "storedAt": "2026-06-13T00:00:00Z",
+                "payloadEncoding": "messagepack",
+                "payload": document.encode_payload(Item("remote", "", 0)),
+            }
+
+        with self.assertRaises(SchemaConflictError):
+            apply_raw_document_record(cache, schema_document_map([first, second]), raw(second))
+        snapshot = {"schemaVersion": "cultnet.snapshot_response_raw.v0", "messageId": "s", "documents": [raw(second)]}
+        with self.assertRaises(SchemaConflictError):
+            apply_raw_snapshot(cache, [first, second], snapshot)
+        self.assertEqual(cache.get_required(first, "shared").name, "held")
+        self.assertIsNone(cache.get(second, "shared"))
+        apply_raw_document_record(cache, schema_document_map([first, second]), raw(first))  # the same type still replaces
+        self.assertEqual(cache.get_required(first, "shared").name, "remote")
+
     def test_cultnet_replication_helpers_apply_raw_snapshot_and_shard_log(self) -> None:
         document = define_database_entry_type(
             "replica.item",