Negotiate exact hot-body transport demand

Directory.Build.props

@@ -14,7 +14,7 @@
     <PackageLicenseExpression>MIT</PackageLicenseExpression>
     <Copyright>© Meta Vens and contributors</Copyright>
     <PublishRepositoryUrl>true</PublishRepositoryUrl>
-	<Version>1.0.19</Version>
+	<Version>1.0.20</Version>
 
     <!-- Symbols/debugging -->
     <EmbedUntrackedSources>true</EmbedUntrackedSources>

src/GameCult.Mesh/CultMeshBodies.cs

@@ -185,7 +185,7 @@ namespace GameCult.Mesh
             CultMeshBodyDescriptor networkFallback,
             CultMeshBodyValidationRequest request)
         {
-            return NegotiateReadOnly(null, preferred, networkFallback, request);
+            return NegotiateReadOnly(null, new[] { preferred, networkFallback }, request);
         }
 
         internal CultMeshBodyNegotiationResult NegotiateReadOnly(
@@ -194,12 +194,33 @@ namespace GameCult.Mesh
             CultMeshBodyDescriptor networkFallback,
             CultMeshBodyValidationRequest request)
         {
-            if (preferred == null) throw new ArgumentNullException(nameof(preferred));
-            if (networkFallback == null) throw new ArgumentNullException(nameof(networkFallback));
-            if (!SameLogicalGeneration(preferred, networkFallback))
-                throw new InvalidOperationException("CultMesh local and network representations describe different logical body generations.");
-            CultMeshBodyDescriptorValidator.Validate(preferred, request);
-            CultMeshBodyDescriptorValidator.Validate(networkFallback, request);
+            return NegotiateReadOnly(producerId, new[] { preferred, networkFallback }, request);
+        }
+
+        /// <summary>
+        /// Selects the first representation the consumer can validate and open. Representation order is
+        /// publisher preference; adapter availability and capability validity decide the actual plane.
+        /// </summary>
+        public CultMeshBodyNegotiationResult NegotiateReadOnly(
+            IReadOnlyList<CultMeshBodyDescriptor> representations,
+            CultMeshBodyValidationRequest request) =>
+            NegotiateReadOnly(null, representations, request);
+
+        internal CultMeshBodyNegotiationResult NegotiateReadOnly(
+            string? producerId,
+            IReadOnlyList<CultMeshBodyDescriptor> representations,
+            CultMeshBodyValidationRequest request)
+        {
+            if (representations == null) throw new ArgumentNullException(nameof(representations));
+            if (representations.Count == 0 || representations.Any(value => value == null))
+                throw new ArgumentException("At least one body representation is required.", nameof(representations));
+            var preferred = representations[0];
+            foreach (var representation in representations)
+            {
+                if (!SameLogicalGeneration(preferred, representation))
+                    throw new InvalidOperationException("CultMesh representations describe different logical body generations.");
+                CultMeshBodyDescriptorValidator.Validate(representation, request);
+            }
             var authorized = _authorizeNamedProducer == null
                 ? _authorizeProducer(preferred)
                 : !string.IsNullOrWhiteSpace(producerId) && _authorizeNamedProducer(producerId!, preferred);
@@ -207,24 +228,29 @@ namespace GameCult.Mesh
                 throw new UnauthorizedAccessException("CultMesh body producer is not authorized for this logical body.");
 
             Exception? preferredFailure = null;
-            try
+            foreach (var representation in representations)
             {
-                if (!_adapters.TryGetValue(preferred.TransportKind, out var local))
-                    throw new NotSupportedException($"No CultMesh {preferred.TransportKind} body adapter is available.");
-                if (!local.CanOpen(preferred))
-                    throw new NotSupportedException($"The CultMesh {preferred.TransportKind} body adapter declined the preferred descriptor.");
-                var lease = local.OpenReadOnly(preferred, request);
-                return new CultMeshBodyNegotiationResult(lease, preferred.TransportKind, null);
-            }
-            catch (Exception error) when (!(error is UnauthorizedAccessException))
-            {
-                preferredFailure = error;
+                try
+                {
+                    if (!_adapters.TryGetValue(representation.TransportKind, out var adapter))
+                        throw new NotSupportedException($"No CultMesh {representation.TransportKind} body adapter is available.");
+                    if (!adapter.CanOpen(representation))
+                        throw new NotSupportedException($"The CultMesh {representation.TransportKind} body adapter declined its descriptor.");
+                    var lease = adapter.OpenReadOnly(representation, request);
+                    return new CultMeshBodyNegotiationResult(
+                        lease,
+                        preferred.TransportKind,
+                        preferredFailure);
+                }
+                catch (Exception error) when (!(error is UnauthorizedAccessException))
+                {
+                    preferredFailure ??= error;
+                }
             }
 
-            if (!_adapters.TryGetValue(CultMeshBodyTransportKind.Network, out var network) || !network.CanOpen(networkFallback))
-                throw new NotSupportedException("No CultMesh network body fallback is available.");
-            var fallback = network.OpenReadOnly(networkFallback, request);
-            return new CultMeshBodyNegotiationResult(fallback, preferred.TransportKind, preferredFailure);
+            throw new NotSupportedException(
+                "No advertised CultMesh body representation could be opened by this consumer.",
+                preferredFailure);
         }
 
         private static bool SameLogicalGeneration(CultMeshBodyDescriptor left, CultMeshBodyDescriptor right) =>

src/GameCult.Mesh/CultMeshBodyDemand.cs

