packages/cultmesh-ts/CHANGELOG.md

@@ -10,6 +10,7 @@ All notable changes to `@gamecult/cultmesh-ts` are recorded here, following
 - The package is renamed `cultmesh-ts` to `@gamecult/cultmesh-ts` (the repo already scopes `@gamecult/cultcache-ts`), so the unscoped name cannot be claimed by someone else. Import from `@gamecult/cultmesh-ts` and its subpaths; the API is unchanged. Depends on `@gamecult/cultnet-ts`.
 - `asSchemaAlias` is removed from `CultMeshDocumentHandle`, `CultMeshCollectionHandle` and their bound forms. Read the handle by the schema you mean.
 - `CultMeshDocumentCatalog` (`document`, `tryDocument`, `latest`, `replace`, `submitPrediction`, `authoritativeWriter`, `predictionWriter`, `watch`) and `CultMeshCollectionCatalog` (`collection`, `tryCollection`, `latest`, `watchChanges`) take `<TDocument = never, TSchema>` and yield `CultMeshCatalogValue<TDocument, TSchema>`: an explicit type argument wins, a CultCache document definition names its own value type, and a bare schema descriptor yields `unknown`. A value type is no longer inferred from a value or an expected result. The `parse` option is removed from these calls, and a lookup by type answers with that type's handle or none, not an alias of a compatible schema.
+- `createRudpDocumentServer` refuses every peer put and delete unless it is given a `writeAuthority`. A refused write is answered with the typed `write_not_authorized` naming the record key (the bytes pinned in `error-vectors.json`), and `onDocumentPutRaw` is never called; pass `writeAuthority: new CultMeshGrantAllWrites()` to keep the old accept-everything behaviour. The authority is asked before the put is sized or decoded and before any handler runs, with the canonical schema id and the peer's address (the schema id is resolved to its canonical form first, which is the timing residual below); one that throws, rejects or answers anything but `granted: true` refuses. A frame the server cannot read is judged as the write it looks like (a put when it does not decode at all), so an ungranted peer gets `write_not_authorized` for it too; a granted peer gets `onError` and a `malformed_document_put` or `malformed_document_delete` refusal. A put carrying a wrong-typed `shardId` or `shardEpoch` is judged with null shard fields and, once granted, refused with `malformed_document_put`. A refusal's record key and reason are clipped (64 and 96 UTF-8 bytes) so it fits `maxPayloadBytes`. Residual: ungranted peers can still distinguish registered from unregistered schemas by timing. A peer `document_delete` now reaches the new `onDocumentDeleteRaw` once granted (without it, a text refusal) instead of the unsupported-request error.
 - Adding a document or collection to a catalog under a schema id or schema name another type already holds is refused with `SchemaConflictError`.
 
 ### Added

packages/cultmesh-ts/README.md

@@ -345,6 +345,8 @@ evidence for the requested provider identity.
 
 ## RUDP Helpers
 
+`createRudpDocumentServer` refuses every peer put and delete unless given a `writeAuthority` (`new CultMeshGrantAllWrites()` accepts all). The authority is asked before a write's payload is sized or decoded and before any handler runs, and a refusal is the typed `write_not_authorized`. Residual: ungranted peers can still distinguish registered from unregistered schemas by timing.
+
 Node runtimes can build the shared CultNet reliable-UDP transport from the
 branded CultMesh entrypoint while the transport semantics stay owned by
 `@gamecult/cultnet-ts`:

packages/cultmesh-ts/src/index.ts

