cultmesh-ts, cultnet-ts: an unanswered put is unconfirmed on every server, session_ended before a result is an error, a failed local send is its own reason

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

packages/cultmesh-ts/CHANGELOG.md

@@ -7,7 +7,7 @@ All notable changes to `@gamecult/cultmesh-ts` are recorded here, following
 
 ### Added
 
-- `publishRudpDocumentOnce` takes `writeReceipts: true`: it sends a hello listing `cultnet.write_admitted.v0` and `cultnet.error.v1` and, against a server that lists both back, resolves only on that write's `write_admitted`, rejects with `CultMeshRudpWriteRefusedError` (now also carrying an `error.v1`) on its refusal and with the new `CultMeshRudpWriteUnconfirmedError` (`reason`: `window_elapsed` after 5000 ms, or `session_ended`) when nothing settles it. Without the option, or against a server that did not advertise, nothing changes: no hello, a 500 ms window, and a window with no refusal still reads as published (interim; not admission). `resultWindowMs` overrides either window. `publishRudpDocumentAndWaitForReceipt` is not changed.
+- `publishRudpDocumentOnce` takes `writeReceipts: true`: it sends a hello listing `cultnet.write_admitted.v0` and `cultnet.error.v1` and, against a server that lists both back, resolves only on that write's `write_admitted`, rejects with `CultMeshRudpWriteRefusedError` (now also carrying an `error.v1`) on its refusal and with the new `CultMeshRudpWriteUnconfirmedError` (`reason`: `window_elapsed` after 5000 ms, `session_ended`, or `local_send_failed`) when nothing settles it. Without the option, or against a server that did not advertise, there is no hello and the window is 500 ms; a put nothing settled in it is unconfirmed too, no longer "published" (**behaviour change**: a server that applies a put but answers nothing now yields `CultMeshRudpWriteUnconfirmedError`, as the spec requires). The legacy success reply (the put's id echoed) still admits it. `resultWindowMs` overrides either window. `publishRudpDocumentAndWaitForReceipt` is not changed.
 
 ### Breaking
 

packages/cultmesh-ts/src/index.ts

