Merge pull request 'C#: SubmitDocumentAsync and the shard write forwarder complete on the result frame' (#58) from libby/cs-client-ack-finding into main

CultNetWriteCompletion.SendAndAwaitResultAsync registers a refusal handler before
the send, then waits a result window (default 500 ms; CultMeshClientOptions.
SubmitResultWindow, CultNetSchemaWriteForwarderOptions.ResultWindow). A
cultnet.error.v0 inside it rejects with the new CultNetWriteRefusedException
(Code: write_not_authorized, message_invalid, ...) whatever its order against the
send or transport ack; a send returning no longer completes the write. The
forwarder's refusal now throws on the same path as an unreachable primary, so
CultNetDatabaseServer.TryForwardPutAsync/DeleteAsync no longer report a refused
forward as done. Raw send primitives (C# schema client/session, Python, Kotlin)
carry the doc line: fire-and-forget, an ack or a return is not admission, read the
result frame.

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

packages/cultmesh-kotlin/src/main/kotlin/org/gamecult/cultmesh/CultMesh.kt

@@ -808,6 +808,7 @@ interface CultNetSchemaMessageTransport : AutoCloseable {
     val profile: CultNetTransportProfile
     val stats: CultNetTransportStats
 
+    /** Fire-and-forget: an ack or a return is not admission; read the result frame. */
     fun sendSchemaMessage(message: CultNetMessage)
 
     fun receiveSchemaMessage(timeoutMs: Long = 1_000, pollIntervalMs: Long = 5): CultNetMessage?
@@ -3167,13 +3168,16 @@ class CultNetRudpSocketTransportConnection(
         return connected
     }
 
+    /** Fire-and-forget: an ack or a return is not admission; read the result frame. */
     fun send(channelId: String, payload: ByteArray) {
         session.sendMany(channelId, payload, channelSendOptions(channelId, nowMs()), maxFragmentBytes).forEach { sendPacket(it) }
         framesSent += 1
     }
 
+    /** Fire-and-forget: an ack or a return is not admission; read the result frame. */
     fun send(channelId: String, payload: String) = send(channelId, payload.toByteArray(StandardCharsets.UTF_8))
 
+    /** Fire-and-forget: an ack or a return is not admission; read the result frame. */
     fun sendSchema(payload: ByteArray) = send("schema", payload)
 
     fun sendSchema(payload: String) = sendSchema(payload.toByteArray(StandardCharsets.UTF_8))

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

@@ -59,6 +59,16 @@ from cultnet_py.cultmesh_contracts import (
 )
 
 _LOGGER = logging.getLogger(__name__)
+
+# How many consecutive refused frames of one kind a RUDP session may send before it is ended: undecodable
+# frames (reset by any frame that decodes) and deletes the write authority refuses (reset by any decoded
+# frame that is not a refused delete). Fewer are answered with a typed refusal and the session is kept; the
+# frame that reaches this number is answered too, then the session ends. The same number is the cap in every
+# runtime.
+MAX_CONSECUTIVE_REFUSED_FRAMES = 8
+
+# How long a session whose capped frame was answered waits for the peer to acknowledge the answer before it ends anyway.
+_RUDP_CAPPED_ANSWER_GRACE_MS = 2000
 _EPHEMERAL_PAIR_ATTEMPTS = 64
 
 
@@ -90,6 +100,10 @@ class _DatabaseSubscription:
 class _RudpPeerConnection:
     session: CultNetRudpSession
     subscriptions: dict[str, _DatabaseSubscription] = field(default_factory=dict)
+    undecodable_frames_in_a_row: int = 0
+    refused_deletes_in_a_row: int = 0
+    # Set once the peer's capped frame has been answered: the session ends when the answers are acknowledged.
+    ending_deadline_ms: int | None = None
 
 
 @dataclass
@@ -346,6 +360,7 @@ class CultMeshLocalServer:
             if packet.connection_id != self.rudp_connection_id:
                 continue
 
+            self._end_answered_peers(rudp_socket, peers)
             now_ms = _now_ms()
             peer = peers.get(remote_addr)
             if packet.packet_type == CultNetRudpPacketType.CONNECT:
@@ -394,18 +409,31 @@ class CultMeshLocalServer:
                 for delivered in result.delivered:
                     if delivered.channel_id != "schema":
                         continue
+                    if peer.ending_deadline_ms is not None:
+                        # The capped frame was answered; nothing after it is served while the answer is delivered.
+                        break
                     try:
                         message = msgpack.unpackb(delivered.payload, raw=False)
                     except (msgpack.ExtraData, msgpack.FormatError, msgpack.StackError, ValueError):
-                        continue
-                    if not isinstance(message, dict):
-                        continue
-                    responses = self._handle_connection_message(
-                        message,
-                        peer.subscriptions,
-                        over_rudp=True,
-                        peer=CultMeshServerPeer(f"{remote_addr[0]}:{remote_addr[1]}", "rudp"),
-                    )
+                        message = None
+                    if isinstance(message, dict):
+                        peer.undecodable_frames_in_a_row = 0
+                        responses = self._handle_connection_message(
+                            message,
+                            peer.subscriptions,
+                            over_rudp=True,
+                            peer=CultMeshServerPeer(f"{remote_addr[0]}:{remote_addr[1]}", "rudp"),
+                        )
+                        refused_delete = message.get("schemaVersion") == "cultnet.document_delete.v0" and any(
+                            response.get("code") == "write_not_authorized" for response in responses
+                        )
+                        peer.refused_deletes_in_a_row = peer.refused_deletes_in_a_row + 1 if refused_delete else 0
+                        refused, run = refused_delete, peer.refused_deletes_in_a_row
+                    else:
+                        # A frame that is no CultNet message is answered, not dropped, and the session kept.
+                        peer.undecodable_frames_in_a_row += 1
+                        refused, run = True, peer.undecodable_frames_in_a_row
+                        responses = [self._message_invalid({}, "message", "CultMesh RUDP server could not read the message.")]
                     for response in responses:
                         if unsendable is None:
                             unsendable = self._send_rudp_schema_frame(
@@ -415,6 +443,11 @@ class CultMeshLocalServer:
                                 msgpack.packb(response, use_bin_type=True),
                                 message_id=str(response.get("messageId") or ""),
                             )
+                    if refused and run >= MAX_CONSECUTIVE_REFUSED_FRAMES:
+                        # The capped frame is answered like any other, reliably. The session ends once the peer
+                        # has acknowledged every answer (or the grace runs out): see _end_answered_peers.
+                        peer.ending_deadline_ms = now_ms + _RUDP_CAPPED_ANSWER_GRACE_MS
+                        break
             except Exception as error:  # noqa: BLE001 - one peer's message never ends the thread
                 # Whatever a peer's message makes the handling raise is that peer's failure.
                 # The session recorded the frame's sequence, so it cannot be kept: a retransmit
@@ -445,11 +478,26 @@ class CultMeshLocalServer:
             if unsendable is not None:
                 self._end_unsendable_peer(rudp_socket, peers, remote_addr, unsendable)
 
+    def _end_answered_peers(
+        self,
+        rudp_socket: socket.socket,
+        peers: dict[tuple[str, int], _RudpPeerConnection],
+    ) -> None:
+        """Ends each session whose capped frame was answered once the peer has acknowledged every answer (a lost
+        answer is resent first), or when the grace has run out."""
+        now_ms = _now_ms()
+        for remote_addr, peer in list(peers.items()):
+            if peer.ending_deadline_ms is None:
+                continue
+            if peer.session.outstanding_reliable_packet_count == 0 or now_ms >= peer.ending_deadline_ms:
+                self._end_refused_peer(rudp_socket, peers, remote_addr)
+
     def _poll_rudp_resends(
         self,
         rudp_socket: socket.socket,
         peers: dict[tuple[str, int], _RudpPeerConnection],
     ) -> None:
+        self._end_answered_peers(rudp_socket, peers)
         now_ms = _now_ms()
         for remote_addr, peer in list(peers.items()):
             for packet in peer.session.due_resends(now_ms):

packages/cultmesh-py/tests/test_client_write_completion.py

@@ -0,0 +1,52 @@
+"""Heimdall's invariant: no client treats a reliable-transport ack as write admission. A put completes only on
+its result frame (write_not_authorized, or the success reply).
+
+The Python write path is a subscription over RUDP: `send` queues the put, `read_next` returns the result frame
+and raises the peer's refusal. The server here refuses the put; its reliable ack of the put arrives (the test
+waits for it) and the client still reports the refusal, never success. The transport's `receive` yields data
+frames only, so the order of the ack against the refusal cannot change what `read_next` reports.
+"""
+
+from __future__ import annotations
+
+import unittest
+
+from cultcache_py import define_database_entry_type
+from cultmesh_py import CultMesh, create_node
+from cultmesh_py.server import CultMeshLocalServer
+from cultnet_py import CultNetDatabaseSubscription, CultNetPeerError, create_rudp_schema_transport
+
+from test_remote_write_authority import KEY_A, NOTE, SCHEMA_ID, put_of
+
+
+class ClientWriteCompletionTests(unittest.TestCase):
+    def test_a_put_the_server_refused_is_reported_refused_after_its_ack_arrived(self) -> None:
+        node = create_node(runtime_id="completion-server")
+        node.database.register_document(NOTE)
+        server = CultMeshLocalServer(node=node, port=0, write_authority=None)
+        server.start()
+        try:
+            subscription = CultNetDatabaseSubscription(
+                host="127.0.0.1",
+                port=server.port,
+                timeout_seconds=4.0,
+                subscription_id="completion",
+                schema_ids=[SCHEMA_ID],
+                include_snapshot=False,
+                create_transport=lambda: create_rudp_schema_transport(
+                    host="127.0.0.1", port=server.port, connection_id=server.rudp_connection_id, timeout_seconds=4.0
+                ),
+            )
+            with subscription as sub:
+                sub.send(put_of(KEY_A, "peer"))
+                sub._transport.flush_reliable(2.0)  # the put's reliable ack has arrived
+                with self.assertRaises(CultNetPeerError) as raised:
+                    sub.read_next()
+        finally:
+            server.stop()
+        self.assertEqual(raised.exception.response["code"], "write_not_authorized")
+        self.assertIsNone(server.node.cache.get(NOTE, KEY_A))
+
+
+if __name__ == "__main__":
+    unittest.main()

packages/cultmesh-py/tests/test_cultmesh.py

@@ -24,7 +24,9 @@ from cultcache_py import (
     define_document_type,
 )
 from cultcache_py.interop import INTEROP_SCHEMA_VERSION, interop_note_document
+from cultmesh_py.server import MAX_CONSECUTIVE_REFUSED_FRAMES
 from cultmesh_py.verify import verify
+from cultmesh_py.write_authority import CultMeshRemoteWriteDecision
 from cultnet_py import (
     compute_simulation_claim_hash,
     CultNetClientAuthorityScope,
@@ -4889,6 +4891,140 @@ class CultMeshTests(unittest.TestCase):
         self.assertEqual(after["documents"], [])
         self.assertTrue(alive)
 
+    def _rudp_open(self, server: Any) -> Any:
+        return create_rudp_schema_transport(
+            host="127.0.0.1",
+            port=server.port,
+            connection_id=server.rudp_connection_id,
+            timeout_seconds=4.0,
+            max_fragment_bytes=1200,
+        )
+
+    def _rudp_answer(self, transport: Any, payload: bytes) -> dict[str, Any]:
+        transport.send("schema", payload)
+        return msgpack.unpackb(transport.receive(timeout_seconds=4.0).payload, raw=False)
+
+    def _assert_rudp_session_ended(self, transport: Any, server: Any) -> None:
+        transport.send("schema", msgpack.packb(snapshot_request(message_id="after-cap").to_wire(), use_bin_type=True))
+        with self.assertRaises(Exception):
+            transport.receive(timeout_seconds=1.0)
+        self.assertTrue(server._rudp_thread.is_alive())
+
+    def test_cultmesh_local_server_answers_unreadable_frames_and_ends_the_session_at_eight_in_a_row(self) -> None:
+        # An undecodable frame (not a msgpack map) is answered message_invalid and the session kept;
+        # eight in a row end it (the eighth is answered too); any decoded frame resets the run.
+        self.assertEqual(MAX_CONSECUTIVE_REFUSED_FRAMES, 8)
+        node = CultMesh.create_node(runtime_id="mesh-rudp-unreadable")
+        server = CultMesh.serve_node(node, write_authority=CultMeshGrantAllWrites())
+        garbage = b"\xc1"
+        snapshot = msgpack.packb(snapshot_request(message_id="ok").to_wire(), use_bin_type=True)
+        transport = self._rudp_open(server)
+        try:
+            for _ in range(2):
+                for _ in range(MAX_CONSECUTIVE_REFUSED_FRAMES - 1):
+                    reply = self._rudp_answer(transport, garbage)
+                    self.assertEqual(reply["code"], "message_invalid")
+                    self.assertEqual(reply["details"]["field"], "message")
+                self.assertEqual(self._rudp_answer(transport, snapshot)["messageId"], "ok")
+            for _ in range(MAX_CONSECUTIVE_REFUSED_FRAMES - 1):
+                self._rudp_answer(transport, garbage)
+            transport.send("schema", garbage)
+            eighth = msgpack.unpackb(transport.receive(timeout_seconds=4.0).payload, raw=False)
+            self.assertEqual(eighth["code"], "message_invalid")
+            self._assert_rudp_session_ended(transport, server)
+        finally:
+            transport.close()
+            server.stop()
+
+    def test_cultmesh_local_server_resends_a_lost_capped_answer_before_it_ends_the_session(self) -> None:
+        # The eighth refusal is answered reliably: the datagram carrying it is lost on the wire, and the
+        # peer still gets the answer (by resend) before the goodbye. An unreliable final answer, or a
+        # goodbye sent at once, would leave the peer without it.
+        node = CultMesh.create_node(runtime_id="mesh-rudp-lost-answer")
+        server = CultMesh.serve_node(node, write_authority=CultMeshGrantAllWrites())
+        send = server._send_rudp_packet
+        dropped: list[Any] = []
+        armed = threading.Event()
+
+        def lossy(rudp_socket: Any, remote_addr: Any, packet: Any) -> Any:
+            if armed.is_set() and not dropped and packet.packet_type == CultNetRudpPacketType.DATA and b"message_invalid" in (packet.payload or b""):
+                dropped.append(packet)
+                return None
+            return send(rudp_socket, remote_addr, packet)
+
+        server._send_rudp_packet = lossy
+        transport = self._rudp_open(server)
+        try:
+            for _ in range(MAX_CONSECUTIVE_REFUSED_FRAMES - 1):
+                self._rudp_answer(transport, b"\xc1")
+            armed.set()
+            reply = self._rudp_answer(transport, b"\xc1")
+            self.assertEqual(len(dropped), 1, "fixture: the first transmission of the last answer was lost")
+            self.assertEqual(reply["code"], "message_invalid")
+            self._assert_rudp_session_ended(transport, server)
+        finally:
+            transport.close()
+            server.stop()
+
+    def test_cultmesh_local_server_serves_nothing_after_the_capped_frame(self) -> None:
+        node = CultMesh.create_node(runtime_id="mesh-rudp-after-cap")
+        server = CultMesh.serve_node(node, write_authority=CultMeshGrantAllWrites())
+        transport = self._rudp_open(server)
+        try:
+            for _ in range(MAX_CONSECUTIVE_REFUSED_FRAMES - 1):
+                self._rudp_answer(transport, b"\xc1")
+            transport.send("schema", b"\xc1")
+            transport.send("schema", msgpack.packb(snapshot_request(message_id="behind-the-cap").to_wire(), use_bin_type=True))
+            first = msgpack.unpackb(transport.receive(timeout_seconds=4.0).payload, raw=False)
+            self.assertEqual(first["code"], "message_invalid")
+            with self.assertRaises(Exception):
+                transport.receive(timeout_seconds=1.0)
+        finally:
+            transport.close()
+            server.stop()
+
+    def test_cultmesh_local_server_ends_the_session_at_eight_refused_deletes_in_a_row(self) -> None:
+        class GrantOnlyKeyB:
+            def decide(self, request: Any) -> Any:
+                return CultMeshRemoteWriteDecision.grant() if request.record_key == "note:b" else CultMeshRemoteWriteDecision.refuse("no")
+
+        document = self._rudp_bound_note()
+        node = CultMesh.create_node(runtime_id="mesh-rudp-deletes")
+        node.database.register_document(document)
+        server = CultMesh.serve_node(node, write_authority=GrantOnlyKeyB())
+
+        def delete(key: str) -> bytes:
+            return msgpack.packb({
+                "schemaVersion": "cultnet.document_delete.v0",
+                "messageId": f"del-{key}",
+                "schemaId": document.schema_id,
+                "recordKey": key,
+            }, use_bin_type=True)
+
+        snapshot = msgpack.packb(snapshot_request(message_id="ok").to_wire(), use_bin_type=True)
+        transport = self._rudp_open(server)
+        try:
+            def refused_run(count: int) -> None:
+                for _ in range(count):
+                    reply = self._rudp_answer(transport, delete("note:a"))
+                    self.assertEqual(reply["code"], "write_not_authorized")
+                    self.assertEqual(reply["details"]["value"], "note:a")
+
+            refused_run(MAX_CONSECUTIVE_REFUSED_FRAMES - 1)
+            # A granted delete resets the run (nothing is stored under it, so the store answers that).
+            self.assertEqual(self._rudp_answer(transport, delete("note:b"))["code"], "unregistered_document_delete")
+            refused_run(MAX_CONSECUTIVE_REFUSED_FRAMES - 1)
+            # So does any decoded non-delete frame.
+            self.assertEqual(self._rudp_answer(transport, snapshot)["messageId"], "ok")
+            refused_run(MAX_CONSECUTIVE_REFUSED_FRAMES - 1)
+            transport.send("schema", delete("note:a"))
+            eighth = msgpack.unpackb(transport.receive(timeout_seconds=4.0).payload, raw=False)
+            self.assertEqual(eighth["code"], "write_not_authorized")
+            self._assert_rudp_session_ended(transport, server)
+        finally:
+            transport.close()
+            server.stop()
+
     def test_cultmesh_local_server_refuses_a_put_its_store_cannot_take_and_serves_on(self) -> None:
         # Well-formed puts the store refuses: a blank key, a payload that is not MessagePack, a
         # payload its schema cannot decode, a global document under another key. Each is refused

packages/cultmesh-rs/src/lib.rs

@@ -82,7 +82,10 @@ pub struct CultMeshRudpDocumentPublishOptions {
     pub target: SocketAddr,
     pub runtime_id: String,
     pub connect_timeout: Duration,
+    /// How long the acknowledgement of every put may take.
     pub flush_timeout: Duration,
+    /// How long after the acknowledgement a refusal may still arrive; a success frame ends the wait early.
+    pub result_window: Duration,
     pub poll_interval: Duration,
     pub resend_delay_ms: u64,
     pub source_agent_id: Option<String>,
@@ -149,6 +152,7 @@ impl Default for CultMeshRudpDocumentPublishOptions {
             runtime_id: "cultmesh-rudp-document-publisher".to_string(),
             connect_timeout: Duration::from_secs(3),
             flush_timeout: Duration::from_millis(300),
+            result_window: Duration::from_millis(500),
             poll_interval: Duration::from_millis(10),
             resend_delay_ms: 50,
             source_agent_id: None,
@@ -882,77 +886,30 @@ fn unix_millis() -> u128 {
 
 /// Publish one raw CultNet schema message to a RUDP catalog.
 ///
-/// Each call owns a fresh connection epoch and returns only after the catalog
-/// acknowledges reliable application admission. Against a Rust CultMesh
-/// catalog, which never acknowledges a message it refuses, a refusal is
-/// returned as an error, naming the refusal when it is one of the fixed texts
-/// such a catalog sends. A catalog that acknowledges a message on receipt and
-/// refuses it afterwards (the TypeScript document server does) is reported as
-/// having admitted it: the acknowledgement arrives first.
+/// A put completes on the catalog's result frame, never on the transport
+/// acknowledgement: an ack proves delivery, not admission. A refusal (a
+/// `cultnet.error.v0` frame) returns an error in either order against the ack,
+/// naming the refusal when it is one of the fixed texts a catalog sends. A
+/// catalog that admits a put sends no success frame, so after the ack the call
+/// waits `result_window` for a refusal and reports the put as admitted when
+/// none comes (interim until the catalog sends a success frame; a success
+/// frame ends the wait early). Each call owns a fresh connection epoch.
 pub fn publish_cultnet_message_to_rudp_catalog(
     message: &cultnet_rs::CultNetMessage,
     options: CultMeshRudpDocumentPublishOptions,
 ) -> Result<()> {
-    if options.runtime_id.trim().is_empty() {
-        anyhow::bail!("runtime_id must be non-empty");
-    }
-    let mut client = connect_rudp_client(
-        options.target,
-        &options.runtime_id,
-        options.connect_timeout,
-        options.poll_interval,
-        options.resend_delay_ms,
-        "cultmesh-rudp-document-publisher",
-    )?;
-
-    let payload = encode_cultnet_message_to_vec(message, CultNetWireContract::CultNetSchemaV0)?;
-    let receipt = client.send_reliable("schema", payload)?;
-    let flush_deadline = Instant::now() + options.flush_timeout;
-    while Instant::now() < flush_deadline {
-        if let Some(frame) = client.receive_once()? {
-            if frame.channel_id == "schema" {
-                if let Ok(cultnet_rs::CultNetMessage::Error { error, .. }) =
-                    decode_cultnet_message_from_slice(
-                        &frame.payload,
-                        CultNetWireContract::CultNetSchemaV0,
-                    )
-                {
-                    let _ = client.disconnect(b"document-refused".to_vec());
-                    anyhow::bail!(
-                        "CultMesh RUDP catalog {} refused the put{}",
-                        options.target,
-                        recognised_refusal(&error)
-                    );
-                }
-            }
-        }
-        match client.reliable_send_status(&receipt) {
-            cultnet_rs::CultNetRudpReliableSendStatus::Acknowledged => {
-                let _ = client.disconnect(b"document-acknowledged".to_vec());
-                return Ok(());
-            }
-            cultnet_rs::CultNetRudpReliableSendStatus::Invalidated => {
-                anyhow::bail!("CultMesh RUDP document receipt was invalidated")
-            }
-            cultnet_rs::CultNetRudpReliableSendStatus::Pending => {}
-        }
-        client.poll_resends()?;
-        thread::sleep(options.poll_interval);
-    }
-    let _ = client.disconnect(b"document-unacknowledged".to_vec());
-    anyhow::bail!(
-        "timed out waiting for reliable acknowledgement from CultMesh RUDP catalog {}",
-        options.target
-    )
+    publish_cultnet_messages_to_rudp_catalog(std::slice::from_ref(message), options)
 }
 
-/// Publish a batch of raw CultNet schema messages to a RUDP catalog.
+/// Publish a batch of raw CultNet schema messages to a RUDP catalog on one
+/// connection.
 ///
-/// Unlike [`publish_cultnet_message_to_rudp_catalog`], which polls for
-/// per-message application-level acknowledgement, this sends every message on
-/// one connection and waits only for a transport-level `flush_reliable`. It
-/// trades per-receipt confirmation for batch throughput; callers that need a
-/// stronger delivery guarantee should use the singular publish path instead.
+/// Completes under the same rule as [`publish_cultnet_message_to_rudp_catalog`]:
+/// every message acknowledged, then no refusal frame within `result_window`
+/// (or one success frame per message). A refusal of any message in the batch
+/// fails the whole call; the batch cannot hide it. The catalog's answers do not
+/// name the message they answer, so a failed batch does not say which message
+/// was refused or whether the others were admitted.
 pub fn publish_cultnet_messages_to_rudp_catalog(
     messages: &[cultnet_rs::CultNetMessage],
     options: CultMeshRudpDocumentPublishOptions,
@@ -969,11 +926,62 @@ pub fn publish_cultnet_messages_to_rudp_catalog(
         "cultmesh-rudp-document-publisher",
     )?;
 
+    let mut receipts = Vec::with_capacity(messages.len());
     for message in messages {
         let payload = encode_cultnet_message_to_vec(message, CultNetWireContract::CultNetSchemaV0)?;
-        client.send("schema", payload)?;
+        receipts.push(client.send_reliable("schema", payload)?);
+    }
+    let ack_deadline = Instant::now() + options.flush_timeout;
+    let mut acknowledged_at: Option<Instant> = None;
+    let mut successes = 0;
+    loop {
+        while let Some(frame) = client.receive_once()? {
+            if frame.channel_id != "schema" {
+                continue;
+            }
+            match decode_cultnet_message_from_slice(&frame.payload, CultNetWireContract::CultNetSchemaV0) {
+                Ok(cultnet_rs::CultNetMessage::Error { error, .. }) => {
+                    let _ = client.disconnect(b"document-refused".to_vec());
+                    anyhow::bail!(
+                        "CultMesh RUDP catalog {} refused the put{}",
+                        options.target,
+                        recognised_refusal(&error)
+                    );
+                }
+                Ok(_) => successes += 1,
+                Err(_) => {}
+            }
+        }
+        if acknowledged_at.is_none() {
+            let mut all_acknowledged = true;
+            for receipt in &receipts {
+                match client.reliable_send_status(receipt) {
+                    cultnet_rs::CultNetRudpReliableSendStatus::Acknowledged => {}
+                    cultnet_rs::CultNetRudpReliableSendStatus::Invalidated => {
+                        anyhow::bail!("CultMesh RUDP document receipt was invalidated")
+                    }
+                    cultnet_rs::CultNetRudpReliableSendStatus::Pending => all_acknowledged = false,
+                }
+            }
+            if all_acknowledged {
+                acknowledged_at = Some(Instant::now());
+            } else if Instant::now() >= ack_deadline {
+                let _ = client.disconnect(b"document-unacknowledged".to_vec());
+                anyhow::bail!(
+                    "timed out waiting for reliable acknowledgement from CultMesh RUDP catalog {}",
+                    options.target
+                )
+            }
+        }
+        if let Some(acknowledged_at) = acknowledged_at {
+            if successes >= messages.len() || acknowledged_at.elapsed() >= options.result_window {
+                let _ = client.disconnect(b"document-completed".to_vec());
+                return Ok(());
+            }
+        }
+        client.poll_resends()?;
+        thread::sleep(options.poll_interval);
     }
-    client.flush_reliable(options.flush_timeout)
 }
 
 fn connect_rudp_client(
@@ -1451,6 +1459,141 @@ mod tests {
         Ok((target, stop, catalog))
     }
 
+    /// A catalog that acknowledges every reliable message it receives and refuses it with a typed
+    /// write_not_authorized, the acknowledgement either before the refusal or after it (the second packet
+    /// follows the first by 100 ms). The refusal carries no acknowledgement of its own.
+    /// What the stub catalog sends for a put, and in which order against the transport ack.
+    #[derive(Clone, Copy, PartialEq, Eq)]
+    enum CatalogReply {
+        RefusalThenAck,
+        AckThenRefusal,
+        AckThenSuccess,
+    }
+
+    fn acknowledging_refusing_catalog(
+        reply: CatalogReply,
+    ) -> Result<(
+        SocketAddr,
+        std::sync::Arc<std::sync::atomic::AtomicBool>,
+        thread::JoinHandle<Result<()>>,
+    )> {
+        use cultnet_rs::{
+            CultNetRudpPacketType, CultNetRudpSendOptions, CultNetRudpSession,
+            CultNetRudpSessionOptions, decode_rudp_packet, encode_rudp_packet,
+        };
+        let socket = UdpSocket::bind("127.0.0.1:0")?;
+        socket.set_read_timeout(Some(Duration::from_millis(5)))?;
+        let target = socket.local_addr()?;
+        let stop = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
+        let catalog_stop = stop.clone();
+        let catalog = thread::spawn(move || -> Result<()> {
+            let mut session: Option<CultNetRudpSession> = None;
+            let mut wire = vec![0_u8; 65_535];
+            while !catalog_stop.load(std::sync::atomic::Ordering::SeqCst) {
+                let Ok((bytes, peer)) = socket.recv_from(&mut wire) else {
+                    continue;
+                };
+                let Ok(packet) = decode_rudp_packet(&wire[..bytes]) else {
+                    continue;
+                };
+                if packet.packet_type == CultNetRudpPacketType::Connect {
+                    let mut accepted = CultNetRudpSession::new(CultNetRudpSessionOptions {
+                        connection_id: packet.connection_id,
+                        initial_sequence: None,
+                        resend_delay_ms: 10_000,
+                        max_pending_reliable_packets: Some(64),
+                    });
+                    let accept = accepted.accept_connect(&packet, 0, Vec::new())?;
+                    socket.send_to(&encode_rudp_packet(&accept)?, peer)?;
+                    session = Some(accepted);
+                    continue;
+                }
+                let Some(session) = session.as_mut() else {
+                    continue;
+                };
+                let Ok(received) = session.receive(&packet, 1) else {
+                    continue;
+                };
+                if !packet.reliable || received.delivered.is_empty() {
+                    continue;
+                }
+                let ack = session.create_ack_for_received(packet.sequence);
+                let refusal = encode_cultnet_message_to_vec(
+                    &if reply == CatalogReply::AckThenSuccess {
+                        cultnet_rs::CultNetMessage::SnapshotResponseRaw {
+                            message_id: "completion".into(),
+                            documents: Vec::new(),
+                        }
+                    } else {
+                        cultnet_rs::CultNetMessage::Error {
+                            error: "write_not_authorized: the catalog refused the document".into(),
+                            code: Some(cultnet_rs::CultNetErrorCode::WriteNotAuthorized),
+                            details: None,
+                        }
+                    },
+                    CultNetWireContract::CultNetSchemaV0,
+                )?;
+                let options = CultNetRudpSendOptions {
+                    reliable: false,
+                    ordered: false,
+                    sequenced: false,
+                    now_ms: 2,
+                    reliable_expire_after_ms: None,
+                };
+                let mut refusals = session.send_many("schema", refusal, options, Some(1200))?;
+                for refusal in &mut refusals {
+                    refusal.ack = 0;
+                    refusal.ack_mask = 0;
+                }
+                let send_refusals = |socket: &UdpSocket| -> Result<()> {
+                    for refusal in &refusals {
+                        socket.send_to(&encode_rudp_packet(refusal)?, peer)?;
+                    }
+                    Ok(())
+                };
+                if reply == CatalogReply::RefusalThenAck {
+                    send_refusals(&socket)?;
+                    thread::sleep(Duration::from_millis(100));
+                    socket.send_to(&encode_rudp_packet(&ack)?, peer)?;
+                } else {
+                    socket.send_to(&encode_rudp_packet(&ack)?, peer)?;
+                    thread::sleep(Duration::from_millis(100));
+                    send_refusals(&socket)?;
+                }
+            }
+            Ok(())
+        });
+        Ok((target, stop, catalog))
+    }
+
+    fn publish_options(target: SocketAddr) -> CultMeshRudpDocumentPublishOptions {
+        CultMeshRudpDocumentPublishOptions {
+            target,
+            connect_timeout: Duration::from_secs(2),
+            flush_timeout: Duration::from_secs(3),
+            poll_interval: Duration::from_millis(2),
+            resend_delay_ms: 5,
+            ..CultMeshRudpDocumentPublishOptions::default()
+        }
+    }
+
+    fn refused_note() -> cultnet_rs::CultNetMessage {
+        cultnet_rs::CultNetMessage::DocumentPutRaw {
+            message_id: "completion".into(),
+            document: cultnet_rs::CultNetRawDocumentRecord {
+                schema_id: "cultmesh.test.raw.v0".into(),
+                record_key: "tests:completion:a".into(),
+                stored_at: "2026-11-08T00:00:00Z".into(),
+                payload_encoding: cultnet_rs::CultNetRawPayloadEncoding::Messagepack,
+                payload: vec![0xc0],
+                source_runtime_id: None,
+                source_agent_id: None,
+                source_role: None,
+                tags: None,
+            },
+        }
+    }
+
     /// A catalog's refusal text is quoted only when it is one of the fixed
     /// refusals a CultMesh catalog sends. Any other text, which could carry
     /// anything, stays out of the publisher's error.
@@ -1901,4 +2044,63 @@ mod tests {
         assert_eq!(lease.bytes(), &[1, 2, 3]);
         Ok(())
     }
+
+    // Heimdall's invariant: no client treats a reliable-transport ack as write admission. The catalog here
+    // acknowledges the put and refuses it, the refusal arriving before or after the ack.
+    #[test]
+    fn a_publish_whose_refusal_arrives_before_the_ack_is_refused() -> Result<()> {
+        let (target, stop, catalog) = acknowledging_refusing_catalog(CatalogReply::RefusalThenAck)?;
+        let result = publish_cultnet_message_to_rudp_catalog(&refused_note(), publish_options(target));
+        stop.store(true, std::sync::atomic::Ordering::SeqCst);
+        catalog.join().expect("catalog thread should not panic")?;
+        let error = result.expect_err("a refused put is never reported as admitted");
+        assert!(error.to_string().contains("refused the put"), "{error:#}");
+        Ok(())
+    }
+
+
+    #[test]
+    fn a_publish_whose_ack_arrives_before_the_refusal_is_refused() -> Result<()> {
+        let (target, stop, catalog) = acknowledging_refusing_catalog(CatalogReply::AckThenRefusal)?;
+        let result = publish_cultnet_message_to_rudp_catalog(&refused_note(), publish_options(target));
+        stop.store(true, std::sync::atomic::Ordering::SeqCst);
+        catalog.join().expect("catalog thread should not panic")?;
+        let error = result.expect_err("a refused put is never reported as admitted");
+        assert!(error.to_string().contains("refused the put"), "{error:#}");
+        Ok(())
+    }
+
+    #[test]
+    fn a_batch_publish_is_not_complete_on_the_ack_of_refused_puts() -> Result<()> {
+        for reply in [CatalogReply::AckThenRefusal, CatalogReply::RefusalThenAck] {
+            let (target, stop, catalog) = acknowledging_refusing_catalog(reply)?;
+            let result = publish_cultnet_messages_to_rudp_catalog(&[refused_note()], publish_options(target));
+            stop.store(true, std::sync::atomic::Ordering::SeqCst);
+            catalog.join().expect("catalog thread should not panic")?;
+            assert!(result.is_err(), "a refused put was reported as published");
+        }
+        Ok(())
+    }
+
+    #[test]
+    fn a_success_frame_completes_the_publish_without_waiting_out_the_result_window() -> Result<()> {
+        for batch in [false, true] {
+            let (target, stop, catalog) = acknowledging_refusing_catalog(CatalogReply::AckThenSuccess)?;
+            let options = CultMeshRudpDocumentPublishOptions {
+                result_window: Duration::from_secs(30),
+                ..publish_options(target)
+            };
+            let started = Instant::now();
+            let result = if batch {
+                publish_cultnet_messages_to_rudp_catalog(&[refused_note()], options)
+            } else {
+                publish_cultnet_message_to_rudp_catalog(&refused_note(), options)
+            };
+            stop.store(true, std::sync::atomic::Ordering::SeqCst);
+            catalog.join().expect("catalog thread should not panic")?;
+            result?;
+            assert!(started.elapsed() < Duration::from_secs(10), "batch={batch}: waited out the window");
+        }
+        Ok(())
+    }
 }

packages/cultmesh-rs/src/rudp_document_server.rs

@@ -19,6 +19,16 @@ const MAX_FRAGMENTS_PER_MESSAGE: usize = u16::MAX as usize;
 /// can carry no shorter id, so no response to it can be smaller.
 const SHORTEST_MESSAGE_ID: &str = "0";
 
+/// How many consecutive refused frames of one kind a session may send before it is ended: undecodable
+/// frames (reset by any frame that decodes) and refused deletes (reset by any decoded non-delete).
+/// Fewer are answered with a typed refusal and the session is kept; the cap that reaches this number
+/// is answered too, then the session ends. The same number is the cap in every runtime.
+pub const MAX_CONSECUTIVE_REFUSED_FRAMES: u32 = 8;
+
+/// How long a session whose capped frame was answered waits for the peer to acknowledge the answer
+/// before it ends anyway.
+const CAPPED_ANSWER_GRACE_MILLIS: u64 = 2_000;
+
 /// The transport identity of one remote CultNet RUDP session.
 ///
 /// Connection identifiers are scoped to a remote socket address. Two clients
@@ -196,7 +206,10 @@ pub struct CultMeshRudpMaintenance {
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 pub enum CultMeshRudpApplicationOperation {
     DocumentPutRaw,
+    DocumentDelete,
     SnapshotRequest,
+    /// A schema frame that did not decode as any CultNet message; there is no message id to name.
+    Unreadable,
 }
 
 /// Why the server refused an application message.
@@ -241,12 +254,18 @@ pub enum CultMeshRudpRejectionReason {
     /// (`send_error_code`). The session has ended with a goodbye carrying the
     /// same name, and no refusal: the server knows the peer cannot be reached.
     ResponseSendFailed(&'static str),
+    /// The server has no delete path, so every delete is refused. Nothing reached the sink.
+    DeleteNotSupported,
+    /// A schema frame did not decode as a CultNet message. Carries the decoder's error text.
+    MessageUnreadable(String),
 }
 
 impl std::fmt::Display for CultMeshRudpRejectionReason {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         match self {
             Self::SinkRefused(error) | Self::SnapshotSourceFailed(error) => f.write_str(error),
+            Self::DeleteNotSupported => f.write_str("the server does not take deletes"),
+            Self::MessageUnreadable(error) => write!(f, "message could not be read: {error}"),
             Self::DocumentUnservable {
                 response_bytes,
                 max_snapshot_response_bytes,
@@ -287,7 +306,7 @@ impl std::fmt::Display for CultMeshRudpRejectionReason {
 /// The texts a refused peer is sent, one per rejection variant. They are fixed:
 /// a sink's or source's error text can quote what a peer sent, so it stays with
 /// the caller that receives the rejection and never reaches the wire.
-pub(crate) const REFUSAL_TEXTS: [&str; 9] = [
+pub(crate) const REFUSAL_TEXTS: [&str; 11] = [
     "the catalog refused the document",
     "the document can never be served",
     "the snapshot source failed",
@@ -297,6 +316,8 @@ pub(crate) const REFUSAL_TEXTS: [&str; 9] = [
     "the retained payload budget is full",
     "the snapshot response could not be queued",
     "the snapshot response could not be sent",
+    "the server does not take deletes",
+    "the message could not be read",
 ];
 
 impl CultMeshRudpRejectionReason {
@@ -312,6 +333,8 @@ impl CultMeshRudpRejectionReason {
             Self::PayloadBudgetFull => 6,
             Self::ResponseQueueFailed(_) => 7,
             Self::ResponseSendFailed(_) => 8,
+            Self::DeleteNotSupported => 9,
+            Self::MessageUnreadable(_) => 10,
         }]
     }
 }
@@ -331,25 +354,46 @@ pub(crate) fn write_refused_message(record_key: &str, reason: &str) -> CultNetMe
     }
 }
 
+/// The typed `message_invalid` refusal naming the request `field` at fault, as C#
+/// `CultNetErrorMessage.ForMessageInvalid` builds it.
+pub(crate) fn message_invalid_message(field: &str, reason: &str) -> CultNetMessage {
+    CultNetMessage::Error {
+        error: format!("message_invalid: {reason}"),
+        code: Some(CultNetErrorCode::MessageInvalid),
+        details: Some(CultNetErrorDetails {
+            field: Some(field.into()),
+            value: None,
+            as_of: None,
+            current: None,
+        }),
+    }
+}
+
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub struct CultMeshRudpApplicationRejection {
     pub session: CultMeshRudpSessionKey,
     pub operation: CultMeshRudpApplicationOperation,
     pub message_id: String,
-    /// The record a rejected put concerned; `None` for a snapshot request. A sink refusal names it to the peer.
+    /// The record a rejected put or delete concerned; `None` for a snapshot request or an unreadable frame.
+    /// A sink or delete refusal names it to the peer.
     pub record_key: Option<String>,
     pub reason: CultMeshRudpRejectionReason,
 }
 
 impl CultMeshRudpApplicationRejection {
-    /// What the refused peer is sent: a sink's refusal of a put is a typed `write_not_authorized` naming
-    /// the record; every other rejection is the variant's fixed text, untyped. Neither carries the
-    /// sink's or source's own text.
+    /// What the refused peer is sent: a refused put or delete is a typed `write_not_authorized` naming
+    /// the record, an unreadable frame a typed `message_invalid` naming `message`; every other rejection
+    /// is the variant's fixed text, untyped. None carries the sink's, source's or decoder's own text.
     fn refusal_message(&self) -> CultNetMessage {
         let text = self.reason.refusal_text();
         match (&self.reason, &self.record_key) {
-            (CultMeshRudpRejectionReason::SinkRefused(_), Some(record_key)) => {
-                write_refused_message(record_key, text)
+            (
+                CultMeshRudpRejectionReason::SinkRefused(_)
+                | CultMeshRudpRejectionReason::DeleteNotSupported,
+                Some(record_key),
+            ) => write_refused_message(record_key, text),
+            (CultMeshRudpRejectionReason::MessageUnreadable(_), _) => {
+                message_invalid_message("message", text)
             }
             _ => CultNetMessage::Error {
                 error: text.into(),
@@ -382,6 +426,11 @@ struct SessionEntry {
     created_at_monotonic_millis: u64,
     last_activity_monotonic_millis: u64,
     admitted_payload_bytes: usize,
+    undecodable_frames_in_a_row: u32,
+    refused_deletes_in_a_row: u32,
+    /// Set once the peer's capped frame has been answered: the session ends when the answer is acknowledged,
+    /// or at this time.
+    ending_deadline_monotonic_millis: Option<u64>,
 }
 
 /// A synchronous, multi-session CultNet RUDP raw-document server.
@@ -556,7 +605,18 @@ where
         };
 
         let mut unsendable = match result.reply {
-            Some(reply) => self.send_packet(key.remote_addr, &reply)?,
+            Some(mut reply) => {
+                // A reply (a Pong) in the grace after the capped frame acknowledges nothing either.
+                if self
+                    .sessions
+                    .get(&key)
+                    .is_some_and(|entry| entry.ending_deadline_monotonic_millis.is_some())
+                {
+                    reply.ack = 0;
+                    reply.ack_mask = 0;
+                }
+                self.send_packet(key.remote_addr, &reply)?
+            }
             None => None,
         };
 
@@ -565,24 +625,125 @@ where
             return Ok(CultMeshRudpPollOutcome::Handled);
         }
 
+        let mut kept_rejection = None;
         for frame in result.delivered {
             if frame.channel_id != "schema" {
                 continue;
             }
-            let message = match decode_cultnet_message_from_slice(
+            // The capped frame was answered: nothing after it is served while the answer is delivered.
+            if self
+                .sessions
+                .get(&key)
+                .is_some_and(|entry| entry.ending_deadline_monotonic_millis.is_some())
+            {
+                break;
+            }
+            let decoded = decode_cultnet_message_from_slice(
                 &frame.payload,
                 CultNetWireContract::CultNetSchemaV0,
-            ) {
-                Ok(message) => message,
-                Err(_) => continue,
+            );
+            // Runs of refusals this frame extends or breaks. A session is kept through a short run and
+            // ended by `MAX_CONSECUTIVE_REFUSED_FRAMES` in a row.
+            let (is_unreadable, is_delete) = (
+                decoded.is_err(),
+                matches!(decoded, Ok(CultNetMessage::DocumentDelete { .. })),
+            );
+            let run = self.sessions.get_mut(&key).map(|entry| {
+                entry.undecodable_frames_in_a_row = if is_unreadable {
+                    entry.undecodable_frames_in_a_row + 1
+                } else {
+                    0
+                };
+                if is_delete {
+                    entry.refused_deletes_in_a_row += 1;
+                } else if !is_unreadable {
+                    entry.refused_deletes_in_a_row = 0;
+                }
+                if is_unreadable {
+                    entry.undecodable_frames_in_a_row
+                } else {
+                    entry.refused_deletes_in_a_row
+                }
+            });
+            let rejection = match decoded {
+                Ok(message) => self.deliver_application_message(
+                    key,
+                    frame.sequence,
+                    message,
+                    now_unix,
+                    now_monotonic,
+                )?,
+                Err(error) => Some(CultMeshRudpApplicationRejection {
+                    session: key,
+                    operation: CultMeshRudpApplicationOperation::Unreadable,
+                    message_id: String::new(),
+                    record_key: None,
+                    reason: CultMeshRudpRejectionReason::MessageUnreadable(format!("{error:#}")),
+                }),
             };
-            if let Some(rejection) = self.deliver_application_message(
-                key,
-                frame.sequence,
-                message,
-                now_unix,
-                now_monotonic,
-            )? {
+            if let Some(rejection) = rejection {
+                // A refused delete or unreadable frame below the cap is answered on the live session,
+                // which goes on: the peer keeps the connection a stray frame did not break.
+                if (is_unreadable || is_delete)
+                    && run.is_some_and(|run| run < MAX_CONSECUTIVE_REFUSED_FRAMES)
+                {
+                    let refusal = rejection.refusal_message();
+                    let packets = match self.sessions.remove(&key) {
+                        Some(mut entry) => {
+                            let packets = self.refusal_packets(&mut entry.session, &refusal, false);
+                            self.sessions.insert(key, entry);
+                            packets
+                        }
+                        None => Vec::new(),
+                    };
+                    for packet in &packets {
+                        if let Some(error) = self.send_packet(key.remote_addr, packet)? {
+                            unsendable = Some(error);
+                            break;
+                        }
+                    }
+                    kept_rejection = Some(rejection);
+                    continue;
+                }
+                // The capped frame is answered reliably, so a lost datagram is resent; the session ends once
+                // the peer has acknowledged the answer or the grace runs out (`end_answered_sessions`). When
+                // the answer cannot be queued, the session ends at once with the refusal as before.
+                if (is_unreadable || is_delete)
+                    && run.is_some()
+                    && !matches!(
+                        rejection.reason,
+                        CultMeshRudpRejectionReason::ResponseSendFailed(_)
+                    )
+                {
+                    let refusal = rejection.refusal_message();
+                    let packets = match self.sessions.remove(&key) {
+                        Some(mut entry) => {
+                            let packets = self.refusal_packets(&mut entry.session, &refusal, true);
+                            if !packets.is_empty() {
+                                entry.ending_deadline_monotonic_millis =
+                                    Some(now_monotonic.saturating_add(CAPPED_ANSWER_GRACE_MILLIS));
+                            }
+                            self.sessions.insert(key, entry);
+                            packets
+                        }
+                        None => Vec::new(),
+                    };
+                    if packets.is_empty() {
+                        self.end_session(key, SessionEnd::Rejected(refusal))?;
+                        return Ok(CultMeshRudpPollOutcome::ApplicationRejected(rejection));
+                    }
+                    for mut packet in packets {
+                        // Acknowledges nothing, as every refusal: the refused frame was not admitted.
+                        packet.ack = 0;
+                        packet.ack_mask = 0;
+                        if let Some(error) = self.send_packet(key.remote_addr, &packet)? {
+                            unsendable = Some(error);
+                            break;
+                        }
+                    }
+                    kept_rejection = Some(rejection);
+                    break;
+                }
                 // The only reply `receive` returns is a Pong, which delivers no
                 // frame, so no earlier send of this poll can precede a rejection.
                 // A response that could not be sent ended its session where the
@@ -597,7 +758,11 @@ where
             }
         }
 
-        if packet.reliable && unsendable.is_none() {
+        let answering_capped_frame = self
+            .sessions
+            .get(&key)
+            .is_some_and(|entry| entry.ending_deadline_monotonic_millis.is_some());
+        if packet.reliable && unsendable.is_none() && !answering_capped_frame {
             let ack = self
                 .sessions
                 .get_mut(&key)
@@ -609,7 +774,11 @@ where
         if let Some(error) = unsendable {
             self.end_session(key, SessionEnd::Unsendable(error))?;
         }
-        Ok(CultMeshRudpPollOutcome::Handled)
+        self.end_answered_sessions()?;
+        Ok(match kept_rejection {
+            Some(rejection) => CultMeshRudpPollOutcome::ApplicationRejected(rejection),
+            None => CultMeshRudpPollOutcome::Handled,
+        })
     }
 
     /// Expire idle sessions and resend reliable packets whose deadline passed.
@@ -623,10 +792,16 @@ where
                 && now.saturating_sub(entry.created_at_monotonic_millis) <= lifetime_ms
         });
         let expired = before - self.sessions.len();
+        self.end_answered_sessions()?;
 
         let mut resends = Vec::new();
         for (key, entry) in &mut self.sessions {
-            for packet in entry.session.due_resends(now) {
+            for mut packet in entry.session.due_resends(now) {
+                if entry.ending_deadline_monotonic_millis.is_some() {
+                    // The resent answer to a capped frame acknowledges nothing either.
+                    packet.ack = 0;
+                    packet.ack_mask = 0;
+                }
                 resends.push((*key, packet));
             }
         }
@@ -645,6 +820,26 @@ where
         })
     }
 
+    /// Ends each session whose capped frame was answered once the peer has acknowledged every answer (a lost
+    /// answer is resent first), or when the grace has run out.
+    fn end_answered_sessions(&mut self) -> Result<()> {
+        let now = self.clock.now_monotonic_millis();
+        let due: Vec<CultMeshRudpSessionKey> = self
+            .sessions
+            .iter()
+            .filter(|(_, entry)| {
+                entry.ending_deadline_monotonic_millis.is_some_and(|deadline| {
+                    entry.session.outstanding_reliable_packet_count() == 0 || now >= deadline
+                })
+            })
+            .map(|(key, _)| *key)
+            .collect();
+        for key in due {
+            self.end_session(key, SessionEnd::Refused)?;
+        }
+        Ok(())
+    }
+
     /// `Ok(false)` when the Connect was refused because the session cap is full;
     /// only a socket failure is an `Err`.
     fn accept_connection(
@@ -694,6 +889,9 @@ where
                         created_at_monotonic_millis: now,
                         last_activity_monotonic_millis: now,
                         admitted_payload_bytes: 0,
+                        undecodable_frames_in_a_row: 0,
+                        refused_deletes_in_a_row: 0,
+                        ending_deadline_monotonic_millis: None,
                     },
                 );
                 accept
@@ -898,6 +1096,19 @@ where
                     }
                 }
             }
+            CultNetMessage::DocumentDelete {
+                message_id,
+                record_key,
+                ..
+            } => {
+                return Ok(Some(CultMeshRudpApplicationRejection {
+                    session: key,
+                    operation: CultMeshRudpApplicationOperation::DocumentDelete,
+                    message_id,
+                    record_key: Some(record_key),
+                    reason: CultMeshRudpRejectionReason::DeleteNotSupported,
+                }));
+            }
             _ => {}
         }
         Ok(None)
@@ -920,7 +1131,7 @@ where
         };
         let (refusal, mut unsendable) = match end {
             SessionEnd::Refused => (Vec::new(), None),
-            SessionEnd::Rejected(refusal) => (self.refusal_packets(&mut entry.session, &refusal), None),
+            SessionEnd::Rejected(refusal) => (self.refusal_packets(&mut entry.session, &refusal, false), None),
             SessionEnd::Unsendable(error) => (Vec::new(), Some(error)),
         };
         entry.session.reset_peer_state();
@@ -943,11 +1154,12 @@ where
         Ok(())
     }
 
-    /// The refusal a rejected peer is sent, as unreliable, unordered packets.
+    /// The refusal a rejected peer is sent: unreliable and unordered, or (`reliable`) queued like any response.
     fn refusal_packets(
         &self,
         session: &mut CultNetRudpSession,
         refusal: &CultNetMessage,
+        reliable: bool,
     ) -> Vec<CultNetRudpPacket> {
         encode_cultnet_message_to_vec(refusal, CultNetWireContract::CultNetSchemaV0)
             .and_then(|payload| {
@@ -955,8 +1167,8 @@ where
                     "schema",
                     payload,
                     CultNetRudpSendOptions {
-                        reliable: false,
-                        ordered: false,
+                        reliable,
+                        ordered: reliable,
                         sequenced: false,
                         now_ms: self.clock.now_monotonic_millis(),
                         reliable_expire_after_ms: None,
@@ -1129,6 +1341,9 @@ mod tests {
     // (`ForRemoteWriteRefused(request for "some:key", "not granted")`) and pinned by C#, TS and Python too.
     const WRITE_NOT_AUTHORIZED_HEX: &str = "85AD736368656D6156657273696F6EB063756C746E65742E6572726F722E7630A56572726F72D92177726974655F6E6F745F617574686F72697A65643A206E6F74206772616E746564AB726F7574696E6748696E74C0A4636F6465B477726974655F6E6F745F617574686F72697A6564A764657461696C7384A56669656C64A97265636F72644B6579A576616C7565A8736F6D653A6B6579A461734F66C0A763757272656E74C0";
 
+    // The same file's `message_invalid` vector: field `document`, reason "Raw put messages must contain a document map."
+    const MESSAGE_INVALID_HEX: &str = "85AD736368656D6156657273696F6EB063756C746E65742E6572726F722E7630A56572726F72D93E6D6573736167655F696E76616C69643A2052617720707574206D65737361676573206D75737420636F6E7461696E206120646F63756D656E74206D61702EAB726F7574696E6748696E74C0A4636F6465AF6D6573736167655F696E76616C6964A764657461696C7384A56669656C64A8646F63756D656E74A576616C7565C0A461734F66C0A763757272656E74C0";
+
     fn hex(bytes: &[u8]) -> String {
         bytes.iter().map(|byte| format!("{byte:02X}")).collect()
     }
@@ -1142,4 +1357,14 @@ mod tests {
         .unwrap();
         assert_eq!(hex(&bytes), WRITE_NOT_AUTHORIZED_HEX);
     }
+
+    #[test]
+    fn the_unreadable_refusal_is_the_pinned_message_invalid_bytes() {
+        let bytes = encode_cultnet_message_to_vec(
+            &message_invalid_message("document", "Raw put messages must contain a document map."),
+            CultNetWireContract::CultNetSchemaV0,
+        )
+        .unwrap();
+        assert_eq!(hex(&bytes), MESSAGE_INVALID_HEX);
+    }
 }

packages/cultmesh-rs/tests/rudp_document_server.rs

@@ -343,14 +343,17 @@ fn refusal_seen_by(
     client: &mut CultNetRudpSocketTransportConnection,
     receipt: &CultNetRudpReliableSendReceipt,
 ) -> Result<String> {
-    let (error, code, details) = typed_refusal_seen_by(client, receipt)?;
+    let (error, code, details) = typed_refusal_seen_by(client, receipt, None)?;
     assert_eq!((code, details), (None, None));
     Ok(error)
 }
 
+/// `server` is polled while the goodbye is awaited when the refusal ends its session only once the peer has
+/// acknowledged it (the capped frame's answer).
 fn typed_refusal_seen_by(
     client: &mut CultNetRudpSocketTransportConnection,
     receipt: &CultNetRudpReliableSendReceipt,
+    mut server: Option<&mut Server>,
 ) -> Result<(String, Option<CultNetErrorCode>, Option<CultNetErrorDetails>)> {
     let mut refusal = None;
     for _ in 0..2_000 {
@@ -381,6 +384,9 @@ fn typed_refusal_seen_by(
         "a refusal must not acknowledge the refused message"
     );
     for _ in 0..2_000 {
+        if let Some(server) = server.as_deref_mut() {
+            server.poll_once()?;
+        }
         client.receive_once()?;
         if !client.connected() {
             break;
@@ -681,7 +687,7 @@ fn application_rejection_is_nonfatal_peer_scoped_refused_to_the_peer_and_unackno
     // The caller keeps the sink's text; the peer is sent fixed text, never it.
     // A refused put is typed write_not_authorized and names the record; nothing of it reached the sink.
     assert_eq!(
-        typed_refusal_seen_by(&mut publisher, &publish_receipt)?,
+        typed_refusal_seen_by(&mut publisher, &publish_receipt, None)?,
         (
             "write_not_authorized: the catalog refused the document".to_string(),
             Some(CultNetErrorCode::WriteNotAuthorized),
@@ -751,6 +757,173 @@ fn application_rejection_is_nonfatal_peer_scoped_refused_to_the_peer_and_unackno
     Ok(())
 }
 
+/// Sends `payload` reliably, polls until the server refuses it, and returns the typed refusal the
+/// peer is sent. The session is not asserted here: the caller says whether it lives.
+fn refusal_on_live_session(
+    server: &mut CultMeshRudpDocumentServer<Sink, Source, Clock>,
+    client: &mut CultNetRudpSocketTransportConnection,
+    payload: Vec<u8>,
+) -> Result<(Option<CultNetErrorCode>, Option<CultNetErrorDetails>, String)> {
+    client.send_reliable("schema", payload)?;
+    let mut rejected = false;
+    for _ in 0..2_000 {
+        if !rejected {
+            rejected = matches!(server.poll_once()?, CultMeshRudpPollOutcome::ApplicationRejected(_));
+        }
+        if let Some(frame) = client.receive_once()? {
+            if let CultNetMessage::Error { error, code, details } = decode_cultnet_message_from_slice(
+                &frame.payload,
+                CultNetWireContract::CultNetSchemaV0,
+            )? {
+                assert!(rejected, "the refusal preceded the server's own outcome");
+                return Ok((code, details, error));
+            }
+        }
+        thread::sleep(Duration::from_millis(1));
+    }
+    anyhow::bail!("no typed refusal reached the peer")
+}
+
+fn garbage() -> Vec<u8> {
+    let mut payload = vec![0xAE];
+    payload.extend_from_slice(b"CANARY-garbage");
+    payload
+}
+
+fn delete_of(key: &str) -> Vec<u8> {
+    encode_cultnet_message_to_vec(
+        &CultNetMessage::DocumentDelete {
+            message_id: "del".into(),
+            schema_id: "sha256:note".into(),
+            record_key: key.into(),
+        },
+        CultNetWireContract::CultNetSchemaV0,
+    )
+    .unwrap()
+}
+
+fn valid_put(key: &str) -> Vec<u8> {
+    encode_cultnet_message_to_vec(
+        &CultNetMessage::DocumentPutRaw {
+            message_id: format!("put-{key}"),
+            document: document(key, vec![1, 2, 3]),
+        },
+        CultNetWireContract::CultNetSchemaV0,
+    )
+    .unwrap()
+}
+
+fn stored_after_poll(
+    server: &mut CultMeshRudpDocumentServer<Sink, Source, Clock>,
+    client: &mut CultNetRudpSocketTransportConnection,
+    sink: &Sink,
+    receipts: usize,
+) -> Result<()> {
+    client.send_reliable("schema", valid_put(&format!("k{receipts}")))?;
+    for _ in 0..2_000 {
+        server.poll_once()?;
+        client.receive_once()?;
+        if sink.0.lock().unwrap().receipts.len() >= receipts {
+            return Ok(());
+        }
+        thread::sleep(Duration::from_millis(1));
+    }
+    anyhow::bail!("a valid put on the kept session was not stored")
+}
+
+#[test]
+fn an_unreadable_frame_is_refused_typed_and_the_session_kept_until_eight_in_a_row() -> Result<()> {
+    let sink = Sink::default();
+    let mut server = server(Default::default(), Clock::new(49_000), sink.clone(), Source::default())?;
+    let target = server.local_addr()?;
+    let mut garbler = client(target, 111)?;
+    let mut survivor = client(target, 113)?;
+    connect(&mut server, &mut [&mut garbler, &mut survivor])?;
+    let message_invalid = (
+        Some(CultNetErrorCode::MessageInvalid),
+        Some(CultNetErrorDetails {
+            field: Some("message".into()),
+            value: None,
+            as_of: None,
+            current: None,
+        }),
+        "message_invalid: the message could not be read".to_string(),
+    );
+
+    // Seven in a row are answered and the session lives, a valid frame resets the run, and seven more
+    // are answered again: a counter that never resets, or a cap below eight, ends the session early.
+    for _ in 0..7 {
+        assert_eq!(refusal_on_live_session(&mut server, &mut garbler, garbage())?, message_invalid);
+        assert!(garbler.connected());
+    }
+    stored_after_poll(&mut server, &mut garbler, &sink, 1)?;
+    for _ in 0..7 {
+        assert_eq!(refusal_on_live_session(&mut server, &mut garbler, garbage())?, message_invalid);
+        assert!(garbler.connected());
+    }
+    assert_eq!(server.session_count(), 2);
+
+    // The eighth consecutive one is answered too, then the session ends with a goodbye.
+    let receipt = garbler.send_reliable("schema", garbage())?;
+    let CultMeshRudpPollOutcome::ApplicationRejected(rejection) = server.poll_once()? else {
+        panic!("the eighth consecutive unreadable frame must be refused");
+    };
+    assert_eq!(rejection.operation, CultMeshRudpApplicationOperation::Unreadable);
+    let (error, code, _) = typed_refusal_seen_by(&mut garbler, &receipt, Some(&mut server))?;
+    assert_eq!((error.as_str(), code), (message_invalid.2.as_str(), message_invalid.0));
+    drain_until_idle(&mut server)?;
+    assert_eq!(server.session_count(), 1, "only the capped peer's session ended");
+    assert_eq!(sink.0.lock().unwrap().receipts.len(), 1);
+    Ok(())
+}
+
+#[test]
+fn a_refused_delete_keeps_the_session_until_eight_in_a_row() -> Result<()> {
+    let sink = Sink::default();
+    let mut server = server(Default::default(), Clock::new(50_000), sink.clone(), Source::default())?;
+    let target = server.local_addr()?;
+    let mut deleter = client(target, 112)?;
+    connect(&mut server, &mut [&mut deleter])?;
+    let refused = |key: &str| {
+        (
+            Some(CultNetErrorCode::WriteNotAuthorized),
+            Some(CultNetErrorDetails {
+                field: Some("recordKey".into()),
+                value: Some(key.into()),
+                as_of: None,
+                current: None,
+            }),
+            "write_not_authorized: the server does not take deletes".to_string(),
+        )
+    };
+    for _ in 0..7 {
+        assert_eq!(refusal_on_live_session(&mut server, &mut deleter, delete_of("note:victim"))?, refused("note:victim"));
+        assert!(deleter.connected());
+    }
+    // A decoded non-delete frame resets the run.
+    stored_after_poll(&mut server, &mut deleter, &sink, 1)?;
+    for _ in 0..7 {
+        assert_eq!(refusal_on_live_session(&mut server, &mut deleter, delete_of("note:victim"))?, refused("note:victim"));
+    }
+    let receipt = deleter.send_reliable("schema", delete_of("note:victim"))?;
+    let CultMeshRudpPollOutcome::ApplicationRejected(rejection) = server.poll_once()? else {
+        panic!("the eighth consecutive delete must be refused");
+    };
+    assert_eq!(rejection.operation, CultMeshRudpApplicationOperation::DocumentDelete);
+    assert_eq!(rejection.reason, CultMeshRudpRejectionReason::DeleteNotSupported);
+    let (error, code, details) = typed_refusal_seen_by(&mut deleter, &receipt, Some(&mut server))?;
+    assert_eq!(
+        (error, code, details),
+        {
+            let (c, d, e) = refused("note:victim");
+            (e, c, d)
+        }
+    );
+    drain_until_idle(&mut server)?;
+    assert_eq!(server.session_count(), 0);
+    Ok(())
+}
+
 #[test]
 fn snapshot_response_resends_until_acknowledged() -> Result<()> {
     let clock = Clock::new(51_000);
@@ -1456,6 +1629,11 @@ impl RawPeer {
     /// Sends `message` reliably and in order; returns its sequences.
     fn send(&mut self, message: &CultNetMessage) -> Result<Vec<u32>> {
         let payload = encode_cultnet_message_to_vec(message, CultNetWireContract::CultNetSchemaV0)?;
+        self.send_payload(payload)
+    }
+
+    /// Sends `payload` as one reliable, ordered schema frame; returns its sequences.
+    fn send_payload(&mut self, payload: Vec<u8>) -> Result<Vec<u32>> {
         let options = cultnet_rs::CultNetRudpSendOptions {
             reliable: true,
             ordered: true,
@@ -1743,3 +1921,138 @@ fn a_response_that_is_too_large_to_send_is_named_emsgsize() -> Result<()> {
     );
     Ok(())
 }
+
+/// A peer on `server` that has sent seven unreadable frames, each answered and processed, and then the
+/// eighth, whose answer is waiting at its socket. Returns the peer and the eighth frame's sequence.
+fn peer_at_the_cap(server: &mut Server, connection_id: u32) -> Result<(RawPeer, u32)> {
+    let mut peer = RawPeer::connect(server, connection_id)?;
+    for _ in 0..7 {
+        peer.send_payload(garbage())?;
+        assert!(matches!(
+            server.poll_once()?,
+            CultMeshRudpPollOutcome::ApplicationRejected(_)
+        ));
+        for packet in peer.drain() {
+            peer.session.receive(&packet, 3)?;
+        }
+    }
+    let capped = peer.send_payload(garbage())?[0];
+    assert!(matches!(
+        server.poll_once()?,
+        CultMeshRudpPollOutcome::ApplicationRejected(_)
+    ));
+    Ok((peer, capped))
+}
+
+fn is_goodbye(packet: &cultnet_rs::CultNetRudpPacket) -> bool {
+    packet.packet_type == cultnet_rs::CultNetRudpPacketType::Disconnect
+}
+
+fn is_answer(packet: &cultnet_rs::CultNetRudpPacket) -> bool {
+    packet.packet_type == cultnet_rs::CultNetRudpPacketType::Data && packet.reliable
+}
+
+/// The capped frame's answer is reliable: its datagram is lost, the server resends it, and only once the
+/// peer has acknowledged it does the goodbye follow. Neither transmission acknowledges the capped frame.
+#[test]
+fn a_lost_capped_answer_is_resent_and_the_goodbye_waits_for_its_acknowledgement() -> Result<()> {
+    let clock = Clock::new(81_000);
+    let mut server = server(Default::default(), clock.clone(), Sink::default(), Source::default())?;
+    let (mut peer, capped) = peer_at_the_cap(&mut server, 81)?;
+
+    let first = peer.drain();
+    assert!(first.iter().any(is_answer), "fixture: the answer was sent, and is lost here");
+    assert!(!first.iter().any(is_goodbye), "no goodbye before the answer is acknowledged");
+    assert!(
+        !first.iter().any(|packet| acknowledges(packet, capped)),
+        "nothing sent for the capped frame acknowledges it"
+    );
+    assert_eq!(server.session_count(), 1);
+
+    clock.set_monotonic(81_000 + 60);
+    server.maintain()?;
+    let resent = peer.drain();
+    let answer = resent.iter().find(|packet| is_answer(packet)).expect("the lost answer is resent");
+    assert!(!acknowledges(answer, capped), "the resent answer acknowledges nothing the peer sent");
+    assert!(!resent.iter().any(is_goodbye), "still no goodbye");
+    let delivered = peer.session.receive(answer, 4)?.delivered;
+    let Ok(CultNetMessage::Error { code, .. }) =
+        decode_cultnet_message_from_slice(&delivered[0].payload, CultNetWireContract::CultNetSchemaV0)
+    else {
+        panic!("the resent answer is a typed refusal");
+    };
+    assert_eq!(code, Some(CultNetErrorCode::MessageInvalid));
+
+    peer.send_packet(&peer.session.create_ack_for(answer.sequence))?;
+    assert_eq!(server.poll_once()?, CultMeshRudpPollOutcome::Handled);
+    assert!(peer.drain().iter().any(is_goodbye), "acknowledged: the goodbye follows");
+    assert_eq!(server.session_count(), 0);
+    Ok(())
+}
+
+/// A peer that never acknowledges the answer is not kept for ever: the session ends after the grace.
+#[test]
+fn a_capped_session_whose_answer_is_never_acknowledged_ends_after_the_grace() -> Result<()> {
+    let clock = Clock::new(82_000);
+    let mut server = server(Default::default(), clock.clone(), Sink::default(), Source::default())?;
+    let (peer, _) = peer_at_the_cap(&mut server, 82)?;
+    peer.drain();
+
+    clock.set_monotonic(82_000 + 1_999);
+    server.maintain()?;
+    assert!(!peer.drain().iter().any(is_goodbye), "inside the grace");
+    assert_eq!(server.session_count(), 1);
+    clock.set_monotonic(82_000 + 2_000);
+    server.maintain()?;
+    assert!(peer.drain().iter().any(is_goodbye), "the grace ran out");
+    assert_eq!(server.session_count(), 0);
+    Ok(())
+}
+
+/// Nothing behind the capped frame is served while its answer is delivered.
+#[test]
+fn nothing_behind_the_capped_frame_is_served() -> Result<()> {
+    let sink = Sink::default();
+    let mut server = server(Default::default(), Clock::new(83_000), sink.clone(), Source::default())?;
+    let (mut peer, _) = peer_at_the_cap(&mut server, 83)?;
+    peer.send_payload(valid_put("behind-the-cap"))?;
+    for _ in 0..3 {
+        server.poll_once()?;
+    }
+    assert!(sink.0.lock().unwrap().receipts.is_empty(), "the put behind the cap was not stored");
+    Ok(())
+}
+
+/// A Ping inside the grace is answered with a Pong that acknowledges nothing: its ack fields are built from
+/// what the session received, which includes the capped frame.
+#[test]
+fn a_pong_in_the_grace_after_the_cap_acknowledges_nothing() -> Result<()> {
+    let mut server = server(Default::default(), Clock::new(84_000), Sink::default(), Source::default())?;
+    let (mut peer, capped) = peer_at_the_cap(&mut server, 84)?;
+    peer.drain();
+
+    let ping = peer.session.create_ping(b"alive".to_vec());
+    peer.send_packet(&ping)?;
+    assert_eq!(server.poll_once()?, CultMeshRudpPollOutcome::Handled);
+    let pong = peer
+        .drain()
+        .into_iter()
+        .find(|packet| packet.packet_type == cultnet_rs::CultNetRudpPacketType::Pong)
+        .expect("the ping is answered");
+    assert_eq!(pong.payload, b"alive");
+    assert!(!acknowledges(&pong, capped), "the Pong acknowledges the capped frame");
+    assert_eq!((pong.ack, pong.ack_mask), (0, 0));
+    Ok(())
+}
+
+/// The grace deadline saturates: at the end of the clock the answer is not given up on at once.
+#[test]
+fn the_grace_deadline_does_not_wrap_at_the_end_of_the_clock() -> Result<()> {
+    let mut server = server(Default::default(), Clock::new(u64::MAX - 10), Sink::default(), Source::default())?;
+    let (peer, _) = peer_at_the_cap(&mut server, 85)?;
+    peer.drain();
+    server.maintain()?;
+    assert!(!peer.drain().iter().any(is_goodbye), "the deadline wrapped and the session ended before its grace");
+    assert_eq!(server.session_count(), 1);
+    Ok(())
+}

packages/cultmesh-ts/CHANGELOG.md

@@ -16,6 +16,7 @@ All notable changes to `@gamecult/cultmesh-ts` are recorded here, following
 
 ### Added
 
+- `createRudpDocumentServer` ends a session after `CULTMESH_RUDP_DOCUMENT_SERVER_MAX_CONSECUTIVE_REFUSED_FRAMES` (8) consecutive undecodable frames or consecutive deletes the write authority refuses; fewer are answered and the session is kept. Any decoded frame resets the undecodable run, any decoded non-refused-delete frame resets the delete run. The same cap holds in every runtime.
 - `CultMeshCatalogValue<TDocument, TSchema>`, exported.
 - `documentById` and `tryDocumentById` on `CultMeshDocumentCatalog`, which address one document by its document id when a catalog holds several documents of one type.
 

packages/cultmesh-ts/src/index.ts

@@ -4121,6 +4121,16 @@ export interface CultMeshRudpDocumentServerOptions extends CultMeshRudpSocketOpt
 /** A write authority that has not answered in this long has refused the write. */
 export const CULTMESH_RUDP_DOCUMENT_SERVER_DEFAULT_DECIDE_TIMEOUT_MS = 5_000;
 
+/**
+ * How many consecutive refused frames of one kind a `createRudpDocumentServer` session may send before it is
+ * ended: undecodable frames (reset by any frame that decodes) and deletes the write authority refuses (reset by any
+ * decoded frame that is not a refused delete). Fewer are answered with a typed refusal and the session is kept; the
+ * frame that reaches this number is answered too, then the session ends. The same number is the cap in every runtime.
+ */
+export const CULTMESH_RUDP_DOCUMENT_SERVER_MAX_CONSECUTIVE_REFUSED_FRAMES = 8;
+/** How long a session whose capped frame was answered waits for the peer to acknowledge the answers before it ends anyway. */
+const CULTMESH_RUDP_DOCUMENT_SERVER_CAPPED_ANSWER_GRACE_MS = 2000;
+
 /** Replies larger than one fragment are split into fragments of at most this many bytes. */
 export const CULTMESH_RUDP_DOCUMENT_SERVER_DEFAULT_MAX_FRAGMENT_BYTES = 1200;
 
@@ -4173,7 +4183,29 @@ export interface CultMeshRudpDocumentServer {
   close(): void;
 }
 
+/**
+ * A write the peer refused: the typed `cultnet.error.v0` it answered with (`write_not_authorized`,
+ * `message_invalid`, ...), whichever of the transport ack and the result frame came first.
+ */
+export class CultMeshRudpWriteRefusedError extends Error {
+  readonly code: string | undefined;
+  readonly response: Extract<CultNetMessage, { schemaVersion: "cultnet.error.v0" }>;
+  constructor(response: Extract<CultNetMessage, { schemaVersion: "cultnet.error.v0" }>) {
+    super(response.error);
+    this.name = "CultMeshRudpWriteRefusedError";
+    this.code = response.code ?? undefined;
+    this.response = response;
+  }
+}
+
 export interface CultMeshRudpDocumentPublishOptions extends CultMeshRudpPeerOptions {
+  /**
+   * How long after the put's transport ack a refusal may still arrive. A reliable-transport ack is not
+   * admission; the outcome is the peer's result frame. A server that answers a put only on refusal sends no
+   * success frame, so the put is reported published once this window passes with no refusal; a success reply
+   * ends the wait early. Default 500 ms.
+   */
+  resultWindowMs?: number;
   messageId?: string;
   sourceRuntimeId?: string;
   sourceAgentId?: string;
@@ -5537,6 +5569,11 @@ export class CultMesh {
       closed: boolean;
       /** A packet it owed could never be sent, so the session ended. */
       unsendable: boolean;
+      /** Consecutive undecodable frames, and consecutive deletes the authority refused. */
+      unreadableInARow: number;
+      refusedDeletesInARow: number;
+      /** Set once the peer's capped frame is answered: the session ends when the answers are acknowledged, or at this time. */
+      endingDeadlineMs?: number;
       work: Promise<void>;
     };
     const sessions = new Map<string, SessionRecord>();
@@ -5591,6 +5628,8 @@ export class CultMesh {
             connectPayload,
             closed: false,
             unsendable: false,
+            unreadableInARow: 0,
+            refusedDeletesInARow: 0,
             session: new CultNetRudpSession({
               connectionId,
               initialSequence: options.initialSequence,
@@ -5655,7 +5694,8 @@ export class CultMesh {
       record: SessionRecord,
       payload: Uint8Array,
     ): Promise<void> {
-      if (record.closed) return;
+      // Nothing after the capped frame is served while its answer is delivered.
+      if (record.closed || record.endingDeadlineMs !== undefined) return;
       let decoded: unknown;
       let message: CultNetMessage;
       try {
@@ -5665,12 +5705,21 @@ export class CultMesh {
         // A frame the server cannot read is judged as the write it looks like (a put when it does not even decode), so
         // an ungranted peer hears the same refusal for it as for any other write, not a parse error.
         const write = readableWrite(decoded);
-        if (decoded !== undefined && !write) throw error;
-        if (await writeRefused(record, write?.operation ?? "put", write?.schemaId ?? "", write?.recordKey ?? "", null, null)) return;
-        reportError(error);
-        sendSchemaMessage(record, cultNetMessageInvalidError("message", "CultMesh RUDP document server could not read the write."));
+        record.unreadableInARow += 1;
+        if (decoded !== undefined && !write) {
+          // It decodes but is no message the parser takes (an unknown schemaVersion, a non-map) and is no write the
+          // authority could judge: answered and counted like any other unreadable frame.
+          reportError(error);
+          sendSchemaMessage(record, cultNetMessageInvalidError("message", "CultMesh RUDP document server could not read the message."));
+        } else if (!await writeRefused(record, write?.operation ?? "put", write?.schemaId ?? "", write?.recordKey ?? "", null, null)) {
+          reportError(error);
+          sendSchemaMessage(record, cultNetMessageInvalidError("message", "CultMesh RUDP document server could not read the write."));
+        }
+        if (record.unreadableInARow >= CULTMESH_RUDP_DOCUMENT_SERVER_MAX_CONSECUTIVE_REFUSED_FRAMES) endCappedSession(record);
         return;
       }
+      record.unreadableInARow = 0;
+      if (message.schemaVersion !== "cultnet.document_delete.v0") record.refusedDeletesInARow = 0;
       switch (message.schemaVersion) {
         case "cultnet.snapshot_request.v0": {
           if (!options.getCache) {
@@ -5744,7 +5793,12 @@ export class CultMesh {
           }
           return;
         case "cultnet.document_delete.v0": {
-          if (await writeRefused(record, "delete", message.schemaId, message.recordKey, null, null)) return;
+          if (await writeRefused(record, "delete", message.schemaId, message.recordKey, null, null)) {
+            record.refusedDeletesInARow += 1;
+            if (record.refusedDeletesInARow >= CULTMESH_RUDP_DOCUMENT_SERVER_MAX_CONSECUTIVE_REFUSED_FRAMES) endCappedSession(record);
+            return;
+          }
+          record.refusedDeletesInARow = 0;
           if (!options.onDocumentDeleteRaw) {
             sendSchemaMessage(record, cultNetTextError("CultMesh RUDP document server has no delete handler."));
             return;
@@ -5895,6 +5949,17 @@ export class CultMesh {
       sendRudpDatagram(socket, goodbye, record.remote.port, record.remote.address, () => {});
     }
 
+    /**
+     * A session that reached the cap of consecutive refused frames: its last refusal went out like any reply
+     * (reliable, so a lost datagram is resent), and the session ends once the peer has acknowledged every reply or
+     * the grace runs out. The resend timer ends it.
+     */
+    function endCappedSession(record: SessionRecord): void {
+      const key = `${record.remote.address}:${record.remote.port}`;
+      if (sessions.get(key) !== record) return;
+      record.endingDeadlineMs ??= Date.now() + CULTMESH_RUDP_DOCUMENT_SERVER_CAPPED_ANSWER_GRACE_MS;
+    }
+
     /** The session owes its peer a packet that can never be sent as built. */
     function endUnsendableSession(record: SessionRecord, error: Error): void {
       if (record.unsendable) return;
@@ -6028,6 +6093,11 @@ export class CultMesh {
               closeSession(record);
               continue;
             }
+            if (record.endingDeadlineMs !== undefined
+              && (record.session.outstandingReliablePacketCount === 0 || nowMs >= record.endingDeadlineMs)) {
+              endRefusedSession(key, record);
+              continue;
+            }
             for (const packet of record.session.dueResends(nowMs)) {
               sendPacket(record, packet);
             }
@@ -6109,11 +6179,23 @@ export class CultMesh {
     requireNonEmpty(runtimeId, "runtimeId");
     requireNonEmpty(recordKey, "recordKey");
     const registry = new CultNetDocumentRegistry([binding]);
+    const messageId = options.messageId ?? `${runtimeId}:${binding.definition.type}:${recordKey}`;
     const peer = await CultMesh.createRudpPeer(runtimeId, connectionId, endpoint, options);
     try {
+      // A transport ack is not admission: the put completes on its result frame. A refusal rejects with the
+      // typed error whether it arrives before the ack, after it, or inside the result window.
+      let refusal: CultMeshRudpWriteRefusedError | undefined;
+      let succeeded = false;
+      let wake: (() => void) | undefined;
+      peer.on("message", (message: CultNetMessage) => {
+        if (message.schemaVersion === "cultnet.error.v0") refusal ??= new CultMeshRudpWriteRefusedError(message);
+        else if ("messageId" in message && message.messageId === messageId) succeeded = true;
+        else return;
+        wake?.();
+      });
       peer.send(registry.createRawDocumentPutMessage(
         binding,
-        options.messageId ?? `${runtimeId}:${binding.definition.type}:${recordKey}`,
+        messageId,
         recordKey,
         value,
         {
@@ -6124,6 +6206,13 @@ export class CultMesh {
         },
       ));
       await peer.flush(options.flushTimeoutMs ?? 30_000);
+      if (!refusal && !succeeded) {
+        await new Promise<void>((resolve) => {
+          const timer = setTimeout(resolve, options.resultWindowMs ?? 500);
+          wake = () => { clearTimeout(timer); resolve(); };
+        });
+      }
+      if (refusal) throw refusal;
     } finally {
       peer.close();
     }
@@ -6142,6 +6231,7 @@ export class CultMesh {
       const receipt = new Promise<CultCacheDocumentValue<TReceiptDefinition>>((resolve, reject) => {
         const timeout = setTimeout(() => { cleanup(); reject(new Error(`Timed out waiting for RUDP application receipt for ${recordKey}.`)); }, options.receiptTimeoutMs ?? 5_000);
         const onMessage = (message: CultNetMessage): void => {
+          if (message.schemaVersion === "cultnet.error.v0") { cleanup(); reject(new CultMeshRudpWriteRefusedError(message)); return; }
           if (message.schemaVersion !== "cultnet.document_put_raw.v0" || message.messageId !== messageId || message.document.schemaId !== receiptBinding.definition.schemaId) return;
           try {
             const decoded = decodeDocumentPayload(receiptBinding.definition, message.document.payload);

packages/cultmesh-ts/test/client-write-completion.test.ts

@@ -0,0 +1,117 @@
+// Heimdall's invariant: no client may treat a reliable-transport ack as write admission. A put completes only
+// on its result frame (write_not_authorized, or the success reply). Each test serves a client a peer that
+// acknowledges the put and refuses it, the ack arriving before the refusal and after it.
+import assert from "node:assert/strict";
+import { createSocket, type Socket } from "node:dgram";
+import test from "node:test";
+import { z } from "zod";
+import { defineDocumentType } from "@gamecult/cultcache-ts";
+import {
+  CultNetDocumentRegistry,
+  CultNetRudpSession,
+  cultNetWriteNotAuthorizedError,
+  decodeRudpPacket,
+  defineCultNetDocumentBinding,
+  encodeCultNetMessageForWire,
+  encodeRudpPacket,
+} from "@gamecult/cultnet-ts";
+import { encodeWireValue } from "@gamecult/cultnet-ts/wire";
+import { CultMesh, CultMeshGrantAllWrites, CultMeshRudpWriteRefusedError } from "../src/index";
+
+const isTheRefusal = (error: unknown): boolean =>
+  error instanceof CultMeshRudpWriteRefusedError && error.code === "write_not_authorized";
+
+const note = defineDocumentType({
+  type: "cultmesh.completion_note",
+  schemaId: "cultmesh.completion_note.v0",
+  schema: z.object({ noteId: z.string(), text: z.string() }),
+  name: "noteId",
+});
+const binding = defineCultNetDocumentBinding({ definition: note });
+const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));
+
+type Order = "ack-then-refusal" | "refusal-then-ack";
+
+/** A peer that acknowledges every reliable put and refuses it with write_not_authorized, in the given order. */
+async function refusingPeer(connectionId: number, order: Order) {
+  const socket: Socket = createSocket("udp4");
+  await new Promise<void>((resolve) => socket.bind(0, "127.0.0.1", resolve));
+  const session = new CultNetRudpSession({ connectionId, resendDelayMs: 50 });
+  const timers = new Set<NodeJS.Timeout>();
+  let closed = false;
+  const send = (packet: Parameters<typeof encodeRudpPacket>[0], port: number, address: string) => {
+    if (!closed) socket.send(encodeRudpPacket(packet), port, address);
+  };
+  const later = (action: () => void) => { const timer = setTimeout(() => { timers.delete(timer); action(); }, 120); timers.add(timer); };
+  socket.on("message", (wire, remote) => {
+    const packet = decodeRudpPacket(wire);
+    const now = Date.now();
+    if (packet.packetType === "connect") {
+      send(session.acceptConnect(packet, now), remote.port, remote.address);
+      return;
+    }
+    const result = session.receive(packet, now);
+    if (result.reply) send(result.reply, remote.port, remote.address);
+    for (const frame of result.delivered) {
+      if (frame.channelId !== "schema") continue;
+      const ack = session.createAckForReceived(packet.sequence);
+      const refusal = session.sendMany(
+        "schema",
+        encodeWireValue(encodeCultNetMessageForWire(cultNetWriteNotAuthorizedError("tests:completion:a"), "cultnet.schema.v0")),
+        { reliable: true, ordered: true, nowMs: now },
+      );
+      if (order === "ack-then-refusal") {
+        send(ack, remote.port, remote.address);
+        later(() => refusal.forEach((p) => send(p, remote.port, remote.address)));
+      } else {
+        // The refusal carries no acknowledgement of its own; the ack follows it.
+        refusal.forEach((p) => send({ ...p, ack: 0, ackMask: 0 }, remote.port, remote.address));
+        later(() => send(ack, remote.port, remote.address));
+      }
+    }
+    if (packet.reliable) send(session.createAckForReceived(packet.sequence), remote.port, remote.address);
+  });
+  return { endpoint: `rudp://127.0.0.1:${socket.address().port}`, close: () => { closed = true; timers.forEach(clearTimeout); socket.close(); } };
+}
+
+const orders: Order[] = ["ack-then-refusal", "refusal-then-ack"];
+let nextConnection = 0x10205000;
+
+for (const order of orders) {
+  test(`publishRudpDocumentOnce does not complete on the ack of a refused put (${order})`, async () => {
+    const peer = await refusingPeer(nextConnection, order);
+    try {
+      await assert.rejects(
+        CultMesh.publishRudpDocumentOnce("completion-once", nextConnection++, peer.endpoint, binding, "tests:completion:a", { noteId: "a", text: "x" }, { resendDelayMs: 50, resendPollMs: 5, flushTimeoutMs: 2_000 }),
+        isTheRefusal,
+      );
+    } finally { peer.close(); }
+  });
+
+  test(`publishRudpDocumentAndWaitForReceipt reports a refused put as refused, never as success (${order})`, async () => {
+    const peer = await refusingPeer(nextConnection, order);
+    try {
+      await assert.rejects(
+        CultMesh.publishRudpDocumentAndWaitForReceipt("completion-receipt", nextConnection++, peer.endpoint, binding, "tests:completion:a", { noteId: "a", text: "x" }, binding, { resendDelayMs: 50, resendPollMs: 5, receiptTimeoutMs: 800 }),
+        isTheRefusal,
+      );
+    } finally { peer.close(); }
+    await sleep(10);
+  });
+}
+
+test("publishRudpDocumentOnce still publishes a put the server admitted", async () => {
+  const connectionId = nextConnection++;
+  const puts: string[] = [];
+  const server = CultMesh.createRudpDocumentServer("completion-admit", connectionId, {
+    documents: new CultNetDocumentRegistry([binding]),
+    writeAuthority: new CultMeshGrantAllWrites(),
+    bindHost: "127.0.0.1", bindPort: 0, resendDelayMs: 50, resendPollMs: 5,
+    onDocumentPutRaw: (put) => { puts.push(put.recordKey); },
+  });
+  await server.start();
+  try {
+    await CultMesh.publishRudpDocumentOnce("completion-admit-client", connectionId, `rudp://127.0.0.1:${server.bind.port}`, binding, "tests:completion:a", { noteId: "a", text: "x" }, { resendDelayMs: 50, resendPollMs: 5, resultWindowMs: 200 });
+    assert.deepEqual(puts, ["tests:completion:a"]);
+  } finally { server.close(); }
+});

packages/cultmesh-ts/test/remote-write-authority.test.ts

@@ -1,11 +1,13 @@
 // A peer's put or delete is applied only when the server's write authority grants it. Ports the behaviours of
 // cultmesh-py test_remote_write_authority.py (no forwarding: this server does not forward non-primary writes).
 import assert from "node:assert/strict";
+import { createSocket, type Socket } from "node:dgram";
 import { readFileSync } from "node:fs";
 import { join } from "node:path";
 import test from "node:test";
 import { z } from "zod";
 import { documentPayloadFromBytes, defineDocumentType } from "@gamecult/cultcache-ts";
+import { decodeRudpPacket } from "@gamecult/cultnet-ts";
 import { encodeWireValue } from "@gamecult/cultnet-ts/wire";
 import {
   CultNetDocumentRegistry,
@@ -16,6 +18,7 @@ import {
 } from "@gamecult/cultnet-ts";
 import {
   CultMesh,
+  CULTMESH_RUDP_DOCUMENT_SERVER_MAX_CONSECUTIVE_REFUSED_FRAMES,
   CultMeshGrantAllWrites,
   type CultMeshRemoteWriteAuthority,
   type CultMeshRemoteWriteDecision,
@@ -505,3 +508,178 @@ test("a granted peer's unreadable write reaches onError and gets a malformed ref
     } finally { r.close(); }
   }
 });
+
+const CAP = 8; // the cap every runtime shares; the exported constant must stay this number
+
+test("the consecutive-refused-frames cap is the number every runtime shares", () => {
+  assert.equal(CULTMESH_RUDP_DOCUMENT_SERVER_MAX_CONSECUTIVE_REFUSED_FRAMES, CAP);
+});
+
+test("unreadable frames are answered and the session kept until eight in a row; any decoded frame resets the run", async () => {
+  const r = await rawRig(new Recording(grant));
+  try {
+    const batch = async (frames: number) => {
+      for (let i = 0; i < frames; i++) r.send(Uint8Array.from([0xc1]));
+      await sleep(400);
+    };
+    await batch(CAP - 1);
+    assert.equal(r.replies.length, CAP - 1, "each is answered");
+    r.send(encodeWireValue(put(KEY_A)));
+    await sleep(250);
+    assert.deepEqual(r.puts.map((p) => p.recordKey), [KEY_A], "the session was kept through seven");
+    await batch(CAP - 1);
+    assert.equal(r.replies.length, 2 * (CAP - 1), "the valid frame reset the run");
+    r.send(Uint8Array.from([0xc1]));
+    await sleep(300);
+    assert.equal(r.replies.length, 2 * (CAP - 1) + 1, "the eighth consecutive one is answered too");
+    assert.ok(r.replies.every((reply) => (reply as { code?: string }).code === "message_invalid"));
+    // The goodbye reached the peer: its client no longer takes a send.
+    assert.throws(() => r.send(encodeWireValue(put(KEY_B))), /before the session is connected/, "then the session ended");
+    assert.deepEqual(r.puts.map((p) => p.recordKey), [KEY_A]);
+  } finally { r.close(); }
+});
+
+test("deletes the authority refuses are answered and the session kept until eight in a row; any other decoded frame resets the run", async () => {
+  const authority = new Recording((request) => request.recordKey === KEY_B ? { granted: true } : { granted: false });
+  await using(await rig(authority), async (r) => {
+    const refuse = async (deletes: number) => {
+      for (let i = 0; i < deletes; i++) r.peer.send(del(KEY_A));
+      await sleep(400);
+    };
+    await refuse(CAP - 1);
+    assert.equal(r.replies.length, CAP - 1);
+    r.peer.send(del(KEY_B)); // granted: resets the run
+    await sleep(250);
+    assert.deepEqual(r.deletes.map((d) => d.recordKey), [KEY_B]);
+    await refuse(CAP - 1);
+    assert.equal(r.replies.length, 2 * (CAP - 1));
+    r.peer.send(put(KEY_B)); // a decoded non-delete resets too
+    await sleep(250);
+    await refuse(CAP - 1);
+    assert.equal(r.replies.length, 3 * (CAP - 1), "a put reset the run");
+    r.peer.send(del(KEY_A));
+    await sleep(300);
+    assert.equal(r.replies.length, 3 * (CAP - 1) + 1, "the eighth consecutive refused delete is answered too");
+    for (const reply of r.replies) assertRefused(reply, KEY_A);
+    assert.throws(() => r.peer.send(del(KEY_B)), /before the session is connected/, "then the session ended");
+    assert.equal(r.deletes.length, 1);
+  });
+});
+
+test("a frame that decodes but is no message the parser takes is answered, counted, and ended at eight in a row like any unreadable frame", async () => {
+  const r = await rawRig(new Recording(grant));
+  const unknown = () => encodeWireValue({ schemaVersion: "cultnet.not-a-message.v0", messageId: "x" });
+  try {
+    const batch = async (frames: number) => {
+      for (let i = 0; i < frames; i++) r.send(unknown());
+      await sleep(400);
+    };
+    await batch(CAP - 1);
+    assert.equal(r.replies.length, CAP - 1, "each is answered");
+    assert.equal(r.errors.length, CAP - 1, "and reported");
+    r.send(encodeWireValue(put(KEY_A)));
+    await sleep(250);
+    assert.deepEqual(r.puts.map((p) => p.recordKey), [KEY_A], "the session was kept through seven");
+    await batch(CAP - 1);
+    assert.equal(r.replies.length, 2 * (CAP - 1), "the decoded put reset the run");
+    r.send(unknown());
+    await sleep(300);
+    assert.equal(r.replies.length, 2 * (CAP - 1) + 1, "the eighth is answered too");
+    for (const reply of r.replies) {
+      assert.equal((reply as { code?: string }).code, "message_invalid");
+      assert.equal((reply as { details: { field: string } }).details.field, "message");
+    }
+    assert.throws(() => r.send(unknown()), /before the session is connected/, "then the session ended");
+  } finally { r.close(); }
+});
+
+/** A UDP relay between a client and a server that drops the first (or, with `dropEvery`, every) server datagram carrying `payloadContains` once armed. */
+async function lossyRelay(serverPort: number, payloadContains: string, dropEvery = false) {
+  const dropped: unknown[] = [];
+  let armed = false;
+  let clientAddress: { port: number; address: string } | undefined;
+  const toClient: Socket = createSocket("udp4");
+  const toServer: Socket = createSocket("udp4");
+  toServer.on("message", (wire) => {
+    if (!clientAddress) return;
+    if (armed && (dropEvery || dropped.length === 0)) {
+      const packet = decodeRudpPacket(wire);
+      if (packet.packetType === "data" && Buffer.from(packet.payload ?? []).includes(payloadContains)) {
+        dropped.push(packet);
+        return;
+      }
+    }
+    toClient.send(wire, clientAddress.port, clientAddress.address);
+  });
+  toClient.on("message", (wire, remote) => {
+    clientAddress = remote;
+    toServer.send(wire, serverPort, "127.0.0.1");
+  });
+  await new Promise<void>((resolve) => toClient.bind(0, "127.0.0.1", resolve));
+  await new Promise<void>((resolve) => toServer.bind(0, "127.0.0.1", resolve));
+  return {
+    port: toClient.address().port,
+    dropped,
+    arm: () => { armed = true; },
+    close: () => { toClient.close(); toServer.close(); },
+  };
+}
+
+async function cappedOverRelay(dropEvery: boolean, body: (x: { relay: Awaited<ReturnType<typeof lossyRelay>>; client: { send(channel: string, payload: Uint8Array): void }; replies: CultNetMessage[] }) => Promise<void>) {
+  const connectionId = nextConnection++;
+  const server = CultMesh.createRudpDocumentServer("authority-lossy", connectionId, {
+    documents: new CultNetDocumentRegistry([defineCultNetDocumentBinding({ definition: note })]),
+    writeAuthority: new Recording(grant), bindHost: "127.0.0.1", bindPort: 0, resendDelayMs: 25, resendPollMs: 5,
+    onDocumentPutRaw: () => {}, onError: () => {},
+  });
+  await server.start();
+  const relay = await lossyRelay(server.bind.port, "message_invalid", dropEvery);
+  const client = await CultMesh.createRudpClient("lossy", connectionId, `rudp://127.0.0.1:${relay.port}`, { resendDelayMs: 25, resendPollMs: 5 });
+  try {
+    client.connect();
+    const started = Date.now();
+    while (!client.connected) {
+      if (Date.now() - started > 2_000) throw new Error("not connected");
+      await sleep(5);
+    }
+    const replies: CultNetMessage[] = [];
+    new CultNetPeer(client, { wireContract: "cultnet.schema.v0" }).on("message", (m) => replies.push(m));
+    for (let i = 0; i < CAP - 1; i++) client.send("schema", Uint8Array.from([0xc1]));
+    await sleep(300);
+    assert.equal(replies.length, CAP - 1);
+    relay.arm();
+    client.send("schema", Uint8Array.from([0xc1]));
+    await body({ relay, client, replies });
+  } finally { client.close(); server.close(); relay.close(); }
+}
+
+test("the capped frame's answer is resent when its datagram is lost, and the goodbye follows only after it arrived", async () => {
+  await cappedOverRelay(false, async ({ relay, client, replies }) => {
+    await sleep(600);
+    assert.equal(relay.dropped.length, 1, "fixture: the first transmission of the last answer was lost");
+    assert.equal(replies.length, CAP, "the lost answer was resent and arrived");
+    assert.equal((replies[CAP - 1] as { code?: string }).code, "message_invalid");
+    assert.throws(() => client.send("schema", Uint8Array.from([0xc1])), /before the session is connected/, "then the session ended");
+  });
+});
+
+test("a capped session whose answer is never acknowledged still ends, after the grace", async () => {
+  await cappedOverRelay(true, async ({ client, replies }) => {
+    await sleep(1200);
+    assert.equal(replies.length, CAP - 1, "fixture: the last answer never arrives");
+    client.send("schema", Uint8Array.from([0xc1])); // still connected: the server is waiting for the acknowledgement
+    await sleep(1500);
+    assert.throws(() => client.send("schema", Uint8Array.from([0xc1])), /before the session is connected/, "the grace ran out and the goodbye went out");
+  });
+});
+
+test("nothing behind the capped frame is served while its answer is delivered", async () => {
+  const r = await rawRig(new Recording(grant));
+  try {
+    for (let i = 0; i < CAP; i++) r.send(Uint8Array.from([0xc1]));
+    r.send(encodeWireValue(put(KEY_A)));
+    await sleep(500);
+    assert.equal(r.replies.length, CAP);
+    assert.deepEqual(r.puts, [], "the put behind the cap was not served");
+  } finally { r.close(); }
+});

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

@@ -58,6 +58,7 @@ class CultNetRawClient:
     create_transport: CultNetSchemaTransportFactory | None = None
 
     def send(self, message: CultNetMessage | dict[str, Any]) -> None:
+        """Fire-and-forget: an ack or a return is not admission; read the result frame."""
         wire = message.to_wire() if isinstance(message, CultNetMessage) else message
         with self.open_transport() as transport:
             transport.send("schema", msgpack.packb(wire, use_bin_type=True))
@@ -319,6 +320,7 @@ class CultNetDatabaseSubscription:
                 self._transport = None
 
     def send(self, message: CultNetMessage | dict[str, Any]) -> None:
+        """Fire-and-forget: an ack or a return is not admission; read the result frame."""
         if self._transport is None:
             raise RuntimeError("CultNet database subscription is not open")
         wire = message.to_wire() if isinstance(message, CultNetMessage) else message

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

@@ -1256,6 +1256,7 @@ class CultNetRudpSocketTransportConnection:
         self._send_packet(self.session.create_connect(_now_ms(), payload))
 
     def send(self, channel_id: str, payload: bytes) -> None:
+        """Fire-and-forget: an ack or a return is not admission; read the result frame."""
         packets = self.session.send_many(
             channel_id,
             payload,
@@ -1567,6 +1568,7 @@ class TcpFramedTransportConnection:
         )
 
     def send(self, channel_id: str, payload: bytes) -> None:
+        """Fire-and-forget: an ack or a return is not admission; read the result frame."""
         if channel_id != "schema":
             raise ValueError(f"tcp_framed transport only supports the schema channel, got {channel_id!r}")
         before = len(payload) + 4

packages/cultnet-rs/src/rudp.rs

@@ -2198,6 +2198,7 @@ impl CultNetRudpSocketTransportConnection {
         self.send_packet(&packet)
     }
 
+    /// Fire-and-forget: an ack or a return is not admission; read the result frame.
     pub fn send(&mut self, channel_id: &str, payload: Vec<u8>) -> Result<()> {
         let options =
             channel_send_options(
@@ -2216,6 +2217,8 @@ impl CultNetRudpSocketTransportConnection {
 
     /// Sends one frame on a reliable channel whose packets cannot expire and
     /// returns the transport receipt used to observe its exact acknowledgement.
+    ///
+    /// Fire-and-forget: an ack or a return is not admission; read the result frame.
     pub fn send_reliable(
         &mut self,
         channel_id: &str,
@@ -2309,6 +2312,9 @@ impl CultNetRudpSocketTransportConnection {
         self.session.outstanding_reliable_packet_count()
     }
 
+    /// Waits until every reliable write is acknowledged by the transport.
+    ///
+    /// Fire-and-forget: an ack or a return is not admission; read the result frame.
     pub fn flush_reliable(&mut self, timeout: Duration) -> Result<()> {
         let original_timeout = self.socket.read_timeout()?;
         let poll_timeout = original_timeout

packages/cultnet-rs/src/transport.rs

@@ -252,6 +252,7 @@ impl<TStream> TcpFramedTransportConnection<TStream>
 where
     TStream: Write,
 {
+    /// Fire-and-forget: an ack or a return is not admission; read the result frame.
     pub fn send(&mut self, channel_id: &str, payload: &[u8]) -> Result<()> {
         if channel_id != "schema" {
             return Err(anyhow!(

packages/cultnet-ts/src/idunn-odin-presence-publisher.ts

@@ -107,26 +107,36 @@ async function publishDocument(
     const wirePayload = encode(encodeCultNetMessageForWire(message as never, "cultnet.schema.v0"));
     const packets = session.sendMany("schema", wirePayload, { reliable: true, ordered: true, nowMs: Date.now() });
     for (const packet of packets) await sendPacket(socket, endpoint, packet);
-    await receiveUntil(
-      receiver,
-      session,
-      endpoint,
-      () => session.outstandingReliablePacketCount === 0,
-      2000,
-      (frame) => {
-        if (frame.channelId !== "schema") return;
-        const response = parseCultNetMessage(decode(frame.payload));
-        if (response.schemaVersion === "cultnet.error.v0") {
-          throw new Error(`Odin rejected runtime presence: ${response.error}`);
-        }
-      },
-    );
+    // The transport ack is not admission: the put completes on Odin's result frame. A refusal rejects whether
+    // it arrives before the ack, after it, or inside the result window; a success reply ends the wait early.
+    let succeeded = false;
+    const onFrame = (frame: { channelId: string; payload: Uint8Array }) => {
+      if (frame.channelId !== "schema") return;
+      const response = parseCultNetMessage(decode(frame.payload));
+      if (response.schemaVersion === "cultnet.error.v0") throw new IdunnPresenceRefusedError(response);
+      succeeded = true;
+    };
+    await receiveUntil(receiver, session, endpoint, () => session.outstandingReliablePacketCount === 0, 2000, onFrame);
+    if (!succeeded) await receiveUntil(receiver, session, endpoint, () => succeeded, RESULT_WINDOW_MS, onFrame, true);
   } finally {
     receiver.close();
     socket.close();
   }
 }
 
+/** How long after the transport ack Odin's refusal may still arrive; a success reply ends the wait early. */
+const RESULT_WINDOW_MS = 500;
+
+/** Odin refused the presence put: the typed `cultnet.error.v0` it answered with. */
+export class IdunnPresenceRefusedError extends Error {
+  readonly code: string | undefined;
+  constructor(readonly response: { error: string; code?: string | null }) {
+    super(`Odin rejected runtime presence: ${response.error}`);
+    this.name = "IdunnPresenceRefusedError";
+    this.code = response.code ?? undefined;
+  }
+}
+
 type PacketReceiver = {
   socket: dgram.Socket;
   next(timeoutMs: number): Promise<CultNetRudpPacket>;
@@ -190,6 +200,7 @@ async function receiveUntil(
   predicate: (packet: CultNetRudpPacket) => boolean,
   timeoutMs: number,
   onDelivered?: (frame: { channelId: string; payload: Uint8Array; sequence: number }) => void,
+  returnOnTimeout = false,
 ): Promise<void> {
   const deadline = Date.now() + timeoutMs;
   while (Date.now() < deadline) {
@@ -204,6 +215,7 @@ async function receiveUntil(
     }
     for (const packet of session.dueResends(Date.now())) await sendPacket(receiver.socket, endpoint, packet);
   }
+  if (returnOnTimeout) return;
   throw new Error(`Timed out waiting for CultNet RUDP response after ${timeoutMs} ms.`);
 }
 

packages/cultnet-ts/src/peer.ts

@@ -95,6 +95,10 @@ export class CultNetPeer extends EventEmitter {
     return this.#transport?.profile;
   }
 
+  /**
+   * Fire-and-forget: an ack or a return is not admission; read the result frame (a write is refused with
+   * `cultnet.error.v0`, `write_not_authorized` or `message_invalid`, as a `message` event).
+   */
   send(message: CultNetMessage): void {
     const wireValue = encodeCultNetMessageForWire(message, this.#wireContract);
     const payload = encode(wireValue);
@@ -106,6 +110,7 @@ export class CultNetPeer extends EventEmitter {
     this.#stream?.write(encodeFrame(payload));
   }
 
+  /** Resolves when the transport has acknowledged what was sent. An ack is not admission; read the result frame. */
   async flush(timeoutMs?: number): Promise<void> {
     await this.#transport?.flush?.(timeoutMs);
   }

packages/cultnet-ts/test/idunn-runtime-authority.test.ts

@@ -276,9 +276,14 @@ async function startOdinPeer(context: TestContext, endpoint: string) {
     peer.once("error", reject);
     peer.bind(port, "127.0.0.1", resolve);
   });
-  context.after(() => peer.close());
+  // The second packet of a split delivery is a timer; the test's end cancels it rather than racing the closed socket.
+  const timers = new Set<NodeJS.Timeout>();
+  const later = (send: () => void) => { timers.add(setTimeout(send, 120)); };
+  context.after(() => { for (const timer of timers) clearTimeout(timer); peer.close(); });
   const state = {
     reject: false,
+    // How a refusal and the ack of the put reach the publisher: in one packet, or the ack first, or the refusal first.
+    order: "piggyback" as "piggyback" | "ack-then-refusal" | "refusal-then-ack",
     received: [] as Array<{ schemaVersion?: string; document: { payload: DocumentPayload } }>,
     responseAck: undefined as number | undefined,
     failure: undefined as Error | undefined,
@@ -305,7 +310,15 @@ async function startOdinPeer(context: TestContext, endpoint: string) {
       });
       assert.ok(response);
       state.responseAck = response.ack;
-      peer.send(encodeRudpPacket(response), remote.port, remote.address);
+      if (state.reject && state.order === "ack-then-refusal") {
+        peer.send(encodeRudpPacket(serverSession.createAckForReceived(packet.sequence)), remote.port, remote.address);
+        later(() => peer.send(encodeRudpPacket(response), remote.port, remote.address));
+      } else if (state.reject && state.order === "refusal-then-ack") {
+        peer.send(encodeRudpPacket({ ...response, ack: 0, ackMask: 0 }), remote.port, remote.address);
+        later(() => peer.send(encodeRudpPacket(serverSession!.createAckForReceived(packet.sequence)), remote.port, remote.address));
+      } else {
+        peer.send(encodeRudpPacket(response), remote.port, remote.address);
+      }
     } catch (error) {
       state.failure = error as Error;
     }
@@ -332,6 +345,19 @@ test("the Odin publisher publishes the signer's document to the endpoint Expecte
   assert.equal(peer.failure, undefined, peer.failure?.message);
 });
 
+for (const order of ["ack-then-refusal", "refusal-then-ack"] as const) {
+  test(`the Odin publisher does not complete on the ack of a refused presence (${order})`, async (context) => {
+    const endpoint = expectedFacts("service").odinEndpoint!;
+    const peer = await startOdinPeer(context, endpoint);
+    peer.reject = true;
+    peer.order = order;
+    const signer = signerFor(context, "service");
+    signer.reportHealth("active");
+    const publisher = createIdunnRuntimePresencePublisher({ signer, endpoint });
+    await assert.rejects(publisher.publish("denied presence"), /Odin rejected runtime presence: test admission denied/);
+  });
+}
+
 test("the Odin publisher leaves a capability shortfall to the Rust correlation instead of refusing to publish", async (context) => {
   const endpoint = expectedFacts("service").odinEndpoint!;
   const peer = await startOdinPeer(context, endpoint);

src/GameCult.Mesh/CultMeshClient.cs

@@ -44,6 +44,12 @@ namespace GameCult.Mesh
 
         /// <summary>Gets or sets the deadline for one correlated typed operation response.</summary>
         public TimeSpan OperationResponseTimeout { get; set; } = TimeSpan.FromSeconds(10);
+
+        /// <summary>
+        /// Gets or sets how long after a submitted document the provider's refusal may still arrive. The provider
+        /// sends no success frame for an admitted submission, so this window bounds <c>SubmitDocumentAsync</c>.
+        /// </summary>
+        public TimeSpan SubmitResultWindow { get; set; } = CultNetWriteCompletion.DefaultResultWindow;
     }
 
     /// <summary>
@@ -149,6 +155,7 @@ namespace GameCult.Mesh
         private readonly CultMeshSessionManager _sessions;
         private readonly TimeSpan _subscriptionResponseTimeout;
         private readonly TimeSpan _operationResponseTimeout;
+        private readonly TimeSpan _submitResultWindow;
         private readonly ConcurrentDictionary<string, SharedResource> _documents = new(StringComparer.Ordinal);
         private readonly ConcurrentDictionary<string, SharedResource> _collections = new(StringComparer.Ordinal);
         private bool _disposed;
@@ -191,6 +198,7 @@ namespace GameCult.Mesh
                 options.Sessions);
             _subscriptionResponseTimeout = options.SubscriptionResponseTimeout;
             _operationResponseTimeout = options.OperationResponseTimeout;
+            _submitResultWindow = options.SubmitResultWindow;
         }
 
         /// <summary>Connects to one stable Verse/provider target using one application protocol.</summary>
@@ -319,7 +327,10 @@ namespace GameCult.Mesh
 
         /// <summary>
         /// Submits one typed document to a provider by stable identity. Submission is uncommitted input;
-        /// only a provider-authored receipt or resulting state can establish acceptance.
+        /// only a provider-authored receipt or resulting state can establish acceptance. The task completes on
+        /// the provider's result frame: a refusal (<c>write_not_authorized</c>, <c>message_invalid</c>, ...) throws
+        /// <see cref="CultNetWriteRefusedException"/>, and a submission not refused within
+        /// <see cref="CultMeshClientOptions.SubmitResultWindow"/> completes.
         /// </summary>
         public async Task SubmitDocumentAsync<TDocument>(
             CultMeshSessionTarget target,
@@ -351,7 +362,11 @@ namespace GameCult.Mesh
             var session = await ConnectAsync(target, CultMeshProtocols.Documents, cancellationToken)
                 .ConfigureAwait(false);
             cancellationToken.ThrowIfCancellationRequested();
-            session.OpenSchemaClient().SendCultNet(message);
+            await CultNetWriteCompletion.SendAndAwaitResultAsync(
+                refuse => session.OnCultNet(refuse),
+                () => session.SendCultNet(message),
+                _submitResultWindow,
+                cancellationToken).ConfigureAwait(false);
         }
 
         /// <summary>

src/GameCult.Mesh/CultMeshSessions.cs

@@ -352,6 +352,7 @@ namespace GameCult.Mesh
         internal long PhysicalGeneration => Interlocked.Read(ref _physicalGeneration);
         public Observable<CultMeshSessionState> WatchState() => _states;
         public ICultNetSchemaClient OpenSchemaClient() => new SessionSchemaClient(this);
+        /// <summary>Fire-and-forget: an ack or a return is not admission; read the result frame.</summary>
         public void SendCultNet<T>(T message) where T : ICultNetSchemaMessage => Channel.SendCultNet(message);
         public IDisposable OnCultNet<T>(Action<T> callback) where T : ICultNetSchemaMessage
         {

src/GameCult.Networking/CultNetRudpSchemaServer.cs

@@ -1,4 +1,5 @@
 using System;
+using System.Runtime.CompilerServices;
 using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Net;
@@ -113,9 +114,24 @@ namespace GameCult.Networking
         private readonly CultNetRudpSocketTransportServer _transport;
         private readonly ConcurrentDictionary<Type, Delegate> _handlers = new();
         private readonly ConcurrentDictionary<Delegate, Delegate> _schemaPeerAdapters = new();
+        private readonly ConditionalWeakTable<CultNetRudpSocketServerPeer, StrongBox<int>> _undecodableFramesInARow = new();
         private bool _disposed;
         private ILogger _logger = new NullLogger();
 
+        /// <summary>
+        /// How many consecutive undecodable frames a peer's session may send before it is ended. Fewer are
+        /// answered with a typed <c>message_invalid</c> and the session is kept; the frame that reaches this
+        /// number is answered too, then the session ends. Any frame that decodes resets the run. The same
+        /// number is the cap in every runtime.
+        /// </summary>
+        public const int MaxConsecutiveRefusedFrames = 8;
+
+        /// <summary>
+        /// How long a session whose capped frame was answered waits for the peer to acknowledge the answer
+        /// before it ends anyway.
+        /// </summary>
+        internal const long CappedAnswerGraceMs = 2000;
+
         /// <summary>
         /// Gets or sets the logger used by the server. R-AM: the dispatch backstop in DispatchAsync
         /// logs every untyped fault through this instead of swallowing it silently.
@@ -300,6 +316,9 @@ namespace GameCult.Networking
         {
             if (!string.Equals(delivered.Frame.ChannelId, "schema", StringComparison.Ordinal))
                 return false;
+            // The capped frame was answered: nothing after it is served while the answer is delivered.
+            if (delivered.Peer.EndingReason != null)
+                return false;
 
             // A peer's message that cannot be decoded (malformed bytes, an unknown schemaVersion) or
             // handled ends that peer's session, never the poll: the drain goes on serving every other
@@ -317,11 +336,23 @@ namespace GameCult.Networking
             }
             catch (Exception ex)
             {
-                // The peer sent something no runtime can read: the refusal is the peer's.
-                var ended = _transport.EndPeerSession(delivered.Peer, CultNetRudpSession.RefusedPacketReason);
-                Logger.LogError($"CultNet RUDP schema server could not decode a peer's message ({ex.GetType().Name}); {SessionOutcome(ended)}");
+                // The peer sent something no runtime can read: the refusal is the peer's. It is answered
+                // message_invalid and the session kept, until MaxConsecutiveRefusedFrames in a row end it.
+                var run = _undecodableFramesInARow.GetOrCreateValue(delivered.Peer);
+                run.Value++;
+                _transport.SendSchemaMessage(delivered.Peer, CultNetErrorMessage.ForMessageInvalid("message", "CultNet RUDP schema server could not read the message."));
+                if (run.Value < MaxConsecutiveRefusedFrames)
+                {
+                    Logger.LogError($"CultNet RUDP schema server could not decode a peer's message ({ex.GetType().Name}); it was refused and that peer's session kept.");
+                    return false;
+                }
+                // The answer was sent reliably: the session ends once the peer has acknowledged it (a lost
+                // datagram is resent first), or after the grace.
+                _transport.EndPeerSessionWhenAnswered(delivered.Peer, CultNetRudpSession.RefusedPacketReason, CappedAnswerGraceMs);
+                Logger.LogError($"CultNet RUDP schema server could not decode a peer's message ({ex.GetType().Name}); that peer's session ends once its answer is acknowledged.");
                 return false;
             }
+            _undecodableFramesInARow.GetOrCreateValue(delivered.Peer).Value = 0;
 
             if (!_handlers.TryGetValue(message.GetType(), out var handler) || handler == null)
             {

src/GameCult.Networking/CultNetSchemaClient.cs

@@ -26,7 +26,8 @@ namespace GameCult.Networking
         void Connect(string host, int port);
 
         /// <summary>
-        /// Sends one schema-v0 message through the transport.
+        /// Sends one schema-v0 message through the transport. Fire-and-forget: an ack or a return is not admission;
+        /// read the result frame (<see cref="CultNetWriteCompletion"/>).
         /// </summary>
         void SendCultNet<T>(T message)
             where T : ICultNetSchemaMessage;

src/GameCult.Networking/CultNetShardWriteForwarding.cs

@@ -68,6 +68,12 @@ namespace GameCult.Networking
         /// rudp:// endpoints use the C# RUDP schema client and cultnet:// endpoints use LiteNetLib.
         /// </summary>
         public Func<ICultNetSchemaClient>? CreateClient { get; set; }
+
+        /// <summary>
+        /// Gets or sets how long after a forwarded write the primary's refusal may still arrive; a refusal fails the
+        /// forward with <see cref="CultNetWriteRefusedException"/>. Defaults to <see cref="CultNetWriteCompletion.DefaultResultWindow"/>.
+        /// </summary>
+        public TimeSpan ResultWindow { get; set; } = CultNetWriteCompletion.DefaultResultWindow;
     }
 
     /// <summary>
@@ -85,7 +91,10 @@ namespace GameCult.Networking
             _options = options ?? new CultNetSchemaWriteForwarderOptions();
         }
 
-        /// <inheritdoc />
+        /// <summary>
+        /// Forwards a put and completes on the primary's result frame: a refusal throws
+        /// <see cref="CultNetWriteRefusedException"/>, the same failure path as an unreachable primary.
+        /// </summary>
         public async Task ForwardPutAsync(CultNetShardDescriptor shard, CultNetDocumentPutRawMessage message)
         {
             if (message == null) throw new ArgumentNullException(nameof(message));
@@ -95,7 +104,10 @@ namespace GameCult.Networking
             await SendAsync(endpoint, message).ConfigureAwait(false);
         }
 
-        /// <inheritdoc />
+        /// <summary>
+        /// Forwards a delete and completes on the primary's result frame: a refusal throws
+        /// <see cref="CultNetWriteRefusedException"/>.
+        /// </summary>
         public async Task ForwardDeleteAsync(CultNetShardDescriptor shard, CultNetDocumentDeleteMessage message)
         {
             if (message == null) throw new ArgumentNullException(nameof(message));
@@ -132,7 +144,14 @@ namespace GameCult.Networking
             using var client = CreateClient(endpoint);
             client.Connect(host, port);
             await WaitForConnectionAsync(client, endpoint).ConfigureAwait(false);
-            client.SendCultNet(message);
+            await CultNetWriteCompletion.SendAndAwaitResultAsync(
+                refuse =>
+                {
+                    client.OnCultNet(refuse);
+                    return null;
+                },
+                () => client.SendCultNet(message),
+                _options.ResultWindow).ConfigureAwait(false);
         }
 
         private ICultNetSchemaClient CreateClient(string endpoint)

src/GameCult.Networking/CultNetTransport.cs

@@ -2759,6 +2759,11 @@ namespace GameCult.Networking
         // under SessionGate by whichever thread's send found the failure; the polling thread reads
         // it when it retires the peer.
         internal byte[]? UnsendableReason { get; set; }
+
+        // Set once the peer's capped frame has been answered: the session ends with this reason when the
+        // peer has acknowledged every answer, or at the deadline. Under SessionGate.
+        internal byte[]? EndingReason { get; set; }
+        internal long EndingDeadlineMs { get; set; }
     }
 
     /// <summary>
@@ -2928,6 +2933,7 @@ namespace GameCult.Networking
         public bool TryReceiveOnce(out CultNetRudpSocketServerFrame? delivered)
         {
             RetireUnsendablePeers();
+            EndAnsweredPeers();
             if (_deliveredFrames.Count > 0)
             {
                 delivered = _deliveredFrames.Dequeue();
@@ -3146,6 +3152,7 @@ namespace GameCult.Networking
         public void PollResends()
         {
             RetireUnsendablePeers();
+            EndAnsweredPeers();
             if (_socket.Poll(0, SelectMode.SelectRead))
                 return;
             foreach (var peer in _peers.Values.ToArray())
@@ -3244,6 +3251,44 @@ namespace GameCult.Networking
             return true;
         }
 
+        // Polling thread only. The application answered a peer's capped frame: the answer is reliable, so a
+        // lost datagram is resent, and the session ends with `reason` once the peer has acknowledged every
+        // answer or `graceMs` have passed (EndAnsweredPeers). Frames of the peer still queued are not delivered.
+        internal void EndPeerSessionWhenAnswered(CultNetRudpSocketServerPeer peer, byte[] reason, long graceMs)
+        {
+            if (peer == null) throw new ArgumentNullException(nameof(peer));
+            if (reason == null) throw new ArgumentNullException(nameof(reason));
+            lock (peer.SessionGate)
+            {
+                if (peer.Session.Ended || peer.EndingReason != null)
+                    return;
+                peer.EndingReason = reason;
+                peer.EndingDeadlineMs = NowMs() + graceMs;
+            }
+            var kept = _deliveredFrames.Where(frame => !ReferenceEquals(frame.Peer, peer)).ToArray();
+            _deliveredFrames.Clear();
+            foreach (var frame in kept)
+                _deliveredFrames.Enqueue(frame);
+        }
+
+        // Polling thread only.
+        private void EndAnsweredPeers()
+        {
+            foreach (var peer in _peers.Values.ToArray())
+            {
+                byte[]? reason;
+                lock (peer.SessionGate)
+                {
+                    reason = peer.EndingReason != null
+                        && (peer.Session.OutstandingReliablePacketCount == 0 || NowMs() >= peer.EndingDeadlineMs)
+                        ? peer.EndingReason
+                        : null;
+                }
+                if (reason != null)
+                    EndPeerSession(peer, reason);
+            }
+        }
+
         // Polling thread only. A peer a new Connect already replaced was reported then, and its key now
         // names the peer that replaced it.
         private void RetireUnsendablePeers()

src/GameCult.Networking/CultNetWriteCompletion.cs

@@ -0,0 +1,64 @@
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace GameCult.Networking
+{
+    /// <summary>
+    /// A peer refused a write: the typed <c>cultnet.error.v0</c> it answered with.
+    /// </summary>
+    public sealed class CultNetWriteRefusedException : Exception
+    {
+        /// <summary>Creates the exception from the refusal frame.</summary>
+        public CultNetWriteRefusedException(CultNetErrorMessage refusal)
+            : base(refusal?.Error ?? throw new ArgumentNullException(nameof(refusal)))
+        {
+            Code = refusal.Code;
+            Refusal = refusal;
+        }
+
+        /// <summary>Gets the machine-readable refusal code, e.g. <c>write_not_authorized</c> or <c>message_invalid</c>.</summary>
+        public string? Code { get; }
+
+        /// <summary>Gets the refusal frame as received.</summary>
+        public CultNetErrorMessage Refusal { get; }
+    }
+
+    /// <summary>
+    /// Completes a write on the peer's result frame, never on a send's return or a transport acknowledgement:
+    /// an ack proves delivery, not admission.
+    /// </summary>
+    public static class CultNetWriteCompletion
+    {
+        /// <summary>How long after a send a refusal may still arrive when the caller names no window.</summary>
+        public static readonly TimeSpan DefaultResultWindow = TimeSpan.FromMilliseconds(500);
+
+        /// <summary>
+        /// Sends one write and waits <paramref name="resultWindow"/> for the peer's answer. A refusal frame rejects with
+        /// <see cref="CultNetWriteRefusedException"/> whatever its order against the transport acknowledgement. A peer
+        /// that admits a write sends no success frame, so the end of the window is the interim bound on a refusal
+        /// (until peers send one); the error frame carries no message id, so any refusal inside the window is this
+        /// write's.
+        /// </summary>
+        /// <param name="subscribe">Registers the refusal handler and returns its registration; called before the send.</param>
+        /// <param name="send">Sends the write.</param>
+        /// <param name="resultWindow">How long a refusal may take to arrive.</param>
+        /// <param name="cancellationToken">Cancels the wait.</param>
+        public static async Task SendAndAwaitResultAsync(
+            Func<Action<CultNetErrorMessage>, IDisposable?> subscribe,
+            Action send,
+            TimeSpan resultWindow,
+            CancellationToken cancellationToken = default)
+        {
+            if (subscribe == null) throw new ArgumentNullException(nameof(subscribe));
+            if (send == null) throw new ArgumentNullException(nameof(send));
+            var refused = new TaskCompletionSource<CultNetErrorMessage>(TaskCreationOptions.RunContinuationsAsynchronously);
+            using var registration = subscribe(error => refused.TrySetResult(error));
+            send();
+            var window = Task.Delay(resultWindow, cancellationToken);
+            await Task.WhenAny(refused.Task, window).ConfigureAwait(false);
+            if (refused.Task.IsCompleted) throw new CultNetWriteRefusedException(await refused.Task.ConfigureAwait(false));
+            await window.ConfigureAwait(false);
+        }
+    }
+}

tests/GameCult.Mesh.Tests/CultMeshClientTests.cs

@@ -291,6 +291,51 @@ public sealed class CultMeshClientTests
         submitted.Document.SourceRole.Should().Be("pilot");
     }
 
+    [TestCase("write_not_authorized", 0)]
+    [TestCase("message_invalid", 0)]
+    [TestCase("write_not_authorized", 150)]
+    public async Task SubmitDocument_ARefusalRejectsWithTheTypedError_WhateverItsOrderAgainstTheSend(string code, int delayMs)
+    {
+        var connector = new DocumentConnector
+        {
+            RefuseSubmissionsWith = code,
+            RefusalDelay = TimeSpan.FromMilliseconds(delayMs)
+        };
+        using var mesh = new CultMeshClient(new CultMeshClientOptions
+        {
+            RendezvousEndpoints = new[] { "rudp://odin:3076" },
+            Discovery = new CultMeshVerseDiscoveryClientOptions { CreateClient = () => new RendezvousClient() },
+            Sessions = CultMeshTestTrust.LocalSessions,
+            Connectors = new[] { connector }
+        });
+        var request = new ClientTestDocument { Id = "command:1", Text = "move" };
+
+        var act = () => mesh.SubmitDocumentAsync(Target, request.Id, request, "pilot-one", "pilot");
+
+        (await act.Should().ThrowAsync<CultNetWriteRefusedException>()).Which.Code.Should().Be(code);
+    }
+
+    [Test]
+    public async Task SubmitDocument_CompletesOnlyAfterTheResultWindow_NotOnTheSendReturning()
+    {
+        var connector = new DocumentConnector();
+        using var mesh = new CultMeshClient(new CultMeshClientOptions
+        {
+            RendezvousEndpoints = new[] { "rudp://odin:3076" },
+            Discovery = new CultMeshVerseDiscoveryClientOptions { CreateClient = () => new RendezvousClient() },
+            Sessions = CultMeshTestTrust.LocalSessions,
+            Connectors = new[] { connector },
+            SubmitResultWindow = TimeSpan.FromMilliseconds(400)
+        });
+        var request = new ClientTestDocument { Id = "command:1", Text = "move" };
+        await mesh.ConnectAsync(Target, CultMeshProtocols.Documents);
+        var started = DateTime.UtcNow;
+
+        await mesh.SubmitDocumentAsync(Target, request.Id, request, "pilot-one", "pilot");
+
+        (DateTime.UtcNow - started).Should().BeGreaterThanOrEqualTo(TimeSpan.FromMilliseconds(380));
+    }
+
     [Test]
     public async Task InvokeOperation_CorrelatesProviderReceiptThroughStableIdentitySession()
     {
@@ -496,6 +541,8 @@ public sealed class CultMeshClientTests
         public bool SuppressFirstSnapshot { get; set; }
         public bool SuppressFirstSubscriptionSnapshot { get; set; }
         public bool RejectOperations { get; set; }
+        public string? RefuseSubmissionsWith { get; set; }
+        public TimeSpan RefusalDelay { get; set; }
         public bool FailFirstOperationBeforeResponse { get; set; }
         public string ResponseSourceRuntimeId { get; set; } = "aetheria-daemon";
         public string ConnectorId => "document-test";
@@ -515,6 +562,8 @@ public sealed class CultMeshClientTests
                 rejectOperations: RejectOperations,
                 failOperationBeforeResponse: FailFirstOperationBeforeResponse && Clients.Count == 0,
                 responseSourceRuntimeId: ResponseSourceRuntimeId);
+            client.RefuseSubmissionsWith = RefuseSubmissionsWith;
+            client.RefusalDelay = RefusalDelay;
             Clients.Add(client);
             return Task.FromResult<ICultNetSchemaClient>(client);
         }
@@ -522,6 +571,9 @@ public sealed class CultMeshClientTests
 
     private sealed class DocumentClient : ICultNetSchemaClient, ICultNetSchemaClientHealth, ICultMeshVerifiedSchemaClient
     {
+        private readonly List<Action<CultNetErrorMessage>> _errors = new();
+        public string? RefuseSubmissionsWith { get; set; }
+        public TimeSpan RefusalDelay { get; set; }
         private readonly List<Action<CultNetSnapshotResponseRawMessage>> _snapshots = new();
         private readonly List<Action<CultNetOperationResponseMessage>> _operationResponses = new();
         private readonly CultNetDocumentRegistry _documents;
@@ -558,6 +610,15 @@ public sealed class CultMeshClientTests
             if (message is CultNetDocumentPutRawMessage submitted)
             {
                 Submitted.Add(submitted);
+                if (RefuseSubmissionsWith != null)
+                {
+                    var refusal = new CultNetErrorMessage { Error = "not allowed", Code = RefuseSubmissionsWith };
+                    // A refusal that follows the send's return (and any transport ack) must still reject the submit.
+                    Task.Delay(RefusalDelay).ContinueWith(_ =>
+                    {
+                        foreach (var handler in _errors.ToArray()) handler(refusal);
+                    });
+                }
                 return;
             }
             if (message is CultNetOperationRequestMessage operation)
@@ -642,6 +703,8 @@ public sealed class CultMeshClientTests
                 _snapshots.Add(response => callback((T)(object)response));
             if (typeof(T) == typeof(CultNetOperationResponseMessage))
                 _operationResponses.Add(response => callback((T)(object)response));
+            if (typeof(T) == typeof(CultNetErrorMessage))
+                _errors.Add(error => callback((T)(object)error));
         }
         public void Dispose() { }
         public void Fail(Exception error) => _failure.TrySetResult(error);

tests/GameCult.Networking.Tests/CultNetClientWriteCompletionTests.cs

@@ -0,0 +1,100 @@
+#nullable enable
+using System;
+using System.Collections.Concurrent;
+using System.Net;
+using System.Net.Sockets;
+using System.Threading;
+using System.Threading.Tasks;
+using NUnit.Framework;
+
+namespace GameCult.Networking.Tests
+{
+    /// <summary>
+    /// Heimdall's invariant: no client treats a reliable-transport ack as write admission. A put completes
+    /// only on its result frame (write_not_authorized, or the success reply). The C# schema client has no
+    /// completion call: SendCultNet returns once the put is queued, and the outcome reaches the caller only
+    /// as the result frame, through OnCultNet. These tests serve it a peer that acknowledges the put and
+    /// refuses it, the ack before the refusal and after it, and require the refusal to arrive as the outcome.
+    /// </summary>
+    public class CultNetClientWriteCompletionTests
+    {
+        private const uint ConnectionId = 0x10203197;
+
+        // Acknowledges the first reliable schema frame it receives and answers it with write_not_authorized,
+        // the second packet 150 ms after the first. The refusal carries no acknowledgement of its own.
+        private static Task RefusingPeer(Socket socket, bool refusalFirst, CancellationToken stop)
+        {
+            return Task.Run(() =>
+            {
+                var session = new CultNetRudpSession(new CultNetRudpSessionOptions { ConnectionId = ConnectionId, ResendDelayMs = 25 });
+                var buffer = new byte[65535];
+                EndPoint from = new IPEndPoint(IPAddress.Any, 0);
+                var answered = false;
+                while (!stop.IsCancellationRequested)
+                {
+                    if (!socket.Poll(10_000, SelectMode.SelectRead))
+                        continue;
+                    var received = socket.ReceiveFrom(buffer, ref from);
+                    var packet = CultNetRudpPacketCodec.Decode(buffer.AsSpan(0, received).ToArray());
+                    void Send(CultNetRudpPacket p) => socket.SendTo(CultNetRudpPacketCodec.Encode(p), from);
+                    if (packet.PacketType == CultNetRudpPacketType.Connect)
+                    {
+                        Send(session.AcceptConnect(packet));
+                        continue;
+                    }
+                    var result = session.Receive(packet, 1);
+                    if (result.Reply != null)
+                        Send(result.Reply);
+                    if (answered || result.Delivered.Count == 0)
+                        continue;
+                    answered = true;
+                    var ack = session.CreateAckForReceived(packet.Sequence);
+                    var refusal = session.Send("schema", CultNetSchemaMessageSerialization.Serialize(new CultNetErrorMessage
+                    {
+                        Error = "write_not_authorized: the catalog refused the document",
+                        Code = "write_not_authorized"
+                    }), new CultNetRudpSendOptions { Reliable = true, Ordered = true });
+                    refusal.Ack = 0;
+                    refusal.AckMask = 0;
+                    if (refusalFirst)
+                    {
+                        Send(refusal);
+                        Thread.Sleep(150);
+                        Send(ack);
+                    }
+                    else
+                    {
+                        Send(ack);
+                        Thread.Sleep(150);
+                        Send(refusal);
+                    }
+                }
+            });
+        }
+
+        [TestCase(false, TestName = "AckThenRefusal")]
+        [TestCase(true, TestName = "RefusalThenAck")]
+        public async Task ARefusedPutIsReportedRefusedWhateverTheOrderOfItsAck(bool refusalFirst)
+        {
+            using var serverSocket = new Socket(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp);
+            serverSocket.Bind(new IPEndPoint(IPAddress.Loopback, 0));
+            using var stop = new CancellationTokenSource();
+            var peer = RefusingPeer(serverSocket, refusalFirst, stop.Token);
+            using var client = new RudpCultNetSchemaClient("completion", ConnectionId);
+            var outcomes = new BlockingCollection<CultNetErrorMessage>();
+            client.OnCultNet<CultNetErrorMessage>(outcomes.Add);
+            client.Connect("127.0.0.1", ((IPEndPoint)serverSocket.LocalEndPoint!).Port);
+            var deadline = DateTime.UtcNow.AddSeconds(3);
+            while (!client.Connected && DateTime.UtcNow < deadline)
+                await Task.Delay(5);
+            Assert.That(client.Connected, "fixture: the client connected");
+
+            client.SendCultNet(new CultNetDocumentPutRawMessage { MessageId = "completion" });
+
+            Assert.That(outcomes.TryTake(out var refusal, 3000), Is.True, "the result frame reached the caller");
+            Assert.That(refusal!.Code, Is.EqualTo("write_not_authorized"));
+            stop.Cancel();
+            await peer;
+        }
+    }
+}

tests/GameCult.Networking.Tests/CultNetRudpSchemaServerPeerInputTests.cs

@@ -14,8 +14,9 @@ namespace GameCult.Networking.Tests
 {
     /// <summary>
     /// A schema server's poll serves every peer whatever one peer sends. A message that cannot be
-    /// decoded or handled is that peer's failure: its session ends, the drop is counted and logged,
-    /// and the drain goes on to the other peers.
+    /// decoded is answered message_invalid and its session kept until MaxConsecutiveRefusedFrames in a
+    /// row end it; one that cannot be handled ends that peer's session. The fault is logged, and the
+    /// drain goes on to the other peers.
     /// </summary>
     public class CultNetRudpSchemaServerPeerInputTests
     {
@@ -161,30 +162,165 @@ namespace GameCult.Networking.Tests
         }
 
         [Test]
-        public async Task APeerWhoseMessageCannotBeDecodedHasItsSessionEnded()
+        public async Task APeerWhoseMessageCannotBeDecodedIsAnsweredAndKept()
         {
             using var scene = await BadThenGood(Malformed());
-            var dropped = scene.Server.Stats.PacketsDropped;
 
             try { await scene.Server.PollAvailableAsync(16); }
             catch (MessagePackSerializationException) { Assert.Fail("the poll threw for a peer's input"); }
 
-            // The frame was already acknowledged, so the session cannot be kept: the peer is told,
-            // the listener forgets it, and the drop is counted.
-            Assert.That(scene.Ended, Is.EqualTo(new[] { scene.Bad.EndPoint }));
-            Assert.That(scene.Server.Peers.Select(peer => peer.RemoteEndPoint), Is.EqualTo(new[] { scene.Good.EndPoint }));
-            Assert.That(scene.Server.Stats.PacketsDropped, Is.EqualTo(dropped + 1));
-            var goodbye = Drain(scene.Bad.Socket).Single(packet => packet.PacketType == CultNetRudpPacketType.Disconnect);
-            Assert.That(scene.Bad.Session.Receive(goodbye, 1).Disconnected, Is.True);
+            Assert.That(scene.Ended, Is.Empty);
+            Assert.That(scene.Server.Peers, Has.Count.EqualTo(2));
+            var refusal = ErrorsReceivedBy(scene.Bad).Single();
+            Assert.That(refusal.Code, Is.EqualTo("message_invalid"));
+            Assert.That(refusal.Details!.Field, Is.EqualTo("message"));
             Assert.That(Drain(scene.Good.Socket).Select(packet => packet.PacketType), Has.None.EqualTo(CultNetRudpPacketType.Disconnect));
             Assert.That(scene.Logger.Errors, Has.Count.EqualTo(1));
         }
 
-        // The peer's malformed message and a well-formed one after it arrive in one datagram's
-        // delivery: the second is held for order, and both are released together. The session ends
-        // on the first, so the second belongs to no session and is never handled.
+        private static List<CultNetErrorMessage> ErrorsReceivedBy(Client client)
+        {
+            var errors = new List<CultNetErrorMessage>();
+            foreach (var packet in Drain(client.Socket))
+            {
+                var received = client.Session.Receive(packet, 1);
+                foreach (var frame in received.Delivered)
+                    if (CultNetSchemaMessageSerialization.Deserialize(frame.Payload) is CultNetErrorMessage error)
+                        errors.Add(error);
+            }
+            return errors;
+        }
+
+        // The bad peer acknowledges what it received and the server polls on.
+        private static async Task Acknowledge(Scene scene)
+        {
+            scene.Bad.Socket.SendTo(CultNetRudpPacketCodec.Encode(scene.Bad.Session.CreateAck()), scene.Server.LocalEndPoint);
+            await scene.Server.PollAvailableAsync(16);
+        }
+
+        // The bad peer has sent seven undecodable frames (BadThenGood sent the first), each answered.
+        private static async Task<Scene> AtTheCap()
+        {
+            var scene = await BadThenGood(Malformed());
+            await scene.Server.PollAvailableAsync(16);
+            ErrorsReceivedBy(scene.Bad);
+            for (var i = 0; i < 6; i++)
+            {
+                SendSchema(scene.Server, scene.Bad, Malformed());
+                await scene.Server.PollAvailableAsync(16);
+                ErrorsReceivedBy(scene.Bad);
+            }
+            Assert.That(scene.Ended, Is.Empty);
+            return scene;
+        }
+
+        // The capped frame's answer is lost on the wire: the server resends it, and the goodbye follows only
+        // once the peer has acknowledged it.
         [Test]
-        public async Task ARefusedPeersFramesStillQueuedAreNotHandled()
+        public async Task ALostCappedAnswerIsResentAndTheGoodbyeWaitsForItsAcknowledgement()
+        {
+            using var scene = await AtTheCap();
+            SendSchema(scene.Server, scene.Bad, Malformed());
+            await scene.Server.PollAvailableAsync(16);
+
+            var lost = Drain(scene.Bad.Socket);
+            Assert.That(lost.Where(p => p.PacketType == CultNetRudpPacketType.Data), Is.Not.Empty, "fixture: the answer was sent and is lost here");
+            Assert.That(lost.Select(p => p.PacketType), Has.None.EqualTo(CultNetRudpPacketType.Disconnect), "no goodbye before the answer is acknowledged");
+            Assert.That(scene.Ended, Is.Empty);
+
+            var resent = new List<CultNetRudpPacket>();
+            for (var attempt = 0; attempt < 40 && resent.Count == 0; attempt++)
+            {
+                await Task.Delay(50);
+                scene.Server.PollResends();
+                resent = Drain(scene.Bad.Socket);
+            }
+            Assert.That(resent.Select(p => p.PacketType), Has.None.EqualTo(CultNetRudpPacketType.Disconnect), "still no goodbye");
+            var answers = new List<CultNetErrorMessage>();
+            foreach (var packet in resent)
+                foreach (var frame in scene.Bad.Session.Receive(packet, 1).Delivered)
+                    if (CultNetSchemaMessageSerialization.Deserialize(frame.Payload) is CultNetErrorMessage error)
+                        answers.Add(error);
+            Assert.That(answers.Select(e => e.Code), Is.EqualTo(new[] { "message_invalid" }), "the lost answer arrived by resend");
+            Assert.That(scene.Ended, Is.Empty);
+
+            await Acknowledge(scene);
+            Assert.That(scene.Ended, Is.EqualTo(new[] { scene.Bad.EndPoint }), "acknowledged: the session ends");
+            Assert.That(Drain(scene.Bad.Socket).Select(p => p.PacketType), Has.Some.EqualTo(CultNetRudpPacketType.Disconnect));
+        }
+
+        // A peer that never acknowledges the answer is not kept for ever: the session ends after the grace.
+        [Test]
+        public async Task ACappedSessionWhoseAnswerIsNeverAcknowledgedEndsAfterTheGrace()
+        {
+            using var scene = await AtTheCap();
+            SendSchema(scene.Server, scene.Bad, Malformed());
+            await scene.Server.PollAvailableAsync(16);
+            Drain(scene.Bad.Socket);
+
+            await Task.Delay(1500); // the grace is two seconds
+            scene.Server.PollResends();
+            Assert.That(scene.Ended, Is.Empty, "inside the grace");
+            await Task.Delay(700);
+            scene.Server.PollResends();
+            Assert.That(scene.Ended, Is.EqualTo(new[] { scene.Bad.EndPoint }), "the grace ran out");
+        }
+
+        // Nothing behind the capped frame is served while its answer is delivered.
+        [Test]
+        public async Task NothingBehindTheCappedFrameIsServed()
+        {
+            using var scene = await AtTheCap();
+            SendSchema(scene.Server, scene.Bad, Malformed());
+            SendSchema(scene.Server, scene.Bad, CultNetSchemaMessageSerialization.Serialize(
+                new CultNetShardCatalogRequestMessage { MessageId = "behind-the-cap" }));
+            await scene.Server.PollAvailableAsync(16);
+            await scene.Server.PollAvailableAsync(16);
+
+            Assert.That(scene.Served.Select(s => s.MessageId), Does.Not.Contain("behind-the-cap"));
+        }
+
+        // The numbers are the shared cross-runtime cap, spelled out so changing the constant fails here.
+        [Test]
+        public async Task EightConsecutiveUndecodableFramesEndTheSessionAfterTheEighthIsAnswered()
+        {
+            Assert.That(RudpCultNetSchemaServer.MaxConsecutiveRefusedFrames, Is.EqualTo(8));
+            using var scene = await BadThenGood(Malformed());
+            await scene.Server.PollAvailableAsync(16);
+            ErrorsReceivedBy(scene.Bad);
+            async Task Refuse(int count)
+            {
+                for (var i = 0; i < count; i++)
+                {
+                    SendSchema(scene.Server, scene.Bad, Malformed());
+                    await scene.Server.PollAvailableAsync(16);
+                }
+            }
+
+            await Refuse(6); // with the first, seven in a row
+            Assert.That(ErrorsReceivedBy(scene.Bad), Has.Count.EqualTo(6));
+            Assert.That(scene.Ended, Is.Empty);
+            SendSchema(scene.Server, scene.Bad, CultNetSchemaMessageSerialization.Serialize(
+                new CultNetShardCatalogRequestMessage { MessageId = "resets-the-run" }));
+            await scene.Server.PollAvailableAsync(16);
+            Assert.That(scene.Served.Select(s => s.MessageId), Does.Contain("resets-the-run"));
+            await Refuse(7);
+            Assert.That(ErrorsReceivedBy(scene.Bad), Has.Count.EqualTo(7));
+            Assert.That(scene.Ended, Is.Empty, "a decoded frame reset the run");
+
+            await Refuse(1);
+            Assert.That(ErrorsReceivedBy(scene.Bad).Select(e => e.Code), Is.EqualTo(new[] { "message_invalid" }), "the eighth is answered too");
+            Assert.That(scene.Ended, Is.Empty, "the session ends once the answer is acknowledged");
+            await Acknowledge(scene);
+            Assert.That(scene.Ended, Is.EqualTo(new[] { scene.Bad.EndPoint }));
+            Assert.That(scene.Server.Peers.Select(peer => peer.RemoteEndPoint), Is.EqualTo(new[] { scene.Good.EndPoint }));
+        }
+
+        // The peer's malformed message and a well-formed one after it arrive in one datagram's
+        // delivery: the second is held for order, and both are released together. The session is
+        // kept through the first, so the second is handled.
+        [Test]
+        public async Task AFrameQueuedBehindAnUndecodableOneIsStillHandled()
         {
             var logger = new CapturingLogger();
             using var server = new RudpCultNetSchemaServer(new RudpCultNetSchemaServerOptions
@@ -205,8 +341,8 @@ namespace GameCult.Networking.Tests
 
             await server.PollAvailableAsync(16);
 
-            Assert.That(served, Is.Empty);
-            Assert.That(server.Peers, Is.Empty);
+            Assert.That(served, Is.EqualTo(new[] { "after-the-refused-one" }));
+            Assert.That(server.Peers, Has.Count.EqualTo(1));
             Assert.That(logger.Errors, Has.Count.EqualTo(1));
             bad.Socket.Dispose();
         }

tests/GameCult.Networking.Tests/NetworkingTests.cs

@@ -5213,6 +5213,52 @@ namespace GameCult.Networking.Tests
             Assert.That(sent.ShardEpoch, Is.EqualTo(3));
         }
 
+        [TestCase("write_not_authorized", 0)]
+        [TestCase("message_invalid", 0)]
+        [TestCase("write_not_authorized", 150)]
+        public void CultNetSchemaWriteForwarder_RejectsWithTheTypedRefusal_WhateverItsOrderAgainstTheSend(string code, int delayMs)
+        {
+            var fake = new CapturingSchemaClient(_ => null);
+            var forwarder = new CultNetSchemaWriteForwarder(new CultNetSchemaWriteForwarderOptions
+            {
+                CreateClient = () => fake,
+                ResultWindow = TimeSpan.FromSeconds(2)
+            });
+            var shard = new CultNetShardDescriptor(
+                "notes", "primary", 3, isPrimary: false,
+                primaryEndpoints: ["cultnet://primary.example.test:4075"]);
+            fake.OnSend = () =>
+            {
+                var refusal = new CultNetErrorMessage { Error = "refused", Code = code };
+                if (delayMs == 0) fake.Deliver(refusal);
+                else Task.Delay(delayMs).ContinueWith(_ => fake.Deliver(refusal));
+            };
+
+            var put = Assert.ThrowsAsync<CultNetWriteRefusedException>(() => forwarder.ForwardPutAsync(shard, new CultNetDocumentPutRawMessage()));
+            Assert.That(put!.Code, Is.EqualTo(code));
+            var delete = Assert.ThrowsAsync<CultNetWriteRefusedException>(() => forwarder.ForwardDeleteAsync(shard, new CultNetDocumentDeleteMessage { SchemaId = "note.v0", RecordKey = "note:1" }));
+            Assert.That(delete!.Code, Is.EqualTo(code));
+        }
+
+        [Test]
+        public async Task CultNetSchemaWriteForwarder_DoesNotCompleteOnTheSendReturning()
+        {
+            var fake = new CapturingSchemaClient(_ => null);
+            var forwarder = new CultNetSchemaWriteForwarder(new CultNetSchemaWriteForwarderOptions
+            {
+                CreateClient = () => fake,
+                ResultWindow = TimeSpan.FromMilliseconds(400)
+            });
+            var shard = new CultNetShardDescriptor(
+                "notes", "primary", 3, isPrimary: false,
+                primaryEndpoints: ["cultnet://primary.example.test:4075"]);
+            var started = DateTime.UtcNow;
+
+            await forwarder.ForwardDeleteAsync(shard, new CultNetDocumentDeleteMessage { SchemaId = "note.v0", RecordKey = "note:1" });
+
+            Assert.That(DateTime.UtcNow - started, Is.GreaterThanOrEqualTo(TimeSpan.FromMilliseconds(380)));
+        }
+
         [Test]
         public void CultMeshVerseCatalog_Finds_CompatibleTransferTargets()
         {
@@ -7458,6 +7504,14 @@ namespace GameCult.Networking.Tests
                 _respond = respond;
             }
 
+            public Action? OnSend { get; set; }
+
+            public void Deliver(ICultNetSchemaMessage message)
+            {
+                if (!_handlers.TryGetValue(message.GetType(), out var handlers)) return;
+                foreach (var handler in handlers.ToArray()) handler.DynamicInvoke(message);
+            }
+
             public bool Connected { get; private set; }
             public string? ConnectedHost { get; private set; }
             public int ConnectedPort { get; private set; }
@@ -7474,6 +7528,7 @@ namespace GameCult.Networking.Tests
                 where T : ICultNetSchemaMessage
             {
                 SentMessages.Add(message);
+                OnSend?.Invoke();
                 var response = _respond(message);
                 if (response == null)
                 {