packages/cultcache-py/CHANGELOG.md

@@ -2,6 +2,10 @@
 
 ## Unreleased
 
+### Changed
+
+- `parse_stored_at` (was `_parse_stored_at`) is public: cultmesh-py judges a peer's `storedAt` by it. A peer's `document_put_raw` whose `storedAt` is missing, empty or not a stamp is refused `message_invalid` (field `document`) and nothing is minted or stored (variants:ruling:raw-put-without-stamp-refused). A subscribed server's stampless put now raises `ValueError` from `CultMeshClient.read_next_change` instead of being stored under a minted stamp.
+
 ### Added
 
 - `CultCache.update` and `update_global` return `UpdateResult(outcome, document)`; `UpdateOutcome` is `UPDATED`, `DECLINED` or `LOST_RACE`. The single-file store gains `read(key)` (the record and its content version, the SHA-256 C# computes) and `commit_if(envelope, expected)`, the version is the SHA-256 of the record re-encoded as `[key, schemaId, storedAt, payload]` with minimal MessagePack encodings, which is the SHA-256 of the record's stored bytes exactly when the writer encoded it minimally (every writer in this library does; a record another writer encoded with a wider-than-needed length or integer has the version of its minimal form, not of its bytes). Both looked up by the cache and not `BackingStore` protocol members (variants:cut_spec:cut-update-with-retry-py).

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

@@ -10,7 +10,7 @@ from typing import Any, Callable, Generic, Iterable, TypeVar
 
 from .backing_store import BackingStore, CultCacheEnvelope
 from .documents import DocumentDefinition, UnreadSlots, extract_value
-from .stores import SchemaConflictError, _parse_stored_at, retyped, stored_at_over, written_owns
+from .stores import SchemaConflictError, parse_stored_at, retyped, stored_at_over, written_owns
 
 T = TypeVar("T")
 F = TypeVar("F", bound=Callable[..., Any])
@@ -650,7 +650,7 @@ class CultCache:
             self._drop(other, key)
         for envelope, value, keys in admitted:
             stamp = stamps.get(envelope.key)
-            if isinstance(stamp, str) and _parse_stored_at(stamp) is not None:
+            if isinstance(stamp, str) and parse_stored_at(stamp) is not None:
                 envelope = replace(envelope, stored_at=stamp)
             self._remove_lookups(type, envelope.key)
             values[envelope.key] = value

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

@@ -780,7 +780,7 @@ _STORED_AT_SHAPE = re.compile(
 )
 
 
-def _parse_stored_at(text: Any) -> datetime | None:
+def parse_stored_at(text: Any) -> datetime | None:
     """The UTC instant a stored_at text names, or None. Only the shape the runtimes write is a stamp: an RFC 3339 date-time with T,
     an optional fraction of 1 to 9 digits (cut to the microsecond) and Z or a numeric offset of hours 00-23 and minutes 00-59. Checked here, not left to
     fromisoformat, whose accepted set differs by interpreter (3.10 and 3.13) and from Rust and TypeScript. An instant outside
@@ -808,7 +808,7 @@ def mint_stored_at(previous: str | None, now: datetime) -> str:
     A previous that does not parse leaves `now`. Raises ValueError, nothing minted, when `previous` is the last instant a stamp
     can name, so a write over it is refused and writes nothing."""
     now = now.astimezone(timezone.utc)
-    last = _parse_stored_at(previous)
+    last = parse_stored_at(previous)
     if last is not None and now <= last:
         try:
             now = last + timedelta(microseconds=1)
@@ -820,7 +820,7 @@ def mint_stored_at(previous: str | None, now: datetime) -> str:
 def stored_at_over(staged: str, replaced: str) -> str:
     """The stored_at a record replacing `replaced` is written with (C# StoredAtOver): `staged` when both parse and it is strictly
     later, else one minted over `replaced`. No content is compared: every replacing write is stamped later."""
-    later, earlier = _parse_stored_at(staged), _parse_stored_at(replaced)
+    later, earlier = parse_stored_at(staged), parse_stored_at(replaced)
     if later is not None and earlier is not None and later > earlier:
         return staged
     return mint_stored_at(replaced, datetime.now(timezone.utc))

packages/cultcache-py/tests/test_update.py

@@ -26,7 +26,7 @@ from cultcache_py import (
     leaves as t,
 )
 from cultcache_py.cache import GLOBAL_KEY, CultCacheError
-from cultcache_py.stores import _parse_stored_at, mint_stored_at, record_version, stored_at_over
+from cultcache_py.stores import parse_stored_at, mint_stored_at, record_version, stored_at_over
 
 
 @dataclass
@@ -578,7 +578,7 @@ class StoreRestampTests(Fixture):
         return next(envelope.stored_at for envelope in store.pull_all() if envelope.key == key)
 
     def later_than_alpha(self, text: str) -> bool:
-        return _parse_stored_at(text) > _parse_stored_at(ALPHA)
+        return parse_stored_at(text) > parse_stored_at(ALPHA)
 
     def test_commit_if_and_the_restamp_at_the_store(self) -> None:
         store = SingleFileMessagePackBackingStore(self.path)
@@ -655,7 +655,7 @@ class StoreRestampTests(Fixture):
         }
         for text, instant in accepted.items():
             with self.subTest(text=text):
