Add publication sync and reconciliation coverage

packages/cultcache-py/tests/test_cultcache.py

@@ -3742,6 +3742,62 @@ class CultCacheTests(unittest.TestCase):
             self.assertEqual(session.pending_predictions(), ())
             self.assertEqual(session.resimulation_inputs(), ())
 
+    def test_cultmesh_game_session_partial_reconciliation_keeps_unmatched_predictions(self) -> None:
+        note_doc = define_database_entry_type(
+            "mesh.partial_input",
+            [("body", 0)],
+            schema_id="mesh.partial_input.v1",
+        )
+        node = CultMesh.create_node(runtime_id="client-partial")
+        node.register_document(note_doc)
+        session = CultMesh.create_game_session(
+            node,
+            CultMeshGameSessionOptions(
+                client_authority_scopes=(
+                    CultNetClientAuthorityScope(
+                        "client-partial",
+                        schema_ids=("mesh.partial_input.v1",),
+                        key_prefix="input:client-partial",
+                    ),
+                )
+            ),
+        )
+        pending = session.predict(note_doc, "input:client-partial:pending", {"body": "pending-local"})
+        accepted = session.predict(note_doc, "input:client-partial:accepted", {"body": "accepted-local"})
+        put = document_put_raw(
+            message_id="authoritative-partial",
+            key=accepted.key,
+            schema_id=accepted.schema_id,
+            stored_at="2026-06-13T00:00:00Z",
+            payload=note_doc.encode_payload({"body": "accepted-authority"}),
+            shard_id="inputs",
+            shard_epoch=1,
+        )
+
+        changes = session.apply_shard_log_response({
+            "schemaVersion": "cultnet.shard_log_response.v0",
+            "messageId": "inputs-partial-log",
+            "shardId": "inputs",
+            "shardEpoch": 1,
+            "entries": [
+                {
+                    "sequence": 1,
+                    "committedAt": "2026-06-13T00:00:00Z",
+                    "changeKind": "updated",
+                    "put": put.to_wire(),
+                }
+            ],
+            "resyncRequired": False,
+        })
+
+        self.assertEqual([(change.change_kind, change.record_key) for change in changes], [
+            ("reconciled", accepted.key),
+        ])
+        self.assertEqual(node.get_required(note_doc, accepted.key)["body"], "accepted-authority")
+        self.assertEqual(node.get_required(note_doc, pending.key)["body"], "pending-local")
+        self.assertEqual([prediction.key for prediction in session.pending_predictions()], [pending.key])
+        self.assertEqual([prediction.key for prediction in session.resimulation_inputs()], [pending.key])
+
     def test_cultmesh_game_session_rolls_back_pending_predictions(self) -> None:
         note_doc = define_database_entry_type(
             "mesh.rollback_input",

src/GameCult.Mesh/CultMesh.cs

@@ -1544,6 +1544,89 @@ namespace GameCult.Mesh
             return DocumentFromPublication<TDocument>(source, key, verse.Context, storeOptions, peerOptions);
         }
 
