GameCult/CultLib
Watch
9
0
Fork
You've already forked CultLib
 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

Add typed Python simulation consensus API

Browse source
This commit is contained in:
Meta Vens 2026-06-14 04:57:41 +01:00
commit fd7389b299
7 changed files with 211 additions and 32 deletions
Download patch file Download diff file Expand all files Collapse all files

4
packages/cultcache-py/PARITY.md
View file

@ -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

8
packages/cultcache-py/README.md
View file

@ -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()
```

2
packages/cultcache-py/src/cultcache_py/verify.py
View file

@ -42,7 +42,9 @@ EXPECTED_EXPORTS: dict[str, tuple[str, ...]] = {
"CultNetShardLogEntry",
"CultNetShardLogResponse",
"CultNetSimulationConsensus",
"CultNetSimulationConsensusCandidate",
"CultNetSimulationConsensusOptions",
"CultNetSimulationObservation",
"CultNetSimulationObservationHub",
"INTEROP_WIRE_CONTRACT",
"apply_raw_document_record",

18
packages/cultcache-py/src/cultmesh_py/session.py
View file

@ -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:

4
packages/cultcache-py/src/cultnet_py/__init__.py
View file

@ -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",

185
packages/cultcache-py/src/cultnet_py/simulation.py
View file

@ -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

22
packages/cultcache-py/tests/test_cultcache.py
View file

@ -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)