Add Python prediction rollback helpers

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

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

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(

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