Return RUDP application receipts

packages/cultmesh-ts/src/index.ts

@@ -4141,11 +4141,25 @@ export interface CultMeshRudpDocumentPut {
   };
 }
 
+export interface CultMeshRudpDocumentReceipt<TDefinition extends AnyCultCacheDocumentDefinition = AnyCultCacheDocumentDefinition> {
+  binding: CultNetDocumentBinding<TDefinition>;
+  recordKey: string;
+  value: CultCacheDocumentValue<TDefinition>;
+  messageId?: string;
+  sourceRuntimeId?: string;
+  sourceAgentId?: string;
+  sourceRole?: string;
+  tags?: string[];
+}
+
 export interface CultMeshRudpDocumentServerOptions extends CultMeshRudpSocketOptions {
   documents: CultNetDocumentRegistry;
   getCache?: () => Promise<CultCache> | CultCache;
   onError?: (error: Error) => void;
-  onDocumentPutRaw?: (document: CultMeshRudpDocumentPut) => void | Promise<void>;
+  onDocumentPutRaw?: (document: CultMeshRudpDocumentPut) =>
+    | CultMeshRudpDocumentReceipt
+    | void
+    | Promise<CultMeshRudpDocumentReceipt | void>;
   onOperationRequest?: (
     request: CultNetOperationRequestMessage,
     session: CultMeshRudpServerSession,
@@ -4176,6 +4190,10 @@ export interface CultMeshRudpDocumentPublishOptions extends CultMeshRudpPeerOpti
   flushTimeoutMs?: number;
 }
 
+export interface CultMeshRudpDocumentReceiptOptions extends CultMeshRudpDocumentPublishOptions {
+  receiptTimeoutMs?: number;
+}
+
 export interface CultMeshRudpEndpoint {
   host: string;
   port: number;
@@ -5709,7 +5727,18 @@ export class CultMesh {
           return;
         case "cultnet.document_put_raw.v0":
           if (options.onDocumentPutRaw) {
-            await options.onDocumentPutRaw(normalizeRudpDocumentPut(message, record.remote));
+            const receipt = await options.onDocumentPutRaw(normalizeRudpDocumentPut(message, record.remote));
+            if (receipt) {
+              sendSchemaMessage(record, options.documents.createRawDocumentPutMessage(
+                receipt.binding, receipt.messageId ?? message.messageId, receipt.recordKey, receipt.value,
+                {
+                  sourceRuntimeId: receipt.sourceRuntimeId ?? runtimeId,
+                  sourceAgentId: receipt.sourceAgentId,
+                  sourceRole: receipt.sourceRole ?? "provider",
+                  tags: receipt.tags ?? ["receipt"],
+                },
+              ));
+            }
           }
           return;
         case "cultnet.operation_request.v0":
@@ -5857,6 +5886,36 @@ export class CultMesh {
     }
   }
 
+  public static async publishRudpDocumentAndWaitForReceipt<TDefinition extends AnyCultCacheDocumentDefinition, TReceiptDefinition extends AnyCultCacheDocumentDefinition>(
+    runtimeId: string, connectionId: number, endpoint: string | CultMeshRudpEndpoint,
+    binding: CultNetDocumentBinding<TDefinition>, recordKey: string, value: CultCacheDocumentValue<TDefinition>,
+    receiptBinding: CultNetDocumentBinding<TReceiptDefinition>, options: CultMeshRudpDocumentReceiptOptions = {},
+  ): Promise<CultCacheDocumentValue<TReceiptDefinition>> {
+    requireNonEmpty(runtimeId, "runtimeId");
+    requireNonEmpty(recordKey, "recordKey");
+    const messageId = options.messageId ?? `${runtimeId}:${binding.definition.type}:${recordKey}`;
+    const peer = await CultMesh.createRudpPeer(runtimeId, connectionId, endpoint, options);
+    try {
+      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.document_put_raw.v0" || message.messageId !== messageId || message.document.schemaId !== receiptBinding.definition.schemaId) return;
+          try {
+            const decoded = decode(new Uint8Array(message.document.payload));
+            cleanup();
+            resolve(receiptBinding.definition.schema.parse(decoded) as CultCacheDocumentValue<TReceiptDefinition>);
+          } catch (error) { cleanup(); reject(error); }
+        };
+        const cleanup = (): void => { clearTimeout(timeout); peer.off("message", onMessage); };
+        peer.on("message", onMessage);
+      });
+      peer.send(new CultNetDocumentRegistry([binding]).createRawDocumentPutMessage(binding, messageId, recordKey, value, {
+        sourceRuntimeId: options.sourceRuntimeId ?? runtimeId, sourceAgentId: options.sourceAgentId, sourceRole: options.sourceRole, tags: options.tags,
+      }));
+      return await receipt;
+    } finally { peer.close(); }
+  }
+
   public static async createRudpServer(
     runtimeId: string,
     connectionId: number,

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

@@ -34,6 +34,13 @@ const noteAliasDocument = defineDocumentType({
   name: "noteId",
 });
 
+const noteReceiptDocument = defineDocumentType({
+  type: "cultmesh.note_receipt",
+  schemaId: "cultmesh.note_receipt.v0",
+  schema: z.object({ commandId: z.string(), state: z.literal("applied"), body: z.string() }),
+  name: "commandId",
+});
+
 const foreignNoteDocument = defineDocumentType({
   type: "runtime.generated.cultmesh.note",
   schemaId: "runtime.generated.cultmesh.note.ui.42",
@@ -2638,6 +2645,29 @@ test("CultMesh TS publishes one registered document to a RUDP catalog", async ()
   }
 });
 
+test("CultMesh TS returns a correlated provider receipt after a RUDP document mutation", async () => {
+  const connectionId = 0x10203050;
+  const commandBinding = defineCultNetDocumentBinding({ definition: noteDocument });
+  const receiptBinding = defineCultNetDocumentBinding({ definition: noteReceiptDocument });
+  const server = CultMesh.createRudpDocumentServer("cultmesh-ts-receipt-server", connectionId, {
+    documents: new CultNetDocumentRegistry([commandBinding, receiptBinding]), bindHost: "127.0.0.1", bindPort: 0,
+    resendDelayMs: 25, resendPollMs: 5,
+    onDocumentPutRaw: async (document) => ({
+      binding: receiptBinding, recordKey: document.recordKey,
+      value: { commandId: document.recordKey, state: "applied", body: (document.payload as { body: string }).body },
+    }),
+  });
+  try {
+    await server.start();
+    const receipt = await CultMesh.publishRudpDocumentAndWaitForReceipt(
+      "cultmesh-ts-receipt-client", connectionId, `rudp://127.0.0.1:${server.bind.port}`,
+      commandBinding, "command:note", { noteId: "command:note", body: "provider applied this" }, receiptBinding,
+      { messageId: "command:note", receiptTimeoutMs: 1_000, resendPollMs: 5 },
+    );
+    assert.deepEqual(receipt, { commandId: "command:note", state: "applied", body: "provider applied this" });
+  } finally { server.close(); }
+});
+
 test("CultMesh TS reads remote RUDP snapshots through document handles", async () => {
   const connectionId = 0x10203048;
   const node = await CultMesh.startNode(