From 1a173f60008f3f040e2ae83c5cb6d3cd3dc11449 Mon Sep 17 00:00:00 2001
From: Meta Vens <rwvens@hotmail.com>
Date: Wed, 30 Sep 2026 17:42:43 +0200
Subject: [PATCH] cultmesh-py: a put that could never be stored is refused, not
 raised

_unservable_put sized a put by building the envelope it would store, and
that read document["recordKey"] and document["payload"] directly: a put
over RUDP with no recordKey raised KeyError out of the RUDP loop and
ended the thread for every peer. A string payload raised TypeError the
same way.

_unservable_put now refuses a put whose document lacks a string recordKey
or a binary payload as malformed_document_put, the code the server
already uses for a put with no document map, before any sizing reads it
and before its early return, so the TCP path refuses it too instead of
raising in apply_raw_put_message.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 .../cultmesh-py/src/cultmesh_py/server.py     | 14 +++++++-
 packages/cultmesh-py/tests/test_cultmesh.py   | 35 +++++++++++++++++++
 2 files changed, 48 insertions(+), 1 deletion(-)

diff --git a/packages/cultmesh-py/src/cultmesh_py/server.py b/packages/cultmesh-py/src/cultmesh_py/server.py
index 4165577b..591a6f46 100644
--- a/packages/cultmesh-py/src/cultmesh_py/server.py
+++ b/packages/cultmesh-py/src/cultmesh_py/server.py
@@ -646,7 +646,19 @@ class CultMeshLocalServer:
         """The refusal for a put this server could never serve, or None. The document alone,
         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."""
+        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."""
+        document_record = message.get("document")
+        if not (
+            isinstance(document_record, dict)
+            and isinstance(document_record.get("recordKey"), str)
+            and isinstance(document_record.get("payload"), (bytes, bytearray))
+        ):
+            return self._error_response(
+                "Raw put documents must carry a string recordKey and a binary payload.",
+                message_id=str(message.get("messageId") or ""),
+                code="malformed_document_put",
+            )
         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)
diff --git a/packages/cultmesh-py/tests/test_cultmesh.py b/packages/cultmesh-py/tests/test_cultmesh.py
index 52191292..95bc7d9a 100644
--- a/packages/cultmesh-py/tests/test_cultmesh.py
+++ b/packages/cultmesh-py/tests/test_cultmesh.py
@@ -4585,6 +4585,41 @@ class CultMeshTests(unittest.TestCase):
         self.assertEqual(hello_reply["schemaVersion"], "cultnet.hello.v0")
         self.assertTrue(alive)
 
+    def test_cultmesh_local_server_refuses_a_malformed_put_over_rudp_and_serves_on(self) -> None:
+        # A put whose document has no string recordKey or no binary payload could never be stored
+        # or served: it is refused as malformed before any sizing reads it, and the RUDP thread
+        # serves the next peer.
+        document = self._rudp_bound_note()
+        node = CultMesh.create_node(runtime_id="mesh-rudp-malformed")
+        node.database.register_document(document)
+        server = CultMesh.serve_node(node)
+        try:
+            refusals = []
+            for field_name, value in (
+                ("recordKey", None),
+                ("recordKey", 5),
+                ("payload", None),
+                ("payload", "not binary"),
+            ):
+                put = self._rudp_note_put(document, f"bad-{field_name}-{value}", "note:bad", 10)
+                if value is None:
+                    del put["document"][field_name]
+                else:
+                    put["document"][field_name] = value
+                refusal, _ = self._rudp_exchange(server, put)
+                refusals.append((put["messageId"], refusal))
+            after, _ = self._rudp_exchange(server, snapshot_request(message_id="after").to_wire())
+            alive = server._rudp_thread is not None and server._rudp_thread.is_alive()
+        finally:
+            server.stop()
+        for message_id, refusal in refusals:
+            self.assertEqual(refusal["schemaVersion"], "cultnet.error.v0", message_id)
+            self.assertEqual(refusal["messageId"], message_id)
+            self.assertEqual(refusal["code"], "malformed_document_put", message_id)
+        self.assertEqual(after["messageId"], "after")
+        self.assertEqual(after["documents"], [])
+        self.assertTrue(alive)
+
     def test_cultmesh_local_server_refuses_options_that_could_answer_nothing(self) -> None:
         # The Rust document server's option rules: every limit is greater than zero, and the limits
         # leave room for an empty snapshot, the smallest response the server sends. One-byte