Cut 4 Soul 4d: a skipped entry is relayed as the primary sent it; probes, wire cases and the closed-enum rule committed

F1: RecordUnappliedEntry keeps the primary's wire entry on the in-memory log
entry (internal Relayed) and ToLogEntryMessage returns it through the same
normalisation the store uses, so a replica with no store no longer relays a
skipped put as a delete; memory and disk relay identically.
Soul's 4d probes are committed (chained replica keeps the row, key-prefixed
reads serve only owned keys, primary-all serves everything; the two
measurements are explicit) with a memory/disk relay-parity test.
Rust: captured-bytes cases for unowned_schema and variant_unsupported (with and
without a subscription).
The error-code set is stated as a closed enum in the schema description (both
copies) and in cultnet-distributed-database.md: adding a code is one
coordinated change across runtimes, since Rust and TypeScript refuse an unknown
code. The Soul 4c probe file's header says what it guards.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

contracts/cultnet/cultnet-distributed-database.md

@@ -104,6 +104,16 @@ Not implemented yet:
 - cross-runtime rollback/resimulation helpers for simulation frames
 - declared CRDT merge policies
 
+## Error Codes
+
+`cultnet.error.v0` carries an optional machine-readable `code`. The set of codes is a closed enum
+(`cultnet.error.schema.json`): Rust and TypeScript refuse a message whose code they do not know, they do
+not guess. Adding a code is therefore a coordinated change across every runtime, made together: the schema (both
+copies), the C# `CultNetErrorMessage` factory, the Rust `CultNetErrorCode`, and a captured-bytes case in
+`packages/cultnet-rs/tests/error_message.rs` for it. TypeScript reads the schema. The current codes are
+`selection_invalid`, `cursor_stale`, `cursor_invalid`, `reference_outside_target`, `variant_unsupported` and
+`unowned_schema` (a write, or a remote put or delete, for a schema and key no shard owns).
+
 ## Live Invariants
 
 - CultCache owns document schema, identity, local indexes, and local diffing.

contracts/cultnet/cultnet.error.schema.json

@@ -26,7 +26,7 @@
       }
     },
     "code": {
-      "description": "Machine-readable refusal code (R-N). Additive: a v0 peer that does not know this field ignores it.",
+      "description": "Machine-readable refusal code (R-N). Additive: a v0 peer that does not know this field ignores it. The set is a closed enum: Rust and TypeScript refuse a message whose code is not listed, so adding a code is one coordinated change across every runtime (this schema, the C# factory, the Rust CultNetErrorCode, and a captured-bytes case for each).",
       "type": ["string", "null"],
       "enum": [null, "selection_invalid", "cursor_stale", "cursor_invalid", "reference_outside_target", "variant_unsupported", "unowned_schema"]
     },

packages/cultnet-rs/contracts/cultnet/cultnet.error.schema.json

@@ -26,7 +26,7 @@
       }
     },
     "code": {
-      "description": "Machine-readable refusal code (R-N). Additive: a v0 peer that does not know this field ignores it.",
+      "description": "Machine-readable refusal code (R-N). Additive: a v0 peer that does not know this field ignores it. The set is a closed enum: Rust and TypeScript refuse a message whose code is not listed, so adding a code is one coordinated change across every runtime (this schema, the C# factory, the Rust CultNetErrorCode, and a captured-bytes case for each).",
       "type": ["string", "null"],
       "enum": [null, "selection_invalid", "cursor_stale", "cursor_invalid", "reference_outside_target", "variant_unsupported", "unowned_schema"]
     },

packages/cultnet-rs/tests/error_message.rs

@@ -118,6 +118,46 @@ fn cases() -> Vec<Case> {
                 details: None,
             },
         },
