cultmesh-py: refuse wrong-typed shard fields before the write gate

A non-string shardId or non-int shardEpoch (bool is not int) on a put or delete
is a malformed_document_put/delete before the authority is asked, so a
shard-scoped authority cannot be bypassed by a shard the store would file under
str(shardId). The authority receives the fields as sent.

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

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

@@ -582,6 +582,9 @@ class CultMeshLocalServer:
                 message_id=str(message.get("messageId") or ""),
                 code="malformed_document_put",
             )]
+        malformed_shard = self._malformed_shard_fields(message, "malformed_document_put")
+        if malformed_shard is not None:
+            return [malformed_shard]
         # Refusal precedes every other judgment of the put. The node resolves the schema first, because the
         # authority must be asked about the canonical schema id (a payload-inferred alias included), as C#
         # does; only a payload of the wire shape is resolved, so nothing of an ungranted peer's is
@@ -640,6 +643,9 @@ class CultMeshLocalServer:
                 message_id=str(message.get("messageId") or ""),
                 code="malformed_document_delete",
             )]
+        malformed_shard = self._malformed_shard_fields(message, "malformed_document_delete")
+        if malformed_shard is not None:
+            return [malformed_shard]
         target = self.node.database.raw_delete_target(message) or (schema_id, record_key)
         refusal = self._refuse_unauthorized_write(peer, CultMeshRemoteWriteOperation.DELETE, target, message)
         if refusal is not None:
@@ -653,6 +659,22 @@ class CultMeshLocalServer:
             )]
         return self._database_delete_notifications(message, subscriptions, resolved_schema_id=change.schema_id)
 
+    def _malformed_shard_fields(self, message: dict[str, Any], code: str) -> dict[str, Any] | None:
+        """The protocol error for a shardId that is not a string or a shardEpoch that is not an int (a bool is
+        not), else None. It precedes the gate, so the authority only ever judges a write whose shard it can
+        read, and the store never files a write under a shard the authority did not see."""
+        shard_id = message.get("shardId")
+        shard_epoch = message.get("shardEpoch")
+        if (shard_id is None or isinstance(shard_id, str)) and (
+            shard_epoch is None or (isinstance(shard_epoch, int) and not isinstance(shard_epoch, bool))
+        ):
+            return None
+        return self._error_response(
+            "shardId must be a string and shardEpoch an integer.",
+            message_id=str(message.get("messageId") or ""),
+            code=code,
+        )
+
     def _refuse_unauthorized_write(
         self,
         peer: CultMeshServerPeer | None,
@@ -675,10 +697,8 @@ class CultMeshLocalServer:
                 operation=operation,
                 schema_id=schema_id,
                 record_key=record_key,
-                shard_id=message.get("shardId") if isinstance(message.get("shardId"), str) else None,
-                shard_epoch=message.get("shardEpoch")
-                if isinstance(message.get("shardEpoch"), int) and not isinstance(message.get("shardEpoch"), bool)
-                else None,
+                shard_id=message.get("shardId"),
+                shard_epoch=message.get("shardEpoch"),
             )
             try:
                 decision = authority.decide(request)

packages/cultmesh-py/tests/test_remote_write_authority.py

@@ -305,15 +305,31 @@ class RemoteWriteAuthorityTests(unittest.TestCase):
         self.assert_refused(replies, KEY_A)
         self.assertLess(peak, 10_000_000)
 
-    def test_shard_fields_of_the_wrong_type_reach_the_authority_as_none(self) -> None:
+    def test_wrong_typed_shard_fields_are_refused_before_the_gate_and_write_nothing(self) -> None:
         authority = Recording(lambda _: CultMeshRemoteWriteDecision.grant())
         server = self.build(authority)
-        message = put_of(KEY_A, "x")
-        message["shardId"] = {"not": "a string"}
-        message["shardEpoch"] = "7"
-        self.send(server, message)
-        self.assertIsNone(authority.requests[0].shard_id)
-        self.assertIsNone(authority.requests[0].shard_epoch)
+        self.seed(server, KEY_B, "server")
+        shards_before = server.node.database.shard_ids()
+        for field, value in (("shardId", {"not": "a string"}), ("shardId", 5), ("shardEpoch", "7"), ("shardEpoch", True), ("shardEpoch", 1.5)):
+            for operation, code in (
+                ("put", "malformed_document_put"),
+                ("delete", "malformed_document_delete"),
+            ):
+                with self.subTest(field=field, value=value, operation=operation):
+                    message = put_of(KEY_A, "x") if operation == "put" else delete_of(KEY_B)
+                    message[field] = value
+                    replies = self.send(server, message)
+                    self.assertEqual([reply["code"] for reply in replies], [code])
+        self.assertEqual(authority.requests, [], "the authority was never asked")
+        self.assertIsNone(self.text_of(server, KEY_A), "nothing was written")
+        self.assertEqual(self.text_of(server, KEY_B), "server", "nothing was deleted")
+        self.assertEqual(server.node.database.shard_ids(), shards_before, "no shard or log entry was filed")
+
+    def test_well_typed_shard_fields_reach_the_authority_as_sent(self) -> None:
+        authority = Recording(lambda _: CultMeshRemoteWriteDecision.grant())
+        server = self.build(authority)
+        self.send(server, put_of(KEY_A, "x"))
+        self.assertEqual((authority.requests[0].shard_id, authority.requests[0].shard_epoch), (SHARD_ID, 1))
 
     def test_grant_all_is_an_explicit_choice_that_grants_every_write(self) -> None:
         server = self.build(CultMeshGrantAllWrites())