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-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(
@@ -1454,8 +1462,16 @@ mod tests {
     /// 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(
-        refusal_first: bool,
+        reply: CatalogReply,
     ) -> Result<(
         SocketAddr,
         std::sync::Arc<std::sync::atomic::AtomicBool>,
@@ -1503,10 +1519,17 @@ mod tests {
                 }
                 let ack = session.create_ack_for_received(packet.sequence);
                 let refusal = encode_cultnet_message_to_vec(
-                    &cultnet_rs::CultNetMessage::Error {
-                        error: "write_not_authorized: the catalog refused the document".into(),
-                        code: Some(cultnet_rs::CultNetErrorCode::WriteNotAuthorized),
-                        details: None,
+                    &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,
                 )?;
@@ -1528,7 +1551,7 @@ mod tests {
                     }
                     Ok(())
                 };
-                if refusal_first {
+                if reply == CatalogReply::RefusalThenAck {
                     send_refusals(&socket)?;
                     thread::sleep(Duration::from_millis(100));
                     socket.send_to(&encode_rudp_packet(&ack)?, peer)?;
@@ -2026,7 +2049,7 @@ mod tests {
     // 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(true)?;
+        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")?;
@@ -2035,4 +2058,49 @@ mod tests {
         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-ts/src/index.ts

@@ -4183,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;
@@ -6157,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,
         {
@@ -6172,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();
     }
@@ -6190,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/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);

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;
+        }
+    }
+}