@@ -21,6 +21,7 @@ import {
   rudpClientBindHost,
   CultNetSchemaCatalog,
   cultNetTextError,
+  cultNetWriteNotAuthorizedError,
   CultNetShardCatalog,
   cultNetBuiltinSchemaRegistry,
   decodeRudpPacket,
@@ -32,6 +33,7 @@ import {
   sendRudpDatagram,
   type CultNetErrorMessage,
   type CultNetDocumentBinding,
+  type CultNetDocumentDeleteMessage,
   type CultNetMessage,
   type CultNetOperationRequestMessage,
   type CultNetOperationResponseMessage,
@@ -3992,6 +3994,63 @@ export type CultMeshRudpDocumentPut = {
   | { readonly decoded: false; readonly payload: DocumentPayload }
 );
 
+/** The peer a write arrived from: the datagram's source, nothing the peer claimed. */
+export interface CultMeshRemoteWritePeer {
+  address: string;
+  family: string;
+  port: number;
+}
+
+export type CultMeshRemoteWriteOperation = "put" | "delete";
+
+/**
+ * One peer write as the authority sees it. `schemaId` is the canonical id the registry resolved for the write (the id
+ * as sent when no binding takes it), never an alias the peer used when a binding resolves it. `shardId` and
+ * `shardEpoch` are as a put carried them, `null` when it sent none (always `null` for a delete, which has none); a
+ * put carrying either with the wrong type is refused before any authority is asked.
+ */
+export interface CultMeshRemoteWriteRequest {
+  peer: CultMeshRemoteWritePeer;
+  operation: CultMeshRemoteWriteOperation;
+  schemaId: string;
+  recordKey: string;
+  shardId: string | null;
+  shardEpoch: number | null;
+}
+
+export interface CultMeshRemoteWriteDecision {
+  granted: boolean;
+  /** Told to the refused peer after `write_not_authorized:`; keep it free of anything secret. */
+  reason?: string;
+}
+
+/**
+ * The one question a document server asks before it applies a peer's put or delete (C# `ICultNetRemoteWriteAuthority`).
+ * A server with no authority refuses every peer write; one whose authority throws, rejects or answers anything but
+ * `granted: true` refuses too.
+ */
+export interface CultMeshRemoteWriteAuthority {
+  decide(request: CultMeshRemoteWriteRequest): CultMeshRemoteWriteDecision | Promise<CultMeshRemoteWriteDecision>;
+}
+
+/**
+ * Grants every peer write. For a host that has already authenticated its peers, or a test rig: it is an explicit
+ * choice because serving without an authority refuses writes.
+ */
+export class CultMeshGrantAllWrites implements CultMeshRemoteWriteAuthority {
+  decide(): CultMeshRemoteWriteDecision {
+    return { granted: true };
+  }
+}
+
+/** A delete as `onDocumentDeleteRaw` receives it, after the authority granted it. */
+export interface CultMeshRudpDocumentDelete {
+  messageId: string;
+  schemaId: string;
+  recordKey: string;
+  remote: CultMeshRemoteWritePeer;
+}
+
 export interface CultMeshRudpDocumentReceipt<TDefinition extends AnyCultCacheDocumentDefinition = AnyCultCacheDocumentDefinition> {
   binding: CultNetDocumentBinding<TDefinition>;
   recordKey: string;
@@ -4031,10 +4090,18 @@ export interface CultMeshRudpDocumentServerOptions extends CultMeshRudpSocketOpt
   documents: CultNetDocumentRegistry;
   getCache?: () => Promise<CultCache> | CultCache;
   onError?: (error: Error) => void;
+  /**
+   * Who may put and delete. Without it every peer put and delete is refused with a typed `write_not_authorized`
+   * (and `onDocumentPutRaw` is never called); pass `new CultMeshGrantAllWrites()` to accept them all. The authority is
+   * asked before the document is sized, resolved or decoded.
+   */
+  writeAuthority?: CultMeshRemoteWriteAuthority;
   onDocumentPutRaw?: (document: CultMeshRudpDocumentPut) =>
     | CultMeshRudpDocumentReceipt
     | void
     | Promise<CultMeshRudpDocumentReceipt | void>;
+  /** Applies a delete the authority granted. Without it a granted delete is answered with a text refusal. */
+  onDocumentDeleteRaw?: (request: CultMeshRudpDocumentDelete) => void | Promise<void>;
   onOperationRequest?: (
     request: CultNetOperationRequestMessage,
     session: CultMeshRudpServerSession,
@@ -5574,7 +5641,21 @@ export class CultMesh {
       payload: Uint8Array,
     ): Promise<void> {
       if (record.closed) return;
-      const message = parseCultNetMessage(decodeWireValue(payload), wireContract);
+      let decoded: unknown;
+      let message: CultNetMessage;
+      try {
+        decoded = decodeWireValue(payload);
+        message = parseCultNetMessage(decoded, wireContract);
+      } catch (error) {
+        // 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, malformedWriteError(write?.operation ?? "put", "CultMesh RUDP document server could not read the write."));
+        return;
+      }
       switch (message.schemaVersion) {
         case "cultnet.snapshot_request.v0": {
           if (!options.getCache) {
@@ -5607,6 +5688,24 @@ export class CultMesh {
           sendSchemaMessage(record, cultNetBuiltinSchemaRegistry.createCatalogResponse(message));
           return;
         case "cultnet.document_put_raw.v0":
+          {
+            // The parser passes shard fields of any type through a put. The authority is asked first, with the shard
+            // fields it can trust (null when wrong-typed), so an ungranted peer hears only the refusal; a granted peer
+            // whose shard fields are wrong-typed is then refused as malformed.
+            const { shardId, shardEpoch } = message as { shardId?: unknown; shardEpoch?: unknown };
+            const malformedShard =
+              (shardId !== undefined && typeof shardId !== "string") ||
+              (shardEpoch !== undefined && !Number.isSafeInteger(shardEpoch));
+            if (await writeRefused(
+              record, "put", message.document.schemaId, message.document.recordKey,
+              malformedShard ? null : (shardId as string | undefined) ?? null,
+              malformedShard ? null : (shardEpoch as number | undefined) ?? null,
+            )) return;
+            if (malformedShard) {
+              sendSchemaMessage(record, malformedWriteError("put", "Raw put messages must carry a string shardId and an integer shardEpoch."));
+              return;
+            }
+          }
           if (options.onDocumentPutRaw) {
             const unservable = unservablePut(message);
             if (unservable) {
@@ -5626,6 +5725,20 @@ export class CultMesh {
             }
           }
           return;
+        case "cultnet.document_delete.v0": {
+          if (await writeRefused(record, "delete", message.schemaId, message.recordKey, null, null)) return;
+          if (!options.onDocumentDeleteRaw) {
+            sendSchemaMessage(record, cultNetTextError("CultMesh RUDP document server has no delete handler."));
+            return;
+          }
+          await options.onDocumentDeleteRaw({
+            messageId: message.messageId,
+            schemaId: message.schemaId,
+            recordKey: message.recordKey,
+            remote: { ...record.remote },
+          });
+          return;
+        }
         case "cultnet.operation_request.v0":
           if (!options.onOperationRequest) {
             sendSchemaMessage(record, cultNetTextError("CultMesh RUDP document server has no operation handler."));
@@ -5638,6 +5751,77 @@ export class CultMesh {
       }
     }
 
+    function clipUtf8(text: string, maxBytes: number): string {
+      const bytes = Buffer.from(text, "utf8");
+      return bytes.byteLength <= maxBytes ? text : new TextDecoder().decode(bytes.subarray(0, maxBytes)).replace(/\uFFFD+$/, "");
+    }
+
+    /** The put or delete a decoded frame claims to be, with the ids it carries when they are strings. */
+    function readableWrite(value: unknown): { operation: CultMeshRemoteWriteOperation; schemaId: string; recordKey: string } | undefined {
+      if (typeof value !== "object" || value === null) return undefined;
+      const frame = value as { schemaVersion?: unknown; schemaId?: unknown; recordKey?: unknown; document?: unknown };
+      const str = (v: unknown): string => (typeof v === "string" ? v : "");
+      if (frame.schemaVersion === "cultnet.document_delete.v0") {
+        return { operation: "delete", schemaId: str(frame.schemaId), recordKey: str(frame.recordKey) };
+      }
+      if (frame.schemaVersion === "cultnet.document_put_raw.v0") {
+        const document = (typeof frame.document === "object" && frame.document !== null ? frame.document : {}) as { schemaId?: unknown; recordKey?: unknown };
+        return { operation: "put", schemaId: str(document.schemaId), recordKey: str(document.recordKey) };
+      }
+      return undefined;
+    }
+
+    // Not a `code`: malformed_document_put/delete are outside the schema's closed code enum, which TS and Rust peers
+    // refuse to decode (docs/cultnet-error-contract-cut.md), so the name leads the text of a codeless error instead.
+    function malformedWriteError(operation: CultMeshRemoteWriteOperation, text: string): CultNetMessage {
+      return cultNetTextError(`malformed_document_${operation}: ${text}`);
+    }
+
+    /**
+     * Asks the write authority about a put or delete, before anything of the write is sized, resolved or decoded, and
+     * answers a refusal with the typed `write_not_authorized` naming the record. True when the write is refused.
+     * No authority refuses; an authority that throws, rejects or does not say `granted: true` refuses too (fail
+     * closed), and a failure is reported without quoting it.
+     */
+    async function writeRefused(
+      record: SessionRecord,
+      operation: CultMeshRemoteWriteOperation,
+      schemaId: string,
+      recordKey: string,
+      shardId: string | null,
+      shardEpoch: number | null,
+    ): Promise<boolean> {
+      let reason: string | undefined;
+      const authority = options.writeAuthority;
+      if (!authority) {
+        reason = "no remote write authority is configured";
+      } else {
+        try {
+          const decision = await authority.decide({
+            peer: { ...record.remote },
+            operation,
+            schemaId: options.documents.canonicalSchemaId(schemaId) ?? schemaId,
+            recordKey,
+            shardId,
+            shardEpoch,
+          });
+          if (decision?.granted === true) return false;
+          reason = typeof decision?.reason === "string" ? decision.reason : undefined;
+        } catch (error) {
+          reportError(new Error(`CultMesh remote write authority failed (${error instanceof Error ? error.name : typeof error}); the write was refused.`));
+          reason = "the write authority failed";
+        }
+      }
+      try {
+        // The key is the peer's and the reason the authority's: both are clipped so the refusal fits `maxPayloadBytes`
+        // whatever they are (the peer is told the clipped key; a `maxPayloadBytes` under about 512 can still not carry it).
+        sendSchemaMessage(record, cultNetWriteNotAuthorizedError(clipUtf8(recordKey, 64), reason === undefined ? undefined : clipUtf8(reason, 96)));
+      } catch (error) {
+        reportError(error);
+      }
+      return true;
+    }
+
     function publicSession(record: SessionRecord): CultMeshRudpServerSession {
       return {
         sessionId: record.sessionId,

packages/cultmesh-ts/test/codec-documents.test.ts

@@ -20,7 +20,7 @@ import {
   defineCultNetDocumentBinding,
   type CultNetPeer,
 } from "@gamecult/cultnet-ts";
-import { CultMesh, cultMeshDocumentFromSingleFile, type CultMeshRudpDocumentPut } from "../src/index";
+import { CultMesh, CultMeshGrantAllWrites, cultMeshDocumentFromSingleFile, type CultMeshRudpDocumentPut } from "../src/index";
 
 // A document written in the reference's positional shape, and a plain map document beside it. Every mesh reader of a
 // payload reads it the way the cache does, so both read through the mesh.
@@ -77,6 +77,7 @@ test("the mesh syncs a codec document from a peer snapshot and returns a codec r
   const receiptBinding = defineCultNetDocumentBinding({ definition: codecReceipt });
   const puts: unknown[] = [];
   const server = CultMesh.createRudpDocumentServer("cultmesh-ts-codec-server", connectionId, {
+    writeAuthority: new CultMeshGrantAllWrites(),
     documents: new CultNetDocumentRegistry([commandBinding, plainBinding, receiptBinding]),
     getCache: () => source.cache,
     bindHost: "127.0.0.1", bindPort: 0, resendDelayMs: 25, resendPollMs: 5,
@@ -131,6 +132,7 @@ test("a RUDP put naming a schema the server has no definition for reaches the ha
   const errors: Error[] = [];
   const puts: CultMeshRudpDocumentPut[] = [];
   const server = CultMesh.createRudpDocumentServer("cultmesh-ts-unregistered-server", connectionId, {
+    writeAuthority: new CultMeshGrantAllWrites(),
     documents: new CultNetDocumentRegistry([known]),
     bindHost: "127.0.0.1", bindPort: 0, resendDelayMs: 25, resendPollMs: 5,
     onError: (error) => errors.push(error),
@@ -171,6 +173,7 @@ test("a payload past the nesting limit is refused at the network doors: a RUDP p
   const errors: unknown[] = [];
   const puts: CultMeshRudpDocumentPut[] = [];
   const server = CultMesh.createRudpDocumentServer("cultmesh-ts-depth-server", connectionId, {
+    writeAuthority: new CultMeshGrantAllWrites(),
     documents: registry,
     bindHost: "127.0.0.1", bindPort: 0, resendDelayMs: 25, resendPollMs: 5,
     onError: (error) => errors.push(error),

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

@@ -21,7 +21,7 @@ import {
   type CultNetMessage,
   type CultNetSnapshotResponseRawMessage,
 } from "@gamecult/cultnet-ts";
-import { CultMesh, CultMeshRudpUnservableDocumentError } from "../src/index";
+import { CultMesh, CultMeshGrantAllWrites, CultMeshRudpUnservableDocumentError } from "../src/index";
 
 // A record's payload as the bytes a peer would put on the wire.
 const payloadOf = (value: unknown) => documentPayloadFromBytes(encodeWireValue(value));
@@ -2857,6 +2857,7 @@ test("CultMesh TS RUDP document server accepts raw document puts from multiple p
     "cultmesh-ts-document-server",
     connectionId,
     {
+      writeAuthority: new CultMeshGrantAllWrites(),
       documents: new CultNetDocumentRegistry(),
       bindHost: "127.0.0.1",
       bindPort: 0,
@@ -2945,6 +2946,7 @@ test("CultMesh TS publishes one registered document to a RUDP catalog", async ()
     "cultmesh-ts-odin-catalog",
     connectionId,
     {
+      writeAuthority: new CultMeshGrantAllWrites(),
       documents: new CultNetDocumentRegistry(),
       bindHost: "127.0.0.1",
       bindPort: 0,
@@ -3020,6 +3022,7 @@ test("CultMesh TS returns a correlated provider receipt after a RUDP document mu
   const commandBinding = defineCultNetDocumentBinding({ definition: noteDocument });
   const receiptBinding = defineCultNetDocumentBinding({ definition: noteReceiptDocument });
   const server = CultMesh.createRudpDocumentServer("cultmesh-ts-receipt-server", connectionId, {
+    writeAuthority: new CultMeshGrantAllWrites(),
     documents: new CultNetDocumentRegistry([commandBinding, receiptBinding]), bindHost: "127.0.0.1", bindPort: 0,
     resendDelayMs: 25, resendPollMs: 5,
     onDocumentPutRaw: async (document) => ({
@@ -3055,6 +3058,7 @@ test("CultMesh TS reads remote RUDP snapshots through document handles", async (
     "cultmesh-ts-rudp-snapshot-server",
     connectionId,
     {
+      writeAuthority: new CultMeshGrantAllWrites(),
       documents: new CultNetDocumentRegistry([
         defineCultNetDocumentBinding({ definition: noteDocument }),
       ]),
@@ -3152,6 +3156,7 @@ test("CultMesh TS syncs remote RUDP snapshots into a local node", async () => {
     "cultmesh-ts-rudp-sync-server",
     connectionId,
     {
+      writeAuthority: new CultMeshGrantAllWrites(),
       documents: new CultNetDocumentRegistry([
         defineCultNetDocumentBinding({ definition: noteDocument }),
       ]),
@@ -3233,6 +3238,7 @@ test("CultMesh TS peer snapshots never take another type's record at the key", a
     "cultmesh-ts-rudp-foreign-snapshot-server",
     connectionId,
     {
+      writeAuthority: new CultMeshGrantAllWrites(),
       documents: new CultNetDocumentRegistry([
         defineCultNetDocumentBinding({ definition: incompatibleNoteDocument }),
         defineCultNetDocumentBinding({ definition: foreignNoteDocument }),
@@ -3474,6 +3480,7 @@ async function withServedBoundRig(
   const messages: CultNetMessage[] = [];
   const unparsed: Error[] = [];
   const server = CultMesh.createRudpDocumentServer(SERVED_BOUND_SERVER, connectionId, {
+    writeAuthority: new CultMeshGrantAllWrites(),
     documents: registry,
     getCache: () => node.cache,
     bindHost: "127.0.0.1",
@@ -3635,6 +3642,7 @@ async function withRawPeerRig(
   for (const put of stored) await registry.applyRawDocumentPutMessage(node.cache, put);
   const errors: Error[] = [];
   const server = CultMesh.createRudpDocumentServer(SERVED_BOUND_SERVER, connectionId, {
+    writeAuthority: new CultMeshGrantAllWrites(),
     documents: registry,
     getCache: () => (getCache ? getCache(node.cache) : node.cache) as typeof node.cache,
     bindHost: "127.0.0.1",

packages/cultmesh-ts/test/provider-rudp-transport.test.ts

@@ -14,6 +14,7 @@ import {
 
 import {
   CultMesh,
+  CultMeshGrantAllWrites,
   CultMeshProviderRudpTransport,
   CultMeshProviderSessionBroker,
   CULTMESH_PROVIDER_SESSION_SERVICE_ID,
@@ -64,6 +65,7 @@ test("RUDP provider transport requires broker acceptance for the full lifecycle"
     },
   });
   const server = CultMesh.createRudpDocumentServer("odin-test", 0x43554c54, {
+    writeAuthority: new CultMeshGrantAllWrites(),
     bindPort: 0,
     documents: new CultNetDocumentRegistry(),
     onOperationRequest: (request, session) => broker.handle(request, session),
@@ -256,6 +258,7 @@ test("a replaced RUDP generation drops operations queued behind an active handle
   let handlerCalls = 0;
   let closedSessions = 0;
   const server = CultMesh.createRudpDocumentServer("odin-queue-test", connectionId, {
+    writeAuthority: new CultMeshGrantAllWrites(),
     bindPort: 0,
     documents: new CultNetDocumentRegistry(),
     onOperationRequest: async request => {
@@ -326,6 +329,7 @@ test("a client restarted on the same address with the same Connect payload repla
   const connectionId = 0x43554c54;
   let closedSessions = 0;
   const server = CultMesh.createRudpDocumentServer("odin-restart-test", connectionId, {
+    writeAuthority: new CultMeshGrantAllWrites(),
     bindPort: 0,
     documents: new CultNetDocumentRegistry(),
     onSessionClosed: () => { closedSessions += 1; },
@@ -364,6 +368,7 @@ test("a client restarted on the same address with the same Connect payload repla
 test("the RUDP document server does not acknowledge by name a reliable packet its session did not receive", async () => {
   const connectionId = 0x43554c55;
   const server = CultMesh.createRudpDocumentServer("odin-ack-test", connectionId, {
+    writeAuthority: new CultMeshGrantAllWrites(),
     bindPort: 0,
     documents: new CultNetDocumentRegistry(),
     onError: () => undefined,

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

@@ -0,0 +1,461 @@
+// 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 { 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 { encodeWireValue } from "@gamecult/cultnet-ts/wire";
+import {
+  CultNetDocumentRegistry,
+  CultNetPeer,
+  cultNetWriteNotAuthorizedError,
+  defineCultNetDocumentBinding,
+  type CultNetMessage,
+} from "@gamecult/cultnet-ts";
+import {
+  CultMesh,
+  CultMeshGrantAllWrites,
+  type CultMeshRemoteWriteAuthority,
+  type CultMeshRemoteWriteDecision,
+  type CultMeshRemoteWriteRequest,
+  type CultMeshRudpDocumentDelete,
+  type CultMeshRudpDocumentPut,
+} from "../src/index";
+
+const SCHEMA_ID = "cultmesh.authority_note.v0";
+const ALIAS = "runtime.generated.authority-note.ui.42";
+const KEY_A = "tests:authority:a";
+const KEY_B = "tests:authority:b";
+
+const note = defineDocumentType({
+  type: "cultmesh.authority_note",
+  schemaId: SCHEMA_ID,
+  compatibleSchemaIds: [ALIAS],
+  schema: z.object({ noteId: z.string(), text: z.string() }),
+  name: "noteId",
+});
+
+const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));
+
+function put(recordKey: string, extra: Record<string, unknown> = {}, schemaId = SCHEMA_ID, payload?: Uint8Array) {
+  return {
+    schemaVersion: "cultnet.document_put_raw.v0" as const,
+    messageId: `put-${recordKey}`,
+    document: {
+      schemaId,
+      recordKey,
+      storedAt: "2026-11-07T00:00:00Z",
+      payloadEncoding: "messagepack" as const,
+      payload: documentPayloadFromBytes(payload ?? encodeWireValue({ noteId: recordKey, text: "peer" })),
+    },
+    ...extra,
+  } as CultNetMessage;
+}
+
+function del(recordKey: string, schemaId = SCHEMA_ID): CultNetMessage {
+  return { schemaVersion: "cultnet.document_delete.v0", messageId: `delete-${recordKey}`, schemaId, recordKey };
+}
+
+class Recording implements CultMeshRemoteWriteAuthority {
+  readonly requests: CultMeshRemoteWriteRequest[] = [];
+  constructor(private readonly answer: (request: CultMeshRemoteWriteRequest) => unknown) {}
+  decide(request: CultMeshRemoteWriteRequest): CultMeshRemoteWriteDecision | Promise<CultMeshRemoteWriteDecision> {
+    this.requests.push(request);
+    return this.answer(request) as CultMeshRemoteWriteDecision | Promise<CultMeshRemoteWriteDecision>;
+  }
+}
+
+const grant = () => ({ granted: true });
+
+interface Rig {
+  puts: CultMeshRudpDocumentPut[];
+  deletes: CultMeshRudpDocumentDelete[];
+  errors: Error[];
+  replies: CultNetMessage[];
+  peer: CultNetPeer;
+  settle(): Promise<unknown>;
+  close(): void;
+}
+
+let nextConnection = 0x10204000;
+
+async function rig(writeAuthority?: CultMeshRemoteWriteAuthority, extra: { maxPayloadBytes?: number } = {}): Promise<Rig> {
+  const connectionId = nextConnection++;
+  const puts: CultMeshRudpDocumentPut[] = [];
+  const deletes: CultMeshRudpDocumentDelete[] = [];
+  const errors: Error[] = [];
+  const server = CultMesh.createRudpDocumentServer("authority-server", connectionId, {
+    documents: new CultNetDocumentRegistry([defineCultNetDocumentBinding({ definition: note })]),
+    writeAuthority,
+    bindHost: "127.0.0.1",
+    bindPort: 0,
+    resendDelayMs: 25,
+    resendPollMs: 5,
+    onDocumentPutRaw: (document) => { puts.push(document); },
+    onDocumentDeleteRaw: (request) => { deletes.push(request); },
+    onError: (error) => errors.push(error),
+    ...extra,
+  });
+  await server.start();
+  const peer: CultNetPeer = await CultMesh.createRudpPeer("authority-peer", connectionId, `rudp://127.0.0.1:${server.bind.port}`, {
+    resendDelayMs: 25,
+    resendPollMs: 5,
+    connectTimeoutMs: 2_000,
+  });
+  const replies: CultNetMessage[] = [];
+  peer.on("message", (message) => replies.push(message));
+  return {
+    puts, deletes, errors, replies, peer,
+    // A refusal is a reply; a grant sends none, so settle on a quiet period after the last send.
+    settle: () => sleep(250),
+    close: () => { peer.close(); server.close(); },
+  };
+}
+
+async function using(r: Rig, body: (r: Rig) => Promise<unknown>): Promise<void> {
+  try { await body(r); } finally { r.close(); }
+}
+
+function assertRefused(reply: CultNetMessage | undefined, key: string, error?: string): void {
+  assert.ok(reply && reply.schemaVersion === "cultnet.error.v0");
+  assert.equal(reply.code, "write_not_authorized");
+  assert.ok(reply.error.startsWith("write_not_authorized"));
+  assert.deepEqual(reply.details, { field: "recordKey", value: key, asOf: null, current: null });
+  if (error !== undefined) assert.equal(reply.error, error);
+}
+
+test("with no authority a peer's put and delete are refused and the handlers never run", async () => {
+  await using(await rig(undefined), async (r) => {
+    r.peer.send(put(KEY_A));
+    r.peer.send(del(KEY_A));
+    r.peer.send(put(KEY_B));
+    await r.settle();
+    assert.equal(r.replies.length, 3);
+    assertRefused(r.replies[0], KEY_A);
+    assertRefused(r.replies[1], KEY_A);
+    assertRefused(r.replies[2], KEY_B);
+    assert.deepEqual([r.puts.length, r.deletes.length], [0, 0]);
+    assert.deepEqual(r.errors, []);
+  });
+});
+
+test("a granted put and delete reach their handlers, send nothing back, and the authority saw this peer", async () => {
+  const authority = new Recording(grant);
+  await using(await rig(authority), async (r) => {
+    r.peer.send(put(KEY_A));
+    r.peer.send(del(KEY_A));
+    await r.settle();
+    assert.deepEqual(r.replies, []);
+    assert.deepEqual(r.puts.map((p) => p.recordKey), [KEY_A]);
+    assert.deepEqual(r.deletes.map((d) => d.recordKey), [KEY_A]);
+    assert.equal(authority.requests.length, 2);
+    for (const request of authority.requests) {
+      assert.equal(request.peer.address, "127.0.0.1");
+      assert.ok(request.peer.port > 0);
+    }
+    assert.deepEqual(authority.requests[1]!.peer, r.deletes[0]!.remote, "the delete handler sees the peer the authority was asked about");
+  });
+});
+
+test("the request carries the resolved write, and scoped grants refuse their neighbours", async () => {
+  const authority = new Recording((request) =>
+    request.operation === "put" && request.recordKey === KEY_A ? { granted: true } : { granted: false, reason: "scoped" });
+  await using(await rig(authority), async (r) => {
+    r.peer.send(put(KEY_A, { shardId: "authority-shard", shardEpoch: 1 }, ALIAS));
+    await r.settle();
+    const request = authority.requests[0]!;
+    assert.equal(request.schemaId, SCHEMA_ID, "the authority sees the canonical id, not the alias as sent");
+    assert.equal(request.operation, "put");
+    assert.equal(request.recordKey, KEY_A);
+    assert.equal(request.shardId, "authority-shard");
+    assert.equal(request.shardEpoch, 1);
+    assert.deepEqual(r.puts.map((p) => p.recordKey), [KEY_A]);
+
+    r.peer.send(put(KEY_B));
+    r.peer.send(del(KEY_A));
+    await r.settle();
+    assertRefused(r.replies[0], KEY_B, "write_not_authorized: scoped");
+    assertRefused(r.replies[1], KEY_A, "write_not_authorized: scoped");
+    assert.deepEqual(r.puts.map((p) => p.recordKey), [KEY_A], "a grant scoped to key a refuses key b");
+    assert.deepEqual(r.deletes, [], "a grant scoped to put refuses delete");
+    assert.equal(authority.requests.at(-1)!.operation, "delete");
+    assert.equal(authority.requests.at(-1)!.schemaId, SCHEMA_ID);
+    assert.equal(authority.requests.at(-1)!.shardId, null);
+  });
+});
+
+test("a delete sent under a schema alias asks the authority with the canonical id", async () => {
+  const authority = new Recording(grant);
+  await using(await rig(authority), async (r) => {
+    r.peer.send(del(KEY_A, ALIAS));
+    await r.settle();
+    assert.equal(authority.requests[0]!.schemaId, SCHEMA_ID);
+    assert.equal(r.deletes.length, 1);
+  });
+});
+
+test("refusal precedes every other judgment of the put", async () => {
+  await using(await rig(undefined, { maxPayloadBytes: 256 }), async (r) => {
+    r.peer.send(put("unrelated:authority", {}, "sha256:nobody-owns-this"));
+    r.peer.send(put(KEY_A, {}, SCHEMA_ID, new Uint8Array(20_000)));
+    await r.settle();
+    assertRefused(r.replies[0], "unrelated:authority");
+    assertRefused(r.replies[1], KEY_A);
+    assert.deepEqual(r.errors, [], "the gate ran before the put was sized or resolved");
+    assert.equal(r.puts.length, 0);
+  });
+  const granting = new Recording(grant);
+  await using(await rig(granting, { maxPayloadBytes: 256 }), async (r) => {
+    r.peer.send(put(KEY_A, {}, SCHEMA_ID, new Uint8Array(20_000)));
+    await r.settle();
+    assert.equal(r.replies[0]?.schemaVersion, "cultnet.error.v0");
+    assert.notEqual((r.replies[0] as { code?: string }).code, "write_not_authorized", "a granted put still meets the size judgment");
+    assert.equal(r.errors.length, 1);
+  });
+});
+
+test("an authority that throws or rejects refuses, fails closed, and is reported without being quoted", async () => {
+  for (const decide of [
+    () => { throw new Error("authority exploded"); },
+    () => Promise.reject(new Error("authority exploded")),
+  ]) {
+    await using(await rig(new Recording(decide)), async (r) => {
+      r.peer.send(put(KEY_A));
+      r.peer.send(del(KEY_A));
+      await r.settle();
+      assertRefused(r.replies[0], KEY_A);
+      assertRefused(r.replies[1], KEY_A);
+      assert.deepEqual([r.puts.length, r.deletes.length], [0, 0]);
+      assert.equal(r.errors.length, 2);
+      assert.ok(r.errors.every((e) => e.message.includes("Error") && !e.message.includes("exploded")));
+    });
+  }
+});
+
+test("a refusal with no reason still says why, and a missing decision refuses", async () => {
+  await using(await rig(new Recording(() => ({ granted: false }))), async (r) => {
+    r.peer.send(put(KEY_A));
+    await r.settle();
+    assertRefused(r.replies[0], KEY_A, "write_not_authorized");
+  });
+  for (const answer of [null, undefined, {}, { granted: "yes" }, { granted: 1 }]) {
+    await using(await rig(new Recording(() => answer)), async (r) => {
+      r.peer.send(put(KEY_A));
+      r.peer.send(del(KEY_A));
+      await r.settle();
+      assertRefused(r.replies[0], KEY_A);
+      assertRefused(r.replies[1], KEY_A);
+      assert.deepEqual([r.puts.length, r.deletes.length], [0, 0]);
+    });
+  }
+});
+
+test("the refusal on the wire is the pinned error-vectors bytes", async () => {
+  const vectors = JSON.parse(readFileSync(join(__dirname, "../../../../contracts/cultnet/interop/error-vectors.json"), "utf8")).vectors;
+  const vector = vectors.find((v: { name: string }) => v.name === "write_not_authorized");
+  const expected = Buffer.from(vector.messagePackHex, "hex");
+  assert.deepEqual(Buffer.from(encodeWireValue(cultNetWriteNotAuthorizedError("some:key", "not granted"))), expected);
+  await using(await rig(new Recording(() => ({ granted: false, reason: "not granted" }))), async (r) => {
+    r.peer.send(put("some:key"));
+    await r.settle();
+    assert.deepEqual(Buffer.from(encodeWireValue(r.replies[0])), expected);
+  });
+});
+
+test("wrong-typed shard fields on a put are judged with null shard fields, then refused as malformed once granted", async () => {
+  const authority = new Recording(grant);
+  await using(await rig(authority), async (r) => {
+    const bad: Record<string, unknown>[] = [
+      { shardId: { not: "a string" } }, { shardId: 5 }, { shardEpoch: "7" }, { shardEpoch: true }, { shardEpoch: 1.5 },
+    ];
+    for (const fields of bad) r.peer.send(put(KEY_A, fields));
+    await r.settle();
+    assert.equal(r.replies.length, bad.length);
+    assert.ok(r.replies.every((reply) => reply.schemaVersion === "cultnet.error.v0" && reply.code === null && reply.error.startsWith("malformed_document_put:")));
+    assert.equal(authority.requests.length, bad.length, "the authority is asked first");
+    assert.ok(authority.requests.every((q) => q.shardId === null && q.shardEpoch === null), "with null shard fields");
+    assert.equal(r.puts.length, 0);
+  });
+  await using(await rig(undefined), async (r) => {
+    r.peer.send(put(KEY_A, { shardId: 5 }));
+    await r.settle();
+    assertRefused(r.replies[0], KEY_A);
+  });
+});
+
+test("a refusal fits maxPayloadBytes however long the record key and reason are", async () => {
+  const longKey = "k".repeat(3_000) + "é";
+  await using(await rig(new Recording(() => ({ granted: false, reason: "r".repeat(100_000) })), { maxPayloadBytes: 1024 }), async (r) => {
+    r.peer.send(put(longKey));
+    r.peer.send(del(longKey));
+    await r.settle();
+    assert.equal(r.replies.length, 2);
+    for (const reply of r.replies) {
+      assert.equal((reply as { code?: string }).code, "write_not_authorized");
+      assert.ok(encodeWireValue(reply).byteLength <= 1024);
+      assert.ok(longKey.startsWith((reply as { details: { value: string } }).details.value));
+    }
+    assert.deepEqual(r.errors, []);
+  });
+});
+
+test("well-typed shard fields reach the authority as sent", async () => {
+  const authority = new Recording(grant);
+  await using(await rig(authority), async (r) => {
+    r.peer.send(put(KEY_A, { shardId: "authority-shard", shardEpoch: 1 }));
+    await r.settle();
+    assert.deepEqual([authority.requests[0]!.shardId, authority.requests[0]!.shardEpoch], ["authority-shard", 1]);
+  });
+});
+
+test("grant-all is an explicit choice that grants every write", async () => {
+  await using(await rig(new CultMeshGrantAllWrites()), async (r) => {
+    r.peer.send(put(KEY_A));
+    r.peer.send(del(KEY_A));
+    await r.settle();
+    assert.deepEqual([r.puts.length, r.deletes.length, r.replies.length], [1, 1, 0]);
+  });
+});
+
+test("a granted delete with no delete handler is answered with a text refusal", async () => {
+  const connectionId = nextConnection++;
+  const server = CultMesh.createRudpDocumentServer("no-delete", connectionId, {
+    documents: new CultNetDocumentRegistry([defineCultNetDocumentBinding({ definition: note })]),
+    writeAuthority: new CultMeshGrantAllWrites(),
+    bindHost: "127.0.0.1", bindPort: 0, resendDelayMs: 25, resendPollMs: 5,
+  });
+  await server.start();
+  const peer = await CultMesh.createRudpPeer("p", connectionId, `rudp://127.0.0.1:${server.bind.port}`, { resendDelayMs: 25, resendPollMs: 5, connectTimeoutMs: 2_000 });
+  const replies: CultNetMessage[] = [];
+  peer.on("message", (m) => replies.push(m));
+  try {
+    peer.send(del(KEY_A));
+    await sleep(250);
+    assert.equal(replies.length, 1);
+    assert.notEqual((replies[0] as { code?: string }).code, "write_not_authorized");
+  } finally { peer.close(); server.close(); }
+});
+
+// Hostile frames, sent as raw bytes the typed peer would never build.
+const bin = (n: number, ...tail: number[]) => [0xc6, (n >>> 24) & 255, (n >>> 16) & 255, (n >>> 8) & 255, n & 255, ...tail]; // bin32 header
+const str = (t: string) => [0xa0 | t.length, ...Buffer.from(t)];
+function rawPut(payloadBytes: number[], key = KEY_A): Uint8Array {
+  // {schemaVersion, messageId, document: {schemaId, recordKey, storedAt, payloadEncoding, payload: <payloadBytes>}}
+  return Uint8Array.from([
+    0x83, ...str("schemaVersion"), ...str("cultnet.document_put_raw.v0"), ...str("messageId"), ...str("raw"),
+    ...str("document"), 0x85, ...str("schemaId"), ...str(SCHEMA_ID), ...str("recordKey"), ...str(key),
+    ...str("storedAt"), ...str("2026-11-07T00:00:00Z"), ...str("payloadEncoding"), ...str("messagepack"),
+    ...str("payload"), ...payloadBytes,
+  ]);
+}
+
+async function rawRig(writeAuthority?: CultMeshRemoteWriteAuthority) {
+  const connectionId = nextConnection++;
+  const puts: CultMeshRudpDocumentPut[] = [];
+  const deletes: CultMeshRudpDocumentDelete[] = [];
+  const errors: Error[] = [];
+  const server = CultMesh.createRudpDocumentServer("authority-raw", connectionId, {
+    documents: new CultNetDocumentRegistry([defineCultNetDocumentBinding({ definition: note })]),
+    writeAuthority, bindHost: "127.0.0.1", bindPort: 0, resendDelayMs: 25, resendPollMs: 5,
+    onDocumentPutRaw: (d) => { puts.push(d); }, onDocumentDeleteRaw: (d) => { deletes.push(d); }, onError: (e) => errors.push(e),
+  });
+  await server.start();
+  const client = await CultMesh.createRudpClient("raw", connectionId, `rudp://127.0.0.1:${server.bind.port}`, { resendDelayMs: 25, resendPollMs: 5 });
+  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));
+  return { puts, deletes, errors, replies, send: (b: Uint8Array) => client.send("schema", b), close: () => { client.close(); server.close(); } };
+}
+
+const float64 = (v: number) => { const b = Buffer.alloc(9); b[0] = 0xcb; b.writeDoubleBE(v, 1); return [...b]; };
+
+// Each hostile frame, the operation it is judged as, and the ids the authority is told (empty when the frame does not carry them).
+const hostile: { name: string; frame: Uint8Array; operation: "put" | "delete"; schemaId: string; recordKey: string }[] = [
+  { name: "an int payload", frame: rawPut([0x2a]), operation: "put", schemaId: SCHEMA_ID, recordKey: KEY_A },
+  { name: "a uint64-max integer payload", frame: rawPut([0xcf, 255, 255, 255, 255, 255, 255, 255, 255]), operation: "put", schemaId: SCHEMA_ID, recordKey: KEY_A },
+  { name: "a float64 2**70 payload", frame: rawPut(float64(2 ** 70)), operation: "put", schemaId: SCHEMA_ID, recordKey: KEY_A },
+  { name: "a bin whose claimed length exceeds the frame", frame: rawPut(bin(1_000_000_000, 1, 2, 3)), operation: "put", schemaId: "", recordKey: "" },
+  { name: "a frame that does not decode at all", frame: Uint8Array.from([0xc1]), operation: "put", schemaId: "", recordKey: "" },
+  { name: "a put whose document is not a map", frame: Uint8Array.from([0x82, ...str("schemaVersion"), ...str("cultnet.document_put_raw.v0"), ...str("document"), 0x05]), operation: "put", schemaId: "", recordKey: "" },
+  { name: "a delete whose recordKey is not a string", frame: Uint8Array.from([0x84, ...str("schemaVersion"), ...str("cultnet.document_delete.v0"), ...str("messageId"), ...str("d"), ...str("schemaId"), ...str(SCHEMA_ID), ...str("recordKey"), 0x07]), operation: "delete", schemaId: SCHEMA_ID, recordKey: "" },
+];
+
+/** Records the size of every buffer a frame makes the process allocate, so a transient allocation is seen after it is freed. */
+function watchAllocations(): { largest: () => number; stop: () => void } {
+  let largest = 0;
+  const note = (n: unknown) => { if (typeof n === "number" && n > largest) largest = n; };
+  const originals = { alloc: Buffer.alloc, unsafe: Buffer.allocUnsafe, slow: Buffer.allocUnsafeSlow, u8: globalThis.Uint8Array, ab: globalThis.ArrayBuffer };
+  Buffer.alloc = ((n: number, ...rest: never[]) => { note(n); return originals.alloc.call(Buffer, n, ...rest); }) as typeof Buffer.alloc;
+  Buffer.allocUnsafe = ((n: number) => { note(n); return originals.unsafe.call(Buffer, n); }) as typeof Buffer.allocUnsafe;
+  Buffer.allocUnsafeSlow = ((n: number) => { note(n); return originals.slow.call(Buffer, n); }) as typeof Buffer.allocUnsafeSlow;
+  // eslint-disable-next-line @typescript-eslint/no-explicit-any
+  const watched = (ctor: any): any => new Proxy(ctor, { construct: (target, args, newTarget) => { note(args[0]); return Reflect.construct(target, args, newTarget); } });
+  (globalThis as { Uint8Array: unknown }).Uint8Array = watched(originals.u8);
+  (globalThis as { ArrayBuffer: unknown }).ArrayBuffer = watched(originals.ab);
+  return {
+    largest: () => largest,
+    stop: () => {
+      Buffer.alloc = originals.alloc; Buffer.allocUnsafe = originals.unsafe; Buffer.allocUnsafeSlow = originals.slow;
+      (globalThis as { Uint8Array: unknown }).Uint8Array = originals.u8;
+      (globalThis as { ArrayBuffer: unknown }).ArrayBuffer = originals.ab;
+    },
+  };
+}
+
+test("an ungranted peer's unreadable write is refused with write_not_authorized naming what it carried, and nothing is allocated for it", async () => {
+  for (const h of hostile) {
+    const authority = new Recording(() => ({ granted: false }));
+    for (const withAuthority of [false, true]) {
+      const r = await rawRig(withAuthority ? authority : undefined);
+      const watch = watchAllocations();
+      try {
+        r.send(h.frame);
+        await sleep(250);
+      } finally { watch.stop(); }
+      try {
+        assert.equal(r.replies.length, 1, h.name);
+        assertRefused(r.replies[0], h.recordKey);
+        assert.ok(watch.largest() < 10_000_000, `${h.name}: allocated ${watch.largest()} bytes`);
+        assert.deepEqual([r.puts.length, r.deletes.length, r.errors.length], [0, 0, 0], h.name);
+      } finally { r.close(); }
+    }
+    assert.equal(authority.requests.length, 1, h.name);
+    assert.deepEqual(
+      [authority.requests[0]!.operation, authority.requests[0]!.schemaId, authority.requests[0]!.recordKey],
+      [h.operation, h.schemaId, h.recordKey], h.name);
+  }
+});
+
+test("the allocation watch sees a transient allocation", () => {
+  const watch = watchAllocations();
+  try { void Buffer.alloc(20_000_000); } finally { watch.stop(); }
+  assert.ok(watch.largest() >= 20_000_000);
+  const typed = watchAllocations();
+  try { void new Uint8Array(20_000_000); } finally { typed.stop(); }
+  assert.ok(typed.largest() >= 20_000_000);
+});
+
+test("a granted peer's unreadable write reaches onError and gets a malformed refusal, and no handler runs", async () => {
+  for (const h of hostile) {
+    const authority = new Recording(grant);
+    const r = await rawRig(authority);
+    try {
+      r.send(h.frame);
+      await sleep(250);
+      assert.equal(r.replies.length, 1, h.name);
+      const reply = r.replies[0] as { code: unknown; error: string };
+      assert.equal(reply.code, null, h.name);
+      assert.ok(reply.error.startsWith(`malformed_document_${h.operation}:`), h.name);
+      assert.equal(r.errors.length, 1, h.name);
+      assert.deepEqual([r.puts.length, r.deletes.length], [0, 0], h.name);
+      assert.deepEqual([authority.requests.length, authority.requests[0]!.operation, authority.requests[0]!.recordKey], [1, h.operation, h.recordKey], h.name);
+    } finally { r.close(); }
+  }
+});

packages/cultmesh-ts/test/rudp-document-server-send-failure.test.ts

@@ -14,7 +14,7 @@ import {
   encodeRudpPacket,
   type CultNetPeer,
 } from "@gamecult/cultnet-ts";
-import { CultMesh } from "../src/index";
+import { CultMesh, CultMeshGrantAllWrites } from "../src/index";
 
 async function waitFor(predicate: () => boolean, description: string): Promise<void> {
   const startedAt = Date.now();
@@ -64,6 +64,7 @@ async function startServer(connectionId: number, socket: Socket): Promise<Server
   const closed: number[] = [];
   const errors: Error[] = [];
   const server = CultMesh.createRudpDocumentServer("send-failure-server", connectionId, {
+    writeAuthority: new CultMeshGrantAllWrites(),
     documents: new CultNetDocumentRegistry(),
     socket,
     bindHost: "127.0.0.1",