Python: refuse unknown store headers and extra record slots by name

Delete the raw_record[:4] slice and the legacy-envelope fallthrough for
unknown cultcache.store.* headers.

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

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

@@ -14,6 +14,8 @@ from .backing_store import (
 )
 
 STORE_FORMAT_VERSION = "cultcache.store.v1"
+_STORE_FORMAT_PREFIX = "cultcache.store."
+_PERSISTED_RECORD_SLOTS = 4
 
 
 class JsonLinesBackingStore:
@@ -153,7 +155,14 @@ def _encode_v1_snapshot(envelopes: list[CultCacheEnvelope]) -> list[Any]:
 
 
 def _decode_v1_snapshot(decoded: Any) -> list[CultCacheEnvelope] | None:
-    if not isinstance(decoded, list) or not decoded or decoded[0] != STORE_FORMAT_VERSION:
+    if not isinstance(decoded, list) or not decoded:
+        return None
+    if isinstance(decoded[0], str) and decoded[0].startswith(_STORE_FORMAT_PREFIX) and decoded[0] != STORE_FORMAT_VERSION:
+        raise ValueError(
+            f"CultCache store format {decoded[0]!r} is not readable; this runtime reads {STORE_FORMAT_VERSION!r} only. "
+            "The store needs a runtime that resolves document variants."
+        )
+    if decoded[0] != STORE_FORMAT_VERSION:
         return None
     if len(decoded) < 3 or not isinstance(decoded[1], list) or not isinstance(decoded[2], list):
         raise ValueError("CultCache v1 snapshot must contain a schema catalog and record array")
@@ -169,7 +178,12 @@ def _decode_v1_snapshot(decoded: Any) -> list[CultCacheEnvelope] | None:
     for raw_record in decoded[2]:
         if not isinstance(raw_record, list) or len(raw_record) < 4:
             raise ValueError("CultCache persisted records must be MessagePack arrays")
-        key, schema_id, stored_at, payload = raw_record[:4]
+        if len(raw_record) > _PERSISTED_RECORD_SLOTS:
+            raise ValueError(
+                f"CultCache record {raw_record[0]!r} (schema {raw_record[1]!r}) has {len(raw_record)} slots; "
+                f"this runtime reads {_PERSISTED_RECORD_SLOTS}. The store needs a runtime that resolves document variants."
+            )
+        key, schema_id, stored_at, payload = raw_record
         if not isinstance(key, str) or not key:
             raise ValueError("CultCache persisted records must declare a key")
         if not isinstance(schema_id, str) or not schema_id:

packages/cultcache-py/tests/test_cultcache.py

@@ -802,5 +802,39 @@ class CultCacheTests(unittest.TestCase):
         self.assertEqual(cache.get_required_envelope(document, "item:one").key, "item:one")
 
 
+    # Shared refusal vectors: tests/vectors/document-variants-c0, read by every runtime's tests.
+    _VECTORS = Path(__file__).resolve().parents[3] / "tests" / "vectors" / "document-variants-c0"
+    _ITEM_SCHEMA_ID = "sha256:88d3fdf0a927acf3b163940d8f8c7fe62b3316542ce771a67ec8bc038f594788"
+
+    def _pull_vector(self, name: str) -> list[CultCacheEnvelope]:
+        with tempfile.TemporaryDirectory() as tmp:
+            store_path = Path(tmp) / "store.msgpack"
+            store_path.write_bytes((self._VECTORS / name).read_bytes())
+            return SingleFileMessagePackBackingStore(store_path).pull_all()
+
+    def test_single_file_refuses_unknown_header_by_name(self) -> None:
+        with self.assertRaises(ValueError) as caught:
+            self._pull_vector("unknown-header.msgpack")
+        self.assertIn("cultcache.store.v9", str(caught.exception))
+
+    def test_single_file_refuses_extra_record_slot_naming_the_record(self) -> None:
+        with self.assertRaises(ValueError) as caught:
+            self._pull_vector("extra-slot-full-payload.msgpack")
+        self.assertIn("item:anvil", str(caught.exception))
+        self.assertIn(self._ITEM_SCHEMA_ID, str(caught.exception))
+
+    def test_single_file_refuses_variant_store_by_version_or_record(self) -> None:
+        with self.assertRaises(ValueError) as caught:
+            self._pull_vector("variant-v2.msgpack")
+        message = str(caught.exception)
+        self.assertTrue("cultcache.store.v2" in message or "item:anvil-big" in message, message)
+
+    def test_v1_store_written_at_the_base_commit_still_reads_byte_for_byte(self) -> None:
+        envelopes = self._pull_vector("v1-base.msgpack")
+        self.assertEqual([(e.key, e.type) for e in envelopes], [("alpha", "vectors.item"), ("beta", "vectors.item")])
+        self.assertEqual(envelopes[0].payload, b"\x92\xa5alpha\x01")
+        self.assertEqual(envelopes[1].payload, b"\x92\xa4beta\x02")
+
+
 if __name__ == "__main__":
     unittest.main()