CultNet refuses a variant on both live-change paths, loudly, by key

A load or write that landed a variant reached CultNet subscribers as a plain
"added" record carrying the full resolved payload; a later base edit published
only the base, so the subscriber's copy of the variant went stale. Under Q6
CultNet carries no variants until its own cut. Both live paths now refuse a
change whose key holds a variant, after selection and authorization match, and
send the matching subscriber a CultNetErrorMessage naming the variant and its
base. Refused loudly rather than omitted: an omission is a silent hole in the
subscriber's view, and the snapshot path already refuses by throwing.

Tests: a real load through a real RUDP subscription never delivers the variant
as a record and delivers the typed error; the database server's change builder
refuses the same change.

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

src/GameCult.Networking/CultNetDatabaseServer.cs

@@ -504,7 +504,18 @@ namespace GameCult.Networking
         {
             return _database.WatchAllChanges().Subscribe(change =>
             {
-                var outbound = CreateChangeMessage(change, subscriptionId, selection);
+                CultNetDatabaseChangeRawMessage? outbound;
+                try
+                {
+                    outbound = CreateChangeMessage(change, subscriptionId, selection);
+                }
+                catch (NotSupportedException refusal)
+                {
+                    // Q6: a variant is not a record CultNet can carry; the subscriber that would have received it is told, by key.
+                    peer.SendCultNet(new CultNetErrorMessage { Error = refusal.Message });
+                    return;
+                }
+
                 if (outbound != null)
                 {
                     peer.SendCultNet(outbound);
@@ -549,11 +560,12 @@ namespace GameCult.Networking
                 };
             }
 
+            CultNetDocumentRegistry.RefuseVariantChange(_database.Cache, key);
             return new CultNetDatabaseChangeRawMessage
             {
                 MessageId = Guid.NewGuid().ToString("N"),
                 SubscriptionId = subscriptionId,
-                ChangeKind = kind == CultNetDatabaseChangeKind.Added ? "added" : "updated",
+                ChangeKind = kind ==CultNetDatabaseChangeKind.Added ? "added" : "updated",
                 Document = _database.Documents.ToRawRecord(descriptor, key, document, DateTimeOffset.UtcNow.ToString("O"))
             };
         }

src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs

@@ -315,7 +315,18 @@ namespace GameCult.Networking
                     }
 
                     var sourceRecordKey = ResolveChangeRecordKey(change);
-                    var matched = CreateMatchedRecord(change, request, peer);
+                    CultNetRawDocumentRecord? matched;
+                    try
+                    {
+                        matched = CreateMatchedRecord(change, request, peer);
+                    }
+                    catch (NotSupportedException refusal)
+                    {
+                        // Q6: a variant is not a record CultNet can carry; the subscriber that would have received it is told, by key.
+                        peer.SendCultNet(new CultNetErrorMessage { Error = refusal.Message });
+                        return;
+                    }
+
                     ApplyProjectedChange(
                         projection,
                         sourceRecordKey,
@@ -541,6 +552,7 @@ namespace GameCult.Networking
             if (kind == CultNetDatabaseChangeKind.Removed || document == null)
                 return null;
 
+            CultNetDocumentRegistry.RefuseVariantChange(_database.Cache, key);
             return _database.Documents.ToRawRecord(descriptor, key, document, DateTimeOffset.UtcNow.ToString("O"));
         }
 

src/GameCult.Networking/CultNetDocumentRegistry.cs

@@ -551,10 +551,17 @@ namespace GameCult.Networking
 
         // CultNet carries no variant deltas yet, and a resolved view sent as a plain record would be a lie about the store.
         private static CultStoredDocument RefuseVariant(CultStoredDocument entry) =>
