From 4da13643dd4d065ea7aac905d8497cf2f32dc81a Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sun, 14 Jun 2026 12:58:18 +0100 Subject: [PATCH] Fan out Python simulation observations --- packages/cultcache-py/PARITY.md | 5 +- .../cultcache-py/src/cultmesh_py/client.py | 35 ++++++++++- packages/cultcache-py/tests/test_cultcache.py | 61 +++++++++++++++++++ .../Contracts/cultnet-distributed-database.md | 5 +- 4 files changed, 101 insertions(+), 5 deletions(-) diff --git a/packages/cultcache-py/PARITY.md b/packages/cultcache-py/PARITY.md index 443bdae8..496042a8 100644 --- a/packages/cultcache-py/PARITY.md +++ b/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, diff --git a/packages/cultcache-py/src/cultmesh_py/client.py b/packages/cultcache-py/src/cultmesh_py/client.py index 884d724c..e5e8e365 100644 --- a/packages/cultcache-py/src/cultmesh_py/client.py +++ b/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) diff --git a/packages/cultcache-py/tests/test_cultcache.py b/packages/cultcache-py/tests/test_cultcache.py index f61ca574..ee9299b2 100644 --- a/packages/cultcache-py/tests/test_cultcache.py +++ b/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", diff --git a/src/GameCult.Networking/Contracts/cultnet-distributed-database.md b/src/GameCult.Networking/Contracts/cultnet-distributed-database.md index fd79ecfa..ab6392f6 100644 --- a/src/GameCult.Networking/Contracts/cultnet-distributed-database.md +++ b/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