diff --git a/research/cultnet-distributed-database/summary.md b/research/cultnet-distributed-database/summary.md
index f85cee62..4d247b0d 100644
--- a/research/cultnet-distributed-database/summary.md
+++ b/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.
diff --git a/src/GameCult.Networking/Contracts/cultnet-distributed-database.md b/src/GameCult.Networking/Contracts/cultnet-distributed-database.md
index 4b5dc69c..ae5e1410 100644
--- a/src/GameCult.Networking/Contracts/cultnet-distributed-database.md
+++ b/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
diff --git a/src/GameCult.Networking/CultNetSimulationObservationServer.cs b/src/GameCult.Networking/CultNetSimulationObservationServer.cs
new file mode 100644
index 00000000..2b301197
--- /dev/null
+++ b/src/GameCult.Networking/CultNetSimulationObservationServer.cs
@@ -0,0 +1,79 @@
+using System;
+using System.Linq;
+using System.Threading.Tasks;
+using LiteNetLib;
+
+namespace GameCult.Networking
+{
+ ///
+ /// Bridges schema-v0 simulation observation messages into a local observation hub.
+ ///
+ public sealed class CultNetSimulationObservationServer : IDisposable
+ {
+ private readonly Server _server;
+ private readonly CultNetSimulationObservationHub _hub;
+ private readonly Func _observationHandler;
+ private bool _disposed;
+
+ ///
+ /// Creates and attaches a simulation observation bridge.
+ ///
+ 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);
+ }
+
+ ///
+ /// Gets the observation hub used by this bridge.
+ ///
+ public CultNetSimulationObservationHub Hub => _hub;
+
+ ///
+ /// Applies an observation message and returns current candidate messages for the same subject.
+ ///
+ 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();
+ }
+
+ ///
+ public void Dispose()
+ {
+ if (_disposed)
+ {
+ return;
+ }
+
+ _disposed = true;
+ _server.RemoveCultNetMessageListener(_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;
+ }
+ }
+}
diff --git a/src/GameCult.Networking/README.md b/src/GameCult.Networking/README.md
index a4b4b6b1..f518cae1 100644
--- a/src/GameCult.Networking/README.md
+++ b/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
diff --git a/tests/GameCult.Networking.Tests/NetworkingTests.cs b/tests/GameCult.Networking.Tests/NetworkingTests.cs
index 19b044f6..b63d9971 100644
--- a/tests/GameCult.Networking.Tests/NetworkingTests.cs
+++ b/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()
{