diff --git a/src/GameCult.Mesh/CultMesh.cs b/src/GameCult.Mesh/CultMesh.cs
index 08ecfe36..0c015818 100644
--- a/src/GameCult.Mesh/CultMesh.cs
+++ b/src/GameCult.Mesh/CultMesh.cs
@@ -107,6 +107,39 @@ namespace GameCult.Mesh
}
}
+ ///
+ /// Options for opening one typed document from a remote CultNet snapshot endpoint.
+ ///
+ public sealed class CultMeshPeerSnapshotDocumentOptions
+ {
+ /// Gets or sets the semantic document id. Defaults to the record key.
+ public string? DocumentId { get; set; }
+
+ /// Gets or sets the source id advertised by diagnostics. Defaults to the record key.
+ public string? SourceId { get; set; }
+
+ /// Gets or sets the route hint for the resulting handle.
+ public CultMeshRouteHint? RouteHint { get; set; }
+
+ /// Gets or sets the response timeout for each snapshot request.
+ public TimeSpan ResponseTimeout { get; set; } = TimeSpan.FromSeconds(5);
+
+ /// Gets or sets the connection timeout for endpoint-created clients.
+ public TimeSpan ConnectTimeout { get; set; } = TimeSpan.FromSeconds(5);
+
+ /// Gets or sets the polling interval for watch fallback.
+ public TimeSpan PollInterval { get; set; } = TimeSpan.FromMilliseconds(250);
+
+ /// Gets or sets the message id prefix used for snapshot requests.
+ public string? MessageIdPrefix { get; set; }
+
+ /// Gets or sets client security options for endpoint-created clients.
+ public ClientSecurityOptions? Security { get; set; }
+
+ /// Gets or sets a callback used to configure endpoint-created LiteNetLib clients.
+ public Action? ConfigureClient { get; set; }
+ }
+
///
/// A locally hosted CultMesh node over CultCache, CultNet transport, and the mesh database facade.
///
@@ -1165,6 +1198,144 @@ namespace GameCult.Mesh
return Document(database, key, verse.Context, documentId, sources, routeHint);
}
+ ///
+ /// Creates a typed document handle over one remote CultNet snapshot response provider.
+ ///
+ public static CultMeshDocumentHandle DocumentFromPeerSnapshot(
+ Func> snapshot,
+ string recordKey,
+ CultMeshVerseContext context,
+ CultMeshPeerSnapshotDocumentOptions? options = null)
+ where TDocument : class
+ {
+ if (snapshot == null) throw new ArgumentNullException(nameof(snapshot));
+ if (string.IsNullOrWhiteSpace(recordKey)) throw new ArgumentException("Value must be non-empty.", nameof(recordKey));
+ if (context == null) throw new ArgumentNullException(nameof(context));
+
+ var descriptor = CultDocumentRegistry.Shared.GetRequired();
+ var resolvedOptions = options ?? new CultMeshPeerSnapshotDocumentOptions();
+ var documentId = string.IsNullOrWhiteSpace(resolvedOptions.DocumentId)
+ ? recordKey
+ : resolvedOptions.DocumentId!;
+ var route = resolvedOptions.RouteHint ?? new CultMeshRouteHint(CultMeshLocalityKind.Network, "CultNet snapshot");
+ var sources = new[]
+ {
+ ProjectionSource(
+ string.IsNullOrWhiteSpace(resolvedOptions.SourceId) ? recordKey : resolvedOptions.SourceId!,
+ descriptor.SchemaId,
+ "CultNet snapshot")
+ };
+ Func> latest = async queryContext =>
+ ReadDocumentFromSnapshotResponse(
+ await snapshot(queryContext).ConfigureAwait(false),
+ descriptor.SchemaId,
+ recordKey);
+ var watch = PollingQueryWatcher(
+ async (_parameters, queryContext) => await latest(queryContext).ConfigureAwait(false),
+ new CultMeshPollingWatchOptions(resolvedOptions.PollInterval));
+
+ return Document(
+ documentId,
+ context,
+ latest,
+ queryContext => watch(CultMeshDocumentQueryParameters.Empty, queryContext),
+ sources,
+ route);
+ }
+
+ ///
+ /// Creates a typed document handle over one remote CultNet snapshot response provider.
+ ///
+ public static CultMeshDocumentHandle DocumentFromPeerSnapshot(
+ Func> snapshot,
+ string recordKey,
+ CultMeshVerse verse,
+ CultMeshPeerSnapshotDocumentOptions? options = null)
+ where TDocument : class
+ {
+ if (verse == null) throw new ArgumentNullException(nameof(verse));
+ return DocumentFromPeerSnapshot(snapshot, recordKey, verse.Context, options);
+ }
+
+ ///
+ /// Creates a typed document handle over one remote CultNet schema endpoint.
+ ///
+ public static CultMeshDocumentHandle DocumentFromPeerSnapshot(
+ string endpoint,
+ string recordKey,
+ CultMeshVerseContext context,
+ CultMeshPeerSnapshotDocumentOptions? options = null)
+ where TDocument : class
+ {
+ if (string.IsNullOrWhiteSpace(endpoint)) throw new ArgumentException("Value must be non-empty.", nameof(endpoint));
+ var resolvedOptions = options ?? new CultMeshPeerSnapshotDocumentOptions();
+ return DocumentFromPeerSnapshot(
+ queryContext => RequestPeerSnapshotAsync(
+ () => CultNetSchemaClients.CreateForEndpoint(endpoint, resolvedOptions.Security, resolvedOptions.ConfigureClient),
+ endpoint,
+ recordKey,
+ queryContext,
+ resolvedOptions),
+ recordKey,
+ context,
+ resolvedOptions);
+ }
+
+ ///
+ /// Creates a typed document handle over one remote CultNet schema endpoint.
+ ///
+ public static CultMeshDocumentHandle DocumentFromPeerSnapshot(
+ string endpoint,
+ string recordKey,
+ CultMeshVerse verse,
+ CultMeshPeerSnapshotDocumentOptions? options = null)
+ where TDocument : class
+ {
+ if (verse == null) throw new ArgumentNullException(nameof(verse));
+ return DocumentFromPeerSnapshot(endpoint, recordKey, verse.Context, options);
+ }
+
+ ///
+ /// Creates a typed document handle over one remote CultNet schema client factory.
+ ///
+ public static CultMeshDocumentHandle DocumentFromPeerSnapshot(
+ Func createClient,
+ string endpoint,
+ string recordKey,
+ CultMeshVerseContext context,
+ CultMeshPeerSnapshotDocumentOptions? options = null)
+ where TDocument : class
+ {
+ if (createClient == null) throw new ArgumentNullException(nameof(createClient));
+ if (string.IsNullOrWhiteSpace(endpoint)) throw new ArgumentException("Value must be non-empty.", nameof(endpoint));
+ var resolvedOptions = options ?? new CultMeshPeerSnapshotDocumentOptions();
+ return DocumentFromPeerSnapshot(
+ queryContext => RequestPeerSnapshotAsync(
+ createClient,
+ endpoint,
+ recordKey,
+ queryContext,
+ resolvedOptions),
+ recordKey,
+ context,
+ resolvedOptions);
+ }
+
+ ///
+ /// Creates a typed document handle over one remote CultNet schema client factory.
+ ///
+ public static CultMeshDocumentHandle DocumentFromPeerSnapshot(
+ Func createClient,
+ string endpoint,
+ string recordKey,
+ CultMeshVerse verse,
+ CultMeshPeerSnapshotDocumentOptions? options = null)
+ where TDocument : class
+ {
+ if (verse == null) throw new ArgumentNullException(nameof(verse));
+ return DocumentFromPeerSnapshot(createClient, endpoint, recordKey, verse.Context, options);
+ }
+
///
/// Creates a typed document handle directly over one CultMesh node database record.
///
@@ -1639,6 +1810,116 @@ namespace GameCult.Mesh
});
}
+ private static async Task RequestPeerSnapshotAsync(
+ Func createClient,
+ string endpoint,
+ string recordKey,
+ CultMeshQueryContext context,
+ CultMeshPeerSnapshotDocumentOptions options)
+ where TDocument : class
+ {
+ var descriptor = CultDocumentRegistry.Shared.GetRequired();
+ var messageId = CreateSnapshotMessageId(options, context);
+ var completion = new TaskCompletionSource(
+ TaskCreationOptions.RunContinuationsAsynchronously);
+ using var client = createClient();
+ client.OnCultNet(response =>
+ {
+ if (string.Equals(response.MessageId, messageId, StringComparison.Ordinal))
+ {
+ completion.TrySetResult(response);
+ }
+ });
+ client.OnCultNet(error =>
+ completion.TrySetException(new InvalidOperationException(error.Error)));
+
+ var (host, port) = CultNetSchemaWriteForwarder.ParseEndpoint(endpoint);
+ client.Connect(host, port);
+ await WaitForSchemaClientConnectionAsync(client, endpoint, options.ConnectTimeout).ConfigureAwait(false);
+ client.SendCultNet(new CultNetSnapshotRequestMessage
+ {
+ MessageId = messageId,
+ SchemaIds = new[] { descriptor.SchemaId },
+ RecordKeys = new[] { recordKey }
+ });
+
+ return await WaitForPeerSnapshotResponseAsync(completion.Task, endpoint, options.ResponseTimeout)
+ .ConfigureAwait(false);
+ }
+
+ private static string CreateSnapshotMessageId(
+ CultMeshPeerSnapshotDocumentOptions options,
+ CultMeshQueryContext context)
+ {
+ var prefix = string.IsNullOrWhiteSpace(options.MessageIdPrefix)
+ ? $"cultmesh:{context.RuntimeId}:snapshot"
+ : options.MessageIdPrefix!;
+ return $"{prefix}:{Guid.NewGuid():N}";
+ }
+
+ private static async Task WaitForSchemaClientConnectionAsync(
+ ICultNetSchemaClient client,
+ string endpoint,
+ TimeSpan timeout)
+ {
+ var deadline = DateTimeOffset.UtcNow + timeout;
+ while (!client.Connected)
+ {
+ if (DateTimeOffset.UtcNow >= deadline)
+ {
+ throw new TimeoutException($"Timed out connecting to CultNet snapshot endpoint {endpoint}.");
+ }
+
+ await Task.Delay(TimeSpan.FromMilliseconds(25)).ConfigureAwait(false);
+ }
+ }
+
+ private static async Task WaitForPeerSnapshotResponseAsync(
+ Task responseTask,
+ string endpoint,
+ TimeSpan timeout)
+ {
+ var timeoutTask = Task.Delay(timeout);
+ var completed = await Task.WhenAny(responseTask, timeoutTask).ConfigureAwait(false);
+ if (completed != responseTask)
+ {
+ throw new TimeoutException($"Timed out waiting for CultNet snapshot response from {endpoint}.");
+ }
+
+ return await responseTask.ConfigureAwait(false);
+ }
+
+ private static TDocument ReadDocumentFromSnapshotResponse(
+ CultNetSnapshotResponseRawMessage response,
+ string schemaId,
+ string recordKey)
+ where TDocument : class
+ {
+ if (response == null) throw new ArgumentNullException(nameof(response));
+ var record = response.Documents.FirstOrDefault(candidate =>
+ string.Equals(candidate.SchemaId, schemaId, StringComparison.Ordinal) &&
+ string.Equals(candidate.RecordKey, recordKey, StringComparison.Ordinal))
+ ?? response.Documents.FirstOrDefault(candidate =>
+ string.Equals(candidate.SchemaId, schemaId, StringComparison.Ordinal))
+ ?? response.Documents.FirstOrDefault(candidate =>
+ string.Equals(candidate.RecordKey, recordKey, StringComparison.Ordinal));
+ if (record == null)
+ {
+ throw new InvalidOperationException(
+ $"CultNet snapshot response '{response.MessageId}' did not contain schema '{schemaId}' record '{recordKey}'.");
+ }
+
+ if (!string.Equals(record.PayloadEncoding, "messagepack", StringComparison.Ordinal))
+ {
+ throw new InvalidOperationException(
+ $"CultNet raw document payloadEncoding must be \"messagepack\", not \"{record.PayloadEncoding}\".");
+ }
+
+ return (TDocument)CultDocumentMessagePackSerialization.DeserializeUntyped(
+ typeof(TDocument),
+ record.Payload);
+ }
+
///
/// Parses a rudp://host:port endpoint into its host/port parts.
///
diff --git a/src/GameCult.Mesh/README.md b/src/GameCult.Mesh/README.md
index a6699bde..77ebb014 100644
--- a/src/GameCult.Mesh/README.md
+++ b/src/GameCult.Mesh/README.md
@@ -221,6 +221,17 @@ player.Value!.Health = 90;
await player.CommitAsync();
```
+Remote publications use the same document handle shape:
+
+```csharp
+var health = CultMesh.DocumentFromPeerSnapshot(
+ "cultnet://daemon.local:3075",
+ "daemon:aetheria.health.v1",
+ verse);
+
+var latest = await health.LatestAsync();
+```
+
The physical cache table remains SoA for CPU-local scans:
```csharp
diff --git a/src/GameCult.Mesh/docs/public-api.md b/src/GameCult.Mesh/docs/public-api.md
index d56a5a88..7f5ef40f 100644
--- a/src/GameCult.Mesh/docs/public-api.md
+++ b/src/GameCult.Mesh/docs/public-api.md
@@ -273,6 +273,21 @@ using var subscription = cockpit.Watch(next => RenderCockpit(next));
var daemonAlias = cockpit.AsSchemaAlias();
```
+Remote CultNet snapshots use the same document handle surface. The caller names
+the endpoint, record key, and typed document; CultMesh owns the snapshot request
+id, response filtering, raw MessagePack payload decode, route metadata, and
+polling watch fallback:
+
+```csharp
+var remoteHealth = CultMesh.DocumentFromPeerSnapshot(
+ "cultnet://daemon.local:3075",
+ "daemon:aetheria.health.v1",
+ verse);
+
+var health = await remoteHealth.LatestAsync();
+var uiHealth = remoteHealth.AsSchemaAlias();
+```
+
A domain facade can collect handles into one schema-aware catalog:
```csharp
@@ -346,11 +361,12 @@ const health = await remoteHealth.latest();
```
Projected, remote, shared-memory, IPC, WASM, and local-cache documents all use
-the same TS handle contract. The caller names the typed document or collection
-it wants; the handle owns route defaults, snapshot reads, reactive watches,
-schema alias validation, and replacement when the backing source is mutable.
-Generated Aetheria and Ymir facades should bind those handles to a Verse once
-and keep transport, quorum, and cache mechanics behind CultMesh.
+the same C# and TS handle contract. The caller names the typed document or
+collection it wants; the handle owns route defaults, snapshot reads, reactive
+watches, schema alias validation, prediction submission, and replacement when
+the backing source supports those mutations. Generated Aetheria and Ymir
+facades should bind those handles to a Verse once and keep transport, quorum,
+and cache mechanics behind CultMesh.
Native view descriptors can name unmanaged columns without hand-entered byte
sizes:
diff --git a/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs b/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs
index e4856572..fed12c54 100644
--- a/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs
+++ b/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs
@@ -848,6 +848,50 @@ public sealed class CultMeshStreamingTests
(await handle.LatestAsync()).Text.Should().Be("predicted-thermal");
}
+ [Test]
+ public async Task DocumentHandle_ReadsRemotePeerSnapshotsAsTypedDocuments()
+ {
+ var cache = new CultCache();
+ var key = new CultRecordKey("mesh-note:remote");
+ await cache.UpsertAsync(new MeshNoteDocument
+ {
+ Schema = "tests.mesh_note.v1",
+ Text = "remote-snapshot",
+ Revision = 9
+ }, new CultRecordHandle(key));
+ var registry = new CultNetDocumentRegistry(CultDocumentRegistry.Shared)
+ .Register(CultNetDocumentBinding.ForDocument(CultDocumentRegistry.Shared));
+ var requests = new List();
+
+ var handle = CultMesh.DocumentFromPeerSnapshot(
+ () => new MeshSnapshotSchemaClient(request =>
+ {
+ requests.Add(request);
+ return registry.CreateRawSnapshotResponse(cache, request.MessageId, request);
+ }),
+ "cultnet://snapshot.test:3075",
+ key.Value,
+ CultMesh.Verse("starbridge", "unity-pilot"),
+ new CultMeshPeerSnapshotDocumentOptions
+ {
+ PollInterval = TimeSpan.FromMilliseconds(10),
+ MessageIdPrefix = "mesh-test-snapshot"
+ });
+ var alias = handle.AsSchemaAlias();
+ var catalog = CultMesh.Documents(handle);
+
+ var snapshot = await alias.LatestAsync();
+
+ snapshot.Text.Should().Be("remote-snapshot");
+ snapshot.Revision.Should().Be(9);
+ handle.RouteHint.Kind.Should().Be(CultMeshLocalityKind.Network);
+ handle.Sources.Should().ContainSingle().Which.SourceId.Should().Be(key.Value);
+ requests.Should().ContainSingle();
+ requests[0].SchemaIds.Should().ContainSingle().Which.Should().Be(handle.SchemaId);
+ requests[0].RecordKeys.Should().ContainSingle().Which.Should().Be(key.Value);
+ (await catalog.LatestAsync()).Text.Should().Be("remote-snapshot");
+ }
+
[Test]
public async Task CollectionHandle_ReadsAndWatchesDatabaseCollectionsByIndex()
{
@@ -1538,3 +1582,58 @@ public sealed class MeshObjectsFacade
return _visibleObjects.ExecuteAsync(request);
}
}
+
+internal sealed class MeshSnapshotSchemaClient : ICultNetSchemaClient
+{
+ private readonly Func _respond;
+ private readonly List> _snapshotHandlers = new();
+ private readonly List> _errorHandlers = new();
+
+ public MeshSnapshotSchemaClient(
+ Func respond)
+ {
+ _respond = respond;
+ }
+
+ public bool Connected { get; private set; }
+
+ public void Connect(string host, int port)
+ {
+ Connected = true;
+ }
+
+ public void SendCultNet(T message) where T : ICultNetSchemaMessage
+ {
+ if (message is not CultNetSnapshotRequestMessage request)
+ {
+ foreach (var handler in _errorHandlers)
+ {
+ handler(new CultNetErrorMessage { Error = $"Unexpected message {typeof(T).Name}." });
+ }
+
+ return;
+ }
+
+ var response = _respond(request);
+ foreach (var handler in _snapshotHandlers)
+ {
+ handler(response);
+ }
+ }
+
+ public void OnCultNet(Action callback) where T : ICultNetSchemaMessage
+ {
+ if (typeof(T) == typeof(CultNetSnapshotResponseRawMessage))
+ {
+ _snapshotHandlers.Add(message => callback((T)(object)message));
+ }
+ else if (typeof(T) == typeof(CultNetErrorMessage))
+ {
+ _errorHandlers.Add(message => callback((T)(object)message));
+ }
+ }
+
+ public void Dispose()
+ {
+ }
+}