diff --git a/packages/cultcache-py/PARITY.md b/packages/cultcache-py/PARITY.md index 66ed2bda..76b03bcd 100644 --- a/packages/cultcache-py/PARITY.md +++ b/packages/cultcache-py/PARITY.md @@ -96,6 +96,10 @@ CultCache, CultNet, and CultMesh. It is evidence, not confetti. The live interop harness sends Python a simulation observation, receives a consensus candidate, then requests the committed `gamecult.mesh.simulation_fact` document back over the raw snapshot lane and checks its MessagePack slots. + 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. - 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/README.md b/packages/cultcache-py/README.md index 7abb0cf9..dc410c04 100644 --- a/packages/cultcache-py/README.md +++ b/packages/cultcache-py/README.md @@ -151,6 +151,7 @@ member order. ```python from cultnet_py import ( CultNetRawClient, + CultNetSimulationObservation, apply_raw_snapshot, apply_shard_log_response, compute_simulation_claim_hash, @@ -172,12 +173,12 @@ catalog = client.fetch_schema_catalog(kinds=["wire_message"]) snapshot = client.fetch_snapshot_response(schema_ids=["cultnet.interop-note"]) shard_catalog = client.fetch_shard_catalog(schema_ids=["cultnet.interop-note"]) # With a local CultCache and the matching registered document definitions: -# applied = apply_raw_snapshot(cache, [note_doc], snapshot.to_wire()) +# applied = apply_raw_snapshot(cache, [note_doc], snapshot) subscription = database_subscribe(subscription_id="ui", schema_ids=["cultnet.interop-note"]) shard_request = shard_catalog_request(message_id="shards", schema_ids=["cultnet.interop-note"]) shard_log = client.fetch_shard_log_response(shard_id="interop", shard_epoch=1, after_sequence=0) -# log_changes = apply_shard_log_response(cache, [note_doc], shard_log.to_wire()) +# log_changes = apply_shard_log_response(cache, [note_doc], shard_log) with client.subscribe_database(subscription_id="ui", schema_ids=["cultnet.interop-note"]) as live: initial_snapshot = live.read_next_snapshot_response() claim_hash = compute_simulation_claim_hash("frame:42", "subject:player-1", "hit") @@ -191,6 +192,7 @@ observation = simulation_observation( claim_kind="hit", claim_hash=claim_hash, ) +typed_observation = CultNetSimulationObservation.from_wire(observation.to_wire()) witness = witness_artifact_bundle( bundle_id="bundle-1", witness_kind="interop-proof", @@ -287,7 +289,7 @@ session = CultMesh.create_game_session( ), ), ) -session_commits = session.submit_and_commit(observation.to_wire()) +session_commits = session.submit_and_commit(typed_observation) prediction = session.predict(note_doc, "input:python:move", {"body": "predicted input"}) unsubscribe() ``` diff --git a/packages/cultcache-py/src/cultcache_py/verify.py b/packages/cultcache-py/src/cultcache_py/verify.py index 1544bc84..7ba8ca0b 100644 --- a/packages/cultcache-py/src/cultcache_py/verify.py +++ b/packages/cultcache-py/src/cultcache_py/verify.py @@ -42,7 +42,9 @@ EXPECTED_EXPORTS: dict[str, tuple[str, ...]] = { "CultNetShardLogEntry", "CultNetShardLogResponse", "CultNetSimulationConsensus", + "CultNetSimulationConsensusCandidate", "CultNetSimulationConsensusOptions", + "CultNetSimulationObservation", "CultNetSimulationObservationHub", "INTEROP_WIRE_CONTRACT", "apply_raw_document_record", diff --git a/packages/cultcache-py/src/cultmesh_py/session.py b/packages/cultcache-py/src/cultmesh_py/session.py index e8f66404..c6398821 100644 --- a/packages/cultcache-py/src/cultmesh_py/session.py +++ b/packages/cultcache-py/src/cultmesh_py/session.py @@ -9,6 +9,8 @@ from cultnet_py import ( CultNetClientAuthorityScope, CultNetRawClient, CultNetShardLogResponse, + CultNetSimulationConsensusCandidate, + CultNetSimulationObservation, CultNetSimulationConsensusOptions, CultNetSimulationObservationHub, ) @@ -101,12 +103,24 @@ class CultMeshGameSession: ) return self.apply_shard_log_response(response) - def submit_observation(self, observation_or_message: dict[str, Any]) -> list[dict[str, Any]]: + def submit_observation_candidates( + self, + observation_or_message: dict[str, Any] | CultNetSimulationObservation, + ) -> list[CultNetSimulationConsensusCandidate]: + return self.observation_hub.submit_candidate_objects(observation_or_message) + + def submit_observation(self, observation_or_message: dict[str, Any] | CultNetSimulationObservation) -> list[dict[str, Any]]: return self.observation_hub.submit(observation_or_message) - def submit_and_commit(self, observation_or_message: dict[str, Any]) -> list[CultMeshSimulationFactCommit]: + def submit_and_commit(self, observation_or_message: dict[str, Any] | CultNetSimulationObservation) -> list[CultMeshSimulationFactCommit]: return self.commit_quorum_candidates(self.submit_observation(observation_or_message)) + def commit_quorum_candidate_objects( + self, + candidates: list[CultNetSimulationConsensusCandidate], + ) -> list[CultMeshSimulationFactCommit]: + return self.commit_quorum_candidates([candidate.to_wire() for candidate in candidates]) + def commit_quorum_candidates(self, candidates: list[dict[str, Any]]) -> list[CultMeshSimulationFactCommit]: commits: list[CultMeshSimulationFactCommit] = [] for candidate in candidates: diff --git a/packages/cultcache-py/src/cultnet_py/__init__.py b/packages/cultcache-py/src/cultnet_py/__init__.py index b89c961b..3b3c17c1 100644 --- a/packages/cultcache-py/src/cultnet_py/__init__.py +++ b/packages/cultcache-py/src/cultnet_py/__init__.py @@ -40,7 +40,9 @@ from .snapshot import CultNetRawDocumentRecord, CultNetRawSnapshotResponse from .subscription import CultNetDatabaseChange from .simulation import ( CultNetSimulationConsensus, + CultNetSimulationConsensusCandidate, CultNetSimulationConsensusOptions, + CultNetSimulationObservation, CultNetSimulationObservationHub, ) @@ -60,7 +62,9 @@ __all__ = [ "CultNetAppliedRecord", "CultNetClientAuthorityScope", "CultNetSimulationConsensus", + "CultNetSimulationConsensusCandidate", "CultNetSimulationConsensusOptions", + "CultNetSimulationObservation", "CultNetSimulationObservationHub", "apply_raw_document_record", "apply_raw_snapshot", diff --git a/packages/cultcache-py/src/cultnet_py/simulation.py b/packages/cultcache-py/src/cultnet_py/simulation.py index 8344803e..71999b5c 100644 --- a/packages/cultcache-py/src/cultnet_py/simulation.py +++ b/packages/cultcache-py/src/cultnet_py/simulation.py @@ -4,6 +4,116 @@ from dataclasses import dataclass, field from typing import Any +@dataclass(frozen=True) +class CultNetSimulationObservation: + witness_runtime_id: str + shard_id: str + shard_epoch: int + frame: int + subject_id: str + claim_kind: str + claim_hash: str + claim_summary: str | None = None + weight: float = 1.0 + observed_at: str = "" + + @classmethod + def from_wire(cls, value: dict[str, Any]) -> "CultNetSimulationObservation": + observation = value.get("observation", value) + if not isinstance(observation, dict): + raise ValueError("CultNet simulation observation must be a map") + return cls( + witness_runtime_id=str(observation.get("witnessRuntimeId") or ""), + shard_id=str(observation.get("shardId") or ""), + shard_epoch=int(observation.get("shardEpoch") or 0), + frame=int(observation.get("frame") or 0), + subject_id=str(observation.get("subjectId") or ""), + claim_kind=str(observation.get("claimKind") or ""), + claim_hash=str(observation.get("claimHash") or ""), + claim_summary=_optional_string(observation.get("claimSummary")), + weight=max(float(observation.get("weight", 1.0) or 0.0), 0.0), + observed_at=str(observation.get("observedAt") or ""), + ) + + def to_wire(self) -> dict[str, Any]: + wire: dict[str, Any] = { + "witnessRuntimeId": self.witness_runtime_id, + "shardId": self.shard_id, + "shardEpoch": self.shard_epoch, + "frame": self.frame, + "subjectId": self.subject_id, + "claimKind": self.claim_kind, + "claimHash": self.claim_hash, + "weight": self.weight, + "observedAt": self.observed_at, + } + if self.claim_summary is not None: + wire["claimSummary"] = self.claim_summary + return wire + + def to_message_wire(self, *, message_id: str) -> dict[str, Any]: + return { + "schemaVersion": "cultnet.simulation_observation.v0", + "messageId": message_id, + "observation": self.to_wire(), + } + + +@dataclass(frozen=True) +class CultNetSimulationConsensusCandidate: + shard_id: str + shard_epoch: int + frame: int + subject_id: str + claim_kind: str + claim_hash: str + witness_count: int + support_weight: float + total_weight: float + has_quorum: bool + confidence: float + message_id: str = "" + claim_summary: str | None = None + + @classmethod + def from_wire(cls, value: dict[str, Any]) -> "CultNetSimulationConsensusCandidate": + if value.get("schemaVersion") != "cultnet.simulation_consensus_candidate.v0": + raise ValueError(f"Expected cultnet.simulation_consensus_candidate.v0, received {value.get('schemaVersion')!r}") + return cls( + message_id=str(value.get("messageId") or ""), + shard_id=str(value.get("shardId") or ""), + shard_epoch=int(value.get("shardEpoch") or 0), + frame=int(value.get("frame") or 0), + subject_id=str(value.get("subjectId") or ""), + claim_kind=str(value.get("claimKind") or ""), + claim_hash=str(value.get("claimHash") or ""), + claim_summary=_optional_string(value.get("claimSummary")), + witness_count=int(value.get("witnessCount") or 0), + support_weight=float(value.get("supportWeight") or 0.0), + total_weight=float(value.get("totalWeight") or 0.0), + has_quorum=value.get("hasQuorum") is True, + confidence=float(value.get("confidence") or 0.0), + ) + + def to_wire(self) -> dict[str, Any]: + return { + "schemaVersion": "cultnet.simulation_consensus_candidate.v0", + "messageId": self.message_id, + "shardId": self.shard_id, + "shardEpoch": self.shard_epoch, + "frame": self.frame, + "subjectId": self.subject_id, + "claimKind": self.claim_kind, + "claimHash": self.claim_hash, + "claimSummary": self.claim_summary, + "witnessCount": self.witness_count, + "supportWeight": self.support_weight, + "totalWeight": self.total_weight, + "hasQuorum": self.has_quorum, + "confidence": self.confidence, + } + + @dataclass(frozen=True) class CultNetSimulationConsensusOptions: minimum_witnesses: int = 1 @@ -15,8 +125,12 @@ class CultNetSimulationConsensusOptions: class CultNetSimulationConsensus: options: CultNetSimulationConsensusOptions = field(default_factory=CultNetSimulationConsensusOptions) - def build_candidates(self, observations: list[dict[str, Any]]) -> list[dict[str, Any]]: - clean = _dedupe_observations([observation for observation in observations if _is_valid_observation(observation)]) + def build_candidate_objects( + self, + observations: list[dict[str, Any] | CultNetSimulationObservation], + ) -> list[CultNetSimulationConsensusCandidate]: + normalized = [_observation_wire(observation) for observation in observations] + clean = _dedupe_observations([observation for observation in normalized if _is_valid_observation(observation)]) groups: dict[tuple[str, int, int, str, str], list[dict[str, Any]]] = {} for observation in clean: groups.setdefault(_subject_key(observation), []).append(observation) @@ -25,7 +139,10 @@ class CultNetSimulationConsensus: for _, group in sorted(groups.items(), key=lambda item: item[0]) ] - def _build_candidate(self, observations: list[dict[str, Any]]) -> dict[str, Any]: + def build_candidates(self, observations: list[dict[str, Any] | CultNetSimulationObservation]) -> list[dict[str, Any]]: + return [candidate.to_wire() for candidate in self.build_candidate_objects(observations)] + + def _build_candidate(self, observations: list[dict[str, Any]]) -> CultNetSimulationConsensusCandidate: total_weight = sum(_weight_of(observation) for observation in observations) claim_groups: dict[str, list[dict[str, Any]]] = {} for observation in observations: @@ -38,27 +155,25 @@ class CultNetSimulationConsensus: support_weight = sum(_weight_of(observation) for observation in best_observations) witness_count = len({str(observation.get("witnessRuntimeId") or "") for observation in best_observations}) confidence = 0.0 if total_weight <= 0.0 else support_weight / total_weight - return { - "schemaVersion": "cultnet.simulation_consensus_candidate.v0", - "messageId": "", - "shardId": str(first.get("shardId") or ""), - "shardEpoch": int(first.get("shardEpoch") or 0), - "frame": int(first.get("frame") or 0), - "subjectId": str(first.get("subjectId") or ""), - "claimKind": str(first.get("claimKind") or ""), - "claimHash": claim_hash, - "claimSummary": first.get("claimSummary"), - "witnessCount": witness_count, - "supportWeight": support_weight, - "totalWeight": total_weight, - "hasQuorum": ( + return CultNetSimulationConsensusCandidate( + shard_id=str(first.get("shardId") or ""), + shard_epoch=int(first.get("shardEpoch") or 0), + frame=int(first.get("frame") or 0), + subject_id=str(first.get("subjectId") or ""), + claim_kind=str(first.get("claimKind") or ""), + claim_hash=claim_hash, + claim_summary=_optional_string(first.get("claimSummary")), + witness_count=witness_count, + support_weight=support_weight, + total_weight=total_weight, + has_quorum=( witness_count >= self.options.minimum_witnesses and support_weight >= self.options.minimum_weight and total_weight > 0.0 and confidence >= self.options.quorum_ratio ), - "confidence": confidence, - } + confidence=confidence, + ) @dataclass @@ -69,17 +184,30 @@ class CultNetSimulationObservationHub: def __post_init__(self) -> None: self._consensus = CultNetSimulationConsensus(self.options) - def submit(self, observation_or_message: dict[str, Any]) -> list[dict[str, Any]]: - observation = observation_or_message.get("observation", observation_or_message) - if not isinstance(observation, dict): - raise ValueError("Simulation observation must be a map") + def submit_candidate_objects( + self, + observation_or_message: dict[str, Any] | CultNetSimulationObservation, + ) -> list[CultNetSimulationConsensusCandidate]: + observation = _observation_wire(observation_or_message) self.observations.append(observation) return [ candidate - for candidate in self._consensus.build_candidates(self.observations) - if _candidate_matches_observation(candidate, observation) + for candidate in self._consensus.build_candidate_objects(self.observations) + if _candidate_matches_observation(candidate.to_wire(), observation) ] + def submit(self, observation_or_message: dict[str, Any] | CultNetSimulationObservation) -> list[dict[str, Any]]: + return [candidate.to_wire() for candidate in self.submit_candidate_objects(observation_or_message)] + + +def _observation_wire(observation_or_message: dict[str, Any] | CultNetSimulationObservation) -> dict[str, Any]: + if isinstance(observation_or_message, CultNetSimulationObservation): + return observation_or_message.to_wire() + observation = observation_or_message.get("observation", observation_or_message) + if not isinstance(observation, dict): + raise ValueError("Simulation observation must be a map") + return CultNetSimulationObservation.from_wire(observation).to_wire() + def _dedupe_observations(observations: list[dict[str, Any]]) -> list[dict[str, Any]]: groups: dict[tuple[str, str, int, int, str, str], list[dict[str, Any]]] = {} @@ -135,3 +263,10 @@ def _candidate_matches_observation(candidate: dict[str, Any], observation: dict[ and candidate.get("subjectId") == observation.get("subjectId") and candidate.get("claimKind") == observation.get("claimKind") ) + + +def _optional_string(value: Any) -> str | None: + if value is None: + return None + text = str(value) + return text if text else None diff --git a/packages/cultcache-py/tests/test_cultcache.py b/packages/cultcache-py/tests/test_cultcache.py index f76cd1db..ecf8468a 100644 --- a/packages/cultcache-py/tests/test_cultcache.py +++ b/packages/cultcache-py/tests/test_cultcache.py @@ -34,6 +34,8 @@ from cultnet_py import ( CultNetShardLogEntry, CultNetShardLogResponse, CultNetSimulationConsensusOptions, + CultNetSimulationConsensusCandidate, + CultNetSimulationObservation, apply_raw_document_record, apply_raw_snapshot, apply_shard_log_response, @@ -1110,6 +1112,10 @@ class CultCacheTests(unittest.TestCase): self.assertEqual(message["observation"]["witnessRuntimeId"], "python-test") self.assertEqual(message["observation"]["claimHash"], claim_hash) self.assertEqual(message["observation"]["weight"], 1.0) + observation = CultNetSimulationObservation.from_wire(message) + self.assertEqual(observation.witness_runtime_id, "python-test") + self.assertEqual(observation.claim_summary, "player-1 hit target-a") + self.assertEqual(observation.to_message_wire(message_id="obs-2")["observation"], message["observation"]) def test_cultnet_witness_artifact_bundle_uses_csharp_slot_order(self) -> None: bundle = witness_artifact_bundle( @@ -1505,12 +1511,21 @@ class CultCacheTests(unittest.TestCase): ] candidates = consensus.build_candidates(observations) + typed_candidates = consensus.build_candidate_objects([ + CultNetSimulationObservation.from_wire(observation) + for observation in observations + ]) self.assertEqual(len(candidates), 1) self.assertEqual(candidates[0]["claimHash"], claim_hash) self.assertEqual(candidates[0]["witnessCount"], 2) self.assertEqual(candidates[0]["supportWeight"], 2.0) self.assertTrue(candidates[0]["hasQuorum"]) + self.assertEqual(len(typed_candidates), 1) + self.assertIsInstance(typed_candidates[0], CultNetSimulationConsensusCandidate) + self.assertEqual(typed_candidates[0].claim_hash, claim_hash) + self.assertEqual(typed_candidates[0].to_wire(), candidates[0]) + self.assertEqual(CultNetSimulationConsensusCandidate.from_wire(candidates[0]), typed_candidates[0]) def test_cultmesh_game_session_submits_observations_and_commits_quorum_once(self) -> None: claim_hash = compute_simulation_claim_hash("hit", "alice", "bob", "frame:100") @@ -1546,10 +1561,13 @@ class CultCacheTests(unittest.TestCase): ), ) - self.assertEqual(session.submit_and_commit(first), []) - commits = session.submit_and_commit(second) + self.assertEqual(session.submit_and_commit(CultNetSimulationObservation.from_wire(first)), []) + typed_candidates = session.submit_observation_candidates(CultNetSimulationObservation.from_wire(second)) + commits = session.commit_quorum_candidate_objects(typed_candidates) replay = session.commit_quorum_candidates(session.submit_observation(second)) + self.assertEqual(len(typed_candidates), 1) + self.assertTrue(typed_candidates[0].has_quorum) self.assertEqual(len(commits), 1) self.assertEqual(replay, []) stored = node.get_required(simulation_fact_document, commits[0].key)