@@ -0,0 +1,142 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using GameCult.Networking;
+
+#nullable enable
+
+namespace GameCult.Mesh
+{
+    /// <summary>Derived transport work for one logical hot body.</summary>
+    public sealed class CultMeshBodyDemandPlan
+    {
+        internal CultMeshBodyDemandPlan(
+            string bodyId,
+            IReadOnlyList<CultMeshBodyConsumerRoute> consumers)
+        {
+            BodyId = bodyId;
+            Consumers = consumers;
+            RequiresSharedMemory = consumers.Any(value => value.Transport == CultMeshBodyTransportKind.SharedMemory);
+            RequiresNetwork = consumers.Any(value => value.Transport == CultMeshBodyTransportKind.Network);
+        }
+
+        public string BodyId { get; }
+        public bool HasConsumers => Consumers.Count > 0;
+        public bool RequiresSharedMemory { get; }
+        public bool RequiresNetwork { get; }
+        public IReadOnlyList<CultMeshBodyConsumerRoute> Consumers { get; }
+    }
+
+    /// <summary>One consumer's owner-derived route for one logical body.</summary>
+    public sealed class CultMeshBodyConsumerRoute
+    {
+        internal CultMeshBodyConsumerRoute(
+            string consumerRuntimeId,
+            string subscriptionId,
+            CultMeshBodyTransportKind transport)
+        {
+            ConsumerRuntimeId = consumerRuntimeId;
+            SubscriptionId = subscriptionId;
+            Transport = transport;
+        }
+
+        public string ConsumerRuntimeId { get; }
+        public string SubscriptionId { get; }
+        public CultMeshBodyTransportKind Transport { get; }
+    }
+
+    /// <summary>
+    /// Owns the live projection from exact state subscriptions to required body transport planes.
+    /// A same-machine claim is transport evidence only; shared memory is selected only when both
+    /// the server peer and consumer capability agree.
+    /// </summary>
+    public sealed class CultMeshBodyDemandTracker : IDisposable
+    {
+        private readonly CultNetDatabaseSubscriptionServer? _subscriptions;
+        private readonly object _gate = new();
+        private readonly Dictionary<string, CultNetDatabaseSubscriptionDemand> _demands =
+            new(StringComparer.Ordinal);
+        private bool _disposed;
+
+        public CultMeshBodyDemandTracker(CultNetDatabaseSubscriptionServer subscriptions)
+        {
+            _subscriptions = subscriptions ?? throw new ArgumentNullException(nameof(subscriptions));
+            _subscriptions.DemandChanged += OnDemandChanged;
+        }
+
+        /// <summary>Creates a tracker fed explicitly by a subscription/session control plane.</summary>
+        public CultMeshBodyDemandTracker()
+        {
+        }
+
+        public CultMeshBodyDemandPlan Plan(string bodyId)
+        {
+            if (string.IsNullOrWhiteSpace(bodyId)) throw new ArgumentException("Body identity is required.", nameof(bodyId));
+            lock (_gate)
+            {
+                ThrowIfDisposed();
+                var routes = _demands.Values
+                    .Where(value => value.BodyIds.Contains(bodyId, StringComparer.Ordinal))
+                    .Select(value => Route(value))
+                    .Where(value => value != null)
+                    .Cast<CultMeshBodyConsumerRoute>()
+                    .OrderBy(value => value.ConsumerRuntimeId, StringComparer.Ordinal)
+                    .ThenBy(value => value.SubscriptionId, StringComparer.Ordinal)
+                    .ToArray();
+                return new CultMeshBodyDemandPlan(bodyId, routes);
+            }
+        }
+
+        public void Dispose()
+        {
+            lock (_gate)
+            {
+                if (_disposed) return;
+                _disposed = true;
+                if (_subscriptions != null) _subscriptions.DemandChanged -= OnDemandChanged;
+                _demands.Clear();
+            }
+        }
+
+        /// <summary>Applies one active or withdrawn typed body-demand observation.</summary>
+        public void Observe(CultNetDatabaseSubscriptionDemand demand)
+        {
+            if (demand == null) throw new ArgumentNullException(nameof(demand));
+            var key = demand.ConsumerRuntimeId + "\u001f" + demand.SubscriptionId;
+            lock (_gate)
+            {
+                if (_disposed) return;
+                if (demand.Active) _demands[key] = demand;
+                else _demands.Remove(key);
+            }
+        }
+
+        private void OnDemandChanged(CultNetDatabaseSubscriptionDemand demand) => Observe(demand);
+
+        private static CultMeshBodyConsumerRoute? Route(CultNetDatabaseSubscriptionDemand demand)
+        {
+            var supported = demand.SupportedBodyTransports;
+            if (demand.SameMachine && Supports(supported, CultMeshBodyTransportKind.SharedMemory))
+                return new CultMeshBodyConsumerRoute(
+                    demand.ConsumerRuntimeId,
+                    demand.SubscriptionId,
+                    CultMeshBodyTransportKind.SharedMemory);
+            if (Supports(supported, CultMeshBodyTransportKind.Network))
+                return new CultMeshBodyConsumerRoute(
+                    demand.ConsumerRuntimeId,
+                    demand.SubscriptionId,
+                    CultMeshBodyTransportKind.Network);
+            return null;
+        }
+
+        private static bool Supports(
+            IReadOnlyList<string> supported,
+            CultMeshBodyTransportKind transport) =>
+            supported.Any(value => string.Equals(value, transport.ToString(), StringComparison.OrdinalIgnoreCase));
+
+        private void ThrowIfDisposed()
+        {
+            if (_disposed) throw new ObjectDisposedException(nameof(CultMeshBodyDemandTracker));
+        }
+    }
+}

src/GameCult.Mesh/CultMeshBodyPublications.cs

@@ -13,7 +13,7 @@ namespace GameCult.Mesh
 {
     public static class CultMeshBodyPublicationSchemaVersions
     {
-        public const string Publication = "gamecult.mesh.body_publication.v1";
+        public const string Publication = "gamecult.mesh.body_publication.v2";
         public const string NetworkBody = "gamecult.mesh.network_body.v1";
     }
 
@@ -241,8 +241,7 @@ namespace GameCult.Mesh
         [Key(7)] public long Sequence { get; set; }
         [Key(8)] public CultMeshBodySynchronization Synchronization { get; set; }
         [Key(9)] public long LivenessExpiresAtUnixMs { get; set; }
-        [Key(10)] public CultMeshBodyDescriptor PreferredLocal { get; set; } = new();
-        [Key(11)] public CultMeshBodyDescriptor NetworkFallback { get; set; } = new();
+        [Key(10)] public CultMeshBodyDescriptor[] Representations { get; set; } = Array.Empty<CultMeshBodyDescriptor>();
 
         public static CultRecordKey CreateRecordKey(string bodyId, long producerEpoch, long sequence)
         {
@@ -326,12 +325,13 @@ namespace GameCult.Mesh
                 throw new InvalidOperationException("CultMesh body publication generation values must be non-negative.");
             if (expectedRecordKey.HasValue && !publication.RecordKey.Equals(expectedRecordKey.Value))
                 throw new InvalidOperationException("CultMesh body publication record key disagrees with its generation envelope.");
-            ValidateDescriptor(publication, publication.PreferredLocal, "preferred local");
-            ValidateDescriptor(publication, publication.NetworkFallback, "network fallback");
-            if (publication.PreferredLocal.TransportKind == CultMeshBodyTransportKind.Network)
-                throw new InvalidOperationException("CultMesh preferred local body descriptor must use a local transport.");
-            if (publication.NetworkFallback.TransportKind != CultMeshBodyTransportKind.Network)
-                throw new InvalidOperationException("CultMesh network fallback body descriptor must use network transport.");
+            var representations = publication.Representations ?? Array.Empty<CultMeshBodyDescriptor>();
+            if (representations.Length == 0)
+                throw new InvalidOperationException("CultMesh body publication has no transport representations.");
+            if (representations.Select(value => value?.TransportKind).Distinct().Count() != representations.Length)
+                throw new InvalidOperationException("CultMesh body publication advertises the same transport representation more than once.");
+            foreach (var representation in representations)
+                ValidateDescriptor(publication, representation, representation?.TransportKind.ToString() ?? "missing");
         }
 
         private static void ValidateDescriptor(
@@ -376,8 +376,7 @@ namespace GameCult.Mesh
                 throw new InvalidOperationException("CultMesh body publication is no longer live.");
             return _transport.NegotiateReadOnly(
                 publication.ProducerId,
-                publication.PreferredLocal,
-                publication.NetworkFallback,
+                publication.Representations,
                 request);
         }
     }

src/GameCult.Mesh/CultMeshFrameBodies.cs

@@ -1,10 +1,12 @@
 using System;
+using System.Buffers;
 using System.Collections.Generic;
 using System.IO;
 using System.IO.MemoryMappedFiles;
 using System.Security.Cryptography;
 using System.Text;
 using System.Threading;
