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/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

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/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/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/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)
                 {