+        Case {
+            name: "variant_unsupported_subscription_ended",
+            // C#: ForVariantUnsupported("variant unsupported here", "sub-1")
+            csharp_hex: "85AD736368656D6156657273696F6EB063756C746E65742E6572726F722E7630A56572726F72B876617269616E7420756E737570706F727465642068657265AB726F7574696E6748696E74C0A4636F6465B376617269616E745F756E737570706F72746564A764657461696C7384A56669656C64AE737562736372697074696F6E4964A576616C7565A57375622D31A461734F66C0A763757272656E74C0",
+            message: || CultNetMessage::Error {
+                error: "variant unsupported here".to_string(),
+                code: Some(CultNetErrorCode::VariantUnsupported),
+                details: Some(CultNetErrorDetails {
+                    field: Some("subscriptionId".to_string()),
+                    value: Some("sub-1".to_string()),
+                    as_of: None,
+                    current: None,
+                }),
+            },
+        },
+        Case {
+            name: "variant_unsupported_no_subscription",
+            // C#: ForVariantUnsupported("variant unsupported here") - no subscription ended, so Details stays null.
+            csharp_hex: "85AD736368656D6156657273696F6EB063756C746E65742E6572726F722E7630A56572726F72B876617269616E7420756E737570706F727465642068657265AB726F7574696E6748696E74C0A4636F6465B376617269616E745F756E737570706F72746564A764657461696C73C0",
+            message: || CultNetMessage::Error {
+                error: "variant unsupported here".to_string(),
+                code: Some(CultNetErrorCode::VariantUnsupported),
+                details: None,
+            },
+        },
+        Case {
+            name: "unowned_schema",
+            // C#: ForUnownedSchema(new CultNetUnownedSchemaException("some.schema.v1", new CultRecordKey("some:key")))
+            csharp_hex: "85AD736368656D6156657273696F6EB063756C746E65742E6572726F722E7630A56572726F72D95C756E6F776E65645F736368656D613A204E6F207368617264206F776E7320736368656D612027736F6D652E736368656D612E763127206B65792027736F6D653A6B6579273B207468652077726974652077617320726566757365642EAB726F7574696E6748696E74C0A4636F6465AE756E6F776E65645F736368656D61A764657461696C7384A56669656C64A8736368656D614964A576616C7565AE736F6D652E736368656D612E7631A461734F66C0A763757272656E74C0",
+            message: || CultNetMessage::Error {
+                error: "unowned_schema: No shard owns schema 'some.schema.v1' key 'some:key'; the write was refused.".to_string(),
+                code: Some(CultNetErrorCode::UnownedSchema),
+                details: Some(CultNetErrorDetails {
+                    field: Some("schemaId".to_string()),
+                    value: Some("some.schema.v1".to_string()),
+                    as_of: None,
+                    current: None,
+                }),
+            },
+        },
     ]
 }
 

src/GameCult.Networking/CultNetDatabase.cs

@@ -437,6 +437,10 @@ namespace GameCult.Networking
         /// Gets the previous document, when present.
         /// </summary>
         public object? PreviousDocument { get; }
+
+        // The primary's own wire entry, kept for an entry this replica changed nothing for (a row its shard does not own, an
+        // absent key): it is relayed downstream exactly as the primary sent it, in memory and on disk alike.
+        internal CultNetShardLogEntryMessage? Relayed { get; set; }
     }
 
     /// <summary>