+using Microsoft.Win32.SafeHandles;
 
 #nullable enable
 
@@ -18,6 +20,7 @@ namespace GameCult.Mesh
         private const int HeaderSize = 128;
         private const int SequenceOffset = 8;
         private const int ReaderCountOffset = 32;
+        private const int WriterCountOffset = 48;
         private const int ContractHashOffset = 64;
         private readonly object _gate = new();
         private readonly MemoryMappedFile _mapping;
@@ -27,6 +30,7 @@ namespace GameCult.Mesh
         private readonly string _token;
         private int _cursor;
         private long _nextSequence;
+        private CultMeshFrameBodyWriteLease? _activeWrite;
         private bool _disposed;
 
         public CultMeshFrameBodyPublisher(
@@ -68,42 +72,132 @@ namespace GameCult.Mesh
         public int SlotByteLength { get; }
 
         public bool TryPublish(ReadOnlySpan<byte> body, DateTimeOffset nowUtc, out CultMeshBodyDescriptor descriptor)
+        {
+            if (body.Length > SlotByteLength) throw new ArgumentOutOfRangeException(nameof(body));
+            if (!TryAcquireWrite(out var write))
+            {
+                descriptor = null!;
+                return false;
+            }
+            using (write)
+            {
+                body.CopyTo(write.Span);
+                descriptor = write.Commit(body.Length, nowUtc);
+                return true;
+            }
+        }
+
+        /// <summary>
+        /// Reserves one producer-owned shared-memory slot for direct writes. The caller must commit or dispose the lease.
+        /// </summary>
+        public bool TryAcquireWrite(out CultMeshFrameBodyWriteLease lease)
         {
             lock (_gate)
             {
                 ThrowIfDisposed();
-                if (body.Length > SlotByteLength) throw new ArgumentOutOfRangeException(nameof(body));
-                var publishedBytes = body.ToArray();
-                CultMeshBodyDescriptor? published = null;
+                if (_activeWrite != null)
+                {
+                    lease = null!;
+                    return false;
+                }
+
+                var slot = -1;
                 WithMutex(() =>
                 {
                     for (var attempt = 0; attempt < CultMeshFrameRegion.SlotCount; attempt++)
                     {
-                        var slot = (_cursor + attempt) % CultMeshFrameRegion.SlotCount;
-                        if (_control.ReadInt32(ReaderCountOffset + slot * sizeof(int)) != 0) continue;
-                        using (var view = _mapping.CreateViewAccessor(SlotOffset(slot, SlotByteLength), SlotByteLength, MemoryMappedFileAccess.Write))
-                            view.WriteArray(0, publishedBytes, 0, publishedBytes.Length);
-                        _control.Write(SequenceOffset + slot * sizeof(long), _nextSequence);
-                        _cursor = (slot + 1) % CultMeshFrameRegion.SlotCount;
-                        published = Descriptor(slot, publishedBytes, nowUtc);
-                        _nextSequence++;
+                        var candidate = (_cursor + attempt) % CultMeshFrameRegion.SlotCount;
+                        if (_control.ReadInt32(ReaderCountOffset + candidate * sizeof(int)) != 0 ||
+                            _control.ReadInt32(WriterCountOffset + candidate * sizeof(int)) != 0)
+                            continue;
+                        _control.Write(WriterCountOffset + candidate * sizeof(int), 1);
+                        slot = candidate;
+                        _cursor = (candidate + 1) % CultMeshFrameRegion.SlotCount;
                         return;
                     }
                 });
-                descriptor = published!;
-                return published != null;
+                if (slot < 0)
+                {
+                    lease = null!;
+                    return false;
+                }
+
+                try
+                {
+                    var view = _mapping.CreateViewAccessor(
+                        SlotOffset(slot, SlotByteLength),
+                        SlotByteLength,
+                        MemoryMappedFileAccess.ReadWrite);
+                    lease = new CultMeshFrameBodyWriteLease(this, view, slot, _nextSequence, SlotByteLength);
+                    _activeWrite = lease;
+                    return true;
+                }
+                catch
+                {
+                    WithMutex(() => _control.Write(WriterCountOffset + slot * sizeof(int), 0));
+                    throw;
+                }
             }
         }
 
-        private CultMeshBodyDescriptor Descriptor(int slot, ReadOnlySpan<byte> body, DateTimeOffset nowUtc) => new()
+        internal CultMeshBodyDescriptor Commit(
+            CultMeshFrameBodyWriteLease lease,
+            int byteLength,
+            DateTimeOffset nowUtc)
         {
-            BodyId = BodyId, SchemaId = SchemaId, LayoutVersion = LayoutVersion, ByteSize = body.Length,
-            Capacity = Capacity, ProducerEpoch = ProducerEpoch, Sequence = _nextSequence,
+            lock (_gate)
+            {
+                ThrowIfDisposed();
+                if (!ReferenceEquals(_activeWrite, lease) || lease.Sequence != _nextSequence)
+                    throw new InvalidOperationException("The CultMesh frame write lease is stale or is not owned by this publisher.");
+                if (byteLength < 0 || byteLength > SlotByteLength)
+                    throw new ArgumentOutOfRangeException(nameof(byteLength));
+                var semanticHash = CultMeshBodyDescriptorValidator.ComputeSemanticHash(lease.Span[..byteLength]);
+                WithMutex(() =>
+                {
+                    var writerOffset = WriterCountOffset + lease.SlotIndex * sizeof(int);
+                    if (_control.ReadInt32(writerOffset) != 1 ||
+                        _control.ReadInt32(ReaderCountOffset + lease.SlotIndex * sizeof(int)) != 0)
+                        throw new InvalidOperationException("The CultMesh frame write reservation was lost before commit.");
+                    _control.Write(SequenceOffset + lease.SlotIndex * sizeof(long), lease.Sequence);
+                    _control.Write(writerOffset, 0);
+                });
+                var descriptor = Descriptor(lease.SlotIndex, byteLength, lease.Sequence, semanticHash, nowUtc);
+                _nextSequence++;
+                _activeWrite = null;
+                lease.Complete();
+                return descriptor;
+            }
+        }
+
+        internal void Abandon(CultMeshFrameBodyWriteLease lease)
+        {
+            lock (_gate)
+            {
+                if (!ReferenceEquals(_activeWrite, lease)) return;
+                if (!_disposed)
+                {
+                    WithMutex(() =>
+                        _control.Write(WriterCountOffset + lease.SlotIndex * sizeof(int), 0));
+                }
+                _activeWrite = null;
+            }
+        }
+
+        private CultMeshBodyDescriptor Descriptor(
+            int slot,
+            int byteLength,
+            long sequence,
+            string semanticHash,
+            DateTimeOffset nowUtc) => new()
+        {
+            BodyId = BodyId, SchemaId = SchemaId, LayoutVersion = LayoutVersion, ByteSize = byteLength,
+            Capacity = Capacity, ProducerEpoch = ProducerEpoch, Sequence = sequence,
             AccessMode = CultMeshBodyAccessMode.ReadOnly, Synchronization = CultMeshBodySynchronization.TripleBuffer,
             LeaseExpiresAtUnixMs = nowUtc.Add(_leaseDuration).ToUnixTimeMilliseconds(),
             TransportKind = CultMeshBodyTransportKind.SharedMemory,
             CapabilityToken = _token + "." + slot,
-            SemanticHash = CultMeshBodyDescriptorValidator.ComputeSemanticHash(body)
+            SemanticHash = semanticHash
         };
 
         internal static bool IsFrameCapability(string capability) => capability.StartsWith(CapabilityPrefix, StringComparison.Ordinal);
@@ -178,6 +272,8 @@ namespace GameCult.Mesh
             {
                 if (_disposed) return;
                 _disposed = true;
+                _activeWrite?.Invalidate();
+                _activeWrite = null;
                 _control.Dispose();
                 _mapping.Dispose();
                 _mutex.Dispose();
@@ -214,6 +310,100 @@ namespace GameCult.Mesh
         private static string Require(string value, string name) => string.IsNullOrWhiteSpace(value) ? throw new ArgumentException("Value is required.", name) : value;
     }
 
+    /// <summary>A bounded producer lease over one directly writable shared-memory frame slot.</summary>
+    public sealed unsafe class CultMeshFrameBodyWriteLease : IDisposable
+    {
+        private CultMeshFrameBodyPublisher? _owner;
+        private readonly MemoryMappedViewAccessor _view;
+        private readonly CultMeshMappedMemoryManager _memory;
+        private bool _completed;
+
+        internal CultMeshFrameBodyWriteLease(
+            CultMeshFrameBodyPublisher owner,
+            MemoryMappedViewAccessor view,
+            int slotIndex,
+            long sequence,
+            int byteLength)
+        {
+            _owner = owner;
+            _view = view;
+            _memory = new CultMeshMappedMemoryManager(view, byteLength);
+            SlotIndex = slotIndex;
+            Sequence = sequence;
+        }
+
+        public int SlotIndex { get; }
+        public long Sequence { get; }
+        public Memory<byte> Memory => _memory.Memory;
+        public Span<byte> Span => Memory.Span;
+
+        public CultMeshBodyDescriptor Commit(int byteLength, DateTimeOffset nowUtc) =>
+            (_owner ?? throw new ObjectDisposedException(nameof(CultMeshFrameBodyWriteLease)))
+            .Commit(this, byteLength, nowUtc);
+
+        internal void Complete()
+        {
+            _completed = true;
+            ReleaseResources();
+        }
+
+        internal void Invalidate()
+        {
+            _owner = null;
+            ReleaseResources();
+        }
+
+        public void Dispose()
+        {
+            var owner = Interlocked.Exchange(ref _owner, null);
+            if (!_completed) owner?.Abandon(this);
+            ReleaseResources();
+        }
+
+        private void ReleaseResources()
+        {
+            ((IDisposable)_memory).Dispose();
+            _view.Dispose();
+            _owner = null;
+        }
+    }
+
+    internal sealed unsafe class CultMeshMappedMemoryManager : MemoryManager<byte>
+    {
+        private SafeMemoryMappedViewHandle? _handle;
+        private byte* _pointer;
+        private readonly int _length;
+
+        public CultMeshMappedMemoryManager(MemoryMappedViewAccessor view, int length)
+        {
+            _handle = view.SafeMemoryMappedViewHandle;
+            _length = length;
+            byte* pointer = null;
+            _handle.AcquirePointer(ref pointer);
+            _pointer = pointer + view.PointerOffset;
+        }
+
+        public override Span<byte> GetSpan() =>
+            _handle == null ? throw new ObjectDisposedException(nameof(CultMeshMappedMemoryManager)) : new Span<byte>(_pointer, _length);
+
+        public override MemoryHandle Pin(int elementIndex = 0)
+        {
+            if ((uint)elementIndex > (uint)_length) throw new ArgumentOutOfRangeException(nameof(elementIndex));
+            if (_handle == null) throw new ObjectDisposedException(nameof(CultMeshMappedMemoryManager));
+            return new MemoryHandle(_pointer + elementIndex);
+        }
+
+        public override void Unpin() { }
+
+        protected override void Dispose(bool disposing)
+        {
+            var handle = Interlocked.Exchange(ref _handle, null);
+            if (handle == null) return;
+            handle.ReleasePointer();
+            _pointer = null;
+        }
+    }
+
     public sealed class CultMeshFrameGeneration
     {
         internal CultMeshFrameGeneration(

src/GameCult.Mesh/CultMeshPrimitives.cs

@@ -3,6 +3,7 @@ using System.Collections;
 using System.Collections.Generic;
 using System.Globalization;
 using System.Linq;
+using System.Linq.Expressions;
 using System.Reflection;
 using System.Threading;
 using System.Threading.Tasks;
@@ -1002,6 +1003,27 @@ namespace GameCult.Mesh
             return Watch().Subscribe(onNext);
         }
 
+        /// <summary>
+        /// Projects one typed document field into a stable reactive state pointer. The expression is
+        /// structural metadata: only member access and constant collection indices are accepted.
+        /// </summary>
+        public CultMeshStatePointer<TValue> Field<TValue>(Expression<Func<TDocument, TValue>> field)
+        {
+            if (field == null) throw new ArgumentNullException(nameof(field));
+            var path = CultMeshFieldPath.Parse(field.Body, field.Parameters[0]);
+            var read = field.Compile();
+            return new CultMeshStatePointer<TValue>(
+                DocumentId + "#" + path,
+                async context => read(await Feed
+                    .SnapshotAsync(CultMeshDocumentQueryParameters.Empty, context)
+                    .ConfigureAwait(false)),
+                context => Feed
+                    .Watch(CultMeshDocumentQueryParameters.Empty, context)
+                    .Select(read),
+                RouteHint,
+                Sources);
+        }
+
         /// <summary>Replaces the underlying document when this handle is backed by mutable state.</summary>
         public Task ReplaceAsync(TDocument value)
         {
@@ -1122,6 +1144,62 @@ namespace GameCult.Mesh
                 payload);
         }
 
+        private static class CultMeshFieldPath
+        {
+            public static string Parse(Expression expression, ParameterExpression root)
+            {
+                var segments = new Stack<string>();
+                while (expression != root)
+                {
+                    if (expression is MemberExpression member)
+                    {
+                        segments.Push(member.Member.Name);
+                        expression = member.Expression
+                            ?? throw Invalid();
+                        continue;
+                    }
+
+                    if (expression is MethodCallExpression call &&
+                        call.Method.Name == "get_Item" &&
+                        call.Object != null &&
+                        call.Arguments.Count == 1)
+                    {
+                        segments.Push("[" + ConstantIndex(call.Arguments[0]) + "]");
+                        expression = call.Object;
+                        continue;
+                    }
+
+                    if (expression is BinaryExpression binary && binary.NodeType == ExpressionType.ArrayIndex)
+                    {
+                        segments.Push("[" + ConstantIndex(binary.Right) + "]");
+                        expression = binary.Left;
+                        continue;
+                    }
+
+                    throw Invalid();
+                }
+
+                if (segments.Count == 0) throw Invalid();
+                var path = string.Empty;
+                foreach (var segment in segments)
+                    path += segment.StartsWith("[", StringComparison.Ordinal) || path.Length == 0
+                        ? segment
+                        : "." + segment;
+                return path;
+            }
+
+            private static string ConstantIndex(Expression expression)
+            {
+                if (expression is ConstantExpression constant && constant.Value != null)
+                    return Convert.ToString(constant.Value, CultureInfo.InvariantCulture)!;
+                throw Invalid();
+            }
+
+            private static ArgumentException Invalid() => new(
+                "CultMesh field selectors may contain only member access and constant collection indices.",
+                "field");
+        }
+
         internal static TTarget ConvertDocument<TSource, TTarget>(TSource document)
             where TSource : class
             where TTarget : class

src/GameCult.Mesh/GameCult.Mesh.csproj

@@ -4,6 +4,7 @@
     <PackageId>GameCult.Mesh</PackageId>
     <PackageTags>game;distributed;realtime;mesh;database;simulation;consensus</PackageTags>
     <PackageReadmeFile>README.md</PackageReadmeFile>
+    <AllowUnsafeBlocks>true</AllowUnsafeBlocks>
   </PropertyGroup>
 
   <ItemGroup>

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

@@ -889,6 +889,42 @@ before fetch, corrupt network bytes never produce a lease, local failure is
 reported, fallback preserves identity, and authorization is evaluated separately
 from content integrity.
 
+### Phase 5 exact-demand body map
+
+**Owner:** `CultMeshBodyDemandTracker` owns the live derivation from retained,
+exact database subscriptions to the transport representations a producer must
+publish. The provider owns body contents. A body adapter owns only access to
+one representation. A renderer owns none of those decisions.
+
+**Inputs:** consumer runtime identity, exact logical body IDs, advertised
+consumer transport capabilities, and server-observed peer locality. A
+same-machine claim selects shared memory only when the transport peer is
+actually loopback and the consumer explicitly supports shared memory.
+
+**Outputs:** a per-body plan requiring shared memory, network, both, or no
+publication. Producers use that plan before allocating, copying, hashing, or
+packaging a representation.
+
+**Derived state:** transport preference and publication work are notification
+state derived from active subscriptions. Heavy database snapshots remain
+bootstrap/recovery state. They are not a frame delivery mechanism.
+
+**Forbidden writers:** providers cannot infer demand from renderer identity;
+renderers cannot choose a provider-specific plane; CDN publication cannot run
+unconditionally beside a local mapped write; body descriptors cannot create
+stream demand merely by existing.
+
+**Shared paths:** subscribe, resubscribe, replacement, explicit unsubscribe,
+and server disposal all update the same typed demand projection. Stable peer
+identity, not per-message wrapper identity, keys subscription lifetime.
+
+**Cut line:** local frame publication writes directly into a mapped write lease
+and commits its descriptor. The compatibility byte-array publisher is copy-in
+only. The current remote frame representation may still use the verified CDN
+body path when remote demand exists; Phase 6 must replace that expiry-sensitive
+traffic with a stream/datagram body plane rather than enlarging snapshot or CDN
+timeouts.
+
 ### Gate
 
 - Two local consumers map one verified CDN artifact without duplicate process

src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs

@@ -48,7 +48,10 @@ namespace GameCult.Networking
             IEnumerable<string>? recordKeys = null,
             IEnumerable<string>? schemaIds = null,
             bool includeSnapshot = true,
-            CancellationToken cancellationToken = default)
+            CancellationToken cancellationToken = default,
+            string? consumerRuntimeId = null,
+            IEnumerable<string>? bodyIds = null,
+            IEnumerable<string>? supportedBodyTransports = null)
         {
             if (_disposed) throw new ObjectDisposedException(nameof(CultNetDatabaseSubscriptionClient));
             if (!_client.Connected) throw new InvalidOperationException("CultNet schema client must be connected before subscribing.");
@@ -82,7 +85,13 @@ namespace GameCult.Networking
                     SubscriptionId = subscriptionId,
                     RecordKeys = recordKeys?.Where(value => !string.IsNullOrWhiteSpace(value)).Distinct(StringComparer.Ordinal).ToArray(),
                     SchemaIds = schemaIds?.Where(value => !string.IsNullOrWhiteSpace(value)).Distinct(StringComparer.Ordinal).ToArray(),
-                    IncludeSnapshot = includeSnapshot
+                    IncludeSnapshot = includeSnapshot,
+                    ConsumerRuntimeId = consumerRuntimeId,
+                    BodyIds = bodyIds?.Where(value => !string.IsNullOrWhiteSpace(value)).Distinct(StringComparer.Ordinal).ToArray(),
+                    SupportedBodyTransports = supportedBodyTransports?
+                        .Where(value => !string.IsNullOrWhiteSpace(value))
+                        .Distinct(StringComparer.OrdinalIgnoreCase)
+                        .ToArray()
                 });
             }
             catch

