From fde8a9e681c8452b4e2e84f0f8bfc65a6c70aa7e Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sun, 14 Jun 2026 13:08:16 +0100 Subject: [PATCH] Add Python prediction rollback helpers --- packages/cultcache-py/PARITY.md | 4 + .../cultcache-py/src/cultmesh_py/session.py | 76 ++++++++++++-- packages/cultcache-py/tests/test_cultcache.py | 98 +++++++++++++++++++ .../Contracts/cultnet-distributed-database.md | 6 +- 4 files changed, 175 insertions(+), 9 deletions(-) diff --git a/packages/cultcache-py/PARITY.md b/packages/cultcache-py/PARITY.md index 496042a8..59802693 100644 --- a/packages/cultcache-py/PARITY.md +++ b/packages/cultcache-py/PARITY.md @@ -149,6 +149,10 @@ CultCache, CultNet, and CultMesh. It is evidence, not confetti. The live interop harness runs a Python CultMesh session client, writes a local prediction, pulls the live Python peer shard log, applies the authoritative entry, and verifies the session reports the record as reconciled. + Python sessions also expose pending prediction inspection, resimulation input + listing, explicit rollback of one or all pending predictions, and + authoritative-delete rollback when a shard log rejects speculative local + input. - Python CultMesh session watches: `CultMeshGameSession.watch_candidates(...)` and `watch_simulation_facts(...)` expose the same gameplay-facing observation and diff --git a/packages/cultcache-py/src/cultmesh_py/session.py b/packages/cultcache-py/src/cultmesh_py/session.py index d438c1be..d29c4d33 100644 --- a/packages/cultcache-py/src/cultmesh_py/session.py +++ b/packages/cultcache-py/src/cultmesh_py/session.py @@ -32,6 +32,8 @@ class CultMeshPrediction: schema_id: str document: DocumentDefinition[Any] value: Any + had_previous_value: bool = False + previous_value: Any | None = None @dataclass(frozen=True) @@ -67,7 +69,7 @@ class CultMeshGameSession: self.fact_committer = CultMeshSimulationFactCommitter(self.node) self._committed_fact_ids: set[str] = set() self._client_authority_scopes = tuple(options.client_authority_scopes) - self._predicted_keys: set[tuple[str, str]] = set() + self._predictions: dict[tuple[str, str], CultMeshPrediction] = {} def predict(self, document: DocumentDefinition[Any], key: str, value: Any) -> CultMeshPrediction: schema_id = document.catalog_entry().schema_id @@ -76,9 +78,58 @@ class CultMeshGameSession: f"Runtime {self.node.runtime_id!r} does not have client prediction authority " f"for schema {schema_id!r} key {key!r}" ) + had_previous_value = self.node.cache.get_envelope(document, key) is not None + previous_value = self.node.get(document, key) self.node.put(document, key, value) - self._predicted_keys.add((schema_id, key)) - return CultMeshPrediction(key=key, schema_id=schema_id, document=document, value=value) + prediction = CultMeshPrediction( + key=key, + schema_id=schema_id, + document=document, + value=value, + had_previous_value=had_previous_value, + previous_value=previous_value, + ) + self._predictions[(schema_id, key)] = prediction + return prediction + + def pending_predictions(self) -> tuple[CultMeshPrediction, ...]: + return tuple( + self._predictions[key] + for key in sorted(self._predictions) + ) + + def resimulation_inputs(self) -> tuple[CultMeshPrediction, ...]: + return self.pending_predictions() + + def rollback_prediction( + self, + prediction_or_schema_id: CultMeshPrediction | str, + key: str | None = None, + ) -> CultMeshSessionChange | None: + schema_id, record_key = self._prediction_identity(prediction_or_schema_id, key) + prediction = self._predictions.pop((schema_id, record_key), None) + if prediction is None: + return None + if prediction.had_previous_value: + self.node.put(prediction.document, prediction.key, prediction.previous_value) + value = prediction.previous_value + else: + self.node.delete(prediction.document, prediction.key) + value = None + return CultMeshSessionChange( + schema_id=prediction.schema_id, + record_key=prediction.key, + change_kind="rolled_back", + value=value, + ) + + def rollback_predictions(self) -> list[CultMeshSessionChange]: + return [ + change + for prediction in list(self.pending_predictions()) + for change in [self.rollback_prediction(prediction)] + if change is not None + ] def watch_candidates( self, @@ -156,9 +207,9 @@ class CultMeshGameSession: for record in applied: key = (record.schema_id, record.record_key) change_kind = record.change_kind - if key in self._predicted_keys and record.change_kind in {"added", "updated"}: - self._predicted_keys.remove(key) - change_kind = "reconciled" + if key in self._predictions: + del self._predictions[key] + change_kind = "reconciled" if record.change_kind in {"added", "updated"} else "rolled_back" changes.append(CultMeshSessionChange( schema_id=record.schema_id, record_key=record.record_key, @@ -166,3 +217,16 @@ class CultMeshGameSession: value=record.value, )) return changes + + @staticmethod + def _prediction_identity( + prediction_or_schema_id: CultMeshPrediction | str, + key: str | None, + ) -> tuple[str, str]: + if isinstance(prediction_or_schema_id, CultMeshPrediction): + if key is not None: + raise ValueError("key must be omitted when rolling back a CultMeshPrediction") + return prediction_or_schema_id.schema_id, prediction_or_schema_id.key + if key is None: + raise ValueError("key is required when rolling back by schema id") + return prediction_or_schema_id, key diff --git a/packages/cultcache-py/tests/test_cultcache.py b/packages/cultcache-py/tests/test_cultcache.py index ee9299b2..ff7865db 100644 --- a/packages/cultcache-py/tests/test_cultcache.py +++ b/packages/cultcache-py/tests/test_cultcache.py @@ -2060,6 +2060,104 @@ class CultCacheTests(unittest.TestCase): self.assertEqual(changes[0].change_kind, "reconciled") self.assertEqual(node.get_required(note_doc, prediction.key)["body"], "authoritative") + self.assertEqual(session.pending_predictions(), ()) + self.assertEqual(session.resimulation_inputs(), ()) + + def test_cultmesh_game_session_rolls_back_pending_predictions(self) -> None: + note_doc = define_database_entry_type( + "mesh.rollback_input", + [("body", 0)], + schema_id="mesh.rollback_input.v1", + ) + node = CultMesh.create_node(runtime_id="client-rollback") + node.register_document(note_doc) + node.put(note_doc, "input:client-rollback:existing", {"body": "previous"}) + session = CultMesh.create_game_session( + node, + CultMeshGameSessionOptions( + client_authority_scopes=( + CultNetClientAuthorityScope( + "client-rollback", + schema_ids=("mesh.rollback_input.v1",), + key_prefix="input:client-rollback", + ), + ) + ), + ) + + existing = session.predict(note_doc, "input:client-rollback:existing", {"body": "predicted-existing"}) + created = session.predict(note_doc, "input:client-rollback:new", {"body": "predicted-new"}) + self.assertEqual( + [prediction.key for prediction in session.pending_predictions()], + ["input:client-rollback:existing", "input:client-rollback:new"], + ) + self.assertEqual([prediction.key for prediction in session.resimulation_inputs()], [ + "input:client-rollback:existing", + "input:client-rollback:new", + ]) + + restored = session.rollback_prediction(existing) + self.assertIsNotNone(restored) + assert restored is not None + self.assertEqual(restored.change_kind, "rolled_back") + self.assertEqual(restored.value, {"body": "previous"}) + self.assertEqual(node.get_required(note_doc, "input:client-rollback:existing")["body"], "previous") + + rolled_back = session.rollback_predictions() + self.assertEqual([(change.change_kind, change.record_key, change.value) for change in rolled_back], [ + ("rolled_back", created.key, None), + ]) + self.assertIsNone(node.get(note_doc, created.key)) + self.assertEqual(session.pending_predictions(), ()) + + def test_cultmesh_game_session_authoritative_delete_rolls_back_prediction(self) -> None: + note_doc = define_database_entry_type( + "mesh.deleted_input", + [("body", 0)], + schema_id="mesh.deleted_input.v1", + ) + node = CultMesh.create_node(runtime_id="client-delete") + node.register_document(note_doc) + session = CultMesh.create_game_session( + node, + CultMeshGameSessionOptions( + client_authority_scopes=( + CultNetClientAuthorityScope( + "client-delete", + schema_ids=("mesh.deleted_input.v1",), + key_prefix="input:client-delete", + ), + ) + ), + ) + + prediction = session.predict(note_doc, "input:client-delete:move", {"body": "predicted"}) + delete = document_delete( + message_id="authoritative-delete", + schema_id=prediction.schema_id, + record_key=prediction.key, + shard_id="inputs", + shard_epoch=1, + ) + changes = session.apply_shard_log_response({ + "schemaVersion": "cultnet.shard_log_response.v0", + "messageId": "inputs-delete-log", + "shardId": "inputs", + "shardEpoch": 1, + "entries": [ + { + "sequence": 1, + "committedAt": "2026-06-13T00:00:00Z", + "changeKind": "removed", + "delete": delete.to_wire(), + } + ], + "resyncRequired": False, + }) + + self.assertEqual(changes[0].change_kind, "rolled_back") + self.assertIsNone(node.get(note_doc, prediction.key)) + self.assertEqual(session.pending_predictions(), ()) def test_cultmesh_database_watchers_observe_authoritative_shard_log_reconciliation(self) -> None: note_doc = define_database_entry_type( diff --git a/src/GameCult.Networking/Contracts/cultnet-distributed-database.md b/src/GameCult.Networking/Contracts/cultnet-distributed-database.md index ab6392f6..4b93bd01 100644 --- a/src/GameCult.Networking/Contracts/cultnet-distributed-database.md +++ b/src/GameCult.Networking/Contracts/cultnet-distributed-database.md @@ -96,7 +96,7 @@ Not implemented yet: - leader election or automatic shard failover - peer-to-peer snapshot fanout and throttling - scheduled peer-to-peer fanout for observation gossip and candidate propagation -- rollback/resimulation helpers for simulation frames +- cross-runtime rollback/resimulation helpers for simulation frames - declared CRDT merge policies ## Live Invariants @@ -146,8 +146,8 @@ Add discovery fanout and operational polish: - add periodic gossip fanout over known peer cards - standardize cross-runtime authority lease signature algorithms beyond the package-local Python HMAC verifier -- add simulation-frame rollback/resimulation helpers around predicted input - streams +- standardize cross-runtime simulation-frame rollback/resimulation helpers beyond + Python's local pending-prediction rollback surface - add scheduled peer-to-peer fanout for observation gossip and candidate propagation beyond Python's one-shot client helper