-            entry.Variant == null
-                ? entry
-                : throw new NotSupportedException(
-                    $"Record '{entry.Key.Value}' is a variant of '{entry.Variant.BaseKey}'; CultNet does not carry document variants yet.");
+            entry.Variant == null ? entry : throw VariantRefusal(entry);
+
+        // The live-change funnel's refusal: a load or write that lands a variant must not reach a subscriber as a record.
+        internal static void RefuseVariantChange(CultCache cache, CultRecordKey key)
+        {
+            if (cache.GetStored(key) is { Variant: not null } entry)
+                throw VariantRefusal(entry);
+        }
+
+        private static NotSupportedException VariantRefusal(CultStoredDocument entry) =>
+            new($"Record '{entry.Key.Value}' is a variant of '{entry.Variant!.BaseKey}'; CultNet does not carry document variants yet.");
 
         /// <summary>
         /// Builds the wire record for one evaluated row. The shared path under snapshot and change

tests/GameCult.Networking.Tests/CultNetDocumentRegistryVariantTests.cs

@@ -1,5 +1,12 @@
 #nullable enable
 using System;
+using System.Collections.Concurrent;
+using System.IO;
+using System.Linq;
+using System.Net;
+using System.Net.Sockets;
+using System.Reflection;
+using System.Threading;
 using System.Threading.Tasks;
 using GameCult.Caching;
 using GameCult.Caching.MessagePack;
@@ -63,6 +70,128 @@ namespace GameCult.Networking.Tests
             Assert.That(documents.CreateRawSnapshotResponse(cache, "m2").Documents, Has.Length.EqualTo(1));
         }
 
