cultcache-py: JSON lines writes refuse a link at one shared entry; pin StoreLinkedError as an OSError; changelog

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

packages/cultcache-py/CHANGELOG.md

@@ -13,6 +13,7 @@
 
 ### Breaking
 
+- 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 the single-file store replaced it while the JSON lines store 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`.
 - A payload is one MessagePack value: `DocumentDefinition.decode_payload` refuses bytes after it, by count, with `ValueError` `<type> payload has N bytes after its first MessagePack value (refused on read).` (never the bytes, never the depth refusal; a payload that ends early stays the decoder's), as the C# reference does (`tests/vectors/union-parity/bytes-after-outcomes.msgpack`). It raised `msgpack.ExtraData` (a `ValueError`, so only a caller naming `msgpack.ExtraData` is affected); this is a minor-version bump while pre-1.0. The refused set is unchanged. Cache loads, `get` and every definition built on the MessagePack decoder go through it; a user `payload_decoder` that raises `ExtraData` is refused the same way. Depth is judged first, so a value nested past 256 with a tail is the depth refusal, as in C#.
 - 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.

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

@@ -101,22 +101,24 @@ class JsonLinesBackingStore:
             return envelopes
 
     def push(self, envelope: CultCacheEnvelope) -> None:
-        with self._lock:
-            existing = {(item.type, item.key): item for item in self.pull_all()}
-            existing[(envelope.type, envelope.key)] = envelope
-            self._replace_all(list(existing.values()))
+        self.push_all([envelope])
 
     def push_all(self, envelopes: list[CultCacheEnvelope]) -> None:
-        with self._lock:
-            existing = {(item.type, item.key): item for item in self.pull_all()}
+        def merge(existing: dict[tuple[str, str], CultCacheEnvelope]) -> None:
             for envelope in envelopes:
                 existing[(envelope.type, envelope.key)] = envelope
-            self._replace_all(list(existing.values()))
+        self._rewrite(merge)
 
     def delete(self, type: str, key: str) -> None:
+        self._rewrite(lambda existing: existing.pop((type, key), None))
+
+    def _rewrite(self, change: Callable[[dict[tuple[str, str], CultCacheEnvelope]], object]) -> None:
+        """The one entry of every write: refuses a link at the path, then reads the file, applies the change and replaces it."""
         with self._lock:
-            existing = [item for item in self.pull_all() if not (item.type == type and item.key == key)]
-            self._replace_all(existing)
+            _refuse_linked(self.path)
+            existing = {(item.type, item.key): item for item in self.pull_all()}
+            change(existing)
+            self._replace_all(list(existing.values()))
 
     def _replace_all(self, envelopes: list[CultCacheEnvelope]) -> None:
         self.path.parent.mkdir(parents=True, exist_ok=True)

packages/cultcache-py/tests/test_cultcache.py

@@ -1461,6 +1461,35 @@ class CultCacheTests(unittest.TestCase):
             self.assertEqual(sorted(p.name for p in volume.iterdir()), volume_before)
             self.assertEqual([item.key for item in store.pull_all()], ["a"])
 
+    @unittest.skipIf(os.name != "posix", "symbolic links need POSIX here")
+    def test_a_json_lines_write_at_a_live_link_is_refused_and_the_link_survives(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            volume = Path(tmp) / "volume"
+            volume.mkdir()
+            target = volume / "store.jsonl"
+            envelope = CultCacheEnvelope(key="a", type="t", payload=b"\x90", stored_at="2026-09-30T00:00:00Z")
+            JsonLinesBackingStore(target).push(envelope)
+            before = target.read_bytes()
+            link = Path(tmp) / "link.jsonl"
+            link.symlink_to(target)
+            store = JsonLinesBackingStore(link)
+            for write in (
+                lambda: store.push(replace(envelope, key="b")),
+                lambda: store.push_all([replace(envelope, key="b")]),
+                lambda: store.delete("t", "a"),
+            ):
+                with self.assertRaises(StoreLinkedError):
+                    write()
+            self.assertTrue(link.is_symlink())
+            self.assertEqual(target.read_bytes(), before)
+            self.assertEqual(sorted(p.name for p in Path(tmp).iterdir()), ["link.jsonl", "volume"])
+            self.assertEqual(sorted(p.name for p in volume.iterdir()), ["store.jsonl"])
+
+    def test_store_linked_error_is_an_os_error_carrying_the_path(self) -> None:
+        error = StoreLinkedError(Path("some/store.cc"))
+        self.assertIsInstance(error, OSError)
+        self.assertEqual(error.path, Path("some/store.cc"))
+
     @unittest.skipIf(os.name != "posix", "symbolic links need POSIX here")
     def test_a_write_at_a_live_directory_link_is_refused_as_a_link(self) -> None:
         with tempfile.TemporaryDirectory() as tmp:
@@ -1485,8 +1514,7 @@ class CultCacheTests(unittest.TestCase):
                 store = store_type(path)
                 with self.assertRaises(FileNotFoundError):
                     store.pull_all()
-                # The single-file store refuses the link itself; the unlocked JSON lines store meets the missing target.
-                with self.assertRaises(StoreLinkedError if store_type is SingleFileMessagePackBackingStore else FileNotFoundError):
+                with self.assertRaises(StoreLinkedError):
                     store.push(CultCacheEnvelope(key="k", type="t", payload=b"\x90", stored_at="2026-09-30T00:00:00Z"))
                 self.assertTrue(path.is_symlink())
                 self.assertEqual(list(volume.iterdir()), [])