+        /// <summary>
+        /// Reads one typed document from a configured publication source and hydrates it into a local node.
+        /// </summary>
+        public static async Task<TDocument> SyncDocumentFromPublicationAsync<TDocument>(
+            CultMeshNode node,
+            CultMeshDocumentPublicationSource source,
+            CultRecordKey key,
+            CultMeshVerseContext context,
+            CultMeshStoreDocumentOptions? storeOptions = null,
+            CultMeshPeerSnapshotDocumentOptions? peerOptions = null,
+            bool flush = false)
+            where TDocument : class
+        {
+            if (node == null) throw new ArgumentNullException(nameof(node));
+            if (source == null) throw new ArgumentNullException(nameof(source));
+            if (context == null) throw new ArgumentNullException(nameof(context));
+
+            var document = await DocumentFromPublication<TDocument>(
+                    source,
+                    key,
+                    context,
+                    storeOptions,
+                    peerOptions)
+                .LatestAsync()
+                .ConfigureAwait(false);
+
+            await Document<TDocument>(node, key, context)
+                .ReplaceAsync(document)
+                .ConfigureAwait(false);
+
+            if (flush)
+                await node.FlushAsync().ConfigureAwait(false);
+
+            return document;
+        }
+
+        /// <summary>
+        /// Reads one typed document from a configured publication source and hydrates it into a local node.
+        /// </summary>
+        public static Task<TDocument> SyncDocumentFromPublicationAsync<TDocument>(
+            CultMeshNode node,
+            CultMeshDocumentPublicationSource source,
+            CultRecordKey key,
+            CultMeshVerse verse,
+            CultMeshStoreDocumentOptions? storeOptions = null,
+            CultMeshPeerSnapshotDocumentOptions? peerOptions = null,
+            bool flush = false)
+            where TDocument : class
+        {
+            if (verse == null) throw new ArgumentNullException(nameof(verse));
+            return SyncDocumentFromPublicationAsync<TDocument>(
+                node,
+                source,
+                key,
+                verse.Context,
+                storeOptions,
+                peerOptions,
+                flush);
+        }
+
+        /// <summary>
+        /// Reads one typed document from a configured publication source and hydrates it into a local node.
+        /// </summary>
+        public static Task<TDocument> SyncDocumentFromPublicationAsync<TDocument>(
+            CultMeshNode node,
+            CultMeshDocumentPublicationSource source,
+            string recordKey,
+            CultMeshVerse verse,
+            CultMeshStoreDocumentOptions? storeOptions = null,
+            CultMeshPeerSnapshotDocumentOptions? peerOptions = null,
+            bool flush = false)
+            where TDocument : class
+        {
+            if (string.IsNullOrWhiteSpace(recordKey)) throw new ArgumentException("Value must be non-empty.", nameof(recordKey));
+            return SyncDocumentFromPublicationAsync<TDocument>(
+                node,
+                source,
+                new CultRecordKey(recordKey),
+                verse,
+                storeOptions,
+                peerOptions,
+                flush);
+        }
         /// <summary>
         /// Creates a typed document handle directly over one distributed CultNet database record.
         /// </summary>

tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs

@@ -2305,6 +2305,62 @@ public sealed class CultMeshStreamingTests
         (await mixedAliasSchemaHandle.LatestAsync()).Text.Should().Be("compatible-record-key");
     }
 
+    [Test]
+    public async Task PublicationHelpers_SyncConfiguredPeerSourceIntoLocalNodeAlias()
+    {
+        var sourceCache = new CultCache();
+        var registry = new CultNetDocumentRegistry(CultDocumentRegistry.Shared);
+        registry.Register(CultNetDocumentBinding.ForDocument<MeshNoteDocument>(sourceCache.Registry));
+        var key = new CultRecordKey("mesh-note:publication-sync-alias");
+        await sourceCache.UpsertAsync(new MeshNoteDocument
+        {
+            Schema = "tests.mesh_note.v1",
+            Text = "publication-synced",
+            Revision = 51
+        }, new CultRecordHandle<MeshNoteDocument>(key));
+        var requests = new List<CultNetSnapshotRequestMessage>();
+        var verse = CultMesh.Verse("starbridge", "unity-support");
+        using var node = await CultMesh.CreateNodeAsync(
+            Path.Combine(Path.GetTempPath(), $"cultmesh-publication-sync-alias-{Guid.NewGuid():N}.ccmp"),
+            new CultMeshNodeOptions
+            {
+                StartServer = false,
+                CacheOptions = new CultCacheOpenOptions
+                {
+                    Registry = sourceCache.Registry,
+                    PullOnOpen = false
+                },
+                DatabaseOptions = new CultNetDatabaseOptions
+                {
+                    DocumentRegistry = registry
+                }
+            });
+
+        var synced = await CultMesh.SyncDocumentFromPublicationAsync<MeshNoteAliasDocument>(
+            node,
+            CultMeshDocumentPublicationSource.PeerSnapshot(
+                () => new MeshSnapshotSchemaClient(request =>
+                {
+                    requests.Add(request);
+                    return registry.CreateRawSnapshotResponse(sourceCache, request.MessageId, request);
+                }),
+                "cultnet://publication-sync.test:3075"),
+            key,
+            verse,
+            peerOptions: new CultMeshPeerSnapshotDocumentOptions
+            {
+                MessageIdPrefix = "mesh-test-publication-sync",
+                PollInterval = TimeSpan.FromMilliseconds(10)
+            });
+        var localAlias = await node.Document<MeshNoteAliasDocument>(key, verse).LatestAsync();
+
+        synced.Text.Should().Be("publication-synced");
+        synced.Revision.Should().Be(51);
+        localAlias.Text.Should().Be("publication-synced");
+        localAlias.Revision.Should().Be(51);
+        requests.Should().ContainSingle().Which.RecordKeys.Should().Equal(key.Value);
+    }
+
     [Test]
     public async Task SnapshotHelpers_FetchApplyAndDecodeScopedSnapshots()
     {