+        // The live-change funnel: a load that lands a variant reaches a subscriber as neither a record nor silence.
+        [Test]
+        public async Task ALoadThatLandsAVariantNeverReachesASubscriberAsARecordAndTheSubscriberIsToldByKey()
+        {
+            var registry = CultDocumentRegistry.Shared;
+            var directory = Path.Combine(Path.GetTempPath(), $"cultlib-variant-live-{Guid.NewGuid():N}");
+            Directory.CreateDirectory(directory);
+            try
+            {
+                var path = Path.Combine(directory, "wire.cc");
+                using (var writer = CultCacheMessagePack.Create(path, new CultCacheOpenOptions { Registry = registry }))
+                {
+                    writer.Commit(batch => batch.Upsert(typeof(VariantWireFixture), new VariantWireFixture { Name = "base", Power = 1 }, BaseKey));
+                    writer.UpsertVariantAsync(VariantKey, BaseKey, new[]
+                    {
+                        writer.Override<VariantWireFixture>(nameof(VariantWireFixture.Name), "big"),
+                        writer.Override<VariantWireFixture>(nameof(VariantWireFixture.Power), 9)
+                    }).GetAwaiter().GetResult();
+                }
+
+                var cache = new CultCache(registry, CultCacheMessagePack.CreateCodec(registry));
+                var documents = new CultNetDocumentRegistry(registry)
+                    .Register(CultNetDocumentBinding.ForDocument<VariantWireFixture>(registry));
+                var database = new CultNetDatabase(cache, new CultNetDatabaseOptions { DocumentRegistry = documents });
+                var socket = new Socket(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp);
+                socket.Bind(new IPEndPoint(IPAddress.Loopback, 0));
+                socket.ReceiveTimeout = 20;
+                using var server = new RudpCultNetSchemaServer(new RudpCultNetSchemaServerOptions
+                {
+                    RuntimeId = "variant-live-server",
+                    Socket = socket
+                });
+                using var subscriptions = new CultNetDatabaseSubscriptionServer(server, database);
+                using var cancellation = new CancellationTokenSource();
+                var serverThread = new Thread(() =>
+                {
+                    while (!cancellation.IsCancellationRequested)
+                    {
+                        _ = server.PollOnceAsync().GetAwaiter().GetResult();
+                        Thread.Sleep(1);
+                    }
+                }) { IsBackground = true };
+                serverThread.Start();
+                try
+                {
+                    using var client = CultNetSchemaClients.CreateRudp("variant-live-client");
+                    var snapshot = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+                    var changes = new ConcurrentQueue<CultNetDatabaseChangeRawMessage>();
+                    var errors = new ConcurrentQueue<CultNetErrorMessage>();
+                    client.OnCultNet<CultNetSnapshotResponseRawMessage>(_ => snapshot.TrySetResult(true));
+                    client.OnCultNet<CultNetDatabaseChangeRawMessage>(changes.Enqueue);
+                    client.OnCultNet<CultNetErrorMessage>(errors.Enqueue);
+                    client.Connect("127.0.0.1", server.LocalEndPoint.Port);
+                    await WaitUntilAsync(() => client.Connected);
+                    client.SendCultNet(new CultNetDatabaseSubscribeMessage
+                    {
+                        MessageId = "subscribe-all",
+                        SubscriptionId = "all",
+                        IncludeSnapshot = true
+                    });
+                    await WaitUntilAsync(() => snapshot.Task.IsCompleted);
+
+                    cache.AddBackingStore(new SingleFileMessagePackBackingStore(path));
+
+                    await WaitUntilAsync(() => !errors.IsEmpty || changes.Any(change => change.Document?.RecordKey == VariantKey.Value));
+                    await Task.Delay(200); // let anything else already in flight arrive
+
+                    Assert.That(changes.Select(change => change.Document?.RecordKey), Does.Not.Contain(VariantKey.Value),
+                        "a variant must not reach a subscriber as a plain record");
+                    Assert.That(errors.Select(error => error.Error), Has.Some.Contains(VariantKey.Value),
+                        "the subscriber is told, naming the variant");
+                }
+                finally
+                {
+                    cancellation.Cancel();
+                    serverThread.Join();
+                }
+            }
+            finally
+            {
+                Directory.Delete(directory, recursive: true);
+            }
+        }
+
+        [Test]
+        public async Task TheDatabaseServersLiveChangeForAVariantIsRefusedNamingIt()
+        {
+            var (cache, documents) = await BuildFixtureAsync(withVariant: true);
+            var database = new CultNetDatabase(cache, new CultNetDatabaseOptions { DocumentRegistry = documents });
+            using var server = new Server(cache, ServerSecurityOptions.Development());
+            using var databaseServer = new CultNetDatabaseServer(server, database);
+            var schemaId = cache.Registry.GetRequired<VariantWireFixture>().SchemaId;
+            var change = new CultNetDatabaseChange<VariantWireFixture>(
+                CultNetDatabaseChangeKind.Added,
+                VariantKey,
+                schemaId,
+                database.Shards[0],
+                cache.Get<VariantWireFixture>(VariantKey)!,
+                previousDocument: null);
+            var method = typeof(CultNetDatabaseServer).GetMethod("CreateChangeMessage", BindingFlags.Instance | BindingFlags.NonPublic)!;
+
+            var refused = Assert.Throws<TargetInvocationException>(() => method.Invoke(databaseServer, new object[]
+            {
+                change, "sub-1", new CultNetSelection { Projection = CultNetSelectionProjections.Document }
+            }))!;
+
+            Assert.That(refused.InnerException, Is.TypeOf<NotSupportedException>());
+            Assert.That(refused.InnerException!.Message, Does.Contain(VariantKey.Value).And.Contain(BaseKey.Value));
+        }
+
+        private static async Task WaitUntilAsync(Func<bool> condition)
+        {
+            var deadline = DateTimeOffset.UtcNow + TimeSpan.FromSeconds(5);
+            while (!condition())
+            {
+                if (DateTimeOffset.UtcNow > deadline)
+                    Assert.Fail("Condition was not satisfied before timeout.");
+                await Task.Delay(5);
+            }
+        }
+
+
         [CultDocument("variants.wire-fixture", "variants.wire-fixture.v1")]
         [MessagePackObject(AllowPrivate = true)]
         public sealed class VariantWireFixture