Add exact reactive value subscriptions

src/GameCult.Mesh/docs/reliability-migration-plan.md

@@ -947,6 +947,13 @@ ownership contract.
 **Outputs:** both modes produce the same typed initial values and change event.
 Only durable replication mutates the consumer's CultCache.
 
+`CultNetDatabaseSubscriptionClient.SubscribeLiveValueAsync<T>(...)` is the
+single-record ergonomic path. Its `CultNetLiveValue<T>` owns the exact
+record/schema filter, current value, typed change/removal notifications, and
+unsubscribe lifetime. Consumers no longer need a process-wide change handler,
+an object type switch, or a renderer-local replica cache to model one reactive
+variable.
+
 **Derived state:** renderer-local presentation and reactive variables are
 derived from live notifications. They are not database records merely because
 CultNet carried them.

src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs

@@ -55,6 +55,44 @@ namespace GameCult.Networking
         public Task<Exception>? BackgroundFailure =>
             (_client as RudpCultNetSchemaClient)?.BackgroundFailure;
 
+        /// <summary>
+        /// Subscribes to one exact typed record as an ephemeral reactive value. The returned value
+        /// owns filtering and unsubscribe; no received payload is written into the local cache.
+        /// </summary>
+        public async Task<CultNetLiveValue<TDocument>> SubscribeLiveValueAsync<TDocument>(
+            string subscriptionId,
+            string recordKey,
+            CancellationToken cancellationToken = default)
+            where TDocument : class
+        {
+            if (string.IsNullOrWhiteSpace(recordKey))
+                throw new ArgumentException("Record key is required.", nameof(recordKey));
+            var schemaId = _documents.GetByDocumentType(typeof(TDocument))?.SchemaId ??
+                _cache.Registry.GetRequired<TDocument>().SchemaId;
+            var value = new CultNetLiveValue<TDocument>(this, subscriptionId, recordKey, schemaId);
+            try
+            {
+                var initial = await SubscribeAsync(
+                        subscriptionId,
+                        recordKeys: new[] { recordKey },
+                        schemaIds: new[] { schemaId },
+                        deliveryMode: CultNetDatabaseSubscriptionDeliveryMode.Live,
+                        cancellationToken: cancellationToken)
+                    .ConfigureAwait(false);
+                var documents = initial.OfType<TDocument>().ToArray();
+                if (documents.Length != 1)
+                    throw new InvalidOperationException(
+                        $"Live value subscription '{subscriptionId}' expected one '{schemaId}' record '{recordKey}' but received {documents.Length}.");
+                value.Initialize(documents[0]);
+                return value;
+            }
+            catch
+            {
+                value.Dispose();
+                throw;
+            }
+        }
+
         /// <summary>
         /// Starts a remote subscription and waits until its initial matching snapshot is in the local cache.
         /// </summary>
@@ -250,6 +288,110 @@ namespace GameCult.Networking
         }
     }
 
+    /// <summary>One exact typed remote record kept as an ephemeral current value.</summary>
+    public sealed class CultNetLiveValue<TDocument> : IDisposable where TDocument : class
+    {
+        private readonly CultNetDatabaseSubscriptionClient _owner;
+        private readonly string _subscriptionId;
+        private readonly string _recordKey;
+        private readonly string _schemaId;
+        private readonly object _gate = new();
+        private TDocument? _current;
+        private bool _hasValue;
+        private bool _disposed;
+
+        internal CultNetLiveValue(
+            CultNetDatabaseSubscriptionClient owner,
+            string subscriptionId,
+            string recordKey,
+            string schemaId)
+        {
+            _owner = owner ?? throw new ArgumentNullException(nameof(owner));
+            _subscriptionId = string.IsNullOrWhiteSpace(subscriptionId)
+                ? throw new ArgumentException("Subscription id is required.", nameof(subscriptionId))
+                : subscriptionId;
+            _recordKey = recordKey;
+            _schemaId = schemaId;
+            _owner.Changed += OnChanged;
+        }
+
+        /// <summary>Raised when the provider publishes a new typed value.</summary>
+        public event Action<TDocument>? Changed;
+
+        /// <summary>Raised when the provider removes the selected record.</summary>
+        public event Action? Removed;
+
+        /// <summary>Gets whether the selected record currently exists.</summary>
+        public bool HasValue
+        {
+            get { lock (_gate) return _hasValue; }
+        }
+
+        /// <summary>Gets the current value, or fails when the record has not been published.</summary>
+        public TDocument Current
+        {
+            get
+            {
+                lock (_gate)
+                    return _hasValue
+                        ? _current!
+                        : throw new InvalidOperationException(
+                            $"Live value '{_recordKey}' is not currently published.");
+            }
+        }
+
+        internal void Initialize(TDocument document)
+        {
+            lock (_gate)
+            {
+                if (_disposed || _hasValue) return;
+                _current = document;
+                _hasValue = true;
+            }
+        }
+
+        public void Dispose()
+        {
+            lock (_gate)
+            {
+                if (_disposed) return;
+                _disposed = true;
+            }
+            _owner.Changed -= OnChanged;
+            _owner.Unsubscribe(_subscriptionId);
+        }
+
+        private void OnChanged(CultNetReplicatedDocumentChange change)
+        {
+            if (!string.Equals(change.SubscriptionId, _subscriptionId, StringComparison.Ordinal) ||
+                !string.Equals(change.RecordKey, _recordKey, StringComparison.Ordinal) ||
+                !string.Equals(change.SchemaId, _schemaId, StringComparison.Ordinal))
+                return;
+
+            if (string.Equals(change.ChangeKind, "removed", StringComparison.Ordinal))
+            {
+                lock (_gate)
+                {
+                    if (_disposed) return;
+                    _current = null;
+                    _hasValue = false;
+                }
+                Removed?.Invoke();
+                return;
+            }
+
+            if (change.Document is not TDocument document)
+                return;
+            lock (_gate)
+            {
+                if (_disposed) return;
+                _current = document;
+                _hasValue = true;
+            }
+            Changed?.Invoke(document);
+        }
+    }
+
     /// <summary>Describes one typed document change after local replication.</summary>
     public sealed class CultNetReplicatedDocumentChange
     {

tests/GameCult.Networking.Tests/NetworkingTests.cs

@@ -1693,6 +1693,64 @@ namespace GameCult.Networking.Tests
             Assert.That(targetCache.Get(new CultRecordKey(recordKey)), Is.Null);
         }
 
