Keep failed Python simulation fanout pending

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

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]

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)