cultmesh-py: refuse a put the local server could never serve

With max_snapshot_bytes set, CultMeshLocalServer stored any raw put while
refusing every snapshot response over that size, so it could hold a
document no request would ever return.

_handle_raw_put now asks the database for the snapshot response that would
serve the document alone, built from the envelope the put would store (the
resolved local schema id, not the put's alias) under an empty message id,
the shortest this server answers. It is measured with the snapshot path's
own encoder. Over the bound the put is refused before it is stored, as
cultnet.error.v0 code document_unservable with responseBytes and
maxSnapshotBytes. At the bound it is stored and served at exactly that size.

The database's put and snapshot paths share one envelope builder and one
served-record builder, so the preview and the stored record cannot drift.

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

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

@@ -651,6 +651,38 @@ class CultMeshDatabase:
         return message
 
     def apply_raw_put_message(self, message: dict[str, Any]) -> CultMeshDatabaseChange | None:
+        prepared = self._raw_put_envelope(message)
+        if prepared is None:
+            return None
+        document, envelope = prepared
+        record_key = envelope.key
+        previous = self.cache.get(document, record_key)
+        value = self.cache.put_envelope(document, envelope)
+        change_kind = "added" if previous is None else "updated"
+        change = self._publish_local_change(document, record_key, change_kind, value, previous)
+        self._append_shard_log_put(CultNetMessage(
+            "cultnet.document_put_raw.v0",
+            {key: value for key, value in message.items() if key != "schemaVersion"},
+        ), change_kind)
+        return change
+
+    def raw_put_served_alone(self, message: dict[str, Any], *, message_id: str) -> dict[str, Any] | None:
+        """The snapshot response that would serve this put's document alone, as a wire map.
+
+        It is built from the envelope the put would store, exactly as a snapshot builds it, so a
+        server can judge a put by the size it would be served at. None when the put would not apply.
+        """
+        prepared = self._raw_put_envelope(message)
+        if prepared is None:
+            return None
+        return CultNetRawSnapshotResponse(
+            message_id=message_id,
+            documents=(self._served_record(prepared[1]),),
+        ).to_wire()
+
+    def _raw_put_envelope(
+        self, message: dict[str, Any]
+    ) -> tuple[DocumentDefinition[Any], CultCacheEnvelope] | None:
         document_record = message.get("document")
         if not isinstance(document_record, dict):
             return None
@@ -664,24 +696,24 @@ class CultMeshDatabase:
             )
         except KeyError:
             return None