+        [Test]
+        public async Task DatabaseSubscriptionClient_LiveValueOwnsExactFilteringCurrentValueAndUnsubscribe()
+        {
+            var sourceCache = new CultCache();
+            var sourceDatabase = new CultNetDatabase(sourceCache);
+            const string recordKey = "tests:subscription-client:reactive-note";
+            await sourceDatabase.PutAsync(new CultRecordKey(recordKey), new NetworkSchemaNote
+            {
+                Schema = "tests.networking_note.v1",
+                Text = "initial"
+            });
+            using var server = new RudpCultNetSchemaServer(new RudpCultNetSchemaServerOptions
+            {
+                RuntimeId = "database-live-value-server",
+                Socket = BindUdpSocket()
+            });
+            using var subscriptions = new CultNetDatabaseSubscriptionServer(server, sourceDatabase);
+            using var cancellation = new CancellationTokenSource();
+            var serverThread = new Thread(() =>
+            {
+                while (!cancellation.IsCancellationRequested)
+                {
+                    _ = server.PollOnceAsync().GetAwaiter().GetResult();
+                    Thread.Sleep(1);
+                }
+            }) { IsBackground = true };
+            serverThread.Start();
+
+            var targetCache = new CultCache();
+            var targetDocuments = new CultNetDocumentRegistry(targetCache.Registry)
+                .Register(CultNetDocumentBinding.ForDocument<NetworkSchemaNote>(targetCache.Registry));
+            var transport = CultNetSchemaClients.CreateRudp("database-live-value-client");
+            using var client = new CultNetDatabaseSubscriptionClient(transport, targetCache, targetDocuments);
+            transport.Connect("127.0.0.1", server.LocalEndPoint.Port);
+            await WaitUntilAsync(() => transport.Connected, TimeSpan.FromSeconds(2));
+
+            using var value = await AwaitWithTimeout(
+                client.SubscribeLiveValueAsync<NetworkSchemaNote>("reactive-note", recordKey),
+                TimeSpan.FromSeconds(2));
+            Assert.That(value.HasValue, Is.True);
+            Assert.That(value.Current.Text, Is.EqualTo("initial"));
+            Assert.That(targetCache.Get(new CultRecordKey(recordKey)), Is.Null);
+
+            var changed = new TaskCompletionSource<NetworkSchemaNote>(TaskCreationOptions.RunContinuationsAsynchronously);
+            value.Changed += document => changed.TrySetResult(document);
+            await sourceDatabase.PutAsync(new CultRecordKey(recordKey), new NetworkSchemaNote
+            {
+                Schema = "tests.networking_note.v1",
+                Text = "updated"
+            });
+            var update = await AwaitWithTimeout(changed.Task, TimeSpan.FromSeconds(2));
+
+            cancellation.Cancel();
+            Assert.That(update.Text, Is.EqualTo("updated"));
+            Assert.That(value.Current.Text, Is.EqualTo("updated"));
+            Assert.That(targetCache.Get(new CultRecordKey(recordKey)), Is.Null);
+        }
+
         [Test]
         public async Task DatabaseSubscription_FiltersLiveChangesByWireSchemaBinding()
         {