Add CultMesh peer snapshot document handles

packages/cultmesh-ts/README.md

@@ -265,6 +265,18 @@ const client = await CultMesh.createRudpPeerForAuthorizedPeer(
 );
 const schemas = CultMesh.createSchemaCatalog();
 await client.syncSchemaCatalog(schemas, { kinds: ["document_payload"] });
+
+const health = CultMesh.documentFromPeerSnapshot(
+  () => client,
+  "gamecult.aetheria.daemon_health.v1",
+  "daemon:aetheria.health.v1",
+  {
+    routeHint: CultMesh.routeHint("network", endpoint.uri),
+    timeoutMs: 5_000,
+  },
+);
+
+const latestHealth = await health.latest();
 ```
 
 `CultMesh.createRudpPeerForPeer(...)` remains available for already trusted
@@ -275,6 +287,14 @@ Discovery-first schema/catalog paths should prefer
 `CultMeshPeerCatalog.firstAuthorized(...)` with the authority lease catalog
 before using the peer-card endpoint as a dial target.
 
+`CultMesh.documentFromPeerSnapshot(...)` is the document-level path for RUDP
+snapshots. It hides the snapshot request message id, response listener, raw
+payload binary normalization, and MessagePack decode behind the same
+`CultMeshDocumentHandle` shape as local documents. The helper prefers an exact
+schema id match and falls back to the requested record key, so runtimes that
+alias the same logical document through different generated schema names can
+still read the publication with one CultMesh call.
+
 ## Streaming Mode
 
 CultMesh streaming mode is for audio, video, tensor, and opaque byte frames that

packages/cultmesh-ts/src/index.ts

@@ -21,11 +21,13 @@ import {
   encodeCultNetMessageForWire,
   encodeRudpPacket,
   parseCultNetMessage,
+  type CultNetErrorMessage,
   type CultNetDocumentBinding,
   type CultNetMessage,
   type CultNetRawDocumentRecord,
   type CultNetReconnectPolicy,
   type CultNetSchemaCatalogOptions,
+  type CultNetSnapshotResponseRawMessage,
   type CultNetWireContract,
 } from "cultnet-ts";
 
@@ -1884,6 +1886,49 @@ export function cultMeshDocumentFromSingleFile(
   );
 }
 
+export function cultMeshDocumentFromPeerSnapshot(
+  peer: CultNetPeer | (() => CultNetPeer | Promise<CultNetPeer>),
+  schemaId: string,
+  recordKey: string,
+  options: {
+    documentId?: string;
+    routeHint?: CultMeshRouteHint;
+    sourceId?: string;
+    timeoutMs?: number;
+    pollMs?: number;
+    messageIdPrefix?: string;
+  } = {},
+): CultMeshDocumentHandle<unknown> {
+  requireNonEmpty(schemaId, "schemaId");
+  requireNonEmpty(recordKey, "recordKey");
+  const documentId = options.documentId ?? recordKey;
+  const read = async () =>
+    decodeCultNetRawDocumentPayload(
+      await requestCultNetRawSnapshotDocument(
+        peer,
+        schemaId,
+        recordKey,
+        {
+          timeoutMs: options.timeoutMs,
+          messageIdPrefix: options.messageIdPrefix ?? documentId,
+        },
+      ),
+    );
+
+  return cultMeshDocument(
+    documentId,
+    { schemaId },
+    async () => read(),
+    {
+      routeHint: options.routeHint ?? cultMeshRouteHint("network", "CultNet snapshot"),
+      sources: [cultMeshProjectionSource(options.sourceId ?? documentId, { schemaId })],
+      watchDocument: cultMeshPollingDocumentWatcher(read, {
+        intervalMs: options.pollMs ?? 250,
+      }),
+    },
+  );
+}
+
 export function cultMeshGlobalDocumentFromCache<TDefinition extends AnyCultCacheDocumentDefinition>(
   cache: CultCache,
   definition: TDefinition,
@@ -3591,6 +3636,22 @@ export class CultMesh {
     return cultMeshDocumentFromSingleFile(path, schemaId, options);
   }
 
+  public static documentFromPeerSnapshot(
+    peer: CultNetPeer | (() => CultNetPeer | Promise<CultNetPeer>),
+    schemaId: string,
+    recordKey: string,
+    options: {
+      documentId?: string;
+      routeHint?: CultMeshRouteHint;
+      sourceId?: string;
+      timeoutMs?: number;
+      pollMs?: number;
+      messageIdPrefix?: string;
+    } = {},
+  ): CultMeshDocumentHandle<unknown> {
+    return cultMeshDocumentFromPeerSnapshot(peer, schemaId, recordKey, options);
+  }
+
   public static globalDocumentFromCache<TDefinition extends AnyCultCacheDocumentDefinition>(
     cache: CultCache,
     definition: TDefinition,
@@ -4595,6 +4656,112 @@ function normalizeRudpDocumentPut(
   };
 }
 
+async function requestCultNetRawSnapshotDocument(
+  peerOrProvider: CultNetPeer | (() => CultNetPeer | Promise<CultNetPeer>),
+  schemaId: string,
+  recordKey: string,
+  options: {
+    timeoutMs?: number;
+    messageIdPrefix?: string;
+  } = {},
+): Promise<CultNetRawDocumentRecord> {
+  const peer = await resolveCultNetPeer(peerOrProvider);
+  const messageId = `${options.messageIdPrefix ?? "cultmesh-snapshot"}:${Date.now()}:${Math.random().toString(16).slice(2)}`;
+  const response = await requestCultNetRawSnapshot(peer, [recordKey], {
+    messageId,
+    timeoutMs: options.timeoutMs,
+  });
+  const candidates = response.documents.filter(candidate => candidate.recordKey === recordKey);
+  const document = candidates.find(candidate => candidate.schemaId === schemaId) ?? candidates[0];
+  if (!document) {
+    throw new Error(`CultMesh peer snapshot did not return ${schemaId} at ${recordKey}.`);
+  }
+  return document;
+}
+
+async function requestCultNetRawSnapshot(
+  peer: CultNetPeer,
+  recordKeys: readonly string[],
+  options: {
+    messageId: string;
+    timeoutMs?: number;
+  },
+): Promise<CultNetSnapshotResponseRawMessage> {
+  return new Promise<CultNetSnapshotResponseRawMessage>((resolve, reject) => {
+    const cleanup = (): void => {
+      clearTimeout(timer);
+      peer.off("message", onMessage);
+      peer.off("invalidMessage", onInvalidMessage);
+      peer.off("error", onError);
+      peer.off("close", onClose);
+    };
+    const rejectWith = (error: Error): void => {
+      cleanup();
+      reject(error);
+    };
+    const onMessage = (message: CultNetMessage): void => {
+      if (isCultNetErrorMessage(message)) {
+        cleanup();
+        reject(new Error(message.error));
+        return;
+      }
+      if (message.schemaVersion !== "cultnet.snapshot_response_raw.v0" ||
+          message.messageId !== options.messageId) {
+        return;
+      }
+
+      cleanup();
+      resolve(message);
+    };
+    const onInvalidMessage = (error: Error): void => rejectWith(error);
+    const onError = (error: Error): void => rejectWith(error);
+    const onClose = (): void => rejectWith(new Error("CultNet peer closed before snapshot response."));
+    const timer = setTimeout(
+      () => rejectWith(new Error(`Timed out waiting for CultNet snapshot ${options.messageId}.`)),
+      options.timeoutMs ?? 4_000,
+    );
+
+    peer.on("message", onMessage);
+    peer.on("invalidMessage", onInvalidMessage);
+    peer.on("error", onError);
+    peer.on("close", onClose);
+    peer.sendSnapshotRequest({
+      schemaVersion: "cultnet.snapshot_request.v0",
+      messageId: options.messageId,
+      recordKeys: [...recordKeys],
+    });
+  });
+}
+
+async function resolveCultNetPeer(
+  peerOrProvider: CultNetPeer | (() => CultNetPeer | Promise<CultNetPeer>),
+): Promise<CultNetPeer> {
+  return typeof peerOrProvider === "function"
+    ? await peerOrProvider()
+    : peerOrProvider;
+}
+
+function decodeCultNetRawDocumentPayload(document: CultNetRawDocumentRecord): unknown {
+  if (document.payloadEncoding !== "messagepack") {
+    throw new Error(`Unsupported CultNet raw document payload encoding ${document.payloadEncoding}.`);
+  }
+  return decode(toUint8Array(document.payload));
+}
+
+function isCultNetErrorMessage(message: CultNetMessage): message is CultNetErrorMessage {
+  return message.schemaVersion === "cultnet.error.v0";
+}
+
+function toUint8Array(value: unknown): Uint8Array {
+  if (value instanceof Uint8Array) {
+    return value;
+  }
+  if (Array.isArray(value)) {
+    return Uint8Array.from(value);
+  }
+  throw new Error("CultNet raw document payload was not binary.");
+}
+
 async function bindRudpSocket(options: CultMeshRudpSocketOptions): Promise<Socket> {
   const socket = createSocket("udp4");
   const host = options.bindHost ?? "127.0.0.1";

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

@@ -1832,6 +1832,87 @@ test("CultMesh TS publishes one registered document to a RUDP catalog", async ()
   }
 });
 
+test("CultMesh TS reads remote RUDP snapshots through document handles", async () => {
+  const connectionId = 0x10203048;
+  const node = await CultMesh.startNode(
+    join(await mkdtemp(join(tmpdir(), "cultmesh-ts-rudp-snapshot-")), "node.ccmp"),
+    {
+      documents: [noteDocument],
+    },
+  );
+  await node.put(noteDocument, "note:remote", {
+    noteId: "note:remote",
+    body: "remote handles feel local",
+  });
+
+  const server = CultMesh.createRudpDocumentServer(
+    "cultmesh-ts-rudp-snapshot-server",
+    connectionId,
+    {
+      documents: new CultNetDocumentRegistry([
+        defineCultNetDocumentBinding({ definition: noteDocument }),
+      ]),
+      getCache: () => node.cache,
+      bindHost: "127.0.0.1",
+      bindPort: 0,
+      resendDelayMs: 25,
+      resendPollMs: 5,
+      maxFragmentBytes: 1024,
+      maxPendingReliablePackets: 16,
+    },
+  );
+
+  let peer: CultNetPeer | undefined;
+  try {
+    await server.start();
+    peer = await CultMesh.createRudpPeer(
+      "cultmesh-ts-rudp-snapshot-client",
+      connectionId,
+      `rudp://127.0.0.1:${server.bind.port}`,
+      {
+        resendDelayMs: 25,
+        resendPollMs: 5,
+        maxFragmentBytes: 1024,
+        maxPendingReliablePackets: 16,
+        connectTimeoutMs: 1_000,
+      },
+    );
+
+    const document = CultMesh.documentFromPeerSnapshot(
+      peer,
+      "cultmesh.note.v0",
+      "note:remote",
+      {
+        documentId: "note:remote",
+        timeoutMs: 1_000,
+      },
+    );
+
+    assert.deepEqual(await document.latest(), {
+      noteId: "note:remote",
+      body: "remote handles feel local",
+    });
+
+    const alias = CultMesh.documentFromPeerSnapshot(
+      peer,
+      "cultmesh.note.alias.v1",
+      "note:remote",
+      {
+        documentId: "note:remote.alias",
+        timeoutMs: 1_000,
+      },
+    );
+
+    assert.deepEqual(await alias.latest(), {
+      noteId: "note:remote",
+      body: "remote handles feel local",
+    });
+  } finally {
+    peer?.close();
+    server.close();
+  }
+});
+
 test("CultMesh TS negotiates streaming frame body transports explicitly", () => {
   const streams = CultMesh.createStreamCatalog();
   const streamUpdates: string[] = [];

src/GameCult.Mesh/docs/public-api.md

@@ -330,6 +330,13 @@ unsubscribe();
 const pilots = station.collection(playerShipDocument);
 const currentPilots = await pilots.latest();
 const stopRoster = pilots.watchChanges(change => refreshRoster(change));
+
+const remoteHealth = CultMesh.documentFromPeerSnapshot(
+  () => daemonPeer,
+  "gamecult.aetheria.daemon_health.v1",
+  "daemon:aetheria.health.v1",
+);
+const health = await remoteHealth.latest();
 ```
 
 Projected, remote, shared-memory, IPC, WASM, and local-cache documents all use
@@ -458,6 +465,15 @@ helpers parse endpoint contact hints and choose authorized peers;
 `GameCult.Networking` still owns the RUDP packet codec, session state, resend,
 fragmentation, and channel semantics.
 
+TypeScript runtimes can also wrap a CultNet peer snapshot as a normal document
+handle with `CultMesh.documentFromPeerSnapshot(...)`. The caller supplies the
+peer or peer factory, schema id, and record key; CultMesh sends the snapshot
+request, waits for the matching response, decodes the raw MessagePack document,
+and returns the same `latest()`/`watch()` document surface used by local cache
+and single-file documents. Exact schema ids are preferred, but record-key
+publications remain readable when two runtimes alias the same logical document
+through different schema names.
+
 ### Shard Authority
 
 Each shard has one primary writer for now. Non-primary writes are rejected or