Schedule Python snapshot fanout

packages/cultcache-py/PARITY.md

@@ -80,10 +80,10 @@ CultCache, CultNet, and CultMesh. It is evidence, not confetti.
   from endpoint lists, de-duplicate discovery targets, fan out one-shot
   discovery from endpoints already advertised in local Verse and peer catalogs,
   fan out raw snapshot fetches and simulation observations to peer-card
-  endpoints, apply peer snapshots through a caller-owned database, run a
-  lightweight scheduled simulation-observation fanout loop, and upsert results
-  into local catalogs with the same count-returning ergonomics as the C#
-  discovery clients.
+  endpoints, apply peer snapshots through a caller-owned database, run
+  lightweight scheduled snapshot and simulation-observation fanout loops, 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/__init__.py

@@ -2,6 +2,7 @@ from .client import (
     CultMeshDiscoveryClient,
     CultMeshPeerExchangeClient,
     CultMeshSimulationObservationFanout,
+    CultMeshSnapshotFanout,
     CultMeshVerseDiscoveryClient,
 )
 from .facade import CultMesh
@@ -69,6 +70,7 @@ __all__ = [
     "CultMeshDiscoveryClient",
     "CultMeshPeerExchangeClient",
     "CultMeshSimulationObservationFanout",
+    "CultMeshSnapshotFanout",
     "CultMeshVerseDiscoveryClient",
     "CultMeshStreamCatalog",
     "CultMeshStreamConsumerProfile",

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

@@ -425,6 +425,65 @@ class CultMeshSimulationObservationFanout:
             self.flush()
 
 
+@dataclass
+class CultMeshSnapshotFanout:
+    client: CultMeshDiscoveryClient
+    database: Any
+    peer_catalog: CultMeshPeerCatalog
+    verse_id: str
+    roles: list[str] | None = None
+    schema_ids: list[str] | None = None
+    record_keys: list[str] | None = None
+    shard_id: str | None = None
+    shard_epoch: int | None = None
+    interval_seconds: float = 1.0
+    on_applied: Callable[[Any], None] | None = None
+    on_error: Callable[[str, Exception], None] | None = None
+    _stop: threading.Event = field(default_factory=threading.Event, init=False, repr=False)
+    _thread: threading.Thread | None = field(default=None, init=False, repr=False)
+
+    def sync_once(self) -> list[Any]:
+        applied = self.client.sync_snapshots(
+            self.database,
+            self.peer_catalog,
+            verse_id=self.verse_id,
+            roles=self.roles,
+            schema_ids=self.schema_ids,
+            record_keys=self.record_keys,
+            shard_id=self.shard_id,
+            shard_epoch=self.shard_epoch,
+            on_error=self.on_error,
+        )
+        for record in applied:
+            if self.on_applied is not None:
+                self.on_applied(record)
+        return applied
+
+    def start(self) -> "CultMeshSnapshotFanout":
+        if self._thread is not None:
+            return self
+        self._stop.clear()
+        self._thread = threading.Thread(target=self._run, daemon=True)
+        self._thread.start()
+        return self
+
+    def stop(self) -> None:
+        self._stop.set()
+        if self._thread is not None:
+            self._thread.join(timeout=max(2.0, self.interval_seconds * 2.0))
+            self._thread = None
+
+    def __enter__(self) -> "CultMeshSnapshotFanout":
+        return self.start()
+
+    def __exit__(self, exc_type: Any, exc: Any, traceback: Any) -> None:
+        self.stop()
+
+    def _run(self) -> None:
+        while not self._stop.wait(self.interval_seconds):
+            self.sync_once()
+
+
 def _parse_endpoint(endpoint: str) -> tuple[str, int]:
     if not endpoint or not endpoint.strip():
         raise ValueError("endpoint must be non-empty")

packages/cultcache-py/tests/test_cultcache.py

@@ -91,6 +91,7 @@ from cultmesh_py import (
     CultMeshDiscoveryClient,
     CultMeshPeerExchangeClient,
     CultMeshSimulationObservationFanout,
+    CultMeshSnapshotFanout,
     CultMeshStreamCatalog,
     CultMeshStreamConsumerProfile,
     CultMeshStreamDescriptor,
@@ -2703,6 +2704,87 @@ class CultCacheTests(unittest.TestCase):
         self.assertEqual(target.database.get_required(document, "note:a")["body"], "from-a")
         self.assertEqual(target.database.get_required(document, "note:b")["body"], "from-b")
 
+    def test_cultmesh_snapshot_fanout_syncs_once_and_reports_applied_records(self) -> None:
+        document = define_database_entry_type(
+            "mesh.snapshot_loop_note",
+            [("body", 0)],
+            schema_id="mesh.snapshot_loop_note.v1",
+        )
+        source = CultMesh.create_node(runtime_id="snapshot-loop-source")
+        source.database.register_document(document)
+        source.database.put_raw_message(document, "note:sync-once", {"body": "from-source"})
+        target = CultMesh.create_node(runtime_id="snapshot-loop-target")
+        target.database.register_document(document)
+        server = CultMesh.serve_node(source)
+        try:
+            peers = CultMeshPeerCatalog()
+            peers.upsert(CultMeshPeerCard(
+                peer_id="snapshot-loop-source",
+                verse_id="mesh",
+                endpoints=(f"cultnet://127.0.0.1:{server.port}",),
+                roles=("read-replica",),
+            ))
+            seen: list[str] = []
+            fanout = CultMeshSnapshotFanout(
+                CultMeshDiscoveryClient("127.0.0.1", server.port, timeout_seconds=2.0),
+                target.database,
+                peers,
+                verse_id="mesh",
+                roles=["read-replica"],
+                schema_ids=["mesh.snapshot_loop_note.v1"],
+                on_applied=lambda record: seen.append(record.record_key),
+            )
+
+            applied = fanout.sync_once()
+        finally:
+            server.stop()
+
+        self.assertEqual([record.record_key for record in applied], ["note:sync-once"])
+        self.assertEqual(seen, ["note:sync-once"])
+        self.assertEqual(target.database.get_required(document, "note:sync-once")["body"], "from-source")
+
+    def test_cultmesh_snapshot_fanout_runs_background_loop(self) -> None:
+        document = define_database_entry_type(
+            "mesh.snapshot_background_note",
+            [("body", 0)],
+            schema_id="mesh.snapshot_background_note.v1",
+        )
+        source = CultMesh.create_node(runtime_id="snapshot-background-source")
+        source.database.register_document(document)
+        source.database.put_raw_message(document, "note:background", {"body": "from-background"})
+        target = CultMesh.create_node(runtime_id="snapshot-background-target")
+        target.database.register_document(document)
+        server = CultMesh.serve_node(source)
+        try:
+            peers = CultMeshPeerCatalog()
+            peers.upsert(CultMeshPeerCard(
+                peer_id="snapshot-background-source",
+                verse_id="mesh",
+                endpoints=(f"cultnet://127.0.0.1:{server.port}",),
+                roles=("read-replica",),
+            ))
+            seen: list[str] = []
+            fanout = CultMeshSnapshotFanout(
+                CultMeshDiscoveryClient("127.0.0.1", server.port, timeout_seconds=2.0),
+                target.database,
+                peers,
+                verse_id="mesh",
+                roles=["read-replica"],
+                schema_ids=["mesh.snapshot_background_note.v1"],
+                interval_seconds=0.05,
+                on_applied=lambda record: seen.append(record.record_key),
+            )
+            fanout.start()
+            deadline = time.monotonic() + 2.0
+            while not seen and time.monotonic() < deadline:
+                time.sleep(0.01)
+            fanout.stop()
+        finally:
+            server.stop()
+
+        self.assertTrue(seen)
+        self.assertEqual(target.database.get_required(document, "note:background")["body"], "from-background")
+
     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

@@ -94,7 +94,8 @@ Not implemented yet:
 
 - membership/failure detection
 - leader election or automatic shard failover
-- scheduled peer-to-peer snapshot fanout beyond Python's one-shot client helper
+- cross-runtime scheduled peer-to-peer snapshot fanout beyond Python's local
+  fanout loop
 - cross-runtime scheduled peer-to-peer fanout for observation gossip and
   candidate propagation beyond Python's local fanout loop
 - cross-runtime rollback/resimulation helpers for simulation frames
@@ -149,8 +150,8 @@ Add discovery fanout and operational polish:
   package-local Python HMAC verifier
 - standardize cross-runtime simulation-frame rollback/resimulation helpers beyond
   Python's local pending-prediction rollback surface
-- add scheduled peer-to-peer snapshot fanout beyond Python's one-shot client
-  helper
+- standardize cross-runtime scheduled peer-to-peer snapshot fanout beyond
+  Python's local fanout loop
 - standardize cross-runtime scheduled peer-to-peer fanout for observation gossip
   and candidate propagation beyond Python's local fanout loop