-                self.assertEqual(_parse_stored_at(text), instant)
+                self.assertEqual(parse_stored_at(text), instant)
         for text in (
             "2026-01-01", "20260101T000000Z", "2026-W01-1T00:00:00Z", "2026-01-01 00:00:00Z", "2026-01-01T00:00:00,5Z",
             "2026-01-01T00Z", "2026-01-01T00:00:00", "2026-01-01T00:00:00.Z", "2026-01-01T00:00:00.1234567890Z",
@@ -665,12 +665,12 @@ class StoreRestampTests(Fixture):
             "", "garbage",
         ):
             with self.subTest(text=text):
-                self.assertIsNone(_parse_stored_at(text))
+                self.assertIsNone(parse_stored_at(text))
 
     def test_a_stamp_outside_the_range_is_unparseable_and_one_at_the_edge_refuses_a_later_stamp(self) -> None:
         for text in ("9999-12-31T23:30:00-05:00", "0001-01-01T00:00:00+05:00"):
             with self.subTest(text=text):
-                self.assertIsNone(_parse_stored_at(text))
+                self.assertIsNone(parse_stored_at(text))
         now = datetime(2030, 1, 1, tzinfo=timezone.utc)
         self.assertEqual(mint_stored_at("9999-12-31T23:30:00-05:00", now), "2030-01-01T00:00:00.000000+00:00")
         edge = "9999-12-31T23:59:59.9999999Z"
@@ -699,10 +699,10 @@ class StoreRestampTests(Fixture):
     def test_stamp_parsing_and_install(self) -> None:
         for text in ("2026-01-01T00:00:00.1234567+00:00", "2026-01-01T00:00:00Z", "2026-01-01T00:00:00.123456+00:00"):
             with self.subTest(text=text):
-                self.assertIsNotNone(_parse_stored_at(text))
-                self.assertEqual(_parse_stored_at(text).tzinfo, timezone.utc)
-        self.assertIsNone(_parse_stored_at("garbage"))
-        self.assertIsNone(_parse_stored_at(None))
+                self.assertIsNotNone(parse_stored_at(text))
+                self.assertEqual(parse_stored_at(text).tzinfo, timezone.utc)
+        self.assertIsNone(parse_stored_at("garbage"))
+        self.assertIsNone(parse_stored_at(None))
         now = datetime(2030, 1, 1, tzinfo=timezone.utc)
         self.assertEqual(mint_stored_at("garbage", now), "2030-01-01T00:00:00.000000+00:00")
         self.assertEqual(stored_at_over("2000-01-01T00:00:00Z", "2000-01-01T00:00:00Z")[:4], str(datetime.now(timezone.utc).year))  # no later: minted at now

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

@@ -12,6 +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 import (
     CultNetAppliedRecord,
     CultNetFileShardMutationLogStore,
@@ -664,10 +665,10 @@ class CultMeshDatabase:
         ):
             return None
         try:
-            prepared = self._raw_put_envelope(message)
+            document = self._raw_put_document(message)
         except Exception:  # noqa: BLE001 - any other malformed shape
             return None
-        return None if prepared is None else (prepared[0].catalog_entry().schema_id, prepared[1].key)
+        return None if document is None else (document.catalog_entry().schema_id, document_record["recordKey"])
 
     def raw_delete_target(self, message: dict[str, Any]) -> tuple[str, str] | None:
         """The canonical schema id and record key a raw delete names, or None when its schema is not registered."""
@@ -706,25 +707,34 @@ class CultMeshDatabase:
             documents=(self._served_record(prepared[1]),),
         ).to_wire()
 
-    def _raw_put_envelope(
-        self, message: dict[str, Any]
-    ) -> tuple[DocumentDefinition[Any], CultCacheEnvelope] | None:
+    def _raw_put_document(self, message: dict[str, Any]) -> DocumentDefinition[Any] | None:
+        """The registered document a raw put names, or None when its record is not a map or its schema is not registered."""
         document_record = message.get("document")
         if not isinstance(document_record, dict):
             return None
         schema_id = str(document_record.get("schemaId") or "")
-        documents_by_schema_id = schema_document_map(self.documents)
         try:
