Fan out Python simulation observations

packages/cultcache-py/PARITY.md

@@ -79,8 +79,9 @@ CultCache, CultNet, and CultMesh. It is evidence, not confetti.
   CultNet frames. It can also parse `cultnet://host:port` endpoints, discover
   from endpoint lists, de-duplicate discovery targets, fan out one-shot
   discovery from endpoints already advertised in local Verse and peer catalogs,
-  and upsert results into local catalogs with the same count-returning
-  ergonomics as the C# discovery clients.
+  fan out simulation observations to peer-card endpoints, and upsert results
+  into local catalogs with the same count-returning ergonomics as the C#
+  discovery clients.
 - Python CultMesh facade:
   Exposes C#-matching client entrypoints for Verse discovery, peer exchange,
   raw CultNet client construction, local nodes, catalogs, sessions, streams,

packages/cultcache-py/src/cultmesh_py/client.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
-from typing import Any
+from typing import Any, Callable
 from urllib.parse import urlparse
 
 from cultnet_py import CultNetRawClient
@@ -193,6 +193,39 @@ class CultMeshDiscoveryClient:
             limit=limit,
         )
 
+    def submit_simulation_observation(self, message: dict[str, Any]) -> dict[str, Any]:
+        return self.request(
+            message,
+            expected_schema_version="cultnet.simulation_consensus_candidate.v0",
+        )
+
+    def fanout_simulation_observation(
+        self,
+        catalog: CultMeshPeerCatalog,
+        message: dict[str, Any],
+        *,
+        verse_id: str,
+        roles: list[str] | None = None,
+        on_error: Callable[[str, Exception], None] | None = None,
+    ) -> list[dict[str, Any]]:
+        requested_roles = set(roles or [])
+        endpoints = [
+            endpoint
+            for peer in catalog.find(verse_id)
+            if not requested_roles or requested_roles.intersection(peer.roles)
+            for endpoint in peer.endpoints
+        ]
+        candidates = []
+        for endpoint in _distinct_non_empty(endpoints):
+            try:
+                client = type(self).from_endpoint(endpoint, timeout_seconds=self.timeout_seconds)
+                candidates.append(client.submit_simulation_observation(message))
+            except Exception as error:
+                if on_error is None:
+                    raise
+                on_error(endpoint, error)
+        return candidates
+
     @classmethod
     def from_endpoint(cls, endpoint: str, *, timeout_seconds: float = 4.0) -> "CultMeshDiscoveryClient":
         host, port = _parse_endpoint(endpoint)

packages/cultcache-py/tests/test_cultcache.py

@@ -2395,6 +2395,67 @@ class CultCacheTests(unittest.TestCase):
             {peer.peer_id for peer in fanout_peers.find("aetheria-main", role="read-replica")},
         )
 
+    def test_cultmesh_discovery_client_fans_out_simulation_observations_to_peers(self) -> None:
+        first_hub = CultNetSimulationObservationHub(
+            CultNetSimulationConsensusOptions(minimum_witnesses=1, quorum_ratio=1.0)
+        )
+        second_hub = CultNetSimulationObservationHub(
+            CultNetSimulationConsensusOptions(minimum_witnesses=1, quorum_ratio=1.0)
+        )
+        first_server = CultMesh.serve_node(
+            CultMesh.create_node(runtime_id="sim-peer-a"),
+            observation_hub=first_hub,
+        )
+        second_server = CultMesh.serve_node(
+            CultMesh.create_node(runtime_id="sim-peer-b"),
+            observation_hub=second_hub,
+        )
+        try:
+            peers = CultMeshPeerCatalog()
+            peers.upsert(CultMeshPeerCard(
+                peer_id="sim-peer-a",
+                verse_id="arena",
+                endpoints=(f"cultnet://127.0.0.1:{first_server.port}",),
+                roles=("simulation-witness",),
+            ))
+            peers.upsert(CultMeshPeerCard(
+                peer_id="sim-peer-b",
+                verse_id="arena",
+                endpoints=(f"cultnet://127.0.0.1:{second_server.port}",),
+                roles=("simulation-witness",),
+            ))
+            claim_hash = compute_simulation_claim_hash("hit", "alice", "bob", "frame:200")
+            message = simulation_observation(
+                message_id="fanout-observation",
+                witness_runtime_id="python-fanout",
+                shard_id="arena",
+                shard_epoch=1,
+                frame=200,
+                subject_id="bob",
+                claim_kind="hit",
+                claim_hash=claim_hash,
+                claim_summary="alice hit bob",
+            ).to_wire()
+
+            client = CultMeshDiscoveryClient("127.0.0.1", first_server.port, timeout_seconds=2.0)
+            candidates = client.fanout_simulation_observation(
+                peers,
+                message,
+                verse_id="arena",
+                roles=["simulation-witness"],
+            )
+        finally:
+            first_server.stop()
+            second_server.stop()
+
+        self.assertEqual([candidate["schemaVersion"] for candidate in candidates], [
+            "cultnet.simulation_consensus_candidate.v0",
+            "cultnet.simulation_consensus_candidate.v0",
+        ])
+        self.assertEqual({candidate["claimHash"] for candidate in candidates}, {claim_hash})
+        self.assertEqual({candidate["messageId"] for candidate in candidates}, {"fanout-observation"})
+        self.assertEqual({candidate["hasQuorum"] for candidate in candidates}, {True})
+
     def test_cultmesh_local_server_serves_node_and_catalogs_over_clients(self) -> None:
         document = define_database_entry_type(
             "mesh.server_note",

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

@@ -95,7 +95,7 @@ Not implemented yet:
 - membership/failure detection
 - leader election or automatic shard failover
 - peer-to-peer snapshot fanout and throttling
-- peer-to-peer fanout for observation gossip and candidate propagation
+- scheduled peer-to-peer fanout for observation gossip and candidate propagation
 - rollback/resimulation helpers for simulation frames
 - declared CRDT merge policies
 
@@ -148,7 +148,8 @@ Add discovery fanout and operational polish:
   package-local Python HMAC verifier
 - add simulation-frame rollback/resimulation helpers around predicted input
   streams
-- add peer-to-peer fanout for observation gossip and candidate propagation
+- add scheduled peer-to-peer fanout for observation gossip and candidate
+  propagation beyond Python's one-shot client helper
 
 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