@@ -1819,13 +1823,18 @@ namespace GameCult.Networking
             {
                 RecordMutationLogEntry(new CultNetShardMutationLogEntry(
                     shard.ShardId, shard.Epoch, entry.Sequence, entry.CommittedAt, kind,
-                    schemaId, key, document: null, previousDocument: null), entry);
+                    schemaId, key, document: null, previousDocument: null) { Relayed = entry }, entry);
                 _lastWriteSequence[(schemaId, key.Value)] = entry.Sequence;
             }
         }
 
         private CultNetShardLogEntryMessage ToLogEntryMessage(CultNetShardMutationLogEntry entry)
         {
+            if (entry.Relayed != null)
+            {
+                return NormalizeWireLogEntry(entry, entry.Relayed);
+            }
+
             if (entry.Kind == CultNetDatabaseChangeKind.Removed || entry.Document == null)
             {
                 return new CultNetShardLogEntryMessage

tests/GameCult.Networking.Tests/SoulP4cProbeTests.cs

@@ -11,7 +11,9 @@ using static GameCult.Networking.Tests.NetworkingTests;
 
 namespace GameCult.Networking.Tests
 {
-    // Soul pass 4c scratch probes (not for merge).
+    // Soul pass 4c probes. They guard: a committed put whose shard log failed is not answered as a refusal (Q1); reads and watch agree
+    // on what a database serves (Q2); a replica log apply leaves out a row its shard does not own (Q3); the unowned refusal is not a
+    // shard-authority exception (Q4); two databases over one cache aggregate their log failures (Q5).
     public sealed class SoulP4cProbeTests
     {
         private const string ShardId = "soulc-shard";

tests/GameCult.Networking.Tests/SoulP4dProbeTests.cs

@@ -0,0 +1,201 @@
+#nullable enable
+using System;
+using System.Collections.Generic;
+using System.Diagnostics;
+using System.IO;
+using System.Linq;
+using System.Threading.Tasks;
+using GameCult.Caching;
+using NUnit.Framework;
+using static GameCult.Networking.Tests.NetworkingTests;
+
+namespace GameCult.Networking.Tests
+{
+    // Soul pass 4d probes. They guard the replica log relay (a skipped entry is relayed as the primary sent it, in memory and on
+    // disk), key-prefixed shard reads, and measure two costs; the measurements are explicit.
+    public sealed class SoulP4dProbeTests
+    {
+        private const string ShardId = "soupd-shard";
+        private static readonly string Now = DateTimeOffset.UtcNow.ToString("O");
+
+        private static string NoteSchema(CultCache cache) => cache.Registry.GetRequired<NetworkSchemaNote>().SchemaId;
+        private static string MeshSchema(CultCache cache) => cache.Registry.GetRequired<MeshQuickstartNote>().SchemaId;
+
+        private static CultNetDatabase Db(CultCache cache, bool primary, string[] schemas, string? keyPrefix = null, ICultNetShardMutationLogStore? store = null) =>
+            new(cache, new CultNetDatabaseOptions
+            {
+                Shards = [new CultNetShardDescriptor(ShardId, "primary", epoch: 1, isPrimary: primary, schemaIds: schemas, keyPrefix: keyPrefix)],
+                MutationLogStore = store
+            });
+
+        private static CultNetShardLogResponseMessage UnownedPutLog(CultNetDatabase anyDb, string key)
+        {
+            var put = anyDb.Documents.CreateRawDocumentPutMessage("wire", new CultRecordHandle<MeshQuickstartNote>(new CultRecordKey(key)), new MeshQuickstartNote { NoteId = "p", Body = "primary-put" });
+            put.ShardId = ShardId;
+            put.ShardEpoch = 1;
+            return new CultNetShardLogResponseMessage
+            {
+                MessageId = "log",
+                ShardId = ShardId,
+                ShardEpoch = 1,
+                Entries = [new CultNetShardLogEntryMessage { Sequence = 1, CommittedAt = Now, ChangeKind = "added", Put = put }]
+            };
+        }
+
+        // Q3a: a replica (no store) that skips an unowned put relays it to a chained replica as a delete.
+        [Test, Category("SoulFinding")]
+        public async Task Q3a_MemoryLogRelaysASkippedPutAsARemoval()
+        {
+            var cache = new CultCache();
+            var replica = Db(cache, primary: false, [NoteSchema(cache)]);
+            await replica.ApplyShardLogResponseAsync(UnownedPutLog(replica, "relay:k"));
+            var relayed = replica.GetMutationLogMessages(ShardId);
+            TestContext.Out.WriteLine($"memory relay: kind={relayed[0].ChangeKind} put={relayed[0].Put != null} delete={relayed[0].Delete != null}");
+            Assert.That(relayed[0].ChangeKind, Is.EqualTo("added"), "the relayed entry keeps the primary's kind");
+        }
+
+        // Q3b: the same replica with a durable store relays the primary's put (disk and memory disagree).
+        [Test, Category("SoulFinding")]
+        public async Task Q3b_StoreBackedLogRelaysTheSkippedPutAsAPut()
+        {
+            var dir = Path.Combine(Path.GetTempPath(), "soulp4d-" + Guid.NewGuid().ToString("N"));
+            var cache = new CultCache();
+            var replica = Db(cache, primary: false, [NoteSchema(cache)], store: new CultNetFileShardMutationLogStore(dir));
+            await replica.ApplyShardLogResponseAsync(UnownedPutLog(replica, "relay:k"));
+            var relayed = replica.GetMutationLogMessages(ShardId);
+            var memory = replica.GetMutationLog(ShardId);
+            TestContext.Out.WriteLine($"store relay: kind={relayed[0].ChangeKind} put={relayed[0].Put != null}; memory entry kind={memory[0].Kind} doc={(memory[0].Document == null ? "null" : "set")}");
+            Assert.That(relayed[0].ChangeKind, Is.EqualTo("added"));
+        }
+
+        // Q3c: a chained replica that does own the row (config drift) and holds it deletes it on the relayed entry.
+        [Test, Category("SoulFinding")]
+        public async Task Q3c_AChainedReplicaThatOwnsTheRowDeletesWhatThePrimaryPut()
+        {
+            var cacheA = new CultCache();
+            var a = Db(cacheA, primary: false, [NoteSchema(cacheA)]);
+            await a.ApplyShardLogResponseAsync(UnownedPutLog(a, "relay:k"));
+
+            var cacheC = new CultCache();
+            var c = Db(cacheC, primary: false, [NoteSchema(cacheC), MeshSchema(cacheC)]);
+            await cacheC.UpsertAsync(new MeshQuickstartNote { NoteId = "p", Body = "earlier" }, new CultRecordHandle<MeshQuickstartNote>(new CultRecordKey("relay:k")));
+            await c.ApplyShardLogResponseAsync(new CultNetShardLogResponseMessage
+            {
+                MessageId = "chain",
+                ShardId = ShardId,
+                ShardEpoch = 1,
+                Entries = a.GetMutationLogMessages(ShardId).ToArray()
+            });
+            var row = cacheC.Get<MeshQuickstartNote>(new CultRecordKey("relay:k"));
+            TestContext.Out.WriteLine($"chained row after relay: {row?.Body ?? "<deleted>"}");
+            Assert.That(row, Is.Not.Null, "the primary put this row; a relay must not delete it");
+        }
+
+        // A skipped entry is relayed exactly as the primary sent it, whether the replica keeps its log in memory or on disk.
+        [Test]
+        public async Task SkippedEntriesAreRelayedIdenticallyFromMemoryAndFromDisk()
+        {
+            var dir = Path.Combine(Path.GetTempPath(), "soulp4d-parity-" + Guid.NewGuid().ToString("N"));
+            try
+            {
+                var memoryCache = new CultCache();
+                var inMemory = Db(memoryCache, primary: false, [NoteSchema(memoryCache)]);
+                var diskCache = new CultCache();
+                var onDisk = Db(diskCache, primary: false, [NoteSchema(diskCache)], store: new CultNetFileShardMutationLogStore(dir));
+                var log = UnownedPutLog(inMemory, "relay:k");
+                var removal = new CultNetShardLogEntryMessage
+                {
+                    Sequence = 2,
+                    CommittedAt = Now,
+                    ChangeKind = "removed",
+                    Delete = new CultNetDocumentDeleteMessage
+                    {
+                        MessageId = "d",
+                        SchemaId = memoryCache.Registry.GetRequired<MeshQuickstartNote>().SchemaId,
+                        RecordKey = "relay:k",
+                        ShardId = ShardId,
+                        ShardEpoch = 1
+                    }
+                };
+                log.Entries = [log.Entries[0], removal];
+
+                await inMemory.ApplyShardLogResponseAsync(log);
+                await onDisk.ApplyShardLogResponseAsync(log);
+
+                var memory = inMemory.GetMutationLogMessages(ShardId);
+                var disk = onDisk.GetMutationLogMessages(ShardId);
+                Assert.That(memory.Select(e => e.ChangeKind), Is.EqualTo(new[] { "added", "removed" }));
+                Assert.That(memory.Select(e => e.ChangeKind), Is.EqualTo(disk.Select(e => e.ChangeKind)));
+                Assert.That(memory.Select(e => e.Put != null), Is.EqualTo(new[] { true, false }));
+                Assert.That(memory.Select(e => e.Put != null), Is.EqualTo(disk.Select(e => e.Put != null)));
+                Assert.That(memory.Select(e => e.Delete != null), Is.EqualTo(new[] { false, true }));
+                Assert.That(memory.Select(e => e.Delete != null), Is.EqualTo(disk.Select(e => e.Delete != null)));
+                Assert.That(memory[0].Put!.Document!.RecordKey, Is.EqualTo(disk[0].Put!.Document!.RecordKey));
+                Assert.That(memory[0].Put!.Document!.Payload, Is.EqualTo(disk[0].Put!.Document!.Payload), "the document bytes are the primary's");
+            }
+            finally
+            {
+                if (Directory.Exists(dir)) Directory.Delete(dir, true);
+            }
+        }
+
+        // Q2: key-prefixed shard: the schema is owned, a key outside the prefix is not.
+        [Test, Category("SoulPin")]
+        public async Task Q2_KeyPrefixedReadsServeOnlyTheOwnedKeys()
+        {
+            var cache = new CultCache();
+            var database = Db(cache, primary: true, [NoteSchema(cache)], keyPrefix: "a:");
+            await cache.UpsertAsync(new NetworkSchemaNote { Text = "a" }, new CultRecordHandle<NetworkSchemaNote>(new CultRecordKey("a:1")));
+            await cache.UpsertAsync(new NetworkSchemaNote { Text = "b" }, new CultRecordHandle<NetworkSchemaNote>(new CultRecordKey("b:1")));
+            Assert.That(database.GetAll<NetworkSchemaNote>().Select(n => n.Text), Is.EqualTo(new[] { "a" }));
+            Assert.That(await database.GetAsync<NetworkSchemaNote>(new CultRecordKey("b:1")), Is.Null);
+            Assert.That(await database.GetAsync<NetworkSchemaNote>(new CultRecordKey("a:1")), Is.Not.Null);
+            using var server = new Server(cache, ServerSecurityOptions.Development());
+            using var databaseServer = new CultNetDatabaseServer(server, database);
+            var snapshot = databaseServer.CreateSnapshotResponse(new CultNetSnapshotRequestMessage { MessageId = "all" });
+            Assert.That(snapshot.Documents.Select(d => d.RecordKey), Is.EqualTo(new[] { "a:1" }));
+        }
+
+        [Test, Category("SoulPin")]
+        public async Task Q2_PrimaryAllServesEveryCachedRow()
+        {
+            var cache = new CultCache();
+            var database = new CultNetDatabase(cache, new CultNetDatabaseOptions { Shards = [CultNetShardDescriptor.PrimaryAll()] });
+            await cache.UpsertAsync(new MeshQuickstartNote { NoteId = "x", Body = "x" }, new CultRecordHandle<MeshQuickstartNote>(new CultRecordKey("any:1")));
+            Assert.That(database.GetAll<object>().Count(), Is.EqualTo(1));
+            Assert.That(database.GetByName<MeshQuickstartNote>("x"), Is.Not.Null);
+        }
+
+        // Q2 cost: GetAll through the database versus the cache, scoped shard (reflection per row).
+        [Test, Category("SoulMeasure"), Explicit("Measurement, no assertion.")]
+        public void Q2_GetAllCost()
+        {
+            var cache = new CultCache();
+            var database = Db(cache, primary: false, [NoteSchema(cache)], keyPrefix: "n:");
+            cache.Commit(batch =>
+            {
+                for (var i = 0; i < 20000; i++)
+                    batch.Upsert(new NetworkSchemaNote { Text = "t" + i }, new CultRecordHandle<NetworkSchemaNote>(new CultRecordKey("n:" + i)));
+            });
+            _ = cache.GetAll<NetworkSchemaNote>().Count();
+            _ = database.GetAll<NetworkSchemaNote>().Count();
+            var sw = Stopwatch.StartNew();
+            for (var r = 0; r < 5; r++) _ = cache.GetAll<NetworkSchemaNote>().Count();
+            var cacheMs = sw.Elapsed.TotalMilliseconds / 5;
+            sw.Restart();
+            for (var r = 0; r < 5; r++) _ = database.GetAll<NetworkSchemaNote>().Count();
+            var dbMs = sw.Elapsed.TotalMilliseconds / 5;
+            TestContext.Out.WriteLine($"GetAll 20000 rows: cache={cacheMs:F2}ms database={dbMs:F2}ms ratio={dbMs / Math.Max(cacheMs, 0.001):F1}x");
+        }
+
+        // Wire: the bytes a peer receives for the unowned refusal.
+        [Test, Category("SoulMeasure"), Explicit("Prints the wire bytes the Rust captured-bytes case pins.")]
+        public void Wire_UnownedSchemaBytes()
+        {
+            var message = CultNetErrorMessage.ForUnownedSchema(new CultNetUnownedSchemaException("some.schema.v1", new CultRecordKey("some:key")));
+            var bytes = MessagePack.MessagePackSerializer.Serialize(message, CultNetSchemaMessageSerialization.Options);
+            TestContext.Out.WriteLine("HEX=" + Convert.ToHexString(bytes));
+            TestContext.Out.WriteLine("JSON=" + MessagePack.MessagePackSerializer.ConvertToJson(bytes));
+        }
+    }
+}