-        record_key = str(document_record["recordKey"])
-        previous = self.cache.get(document, record_key)
-        envelope = CultCacheEnvelope(
-            key=record_key,
+        return document, CultCacheEnvelope(
+            key=str(document_record["recordKey"]),
             type=document.type,
             schema_id=resolved_schema_id,
             payload=bytes(document_record["payload"]),
             stored_at=str(document_record.get("storedAt") or ""),
             catalog_entry=document.catalog_entry(),
         )
-        value = self.cache.put_envelope(document, envelope)
-        change_kind = "added" if previous is None else "updated"
-        change = self._publish_local_change(document, record_key, change_kind, value, previous)
-        self._append_shard_log_put(CultNetMessage(
-            "cultnet.document_put_raw.v0",
-            {key: value for key, value in message.items() if key != "schemaVersion"},
-        ), change_kind)
-        return change
+
+    @staticmethod
+    def _served_record(envelope: CultCacheEnvelope) -> CultNetRawDocumentRecord:
+        return CultNetRawDocumentRecord(
+            schema_id=envelope.schema_id or envelope.type,
+            record_key=envelope.key,
+            stored_at=envelope.stored_at,
+            payload_encoding="messagepack",
+            payload=envelope.payload,
+        )
 
     def apply_raw_delete_message(self, message: dict[str, Any]) -> CultMeshDatabaseChange | None:
         schema_id = str(message.get("schemaId") or "")
@@ -774,13 +806,7 @@ class CultMeshDatabase:
                 continue
             if shard_record_keys is not None and (schema_id, envelope.key) not in shard_record_keys:
                 continue
-            documents.append(CultNetRawDocumentRecord(
-                schema_id=schema_id,
-                record_key=envelope.key,
-                stored_at=envelope.stored_at,
-                payload_encoding="messagepack",
-                payload=envelope.payload,
-            ))
+            documents.append(self._served_record(envelope))
 
         return CultNetRawSnapshotResponse(
             message_id=message_id,

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

@@ -24,6 +24,10 @@ from cultnet_py import (
 )
 
 from .node import CultMeshNode
+
+# The shortest message id this server answers a snapshot request under. It takes a
+# request's messageId as given, an empty one included, so no response it sends is smaller.
+_SHORTEST_SERVED_MESSAGE_ID = ""
 from cultnet_py.cultmesh_contracts import (
     PEER_EXCHANGE_REQUEST,
     VERSE_CATALOG_REQUEST,
@@ -357,6 +361,19 @@ class CultMeshLocalServer:
                 message_id=str(message.get("messageId") or ""),
                 code="malformed_document_put",
             )]
+        if self.max_snapshot_bytes is not None:
+            alone = self.node.database.raw_put_served_alone(
+                message, message_id=_SHORTEST_SERVED_MESSAGE_ID
+            )
+            response_bytes = None if alone is None else _snapshot_response_bytes(alone)
+            if response_bytes is not None and response_bytes > self.max_snapshot_bytes:
+                return [self._error_response(
+                    f"Document can never be served: its snapshot response is {response_bytes} bytes; "
+                    f"limit is {self.max_snapshot_bytes}.",
+                    message_id=str(message.get("messageId") or ""),
+                    code="document_unservable",
+                    details={"responseBytes": response_bytes, "maxSnapshotBytes": self.max_snapshot_bytes},
+                )]
         change = self.node.database.apply_raw_put_message(message)
         if change is None:
             return [self._error_response(
@@ -559,7 +576,7 @@ class CultMeshLocalServer:
                 details={"documentCount": document_count, "maxSnapshotDocuments": self.max_snapshot_documents},
             )
         if self.max_snapshot_bytes is not None:
-            response_bytes = len(msgpack.packb(response, use_bin_type=True))
+            response_bytes = _snapshot_response_bytes(response)
             if response_bytes > self.max_snapshot_bytes:
                 return self._error_response(
                     f"Snapshot byte limit exceeded: {response_bytes} > {self.max_snapshot_bytes}.",
@@ -723,3 +740,8 @@ def _infer_schema_name(schema_id: str) -> str | None:
         return None
     version = schema_id[marker + 2:]
     return schema_id[:marker] if version.isdigit() else None
+
+
+def _snapshot_response_bytes(response: dict[str, Any]) -> int:
+    """The one measure of a snapshot response, shared by the snapshot path and put admission."""
+    return len(msgpack.packb(response, use_bin_type=True))

packages/cultmesh-py/tests/test_cultmesh.py

@@ -14,6 +14,8 @@ from dataclasses import dataclass
 from datetime import UTC, datetime, timedelta
 from pathlib import Path
 from typing import Any, Callable
+
+import msgpack
 from cultcache_py import (
     CultCache,
     SingleFileMessagePackBackingStore,
@@ -4293,6 +4295,80 @@ class CultMeshTests(unittest.TestCase):
         self.assertEqual(error["details"]["documentCount"], 2)
         self.assertEqual(error["details"]["maxSnapshotDocuments"], 1)
 
+    def test_cultmesh_local_server_refuses_a_put_it_could_never_serve(self) -> None:
+        # A put is admitted only if some snapshot request could return it. The put arrives under a
+        # long foreign schema alias and is stored and served under the shorter local schema id, so
+        # the size that decides is the served one, not the put's own.
+        document = define_database_entry_type(
+            "mesh.served_bound_note",
+            [("schema_version", 0), ("body", 1)],
+            schema_id="sha256:mesh-served-bound",
+            schema_name="mesh.served_bound_note",
+            schema_version="mesh.served_bound_note.v1",
+        )
+
+        def put(message_id: str, key: str, body_length: int) -> dict[str, Any]:
+            return document_put_raw(
+                message_id=message_id,
+                key=key,
+                schema_id="runtime.generated.mesh.served_bound_note.ui.99",
+                stored_at="2026-09-30T00:00:00Z",
+                payload=document.encode_payload({
+                    "schema_version": "mesh.served_bound_note.v1",
+                    "body": "x" * body_length,
+                }),
+            ).to_wire()
+
+        def exchange(server: Any, *messages: dict[str, Any]) -> tuple[dict[str, Any], bytes]:
+            client = CultMesh.create_client("127.0.0.1", server.port, timeout_seconds=2.0)
+            with client.open_transport() as transport:
+                for message in messages:
+                    transport.send("schema", msgpack.packb(message, use_bin_type=True))
+                payload = transport.receive().payload
+            return msgpack.unpackb(payload, raw=False), payload
+
+        empty_id_snapshot = snapshot_request(message_id="").to_wire()
+
+        # The bound is what the server actually serves for the at-bound document alone, under the
+        # shortest message id it answers (an empty one), measured on an unbounded server.
+        def served_alone_bytes(key: str, body_length: int) -> int:
+            node = CultMesh.create_node(runtime_id="mesh-served-bound-probe")
+            node.database.register_document(document)
+            server = CultMesh.serve_node(node, enable_rudp=False)
+            try:
+                response, payload = exchange(server, put("probe", key, body_length), empty_id_snapshot)
+            finally:
+                server.stop()
+            self.assertEqual(response["schemaVersion"], "cultnet.snapshot_response_raw.v0")
+            self.assertEqual(len(response["documents"]), 1)
+            return len(payload)
+
+        limit = served_alone_bytes("note:fits", 100)
+        self.assertEqual(served_alone_bytes("note:over", 101), limit + 1, "fixture: the documents must straddle the bound by one byte")
+
+        node = CultMesh.create_node(runtime_id="mesh-served-bound")
+        node.database.register_document(document)
+        server = CultMesh.serve_node(node, max_snapshot_bytes=limit, enable_rudp=False)
+        try:
+            refused, _ = exchange(server, put("put-over", "note:over", 101))
+            stored_after_refusal = node.database.get(document, "note:over")
+            served, served_payload = exchange(server, put("put-at", "note:fits", 100), empty_id_snapshot)
+        finally:
+            server.stop()
+
+        self.assertEqual(refused["schemaVersion"], "cultnet.error.v0")
+        self.assertEqual(refused["messageId"], "put-over")
+        self.assertEqual(refused["code"], "document_unservable")
+        self.assertEqual(refused["details"], {"responseBytes": limit + 1, "maxSnapshotBytes": limit})
+        self.assertIn(str(limit + 1), refused["error"])
+        self.assertIn(str(limit), refused["error"])
+        self.assertIsNone(stored_after_refusal)
+
+        self.assertEqual(served["schemaVersion"], "cultnet.snapshot_response_raw.v0")
+        self.assertEqual([record["recordKey"] for record in served["documents"]], ["note:fits"])
+        self.assertEqual(served["documents"][0]["schemaId"], document.catalog_entry().schema_id)
+        self.assertEqual(len(served_payload), limit)
+
     def test_cultmesh_local_server_returns_errors_for_bad_requests(self) -> None:
         document = define_database_entry_type(
             "mesh.error_note",