src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs

@@ -4,6 +4,7 @@ using System.Collections.Generic;
 using System.Linq;
 using System.Reflection;
 using System.Runtime.CompilerServices;
+using System.Net;
 using System.Threading.Tasks;
 using GameCult.Caching;
 using R3;
@@ -21,6 +22,8 @@ namespace GameCult.Networking
         private readonly Func<CultNetDatabaseUnsubscribeMessage, ICultNetSchemaServerPeer, Task> _unsubscribe;
         private readonly ConcurrentDictionary<SubscriptionKey, IDisposable> _subscriptions =
             new ConcurrentDictionary<SubscriptionKey, IDisposable>(SubscriptionKeyComparer.Instance);
+        private readonly ConcurrentDictionary<SubscriptionKey, CultNetDatabaseSubscribeMessage> _requests =
+            new ConcurrentDictionary<SubscriptionKey, CultNetDatabaseSubscribeMessage>(SubscriptionKeyComparer.Instance);
         private bool _disposed;
 
         /// <summary>Attaches live database subscription handlers to a schema server.</summary>
@@ -34,6 +37,9 @@ namespace GameCult.Networking
             _server.OnCultNet(_unsubscribe);
         }
 
+        /// <summary>Raised when exact document subscriptions add or remove typed hot-body demand.</summary>
+        public event Action<CultNetDatabaseSubscriptionDemand>? DemandChanged;
+
         /// <summary>Detaches handlers and releases all active watches.</summary>
         public void Dispose()
         {
@@ -41,8 +47,14 @@ namespace GameCult.Networking
             _disposed = true;
             _server.RemoveCultNetMessageListener<CultNetDatabaseSubscribeMessage>(_subscribe);
             _server.RemoveCultNetMessageListener<CultNetDatabaseUnsubscribeMessage>(_unsubscribe);
-            foreach (var subscription in _subscriptions.Values) subscription.Dispose();
+            foreach (var entry in _subscriptions)
+            {
+                entry.Value.Dispose();
+                if (_requests.TryRemove(entry.Key, out var request))
+                    PublishDemand(request, entry.Key, active: false);
+            }
             _subscriptions.Clear();
+            _requests.Clear();
         }
 
         private Task HandleSubscribeAsync(CultNetDatabaseSubscribeMessage request, ICultNetSchemaServerPeer peer)
