cultcache-py: a record refusal names only a string key; pin link writes for R3

The five-slot refusal echoed a non-string key or schema id with repr, quoting
stored bytes. A string is named; anything else is described by its type.

A push through a live link replaces the link with a regular file. A test
named for R3 pins that until R3 decides whether writes resolve or refuse
links.

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

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

@@ -170,6 +170,14 @@ def _read_store(path: Path, read: Callable[[Path], _Contents]) -> _Contents | No
         return None
 
 
+def _describe_identity(value: Any) -> str:
+    """A record key or schema id as a refusal may show it: a string is an identity and is
+    named; anything else is described by its type, since the value is the store's."""
+    if isinstance(value, str):
+        return repr(value)
+    return f"<{type(value).__name__}>"
+
+
 def _describe_header(header: str) -> str:
     """Echoes a header only in the shape cultcache.store.v<digits>; the bytes are the store's."""
     version = header[len(_STORE_FORMAT_PREFIX) + 1:] if header.startswith(_STORE_FORMAT_PREFIX + "v") else ""
@@ -204,7 +212,8 @@ def _decode_v1_snapshot(decoded: Any) -> list[CultCacheEnvelope] | None:
             raise ValueError("CultCache persisted records must be MessagePack arrays")
         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, more than "
+                f"CultCache record {_describe_identity(raw_record[0])} (schema {_describe_identity(raw_record[1])}) "
+                f"has {len(raw_record)} slots, more than "
                 f"the {_PERSISTED_RECORD_SLOTS} of a {STORE_FORMAT_VERSION} record, so this is not a valid store"
             )
         key, schema_id, stored_at, payload = raw_record

packages/cultcache-py/tests/test_cultcache.py

@@ -893,6 +893,37 @@ class CultCacheTests(unittest.TestCase):
                         self.assertNotIn(repr(value[0]), message)
                         self.assertIn(f"of {len(value[0])} bytes", message)
 
+    def test_single_file_record_refusal_names_only_a_string_key(self) -> None:
+        import msgpack  # type: ignore
+
+        with tempfile.TemporaryDirectory() as tmp:
+            path = Path(tmp) / "store.cc"
+            for key, schema in ((["SECRET-IN-LIST"], "s"), (b"SECRET-BYTES", "s"), ("k", {"SECRET": 1})):
+                path.write_bytes(msgpack.packb(
+                    ["cultcache.store.v1", [], [[key, schema, "t", b"\x90", "extra"]]], use_bin_type=True))
+                with self.assertRaises(ValueError) as caught:
+                    SingleFileMessagePackBackingStore(path).pull_all()
+                self.assertNotIn("SECRET", str(caught.exception))
+                self.assertIn("not a valid store", str(caught.exception))
+
+    # What a write does through a symbolic link today. R3 decides whether a write resolves
+    # the link or refuses it; this pins the current behaviour so that change is deliberate.
+    @unittest.skipIf(os.name != "posix", "symbolic links need POSIX here")
+    def test_r3_decides_resolve_or_refuse_a_push_through_a_live_link_replaces_the_link(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            volume = Path(tmp) / "volume"
+            volume.mkdir()
+            target = volume / "store.cc"
+            envelope = CultCacheEnvelope(key="a", type="t", payload=b"\x90", stored_at="2026-09-30T00:00:00Z")
+            SingleFileMessagePackBackingStore(target).push(envelope)
+            before = target.read_bytes()
+            link = Path(tmp) / "store.cc"
+            link.symlink_to(target)
+            SingleFileMessagePackBackingStore(link).push(replace(envelope, key="b"))
+            self.assertFalse(link.is_symlink())
+            self.assertTrue(link.is_file())
+            self.assertEqual(target.read_bytes(), before)
+
     @unittest.skipIf(os.name != "posix", "symbolic links need POSIX here")
     def test_a_dangling_symlink_at_a_store_path_is_an_os_error_and_nothing_is_written(self) -> None:
         for store_type in (SingleFileMessagePackBackingStore, JsonLinesBackingStore):