Add CultNet simulation observation server

research/cultnet-distributed-database/summary.md

@@ -176,6 +176,7 @@ Shard catalog exchange now exists:
 - simulation witness observations and deterministic consensus candidates
 - schema-v0 simulation observation and candidate messages
 - reactive observation hub for witness gossip and consensus candidates
+- server-side observation bridge for observation messages and candidate replies
 
 ## Next Cut
 
@@ -186,7 +187,7 @@ Build durable authoritative mutation logs:
 - return explicit resync-required responses when requested history is gone
 - then add simulation-frame rollback and resimulation helpers on top of
   predicted input streams
-- add network fanout for observation gossip and candidate propagation
+- add peer-to-peer fanout for observation gossip and candidate propagation
 
 That is the next foundation needed before membership, Raft-style failover, or
 CRDT policy work can be honest.

src/GameCult.Networking/Contracts/cultnet-distributed-database.md

@@ -66,6 +66,8 @@ Implemented:
 - schema-v0 simulation observation and consensus candidate messages
 - reactive simulation observation hub that receives witness gossip and emits
   consensus candidates
+- server-side observation bridge that accepts observation messages and replies
+  with current candidate messages
 
 Not implemented yet:
 
@@ -73,7 +75,7 @@ Not implemented yet:
 - leader election or automatic shard failover
 - durable log persistence and compaction
 - snapshot fallback for compacted log history
-- network fanout for observation gossip and candidate propagation
+- peer-to-peer fanout for observation gossip and candidate propagation
 - rollback/resimulation helpers for simulation frames
 - declared CRDT merge policies
 
@@ -108,7 +110,7 @@ Add durable mutation-log storage and snapshot fallback:
 - return `ResyncRequired` when a replica asks for compacted history
 - after that, add simulation-frame rollback/resimulation helpers around
   predicted input streams
-- add network fanout for observation gossip and candidate propagation
+- add peer-to-peer fanout for observation gossip and candidate propagation
 
 The log is wire-readable, replicas can apply it explicitly, a pull loop can
 drive non-primary shards from primary endpoints, and replica cursors can survive

src/GameCult.Networking/CultNetSimulationObservationServer.cs

@@ -0,0 +1,79 @@
+using System;
+using System.Linq;
+using System.Threading.Tasks;
+using LiteNetLib;
+
+namespace GameCult.Networking
+{
+    /// <summary>
+    /// Bridges schema-v0 simulation observation messages into a local observation hub.
+    /// </summary>
+    public sealed class CultNetSimulationObservationServer : IDisposable
+    {
+        private readonly Server _server;
+        private readonly CultNetSimulationObservationHub _hub;
+        private readonly Func<CultNetSimulationObservationMessage, NetPeer, Task> _observationHandler;
+        private bool _disposed;
+
+        /// <summary>
+        /// Creates and attaches a simulation observation bridge.
+        /// </summary>
+        public CultNetSimulationObservationServer(
+            Server server,
+            CultNetSimulationObservationHub hub)
+        {
+            _server = server ?? throw new ArgumentNullException(nameof(server));
+            _hub = hub ?? throw new ArgumentNullException(nameof(hub));
+            _observationHandler = HandleObservationAsync;
+            _server.OnCultNet(_observationHandler);
+        }
+
+        /// <summary>
+        /// Gets the observation hub used by this bridge.
+        /// </summary>
+        public CultNetSimulationObservationHub Hub => _hub;
+
+        /// <summary>
+        /// Applies an observation message and returns current candidate messages for the same subject.
+        /// </summary>
+        public CultNetSimulationConsensusCandidateMessage[] CreateCandidateMessages(
+            CultNetSimulationObservationMessage message)
+        {
+            if (message == null) throw new ArgumentNullException(nameof(message));
+            return _hub.Submit(message)
+                .Select(candidate => CultNetSimulationConsensusCandidateMessage.FromCandidate(
+                    message.MessageId,
+                    candidate))
+                .ToArray();
+        }
+
+        /// <inheritdoc />
+        public void Dispose()
+        {
+            if (_disposed)
+            {
+                return;
+            }
+
+            _disposed = true;
+            _server.RemoveCultNetMessageListener<CultNetSimulationObservationMessage>(_observationHandler);
+        }
+
+        private Task HandleObservationAsync(CultNetSimulationObservationMessage message, NetPeer peer)
+        {
+            try
+            {
+                foreach (var candidate in CreateCandidateMessages(message))
+                {
+                    peer.SendCultNet(candidate);
+                }
+            }
+            catch (Exception ex)
+            {
+                peer.SendCultNet(new CultNetErrorMessage { Error = ex.Message });
+            }
+
+            return Task.CompletedTask;
+        }
+    }
+}

src/GameCult.Networking/README.md

@@ -163,6 +163,9 @@ path, not committed world state by themselves. The schema-v0 wire contracts are
 `cultnet.simulation_observation.v0` and
 `cultnet.simulation_consensus_candidate.v0`. `CultNetSimulationObservationHub`
 collects incoming observations and emits candidate updates reactively.
+`CultNetSimulationObservationServer` attaches that hub to a server so incoming
+observation messages update the hub and receive current candidate messages in
+reply.
 
 The first coherent shard policy is primary ownership. Each shard has one
 authoritative writer at a time. Clients may connect to any node; a non-owner

tests/GameCult.Networking.Tests/NetworkingTests.cs

@@ -1431,6 +1431,56 @@ namespace GameCult.Networking.Tests
             Assert.That(candidates.Exists(candidate => candidate.HasQuorum), Is.True);
         }
 
+        [Test]
+        public void CultNetSimulationObservationServer_Creates_CandidateMessages()
+        {
+            var cache = new CultCache();
+            using var server = new Server(cache, DevelopmentServerSecurity);
+            using var hub = new CultNetSimulationObservationHub(new CultNetSimulationConsensusOptions
+            {
+                MinimumWitnesses = 2,
+                QuorumRatio = 1d
+            });
+            using var observationServer = new CultNetSimulationObservationServer(server, hub);
+            var claimHash = CultNetSimulationObservation.ComputeClaimHash("hit", "alice", "bob", "frame:100");
+            var first = new CultNetSimulationObservationMessage
+            {
+                MessageId = "observe-1",
+                Observation = new CultNetSimulationObservation
+                {
+                    WitnessRuntimeId = "watcher-1",
+                    ShardId = "arena",
+                    ShardEpoch = 1,
+                    Frame = 100,
+                    SubjectId = "bob",
+                    ClaimKind = "hit",
+                    ClaimHash = claimHash
+                }
+            };
+            var second = new CultNetSimulationObservationMessage
+            {
+                MessageId = "observe-2",
+                Observation = new CultNetSimulationObservation
+                {
+                    WitnessRuntimeId = "watcher-2",
+                    ShardId = "arena",
+                    ShardEpoch = 1,
+                    Frame = 100,
+                    SubjectId = "bob",
+                    ClaimKind = "hit",
+                    ClaimHash = claimHash
+                }
+            };
+
+            observationServer.CreateCandidateMessages(first);
+            var candidates = observationServer.CreateCandidateMessages(second);
+
+            Assert.That(candidates, Has.Length.EqualTo(1));
+            Assert.That(candidates[0].MessageId, Is.EqualTo("observe-2"));
+            Assert.That(candidates[0].ClaimHash, Is.EqualTo(claimHash));
+            Assert.That(candidates[0].HasQuorum, Is.True);
+        }
+
         [Test]
         public void CultNetDatabaseServer_Creates_Filtered_SubscriptionChange()
         {