@@ -57,6 +69,8 @@ namespace GameCult.Networking
             }
 
             var key = new SubscriptionKey(peer, subscriptionId);
+            if (_requests.TryGetValue(key, out var previous))
+                PublishDemand(previous, key, active: false);
             _subscriptions.AddOrUpdate(
                 key,
                 _ => Watch(request, subscriptionId, peer),
@@ -65,6 +79,8 @@ namespace GameCult.Networking
                     current.Dispose();
                     return Watch(request, subscriptionId, peer);
                 });
+            _requests[key] = request;
+            PublishDemand(request, key, active: true);
             if (request.IncludeSnapshot)
             {
                 peer.SendCultNet(_database.Documents.CreateRawSnapshotResponse(
@@ -93,12 +109,41 @@ namespace GameCult.Networking
             var subscriptionId = string.IsNullOrWhiteSpace(request.SubscriptionId)
                 ? request.MessageId
                 : request.SubscriptionId;
-            if (!string.IsNullOrWhiteSpace(subscriptionId) &&
-                _subscriptions.TryRemove(new SubscriptionKey(peer, subscriptionId), out var subscription))
-                subscription.Dispose();
+            if (!string.IsNullOrWhiteSpace(subscriptionId))
+            {
+                var key = new SubscriptionKey(peer, subscriptionId);
+                if (_subscriptions.TryRemove(key, out var subscription)) subscription.Dispose();
+                if (_requests.TryRemove(key, out var subscribed))
+                    PublishDemand(subscribed, key, active: false);
+            }
             return Task.CompletedTask;
         }
 
+        private void PublishDemand(
+            CultNetDatabaseSubscribeMessage request,
+            SubscriptionKey key,
+            bool active)
+        {
+            if (request.BodyIds is not { Length: > 0 }) return;
+            DemandChanged?.Invoke(new CultNetDatabaseSubscriptionDemand(
+                string.IsNullOrWhiteSpace(request.ConsumerRuntimeId) ? key.Id : request.ConsumerRuntimeId!,
+                key.Id,
+                request.BodyIds.Where(value => !string.IsNullOrWhiteSpace(value)).Distinct(StringComparer.Ordinal).ToArray(),
+                (request.SupportedBodyTransports ?? Array.Empty<string>())
+                    .Where(value => !string.IsNullOrWhiteSpace(value))
+                    .Distinct(StringComparer.OrdinalIgnoreCase)
+                    .ToArray(),
+                IsSameMachine(key.Peer),
+                active));
+        }
+
+        private static bool IsSameMachine(ICultNetSchemaServerPeer peer)
+        {
+            if (peer is RudpCultNetSchemaServerPeer rudp && rudp.RemoteEndPoint is IPEndPoint ip)
+                return IPAddress.IsLoopback(ip.Address);
+            return peer is CultNetServerPeer liteNet && IPAddress.IsLoopback(liteNet.Peer.Address);
+        }
+
         private IDisposable Watch(
             CultNetDatabaseSubscribeMessage request,
             string subscriptionId,
@@ -177,9 +222,55 @@ namespace GameCult.Networking
         {
             public static SubscriptionKeyComparer Instance { get; } = new SubscriptionKeyComparer();
             public bool Equals(SubscriptionKey? x, SubscriptionKey? y) =>
-                x != null && y != null && ReferenceEquals(x.Peer, y.Peer) && string.Equals(x.Id, y.Id, StringComparison.Ordinal);
+                x != null && y != null && SamePeer(x.Peer, y.Peer) && string.Equals(x.Id, y.Id, StringComparison.Ordinal);
             public int GetHashCode(SubscriptionKey value) =>
-                (RuntimeHelpers.GetHashCode(value.Peer) * 397) ^ StringComparer.Ordinal.GetHashCode(value.Id);
+                (PeerHash(value.Peer) * 397) ^ StringComparer.Ordinal.GetHashCode(value.Id);
+
+            private static bool SamePeer(ICultNetSchemaServerPeer left, ICultNetSchemaServerPeer right)
+            {
+                if (ReferenceEquals(left, right)) return true;
+                if (left is RudpCultNetSchemaServerPeer leftRudp && right is RudpCultNetSchemaServerPeer rightRudp)
+                    return ReferenceEquals(leftRudp.TransportPeer, rightRudp.TransportPeer);
+                if (left is CultNetServerPeer leftLite && right is CultNetServerPeer rightLite)
+                    return ReferenceEquals(leftLite.Peer, rightLite.Peer);
+                return false;
+            }
+
+            private static int PeerHash(ICultNetSchemaServerPeer peer)
+            {
+                if (peer is RudpCultNetSchemaServerPeer rudp)
+                    return RuntimeHelpers.GetHashCode(rudp.TransportPeer);
+                if (peer is CultNetServerPeer lite)
+                    return RuntimeHelpers.GetHashCode(lite.Peer);
+                return RuntimeHelpers.GetHashCode(peer);
+            }
         }
     }
+
+    /// <summary>One active or withdrawn consumer demand for logical hot bodies.</summary>
+    public sealed class CultNetDatabaseSubscriptionDemand
+    {
+        public CultNetDatabaseSubscriptionDemand(
+            string consumerRuntimeId,
+            string subscriptionId,
+            IReadOnlyList<string> bodyIds,
+            IReadOnlyList<string> supportedBodyTransports,
+            bool sameMachine,
+            bool active)
+        {
+            ConsumerRuntimeId = consumerRuntimeId;
+            SubscriptionId = subscriptionId;
+            BodyIds = bodyIds;
+            SupportedBodyTransports = supportedBodyTransports;
+            SameMachine = sameMachine;
+            Active = active;
+        }
+
+        public string ConsumerRuntimeId { get; }
+        public string SubscriptionId { get; }
+        public IReadOnlyList<string> BodyIds { get; }
+        public IReadOnlyList<string> SupportedBodyTransports { get; }
+        public bool SameMachine { get; }
+        public bool Active { get; }
+    }
 }

src/GameCult.Networking/CultNetSchemaMessages.cs

@@ -787,6 +787,18 @@ namespace GameCult.Networking
         /// Gets or sets whether the server should send a matching snapshot before live changes.
         /// </summary>
         [Key("includeSnapshot")] public bool IncludeSnapshot { get; set; } = true;
+        /// <summary>
+        /// Gets or sets the stable consumer runtime identity used for subscription diagnostics and body-plane demand.
+        /// </summary>
+        [Key("consumerRuntimeId")] public string? ConsumerRuntimeId { get; set; }
+        /// <summary>
+        /// Gets or sets logical hot-body identities needed by this exact state subscription.
+        /// </summary>
+        [Key("bodyIds")] public string[]? BodyIds { get; set; }
+        /// <summary>
+        /// Gets or sets body transports the consumer can open, ordered by consumer preference.
+        /// </summary>
+        [Key("supportedBodyTransports")] public string[]? SupportedBodyTransports { get; set; }
     }
 
     /// <summary>