@@ -4205,8 +4205,8 @@ export class CultMeshRudpWriteRefusedError extends Error {
 }
 
 /**
- * Nothing settled the write (write-receipt.md, "unconfirmed"): the result window ran out with no answer, or the
- * session ended first. Not success and not refusal; a caller that needs admission treats it as not admitted.
+ * Nothing settled the write (write-receipt.md, "unconfirmed"): the result window ran out with no answer, the
+ * session ended first, or the client's own send failed. Not success and not refusal; a caller that needs admission treats it as not admitted.
  */
 export class CultMeshRudpWriteUnconfirmedError extends Error {
   readonly reason: WriteUnconfirmedReason;
@@ -4223,9 +4223,9 @@ export interface CultMeshRudpDocumentPublishOptions extends CultMeshRudpPeerOpti
   /**
    * How long, counted from the send, the put may wait for its result frame. A reliable-transport ack is not
    * admission; the outcome is the peer's result frame. Default: 5000 ms against a server whose hello lists the
-   * write-receipt frames, 500 ms otherwise. Against a server that did not advertise them the put is reported
-   * published once the window passes with no refusal (interim; the caller must not read it as admission), and
-   * a success reply ends the wait early.
+   * write-receipt frames, 500 ms otherwise. A put nothing settled within the window is unconfirmed, never
+   * published, whatever the server advertised; a legacy server's success reply (the put's id echoed back) ends
+   * the wait early as admitted.
    */
   resultWindowMs?: number;
   /**
@@ -6250,19 +6250,24 @@ export class CultMesh {
       });
       tracker = writeTracker;
       writeTracker.send(messageId, Date.now());
-      peer.send(registry.createRawDocumentPutMessage(
-        binding,
-        messageId,
-        recordKey,
-        value,
-        {
-          sourceRuntimeId: options.sourceRuntimeId ?? runtimeId,
-          sourceAgentId: options.sourceAgentId,
-          sourceRole: options.sourceRole,
-          tags: options.tags,
-        },
-      ));
-      await peer.flush(options.flushTimeoutMs ?? 30_000);
+      try {
+        peer.send(registry.createRawDocumentPutMessage(
+          binding,
+          messageId,
+          recordKey,
+          value,
+          {
+            sourceRuntimeId: options.sourceRuntimeId ?? runtimeId,
+            sourceAgentId: options.sourceAgentId,
+            sourceRole: options.sourceRole,
+            tags: options.tags,
+          },
+        ));
+        await peer.flush(options.flushTimeoutMs ?? 30_000);
+      } catch {
+        // Our own send failed or was never acknowledged: the put may or may not have left. Never a refusal.
+        writeTracker.sendFailed(messageId, Date.now());
+      }
       while (!settlement) {
         const remaining = writeTracker.nextDeadlineMs(Date.now()) ?? 0;
         await new Promise<void>((resolve) => {
@@ -6272,8 +6277,7 @@ export class CultMesh {
         writeTracker.tick(Date.now());
       }
       if (settlement.outcome === "refused") throw new CultMeshRudpWriteRefusedError(settlement.response);
-      // Interim, for a server that did not advertise receipts: a window with no refusal reads as published.
-      if (settlement.outcome === "unconfirmed" && receiptCapable) throw new CultMeshRudpWriteUnconfirmedError(messageId, settlement.reason);
+      if (settlement.outcome === "unconfirmed") throw new CultMeshRudpWriteUnconfirmedError(messageId, settlement.reason);
     } finally {
       peer.close();
     }

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

@@ -16,7 +16,7 @@ import {
   encodeRudpPacket,
 } from "@gamecult/cultnet-ts";
 import { encodeWireValue } from "@gamecult/cultnet-ts/wire";
-import { CultMesh, CultMeshGrantAllWrites, CultMeshRudpWriteRefusedError } from "../src/index";
+import { CultMesh, CultMeshGrantAllWrites, CultMeshRudpWriteRefusedError, CultMeshRudpWriteUnconfirmedError } from "../src/index";
 
 const isTheRefusal = (error: unknown): boolean =>
   error instanceof CultMeshRudpWriteRefusedError && error.code === "write_not_authorized";
@@ -100,7 +100,7 @@ for (const order of orders) {
   });
 }
 
-test("publishRudpDocumentOnce still publishes a put the server admitted", async () => {
+test("a put the server applied but did not answer is unconfirmed, never published", async () => {
   const connectionId = nextConnection++;
   const puts: string[] = [];
   const server = CultMesh.createRudpDocumentServer("completion-admit", connectionId, {
@@ -111,7 +111,10 @@ test("publishRudpDocumentOnce still publishes a put the server admitted", async
   });
   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"]);
+    await assert.rejects(
+      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 }),
+      (error: unknown) => error instanceof CultMeshRudpWriteUnconfirmedError && error.reason === "window_elapsed",
+    );
+    assert.deepEqual(puts, ["tests:completion:a"], "the server did apply it; this server just does not answer");
   } finally { server.close(); }
 });

packages/cultmesh-ts/test/cultmesh.test.ts

@@ -21,7 +21,7 @@ import {
   type CultNetMessage,
   type CultNetSnapshotResponseRawMessage,
 } from "@gamecult/cultnet-ts";
-import { CultMesh, CultMeshGrantAllWrites, CultMeshRudpUnservableDocumentError } from "../src/index";
+import { CultMesh, CultMeshGrantAllWrites, CultMeshRudpUnservableDocumentError, CultMeshRudpWriteUnconfirmedError } from "../src/index";
 
 // A record's payload as the bytes a peer would put on the wire.
 const payloadOf = (value: unknown) => documentPayloadFromBytes(encodeWireValue(value));
@@ -2968,25 +2968,30 @@ test("CultMesh TS publishes one registered document to a RUDP catalog", async ()
 
   try {
     await server.start();
-    await CultMesh.publishRudpDocumentOnce(
-      "cultmesh-ts-muninn",
-      connectionId,
-      `rudp://127.0.0.1:${server.bind.port}`,
-      defineCultNetDocumentBinding({ definition: noteDocument }),
-      "note:odin",
-      {
-        noteId: "note:odin",
-        body: "pay respects once",
-      },
-      {
-        resendDelayMs: 25,
-        resendPollMs: 5,
-        maxFragmentBytes: 1024,
-        maxPendingReliablePackets: 16,
-        connectTimeoutMs: 1_000,
-        flushTimeoutMs: 25,
-        sourceRole: "test-provider",
-      },
+    // This server answers a put with nothing, so the writer cannot know the put landed: unconfirmed, though it did.
+    await assert.rejects(
+      CultMesh.publishRudpDocumentOnce(
+        "cultmesh-ts-muninn",
+        connectionId,
+        `rudp://127.0.0.1:${server.bind.port}`,
+        defineCultNetDocumentBinding({ definition: noteDocument }),
+        "note:odin",
+        {
+          noteId: "note:odin",
+          body: "pay respects once",
+        },
+        {
+          resendDelayMs: 25,
+          resendPollMs: 5,
+          maxFragmentBytes: 1024,
+          maxPendingReliablePackets: 16,
+          connectTimeoutMs: 1_000,
+          flushTimeoutMs: 25,
+          resultWindowMs: 100,
+          sourceRole: "test-provider",
+        },
+      ),
+      (error: unknown) => error instanceof CultMeshRudpWriteUnconfirmedError && error.reason === "window_elapsed",
     );
 
     await waitFor(() => received !== undefined, "published RUDP document");

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

@@ -31,8 +31,10 @@ interface Script {
   /** Does the server's hello list both receipt frames? */
   advertise: boolean;
   /** What the server answers a put with, `delayMs` after it arrives. */
-  answer: "admit" | "refuse" | "wrong-id" | "silence";
+  answer: "admit" | "refuse" | "wrong-id" | "silence" | "goodbye";
   delayMs?: number;
+  /** Acknowledge the reliable packets it receives? Default true. */
+  ack?: boolean;
 }
 
 async function fakeServer(connectionId: number, script: Script) {
@@ -72,7 +74,9 @@ async function fakeServer(connectionId: number, script: Script) {
       if (message.schemaVersion !== "cultnet.document_put_raw.v0" || script.answer === "silence") continue;
       const timer = setTimeout(() => {
         timers.delete(timer);
-        if (script.answer === "refuse") {
+        if (script.answer === "goodbye") {
+          send(session.endUnsendableSession(new Error("scripted goodbye")), remote.port, remote.address);
+        } else if (script.answer === "refuse") {
           say({ schemaVersion: "cultnet.error.v1", error: "write_not_authorized", routingHint: null, code: "write_not_authorized", details: null, messageId: message.messageId! }, remote.port, remote.address);
         } else {
           const messageId = script.answer === "wrong-id" ? "someone-else" : message.messageId!;
@@ -81,7 +85,7 @@ async function fakeServer(connectionId: number, script: Script) {
       }, script.delayMs ?? 20);
       timers.add(timer);
     }
-    if (packet.reliable) send(session.createAckForReceived(packet.sequence), remote.port, remote.address);
+    if (packet.reliable && script.ack !== false) send(session.createAckForReceived(packet.sequence), remote.port, remote.address);
   });
   return { endpoint: `rudp://127.0.0.1:${socket.address().port}`, seen, close: () => { closed = true; timers.forEach(clearTimeout); socket.close(); } };
 }
@@ -121,19 +125,38 @@ test("a receipt for another id settles nothing", async () => {
   } finally { server.close(); }
 });
 
