Add CultNet simulation observation hub

research/cultnet-distributed-database/summary.md

@@ -175,6 +175,7 @@ Shard catalog exchange now exists:
 - predicted/reconciled change events for client-side prediction
 - simulation witness observations and deterministic consensus candidates
 - schema-v0 simulation observation and candidate messages
+- reactive observation hub for witness gossip and consensus candidates
 
 ## Next Cut
 
@@ -185,7 +186,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 an observation hub that receives gossip and emits consensus candidates
+- add network 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

@@ -64,6 +64,8 @@ Implemented:
 - simulation witness observations for mesh-side opinions about frame facts
 - deterministic consensus candidates derived from witness observations
 - schema-v0 simulation observation and consensus candidate messages
+- reactive simulation observation hub that receives witness gossip and emits
+  consensus candidates
 
 Not implemented yet:
 
@@ -71,7 +73,7 @@ Not implemented yet:
 - leader election or automatic shard failover
 - durable log persistence and compaction
 - snapshot fallback for compacted log history
-- observation hub/fanout that receives gossip and emits candidates reactively
+- network fanout for observation gossip and candidate propagation
 - rollback/resimulation helpers for simulation frames
 - declared CRDT merge policies
 
@@ -106,8 +108,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 an observation hub that receives schema-v0 observation messages and emits
+- add network fanout for observation gossip and candidate propagation
-  consensus candidates reactively
 
 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/CultNetSimulationConsensus.cs

@@ -4,6 +4,7 @@ using System.Linq;
 using System.Security.Cryptography;
 using System.Text;
 using MessagePack;
+using R3;
 
 namespace GameCult.Networking
 {
@@ -290,4 +291,109 @@ namespace GameCult.Networking
                 observation.ClaimKind);
         }
     }
+
+    /// <summary>
+    /// Collects witness observations and publishes updated consensus candidates.
+    /// </summary>
+    public sealed class CultNetSimulationObservationHub : IDisposable
+    {
+        private readonly List<CultNetSimulationObservation> _observations = new();
+        private readonly CultNetSimulationConsensus _consensus;
+        private readonly Subject<CultNetSimulationObservation> _observationStream = new();
+        private readonly Subject<CultNetSimulationConsensusCandidate> _candidateStream = new();
+        private bool _disposed;
+
+        /// <summary>
+        /// Creates an observation hub.
+        /// </summary>
+        public CultNetSimulationObservationHub(CultNetSimulationConsensusOptions? options = null)
+        {
+            _consensus = new CultNetSimulationConsensus(options);
+        }
+
+        /// <summary>
+        /// Gets all observations currently held by the hub.
+        /// </summary>
+        public IReadOnlyList<CultNetSimulationObservation> Observations => _observations;
+
+        /// <summary>
+        /// Watches accepted witness observations.
+        /// </summary>
+        public Observable<CultNetSimulationObservation> WatchObservations()
+        {
+            ThrowIfDisposed();
+            return _observationStream;
+        }
+
+        /// <summary>
+        /// Watches consensus candidate updates.
+        /// </summary>
+        public Observable<CultNetSimulationConsensusCandidate> WatchCandidates()
+        {
+            ThrowIfDisposed();
+            return _candidateStream;
+        }
+
+        /// <summary>
+        /// Adds an observation and returns current candidates for the same observed subject.
+        /// </summary>
+        public IReadOnlyList<CultNetSimulationConsensusCandidate> Submit(CultNetSimulationObservation observation)
+        {
+            ThrowIfDisposed();
+            if (observation == null) throw new ArgumentNullException(nameof(observation));
+            _observations.Add(observation);
+            _observationStream.OnNext(observation);
+
+            var candidates = _consensus.BuildCandidates(_observations)
+                .Where(candidate => MatchesSubject(candidate, observation))
+                .ToArray();
+            foreach (var candidate in candidates)
+            {
+                _candidateStream.OnNext(candidate);
+            }
+
+            return candidates;
+        }
+
+        /// <summary>
+        /// Adds an observation message and returns current candidates for the same observed subject.
+        /// </summary>
+        public IReadOnlyList<CultNetSimulationConsensusCandidate> Submit(CultNetSimulationObservationMessage message)
+        {
+            if (message == null) throw new ArgumentNullException(nameof(message));
+            return Submit(message.Observation);
+        }
+
+        /// <inheritdoc />
+        public void Dispose()
+        {
+            if (_disposed)
+            {
+                return;
+            }
+
+            _disposed = true;
+            _observationStream.Dispose();
+            _candidateStream.Dispose();
+        }
+
+        private static bool MatchesSubject(
+            CultNetSimulationConsensusCandidate candidate,
+            CultNetSimulationObservation observation)
+        {
+            return string.Equals(candidate.ShardId, observation.ShardId, StringComparison.Ordinal) &&
+                   candidate.ShardEpoch == observation.ShardEpoch &&
+                   candidate.Frame == observation.Frame &&
+                   string.Equals(candidate.SubjectId, observation.SubjectId, StringComparison.Ordinal) &&
+                   string.Equals(candidate.ClaimKind, observation.ClaimKind, StringComparison.Ordinal);
+        }
+
+        private void ThrowIfDisposed()
+        {
+            if (_disposed)
+            {
+                throw new ObjectDisposedException(nameof(CultNetSimulationObservationHub));
+            }
+        }
+    }
 }