tests/GameCult.Mesh.Tests/CultMeshBodyPublicationTests.cs

@@ -49,8 +49,7 @@ public sealed class CultMeshBodyPublicationTests
         var first = Publication(new byte[] { 1, 2, 3, 4 }, now);
         var second = Publication(new byte[] { 5, 6, 7, 8 }, now);
         second.Sequence++;
-        second.PreferredLocal.Sequence++;
-        second.NetworkFallback.Sequence++;
+        foreach (var representation in second.Representations) representation.Sequence++;
         var cache = new CultCache();
 
         await cache.UpsertAsync(first, new CultRecordHandle<CultMeshBodyPublicationDocument>(first.RecordKey));
@@ -70,8 +69,7 @@ public sealed class CultMeshBodyPublicationTests
         var staleLatestKey = CultMeshBodyPublicationDocument.CreateLatestRecordKey(publication.BodyId);
         var next = Publication(new byte[4], DateTimeOffset.UtcNow);
         next.Sequence++;
-        next.PreferredLocal.Sequence++;
-        next.NetworkFallback.Sequence++;
+        foreach (var representation in next.Representations) representation.Sequence++;
 
         Action staleLatest = () => CultMeshBodyPublicationValidator.Validate(
             publication, publication.BodyId, publication.ProducerEpoch, publication.Sequence, staleLatestKey);