-test("a server whose hello does not advertise keeps the interim 500 ms window-success", async () => {
+test("a server whose hello does not advertise leaves a silent put unconfirmed after the 500 ms window, never published", async () => {
   const server = await fakeServer(nextConnection, { advertise: false, answer: "silence" });
   const started = Date.now();
   try {
-    await publish(server, { writeReceipts: true });
+    await assert.rejects(publish(server, { writeReceipts: true }), (error: unknown) =>
+      error instanceof CultMeshRudpWriteUnconfirmedError && error.reason === "window_elapsed");
     assert.ok(Date.now() - started >= 500, "the legacy window ran from the send");
   } finally { server.close(); }
 });
 
+test("a session that ends before any result leaves the put unconfirmed (session_ended), against a legacy server too", async () => {
+  for (const advertise of [false, true]) {
+    const server = await fakeServer(nextConnection, { advertise, answer: "goodbye" });
+    try {
+      await assert.rejects(publish(server, { writeReceipts: true, resultWindowMs: 3_000 }), (error: unknown) =>
+        error instanceof CultMeshRudpWriteUnconfirmedError && error.reason === "session_ended");
+    } finally { server.close(); }
+  }
+});
+
+test("a put whose send is never acknowledged is unconfirmed as local_send_failed, not an untyped error", async () => {
+  const server = await fakeServer(nextConnection, { advertise: true, answer: "silence", ack: false });
+  try {
+    await assert.rejects(publish(server, { writeReceipts: true, flushTimeoutMs: 300 }), (error: unknown) =>
+      error instanceof CultMeshRudpWriteUnconfirmedError && error.reason === "local_send_failed");
+  } finally { server.close(); }
+});
+
 test("without writeReceipts no hello is sent and the receipt frames are never asked for", async () => {
   const server = await fakeServer(nextConnection, { advertise: true, answer: "silence" });
   try {
-    await publish(server, { resultWindowMs: 100 });
+    await assert.rejects(publish(server, { resultWindowMs: 100 }), CultMeshRudpWriteUnconfirmedError);
     assert.deepEqual(server.seen, ["cultnet.document_put_raw.v0"]);
   } finally { server.close(); }
 });

packages/cultnet-ts/src/write-tracker.ts

@@ -14,7 +14,11 @@ export function resultWindowMs(serverAdvertisesReceipts: boolean): number {
   return serverAdvertisesReceipts ? RECEIPT_RESULT_WINDOW_MS : LEGACY_RESULT_WINDOW_MS;
 }
 
-export type WriteUnconfirmedReason = "window_elapsed" | "session_ended";
+/**
+ * Why nothing settled a write: its window ran out, the session ended first, or the client's own send of it failed (it may or
+ * may not have left the machine, and is never a refusal).
+ */
+export type WriteUnconfirmedReason = "window_elapsed" | "session_ended" | "local_send_failed";
 /** What a late answer is recorded as on a write that already settled unconfirmed. */
 export type WriteRecorded = "admitted_late" | "refused_late";
 
@@ -97,6 +101,13 @@ export class WriteTracker {
     }
   }
 
+  /** This client's own send of `messageId` failed: that write settles unconfirmed; the others are untouched. */
+  sendFailed(messageId: string, atMs: number): void {
+    this.tick(atMs);
+    const entry = this.#writes.get(messageId);
+    if (entry?.state.outcome === "pending") this.#settle(messageId, entry, { outcome: "unconfirmed", reason: "local_send_failed" });
+  }
+
   state(messageId: string): WriteState | undefined {
     return this.#writes.get(messageId)?.state;
   }

packages/cultnet-ts/test/write-receipt.test.ts

@@ -194,3 +194,29 @@ test("the replay runner passes every vector against this runtime", () => {
   assert.match(run.stdout, /^PASS scenario\/receipt_admits$/m);
   assert.doesNotMatch(run.stdout, /^FAIL/m);
 });
+
+test("a failed local send settles only that write unconfirmed as local_send_failed, and an answer after it is recorded late", () => {
+  const settled: [string, WriteSettlement][] = [];
+  const tracker = new WriteTracker({ windowMs: 1000, onSettle: (id, settlement) => settled.push([id, settlement]) });
+  tracker.send("w1", 0);
+  tracker.send("w2", 0);
+  tracker.sendFailed("w1", 10);
+  assert.deepEqual(settled, [["w1", { outcome: "unconfirmed", reason: "local_send_failed" }]]);
+  assert.equal(tracker.state("w2")?.outcome, "pending");
+  tracker.frame(admitted("w1"), 20);
+  assert.deepEqual(tracker.state("w1"), { outcome: "unconfirmed", reason: "local_send_failed", recorded: "admitted_late" });
+  tracker.sendFailed("w2", 30);
+  tracker.sendFailed("w2", 31);
+  assert.equal(settled.length, 2, "a write settles once");
+  tracker.sendFailed("unknown", 40);
+  assert.equal(tracker.state("unknown"), undefined);
+});
+
+test("a failed local send advances the clock: other writes whose windows have run settle window_elapsed", () => {
+  const tracker = new WriteTracker({ windowMs: 1000 });
+  tracker.send("w1", 0);
+  tracker.send("w2", 500);
+  tracker.sendFailed("w2", 1200);
+  assert.deepEqual(tracker.state("w1"), { outcome: "unconfirmed", reason: "window_elapsed" });
+  assert.deepEqual(tracker.state("w2"), { outcome: "unconfirmed", reason: "local_send_failed" });
+});