src/GameCult.Networking/README.md

@@ -161,7 +161,8 @@ into candidates with support weight, total observed weight, confidence, and
 quorum status. Those candidates are opinions ready for an authoritative commit
 path, not committed world state by themselves. The schema-v0 wire contracts are
 `cultnet.simulation_observation.v0` and
-`cultnet.simulation_consensus_candidate.v0`.
+`cultnet.simulation_consensus_candidate.v0`. `CultNetSimulationObservationHub`
+collects incoming observations and emits candidate updates reactively.
 
 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

@@ -1385,6 +1385,52 @@ namespace GameCult.Networking.Tests
             Assert.That(candidates[0].HasQuorum, Is.True);
         }
 
+        [Test]
+        public void CultNetSimulationObservationHub_Emits_ReactiveCandidates()
+        {
+            var claimHash = CultNetSimulationObservation.ComputeClaimHash("hit", "alice", "bob", "frame:100");
+            using var hub = new CultNetSimulationObservationHub(new CultNetSimulationConsensusOptions
+            {
+                MinimumWitnesses = 2,
+                QuorumRatio = 1d
+            });
+            var observations = new List<CultNetSimulationObservation>();
+            var candidates = new List<CultNetSimulationConsensusCandidate>();
+            using var observationSubscription = hub.WatchObservations()
+                .Subscribe(observation => observations.Add(observation));
+            using var candidateSubscription = hub.WatchCandidates()
+                .Subscribe(candidate => candidates.Add(candidate));
+
+            hub.Submit(new CultNetSimulationObservation
+            {
+                WitnessRuntimeId = "watcher-1",
+                ShardId = "arena",
+                ShardEpoch = 1,
+                Frame = 100,
+                SubjectId = "bob",
+                ClaimKind = "hit",
+                ClaimHash = claimHash
+            });
+            var current = hub.Submit(new CultNetSimulationObservationMessage
+            {
+                Observation = new CultNetSimulationObservation
+                {
+                    WitnessRuntimeId = "watcher-2",
+                    ShardId = "arena",
+                    ShardEpoch = 1,
+                    Frame = 100,
+                    SubjectId = "bob",
+                    ClaimKind = "hit",
+                    ClaimHash = claimHash
+                }
+            });
+
+            Assert.That(observations, Has.Count.EqualTo(2));
+            Assert.That(current, Has.Count.EqualTo(1));
+            Assert.That(current[0].HasQuorum, Is.True);
+            Assert.That(candidates.Exists(candidate => candidate.HasQuorum), Is.True);
+        }
+
         [Test]
         public void CultNetDatabaseServer_Creates_Filtered_SubscriptionChange()
         {