@@ -95,13 +93,30 @@ public sealed class CultMeshBodyPublicationTests
         lease.ReadSingle(0).Should().Be(12.5f);
     }
 
+    [Test]
+    public void Resolver_AcceptsLocalOnlyPublicationWithoutInventingNetworkWork()
+    {
+        var now = DateTimeOffset.UtcNow;
+        var bytes = BitConverter.GetBytes(21.25f);
+        var publication = Publication(bytes, now);
+        publication.Representations = new[] { publication.Representations[0] };
+        var fetched = false;
+        var resolver = Resolver(bytes, (_, _) => true, () => fetched = true);
+
+        using var lease = resolver.ResolveReadOnly(publication, Request(publication, now));
+
+        lease.TransportKind.Should().Be(publication.Representations[0].TransportKind);
+        lease.ReadSingle(0).Should().Be(21.25f);
+        fetched.Should().BeFalse();
+    }
+
     [Test]
     public void Resolver_FallsBackToEquivalentNetworkRepresentation()
     {
         var now = DateTimeOffset.UtcNow;
         var bytes = new byte[] { 4, 8, 15, 16 };
         var publication = Publication(bytes, now);
-        new CultMeshMappedBodyPublisher(_root).Revoke(publication.PreferredLocal);
+        new CultMeshMappedBodyPublisher(_root).Revoke(publication.Representations[0]);
 
         var result = Resolver(bytes, (_, _) => true).NegotiateReadOnly(publication, Request(publication, now));
         using var lease = result.Lease;
@@ -120,7 +135,7 @@ public sealed class CultMeshBodyPublicationTests
     {
         var now = DateTimeOffset.UtcNow;
         var publication = Publication(new byte[4], now);
-        var descriptor = publication.NetworkFallback;
+        var descriptor = publication.Representations[1];
         if (field == "body") descriptor.BodyId += ":wrong";
         if (field == "schema") descriptor.SchemaId += ".wrong";
         if (field == "capacity") descriptor.Capacity++;
@@ -184,8 +199,7 @@ public sealed class CultMeshBodyPublicationTests
             Sequence = local.Sequence,
             Synchronization = local.Synchronization,
             LivenessExpiresAtUnixMs = local.LeaseExpiresAtUnixMs,
-            PreferredLocal = local,
-            NetworkFallback = Network(local)
+            Representations = new[] { local, Network(local) }
         };
     }
 

tests/GameCult.Mesh.Tests/CultMeshBodyTransportTests.cs

@@ -3,6 +3,7 @@ using System.IO;
 using FluentAssertions;
 using NUnit.Framework;
 using System.IO.MemoryMappedFiles;
+using GameCult.Networking;
 
 namespace GameCult.Mesh.Tests;
 
@@ -263,6 +264,99 @@ public sealed class CultMeshBodyTransportTests
         lease.TransportKind.Should().Be(CultMeshBodyTransportKind.SharedMemory);
     }
 
+    [Test]
+    public void FramePublisher_DirectWriteLeaseCommitsMappedBytesWithoutCompatibilityCopy()
+    {
+        using var publisher = new CultMeshFrameBodyPublisher(
+            "aetheria:entities",
+            "gamecult.eve.entity_soa.body.v2",
+            layoutVersion: 2,
+            capacity: 128,
+            producerEpoch: 7,
+            slotByteLength: 64,
+            leaseDuration: TimeSpan.FromMinutes(1));
+        publisher.TryAcquireWrite(out var write).Should().BeTrue();
+        using (write)
+        {
+            BitConverter.TryWriteBytes(write.Span, 42.5f).Should().BeTrue();
+            var descriptor = write.Commit(sizeof(float), DateTimeOffset.UtcNow);
+            using var read = new CultMeshSharedMemoryBodyAdapter().OpenReadOnly(
+                descriptor,
+                Request(descriptor, DateTimeOffset.UtcNow));
+
+            descriptor.Sequence.Should().Be(0);
+            descriptor.SemanticHash.Should().NotBeNullOrWhiteSpace();
+            read.ReadSingle(0).Should().Be(42.5f);
+        }
+    }
+
+    [Test]
+    public void FramePublisher_AbandonedWriteDoesNotAdvanceGenerationOrHoldSlot()
+    {
+        using var publisher = new CultMeshFrameBodyPublisher(
+            "aetheria:entities",
+            "gamecult.eve.entity_soa.body.v2",
+            layoutVersion: 2,
+            capacity: 128,
+            producerEpoch: 7,
+            slotByteLength: 64);
+        publisher.TryAcquireWrite(out var abandoned).Should().BeTrue();
+        abandoned.Dispose();
+
+        publisher.TryAcquireWrite(out var replacement).Should().BeTrue();
+        using (replacement)
+        {
+            replacement.Span[0] = 23;
+            var descriptor = replacement.Commit(1, DateTimeOffset.UtcNow);
+            descriptor.Sequence.Should().Be(0);
+        }
+    }
+
+    [Test]
+    public void BodyDemand_SelectsSharedMemoryLocallyAndNetworkRemotelyWithoutDuplicateFallbackWork()
+    {
+        using var tracker = new CultMeshBodyDemandTracker();
+        tracker.Observe(Demand("unity-local", "local", sameMachine: true));
+        tracker.Observe(Demand("unity-remote", "remote", sameMachine: false));
+
+        var mixed = tracker.Plan("eve:entity-soa:aetheria.daemon:pilot");
+
+        mixed.HasConsumers.Should().BeTrue();
+        mixed.RequiresSharedMemory.Should().BeTrue();
+        mixed.RequiresNetwork.Should().BeTrue();
+        mixed.Consumers.Should().ContainSingle(value =>
+            value.ConsumerRuntimeId == "unity-local" &&
+            value.Transport == CultMeshBodyTransportKind.SharedMemory);
+        mixed.Consumers.Should().ContainSingle(value =>
+            value.ConsumerRuntimeId == "unity-remote" &&
+            value.Transport == CultMeshBodyTransportKind.Network);
+
+        tracker.Observe(Demand("unity-remote", "remote", sameMachine: false, active: false));
+        var localOnly = tracker.Plan("eve:entity-soa:aetheria.daemon:pilot");
+        localOnly.RequiresSharedMemory.Should().BeTrue();
+        localOnly.RequiresNetwork.Should().BeFalse();
+    }
+
+    [Test]
+    public void BodyDemand_UnknownLocalityFailsTowardNetworkAndUnsupportedDemandCreatesNoRoute()
+    {
+        using var tracker = new CultMeshBodyDemandTracker();
+        tracker.Observe(Demand("unknown", "network", sameMachine: false));
+        tracker.Observe(new CultNetDatabaseSubscriptionDemand(
+            "unsupported",
+            "unsupported",
+            new[] { "eve:entity-soa:aetheria.daemon:pilot" },
+            new[] { CultMeshBodyTransportKind.SharedMemory.ToString() },
+            sameMachine: false,
+            active: true));
+
+        var plan = tracker.Plan("eve:entity-soa:aetheria.daemon:pilot");
+
+        plan.RequiresSharedMemory.Should().BeFalse();
+        plan.RequiresNetwork.Should().BeTrue();
+        plan.Consumers.Should().ContainSingle().Which.ConsumerRuntimeId.Should().Be("unknown");
+    }
+
     private static CultMeshBodyValidationRequest Request(CultMeshBodyDescriptor descriptor, DateTimeOffset now) => new()
     {
         BodyId = descriptor.BodyId,
@@ -275,6 +369,22 @@ public sealed class CultMeshBodyTransportTests
         NowUtc = now
     };
 