-            document = resolve_document_for_raw_record(documents_by_schema_id, schema_id, document_record)
+            return resolve_document_for_raw_record(schema_document_map(self.documents), schema_id, document_record)
         except KeyError:
             return None
+
+    def _raw_put_envelope(
+        self, message: dict[str, Any]
+    ) -> tuple[DocumentDefinition[Any], CultCacheEnvelope] | None:
+        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=str(document_record.get("storedAt") or ""),
+            stored_at=document_record["storedAt"],
             catalog_entry=document.catalog_entry(),
         )
 

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

@@ -10,6 +10,7 @@ from typing import Any
 
 import msgpack
 
+from cultcache_py.stores import parse_stored_at
 from cultnet_py import (
     error_message,
     message_invalid_error,
@@ -969,7 +970,7 @@ class CultMeshLocalServer:
         as a snapshot would serve it under the shortest message id, must fit max_snapshot_bytes;
         a put arriving over RUDP must also fit, as fragments of rudp_max_fragment_bytes, in one
         message and the peer's reliable queue. A put whose document has no string recordKey or
-        binary payload could never be stored or served, and is refused as malformed."""
+        binary payload, or a storedAt that is no stamp, could never be stored or served, and is refused as malformed."""
         document_record = message.get("document")
         if not (
             isinstance(document_record, dict)
@@ -977,6 +978,8 @@ class CultMeshLocalServer:
             and isinstance(document_record.get("payload"), (bytes, bytearray))
         ):
             return self._message_invalid(message, "document", "Raw put documents must carry a string recordKey and a binary payload.")
+        if parse_stored_at(document_record.get("storedAt")) is None:
+            return self._message_invalid(message, "document", "Raw put documents must carry a storedAt stamp.")
         if self.max_snapshot_bytes is None and not over_rudp:
             return None
         alone = self.node.database.raw_put_served_alone(message, message_id=_SHORTEST_SERVED_MESSAGE_ID)

packages/cultmesh-py/tests/test_remote_write_authority.py

@@ -94,6 +94,44 @@ class NotificationStampTests(unittest.TestCase):
                     self.assertEqual(node.database._shard_logs[SHARD_ID][-1]["put"]["document"]["storedAt"], held)
 
 
+class StamplessPutTests(unittest.TestCase):
+    """A peer's put with no storedAt, or one that is no stamp, is refused message_invalid; nothing is minted or stored."""
+
+    def test_a_put_without_a_stamp_is_refused_and_the_store_stays_writable(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            node = create_node(Path(directory) / "node.cc", runtime_id="stampless-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)
+            rudp = CultMeshLocalServer(node=node, port=0, write_authority=CultMeshGrantAllWrites())
+            for label, stamp in (("missing", None), ("empty", ""), ("garbage", "yesterday"), ("no zone", "2026-11-07T00:00:00")):
+                with self.subTest(label):
+                    message = put_of(KEY_A, "peer")
+                    del message["document"]["storedAt"]
+                    if stamp is not None:
+                        message["document"]["storedAt"] = stamp
+                    # The size-limited and RUDP paths measure the document as a snapshot would serve it, so they must refuse first.
+                    for limited in (server, sized, rudp):
+                        [refusal] = limited._handle_connection_message(message, {}, over_rudp=limited is rudp, peer=PEER)
+                        self.assertEqual(refusal["code"], "message_invalid")
+                        self.assertEqual(refusal["details"]["field"], "document")
+                    with self.assertRaises(ValueError):
+                        node.database.apply_raw_put_message(message)
+                    self.assertIsNone(node.cache.get(NOTE, KEY_A))
+                    self.assertNotIn(SHARD_ID, node.database._shard_logs)
+            # An aliased put with a bad stamp still asks the authority about the canonical id, and is refused before the stamp is judged.
+            authority = Recording(lambda request: CultMeshRemoteWriteDecision.refuse("no"))
+            guarded = CultMeshLocalServer(node=node, port=0, write_authority=authority)
+            aliased = put_of(KEY_A, "peer", schema_id=ALIAS, stored_at="yesterday")
+            [refusal] = guarded._handle_connection_message(aliased, {}, peer=PEER)
+            self.assertEqual(refusal["code"], "write_not_authorized")
+            self.assertEqual(authority.requests[0].schema_id, SCHEMA_ID)
+            # The refused puts left the file readable: the next put of any key is stored.
+            replies = server._handle_connection_message(put_of(KEY_B, "stamped"), {}, peer=PEER)
+            self.assertEqual(replies, [])
+            self.assertEqual(node.cache.get(NOTE, KEY_B)["text"], "stamped")
+
+
 class RemoteWriteAuthorityTests(unittest.TestCase):
     def build(self, authority: Any = None) -> CultMeshLocalServer:
         node = create_node(runtime_id="authority-server")