cultnet-py: refuse a replicated record with no stamp where it becomes an envelope (py-snapshot-stampless)

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

packages/cultmesh-py/src/cultmesh_py/node.py

@@ -12,7 +12,7 @@ from typing import Any, Callable
 from cultcache_py import CultCache, CultCacheEnvelope, SingleFileMessagePackBackingStore
 from cultcache_py.cache import CultCacheError, index_text
 from cultcache_py.documents import DocumentDefinition, extract_value
-from cultcache_py.stores import parse_stored_at
+from cultnet_py.replication import raw_record_to_envelope
 from cultnet_py import (
     CultNetAppliedRecord,
     CultNetFileShardMutationLogStore,
@@ -724,19 +724,8 @@ class CultMeshDatabase:
         document = self._raw_put_document(message)
         if document is None:
             return None
-        document_record = message["document"]
-        # A peer's put is stored under the stamp it carries, so one that is no stamp is refused: nothing is minted for it.
-        if parse_stored_at(document_record.get("storedAt")) is None:
-            raise ValueError("A raw put must carry a storedAt stamp.")
-        # One id: the local schema id, under which the cache stores the record.
-        return document, CultCacheEnvelope(
-            key=str(document_record["recordKey"]),
-            type=document.type,
-            schema_id=document.catalog_entry().schema_id,
-            payload=bytes(document_record["payload"]),
-            stored_at=document_record["storedAt"],
-            catalog_entry=document.catalog_entry(),
-        )
+        # One id: the local schema id, under which the cache stores the record. The stamp is judged where a record becomes an envelope.
+        return document, raw_record_to_envelope(document, document.catalog_entry().schema_id, message["document"])
 
     @staticmethod
     def _served_record(envelope: CultCacheEnvelope) -> CultNetRawDocumentRecord:

packages/cultmesh-py/tests/test_stampless_replication.py

@@ -0,0 +1,194 @@
+"""A record a peer replicates to this node is stored under the stamp it carries, so one with no stamp is refused where it becomes an envelope.
+
+The stamp shape is tests/vectors/stored-at-stamp/stamps.msgpack, held to the same answers as the raw put of the server
+(variants:ruling:raw-put-without-stamp-refused). A snapshot response or shard log that carries a record with no stamp is refused whole
+with ValueError before anything is stored (as a record of an unregistered schema is refused whole); the refusal of a put is
+message_invalid (field `document`), tested in test_remote_write_authority.py.
+"""
+
+from __future__ import annotations
+
+import tempfile
+import unittest
+from pathlib import Path
+from typing import Any
+
+import msgpack
+from cultcache_py import define_database_entry_type
+from cultcache_py.stores import parse_stored_at
+from cultmesh_py import CultMeshGrantAllWrites, CultMeshServerPeer, create_node
+from cultmesh_py.client import CultMeshDocumentSubscription
+from cultmesh_py.server import CultMeshLocalServer
+from cultnet_py import CultNetDatabaseChange, document_put_raw
+
+SCHEMA_ID = "sha256:stampless-note"
+SHARD_ID = "stampless-shard"
+PEER = CultMeshServerPeer("127.0.0.1:50000", "tcp")
+VECTOR = Path(__file__).resolve().parents[3] / "tests/vectors/stored-at-stamp/stamps.msgpack"
+ROWS: list[tuple[str, bool]] = [(text, parses) for text, parses in msgpack.unpackb(VECTOR.read_bytes(), raw=False)]
+# The vector's texts, plus the shapes a record can lack a stamp in: the field missing, and not a string at all.
+REFUSED: list[tuple[str, Any]] = [(f"vector {text!r}", text) for text, parses in ROWS if not parses] + [
+    ("missing", None),
+    ("not a string", 12345),
+]
+
+NOTE = define_database_entry_type(
+    "tests.stampless_note",
+    [("schema_version", 0), ("text", 1)],
+    schema_id=SCHEMA_ID,
+    schema_name="tests.stampless_note",
+    schema_version="tests.stampless_note.v1",
+)
+
+
+def record_of(key: str, text: str, stamp: Any = "2026-11-07T00:00:00Z") -> dict[str, Any]:
+    record = {
+        "schemaId": SCHEMA_ID,
+        "recordKey": key,
+        "payloadEncoding": "messagepack",
+        "payload": NOTE.encode_payload({"schema_version": "tests.stampless_note.v1", "text": text}),
+    }
+    if stamp is not None:
+        record["storedAt"] = stamp
+    return record
+
+
+def snapshot_of(*records: dict[str, Any]) -> dict[str, Any]:
+    return {"schemaVersion": "cultnet.snapshot_response_raw.v0", "messageId": "snapshot", "documents": list(records)}
+
+
+def shard_log_of(*records: dict[str, Any]) -> dict[str, Any]:
+    entries = [
+        {
+            "sequence": index + 1,
+            "committedAt": "2026-11-07T00:00:00Z",
+            "changeKind": "added",
+            "put": document_put_raw(
+                message_id=f"put-{index}", key=record["recordKey"], schema_id=SCHEMA_ID, stored_at="2026-11-07T00:00:00Z",
+                payload=record["payload"], shard_id=SHARD_ID, shard_epoch=1,
+            ).to_wire() | {"document": record},
+        }
+        for index, record in enumerate(records)
+    ]
+    return {
+        "schemaVersion": "cultnet.shard_log_response.v0", "messageId": "log", "shardId": SHARD_ID, "shardEpoch": 1,
+        "entries": entries, "resyncRequired": False,
+    }
+
+
+class StampVectorTests(unittest.TestCase):
+    def test_python_parses_every_row_of_the_vector_as_the_vector_says(self) -> None:
+        self.assertGreater(len(ROWS), 30)
+        for text, parses in ROWS:
+            with self.subTest(text=text):
+                self.assertEqual(parse_stored_at(text) is not None, parses)
+
+    def test_every_vector_row_through_the_server_put_handler_is_stored_or_refused_as_the_vector_says(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            node = create_node(Path(directory) / "node.cc", runtime_id="vector-server")
+            node.database.register_document(NOTE)
+            server = CultMeshLocalServer(node=node, port=0, write_authority=CultMeshGrantAllWrites())
+            sized = CultMeshLocalServer(node=node, port=0, write_authority=CultMeshGrantAllWrites(), max_snapshot_bytes=10**6)
+            for index, (text, parses) in enumerate(ROWS):
+                for label, handler, over_rudp in (("plain", server, False), ("sized", sized, False), ("rudp", server, True)):
+                    with self.subTest(text=text, path=label):
+                        key = f"vector:{index}:{label}"
+                        message = document_put_raw(
+                            message_id="wire", key=key, schema_id=SCHEMA_ID, stored_at=text,
+                            payload=record_of(key, "peer")["payload"], shard_id=SHARD_ID, shard_epoch=1,
+                        ).to_wire()
+                        replies = handler._handle_connection_message(message, {}, over_rudp=over_rudp, peer=PEER)
+                        if parses:
+                            self.assertEqual(replies, [])
+                            self.assertEqual(node.cache.get(NOTE, key)["text"], "peer")
+                        else:
+                            self.assertEqual([(r["code"], r["details"]["field"]) for r in replies], [("message_invalid", "document")])
+                            self.assertIsNone(node.cache.get(NOTE, key))
+
+
+class StamplessSnapshotTests(unittest.TestCase):
+    def build(self, directory: str) -> Any:
+        node = create_node(Path(directory) / "node.cc", runtime_id="stampless-client")
+        node.database.register_document(NOTE)
+        node.database.pull()
+        return node
+
+    def test_a_snapshot_with_a_stampless_record_is_refused_whole_and_the_store_survives_a_reopen(self) -> None:
+        for label, stamp in REFUSED:
+            with self.subTest(label), tempfile.TemporaryDirectory() as directory:
+                node = self.build(directory)
+                good, bad = record_of("note:good", "good"), record_of("note:bad", "bad", stamp)
+                for order in ((good, bad), (bad, good)):
+                    with self.assertRaises(ValueError):
+                        node.database.apply_snapshot_response(snapshot_of(*order))
+                    self.assertIsNone(node.cache.get(NOTE, "note:good"), "a refused snapshot stores none of its records")
+                    self.assertIsNone(node.cache.get(NOTE, "note:bad"))
+                # The store file reads back, and is still writable by a snapshot of stamped records.
+                reopened = self.build(directory)
+                self.assertIsNone(reopened.cache.get(NOTE, "note:bad"))
+                applied = reopened.database.apply_snapshot_response(snapshot_of(good))
+                self.assertEqual([a.record_key for a in applied], ["note:good"])
+                self.assertEqual(self.build(directory).cache.get(NOTE, "note:good")["text"], "good")
+
+    def test_a_snapshot_of_every_stamped_vector_row_is_stored(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            node = self.build(directory)
+            rows = [(index, text) for index, (text, parses) in enumerate(ROWS) if parses]
+            node.database.apply_snapshot_response(snapshot_of(*[record_of(f"stamped:{i}", "ok", text) for i, text in rows]))
+            self.assertEqual(len(self.build(directory).cache.get_all(NOTE)), len(rows))
+            for index, _ in rows:
+                self.assertEqual(node.cache.get(NOTE, f"stamped:{index}")["text"], "ok")
+
+
+class StamplessShardLogTests(unittest.TestCase):
+    def test_a_shard_log_put_with_no_stamp_is_refused_and_the_store_stays_writable(self) -> None:
+        for label, stamp in REFUSED:
+            with self.subTest(label), tempfile.TemporaryDirectory() as directory:
+                node = create_node(Path(directory) / "node.cc", runtime_id="stampless-log")
+                node.database.register_document(NOTE)
+                with self.assertRaises(ValueError):
+                    node.database.apply_shard_log_response(shard_log_of(record_of("note:bad", "bad", stamp)))
+                self.assertIsNone(node.cache.get(NOTE, "note:bad"))
+                reopened = create_node(Path(directory) / "node.cc", runtime_id="stampless-log")
+                reopened.database.register_document(NOTE)
+                reopened.database.pull()
+                reopened.database.apply_shard_log_response(shard_log_of(record_of("note:good", "good")))
+                self.assertEqual(reopened.cache.get(NOTE, "note:good")["text"], "good")
+                again = create_node(Path(directory) / "node.cc", runtime_id="stampless-log")
+                again.database.register_document(NOTE)
+                again.database.pull()
+                self.assertEqual(again.cache.get(NOTE, "note:good")["text"], "good")
+
+
+class _StubSubscription:
+    def __init__(self, change: CultNetDatabaseChange) -> None:
+        self._change = change
+
+    def read_next_change(self) -> CultNetDatabaseChange:
+        return self._change
+
+
+class StamplessClientReplayTests(unittest.TestCase):
+    def test_a_subscribed_put_with_no_stamp_is_refused_by_read_next_change_and_nothing_is_stored(self) -> None:
+        for label, stamp in REFUSED:
+            with self.subTest(label), tempfile.TemporaryDirectory() as directory:
+                node = create_node(Path(directory) / "node.cc", runtime_id="stampless-subscriber")
+                node.database.register_document(NOTE)
+                subscription = CultMeshDocumentSubscription(client=None, database=node.database, document=NOTE, key="note:bad")  # type: ignore[arg-type]
+
+                def deliver(record: dict[str, Any]) -> Any:
+                    message = {
+                        "schemaVersion": "cultnet.database_change_raw.v0", "messageId": "change", "changeKind": "added",
+                        "schemaId": SCHEMA_ID, "recordKey": record["recordKey"], "document": record,
+                    }
+                    subscription._subscription = _StubSubscription(CultNetDatabaseChange.from_wire(message))  # type: ignore[assignment]
+                    return subscription.read_next_change()
+
+                with self.assertRaises(ValueError):
+                    deliver(record_of("note:bad", "bad", stamp))
+                self.assertIsNone(node.cache.get(NOTE, "note:bad"))
+                self.assertEqual(deliver(record_of("note:bad", "stamped")).value["text"], "stamped")
+
+
+if __name__ == "__main__":
+    unittest.main()

packages/cultnet-py/CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ## Unreleased (breaking; version 0.x, so the next minor)
 
+- `raw_record_to_envelope`, where a replicated record becomes an envelope, refuses a record whose `storedAt` is missing, empty or not a stamp (`cultcache_py.stores.parse_stored_at`, the shape pinned by `tests/vectors/stored-at-stamp`) with `ValueError`, nothing minted or stored (`variants:ruling:raw-put-without-stamp-refused`). `apply_raw_document_record`, `apply_raw_snapshot` and `apply_shard_log_response` come through it: a snapshot response with such a record is refused whole before any of its records is stored, as one naming an unregistered schema is; a shard log refuses at that entry, the entries before it already applied. It stored the record under `""`, and the next open of the store raised `CultCacheError` `Missing envelope`. `cultmesh-py`'s raw put builds its envelope through the same function, so there is one owner of the check there; its server gate still answers a peer's put `message_invalid`. A client's `read_next_change` lets the `ValueError` out.
 - `decode_witness_artifact_bundle_payload` (and `CultNetWitnessArtifactBundle.from_payload`) read the payload through `cultcache_py.read_payload_tree`: bytes after its first MessagePack value raise `ValueError` `cultnet.witness_artifact_bundle payload has N bytes after its first MessagePack value (refused on read).` instead of `msgpack.ExtraData`, and a payload nested past 256 raises `CultCachePayloadDepthError`. Needs `cultcache-py` 0.4.0, the release that exports `read_payload_tree`: the dependency is now `cultcache-py>=0.4.0` (it was `>=0.2.0`, which admits releases whose `import cultnet_py` fails), and `tests/test_declared_dependency_ranges.py` fails when a declared bound admits a tagged release lacking a name the package imports. The same check found `cultmesh-py` importing `cache.index_text`, which `cultcache-py` 0.2.0 lacks; its bound is `cultcache-py>=0.4.0` too (it has no changelog of its own). `resolve_document_for_raw_record` reads the schema hint from a raw record's first value whether or not bytes follow it, so a record routed by its stamp is refused by count when applied (`ValueError`, in a snapshot, a shard log and a direct apply) like one routed by a registered schema id; `apply_shard_log_response` used to skip it silently as an unknown schema. A tail on a payload whose first value names no registered schema is still `KeyError`, which a shard log skips.
 - `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.

packages/cultnet-py/src/cultnet_py/replication.py

@@ -5,6 +5,7 @@ from typing import Any, Iterable
 
 from cultcache_py import CultCache, CultCacheEnvelope
 from cultcache_py.documents import DocumentDefinition
+from cultcache_py.stores import parse_stored_at
 
 from .shard_log import CultNetShardLogResponse
 from .snapshot import CultNetRawDocumentRecord, CultNetRawSnapshotResponse
@@ -55,12 +56,18 @@ def raw_record_to_envelope(
     schema_id: str,
     record: dict[str, Any],
 ) -> CultCacheEnvelope:
+    """The envelope a replicated raw record is stored as; the one place a record becomes an envelope (a direct apply, a snapshot and a
+    shard log all come through here). A peer's record is stored under the stamp it carries, so one whose storedAt is missing, empty or
+    no stamp is refused with ValueError and nothing is minted or stored (variants:ruling:raw-put-without-stamp-refused)."""
+    stored_at = record.get("storedAt")
+    if parse_stored_at(stored_at) is None:
+        raise ValueError("A replicated raw record must carry a storedAt stamp.")
     return CultCacheEnvelope(
         key=str(record["recordKey"]),
         type=document.type,
         schema_id=schema_id,
         payload=bytes(record["payload"]),
-        stored_at=str(record.get("storedAt") or ""),
+        stored_at=stored_at,
         catalog_entry=document.catalog_entry(),
     )