+    private static CultNetDatabaseSubscriptionDemand Demand(
+        string runtimeId,
+        string subscriptionId,
+        bool sameMachine,
+        bool active = true) => new(
+        runtimeId,
+        subscriptionId,
+        new[] { "eve:entity-soa:aetheria.daemon:pilot" },
+        new[]
+        {
+            CultMeshBodyTransportKind.SharedMemory.ToString(),
+            CultMeshBodyTransportKind.Network.ToString()
+        },
+        sameMachine,
+        active);
+
     private static CultMeshBodyDescriptor NetworkRepresentation(CultMeshBodyDescriptor descriptor) => new()
     {
         BodyId = descriptor.BodyId,

tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs

@@ -723,6 +723,40 @@ public sealed class CultMeshStreamingTests
         observed.Revision.Should().Be(2);
     }
 
+    [Test]
+    public async Task DocumentHandle_Field_ProjectsTypedReactiveStateWithoutRepublishingSurfaceTrees()
+    {
+        var subject = new Subject<MeshNoteDocument>();
+        var route = new CultMeshRouteHint(CultMeshLocalityKind.SharedMemory, "co-located document slab");
+        var current = new MeshNoteDocument
+        {
+            Schema = "tests.mesh_note.v1",
+            Text = "initial",
+            Revision = 1
+        };
+        var handle = CultMesh.Document(
+            "mesh.note.current",
+            CultMesh.Verse("starbridge", "unity-pilot", route),
+            _ => Task.FromResult(current),
+            _ => subject,
+            routeHint: route);
+
+        var text = handle.Field(value => value.Text);
+
+        text.PointerId.Should().Be("mesh.note.current#Text");
+        text.RouteHint.Kind.Should().Be(CultMeshLocalityKind.SharedMemory);
+        (await text.ResolveAsync()).Should().Be("initial");
+        string observed = null!;
+        using var subscription = text.Watch().Subscribe(value => observed = value);
+        subject.OnNext(new MeshNoteDocument
+        {
+            Schema = "tests.mesh_note.v1",
+            Text = "reactive",
+            Revision = 2
+        });
+        observed.Should().Be("reactive");
+    }
+
     [Test]
     public void DocumentHandle_RejectsAliasTypesWithDifferentSchemaIdentity()
     {

tests/GameCult.Networking.Tests/NetworkingTests.cs

@@ -1425,6 +1425,68 @@ namespace GameCult.Networking.Tests
             Assert.That(update.Document!.RecordKey, Is.EqualTo(recordKey));
         }
 
+        [Test]
+        public async Task DatabaseSubscriptionServer_ProjectsBodyDemandAndWithdrawalFromExactSubscription()
+        {
+            var database = new CultNetDatabase(new CultCache());
+            using var server = new RudpCultNetSchemaServer(new RudpCultNetSchemaServerOptions
+            {
+                RuntimeId = "database-body-demand-server",
+                Socket = BindUdpSocket()
+            });
+            using var subscriptions = new CultNetDatabaseSubscriptionServer(server, database);
+            var active = new TaskCompletionSource<CultNetDatabaseSubscriptionDemand>(
+                TaskCreationOptions.RunContinuationsAsynchronously);
+            var withdrawn = new TaskCompletionSource<CultNetDatabaseSubscriptionDemand>(
+                TaskCreationOptions.RunContinuationsAsynchronously);
+            subscriptions.DemandChanged += demand =>
+            {
+                if (demand.Active) active.TrySetResult(demand);
+                else withdrawn.TrySetResult(demand);
+            };
+            using var cancellation = new CancellationTokenSource();
+            var serverThread = new Thread(() =>
+            {
+                while (!cancellation.IsCancellationRequested)
+                {
+                    _ = server.PollOnceAsync().GetAwaiter().GetResult();
+                    Thread.Sleep(1);
+                }
+            }) { IsBackground = true };
+            serverThread.Start();
+
+            using var client = CultNetSchemaClients.CreateRudp("database-body-demand-client");
+            client.Connect("127.0.0.1", server.LocalEndPoint.Port);
+            await WaitUntilAsync(() => client.Connected, TimeSpan.FromSeconds(2));
+            client.SendCultNet(new CultNetDatabaseSubscribeMessage
+            {
+                MessageId = "subscribe-body",
+                SubscriptionId = "world-body",
+                RecordKeys = ["world:entities", "mesh:body:world:latest"],
+                ConsumerRuntimeId = "eve-unity",
+                BodyIds = ["world"],
+                SupportedBodyTransports = ["SharedMemory", "Network"],
+                IncludeSnapshot = false
+            });
+
+            var observed = await AwaitWithTimeout(active.Task, TimeSpan.FromSeconds(2));
+            Assert.That(observed.ConsumerRuntimeId, Is.EqualTo("eve-unity"));
+            Assert.That(observed.SubscriptionId, Is.EqualTo("world-body"));
+            Assert.That(observed.BodyIds, Is.EqualTo(new[] { "world" }));
+            Assert.That(observed.SupportedBodyTransports, Is.EqualTo(new[] { "SharedMemory", "Network" }));
+            Assert.That(observed.SameMachine, Is.True);
+
+            client.SendCultNet(new CultNetDatabaseUnsubscribeMessage
+            {
+                MessageId = "unsubscribe-body",
+                SubscriptionId = "world-body"
+            });
+            var removed = await AwaitWithTimeout(withdrawn.Task, TimeSpan.FromSeconds(2));
+            cancellation.Cancel();
+            Assert.That(removed.ConsumerRuntimeId, Is.EqualTo("eve-unity"));
+            Assert.That(removed.Active, Is.False);
+        }
+
         [Test]
         public async Task DatabaseSubscriptionClient_ReplicatesInitialAndLiveTypedDocumentsOverRudp()
         {

unity/org.gamecult.cultlib/package.json

@@ -1,7 +1,7 @@
 {
   "name": "org.gamecult.cultlib",
   "displayName": "CultLib",
-  "version": "1.0.19",
+  "version": "1.0.20",
   "unity": "2021.3",
   "description": "CultCache, CultNet, and CultMesh managed runtime assemblies for Unity.",
   "license": "MIT",