diff --git a/packages/cultcache-py/PARITY.md b/packages/cultcache-py/PARITY.md index 57c01f9b..565bb04d 100644 --- a/packages/cultcache-py/PARITY.md +++ b/packages/cultcache-py/PARITY.md @@ -156,7 +156,9 @@ CultCache, CultNet, and CultMesh. It is evidence, not confetti. Python exposes typed `CultNetSimulationObservation` and `CultNetSimulationConsensusCandidate` objects for local consensus and game-session commit paths while preserving the schema-v0 dict projections - used by the wire harness. + used by the wire harness. Scheduled simulation-observation fanout keeps + queued observations pending when no peer produces a candidate, even when + endpoint failures are reported through a caller-owned error callback. - Python CultMesh prediction reconciliation: The live interop harness runs a Python CultMesh session client, writes a local prediction, pulls the live Python peer shard log, applies the authoritative diff --git a/packages/cultcache-py/src/cultmesh_py/client.py b/packages/cultcache-py/src/cultmesh_py/client.py index 69755128..b95a08ab 100644 --- a/packages/cultcache-py/src/cultmesh_py/client.py +++ b/packages/cultcache-py/src/cultmesh_py/client.py @@ -383,16 +383,20 @@ class CultMeshSimulationObservationFanout: failed_messages: list[dict[str, Any]] = [] for message in messages: try: - candidates.extend(self.client.fanout_simulation_observation( + message_candidates = self.client.fanout_simulation_observation( self.peer_catalog, message, verse_id=self.verse_id, roles=self.roles, on_error=self.on_error, - )) + ) except Exception: failed_messages.append(message) continue + if not message_candidates: + failed_messages.append(message) + continue + candidates.extend(message_candidates) if failed_messages: with self._lock: self._pending = [*failed_messages, *self._pending] diff --git a/packages/cultcache-py/tests/test_cultcache.py b/packages/cultcache-py/tests/test_cultcache.py index 766a9972..e062c282 100644 --- a/packages/cultcache-py/tests/test_cultcache.py +++ b/packages/cultcache-py/tests/test_cultcache.py @@ -2709,6 +2709,45 @@ class CultCacheTests(unittest.TestCase): self.assertEqual([candidate["claimHash"] for candidate in candidates], [claim_hash]) self.assertEqual(seen_candidates[0]["messageId"], "queued-observation") + def test_cultmesh_simulation_observation_fanout_keeps_failed_messages_pending(self) -> None: + closed_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + closed_socket.bind(("127.0.0.1", 0)) + closed_port = closed_socket.getsockname()[1] + closed_socket.close() + peers = CultMeshPeerCatalog() + peers.upsert(CultMeshPeerCard( + peer_id="closed-simulation-peer", + verse_id="arena", + endpoints=(f"cultnet://127.0.0.1:{closed_port}",), + roles=("simulation-witness",), + )) + errors: list[tuple[str, str]] = [] + fanout = CultMeshSimulationObservationFanout( + CultMeshDiscoveryClient("127.0.0.1", closed_port, timeout_seconds=0.2), + peers, + verse_id="arena", + roles=["simulation-witness"], + on_error=lambda endpoint, error: errors.append((endpoint, str(error))), + ) + fanout.enqueue(simulation_observation( + message_id="retry-observation", + witness_runtime_id="python-fanout", + shard_id="arena", + shard_epoch=1, + frame=212, + subject_id="bob", + claim_kind="miss", + claim_hash=compute_simulation_claim_hash("miss", "alice", "bob", "frame:212"), + claim_summary="alice missed bob", + ).to_wire()) + + candidates = fanout.flush() + + self.assertEqual(candidates, []) + self.assertEqual(fanout.pending_count(), 1) + self.assertEqual(len(errors), 1) + self.assertIn(f"127.0.0.1:{closed_port}", errors[0][0]) + def test_cultmesh_simulation_observation_fanout_runs_background_loop(self) -> None: hub = CultNetSimulationObservationHub( CultNetSimulationConsensusOptions(minimum_witnesses=1, quorum_ratio=1.0)