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

Add Python CultMesh game session loop

Browse source
This commit is contained in:
Meta Vens 2026-06-14 00:40:31 +01:00
commit e863902e76
7 changed files with 336 additions and 2 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -254,6 +254,9 @@ committed = facts.commit({
"confidence": 1.0,
"hasQuorum": True,
})
session = CultMesh.create_game_session(node)
session_commits = session.submit_and_commit(observation.to_wire())
```
## Wire Parity
@ -283,8 +286,8 @@ Current receipts:
- `packages/cultcache-py/tests/test_cultcache.py` covers Python CultMesh
Verse catalog, peer exchange, authority lease, stream negotiation, CultNet
helper shapes, raw snapshot/shard-log application, simulation claim hashing,
committed simulation fact payload slots, and witness artifact bundle payload
slots.
consensus aggregation, game-session fact commits, committed simulation fact
payload slots, and witness artifact bundle payload slots.
- `cultcache_py`, `cultnet_py`, and `cultmesh_py` ship `py.typed` markers so
downstream type checkers can inspect the package surface instead of treating
the runtime as an untyped xenos swamp.

8
packages/cultcache-py/src/cultmesh_py/__init__.py
View file

@ -1,6 +1,11 @@
from .client import CultMeshDiscoveryClient
from .facade import CultMesh
from .node import CultMeshNode, create_node
from .session import (
CultMeshGameSession,
CultMeshGameSessionOptions,
CultMeshPrediction,
)
from .simulation import (
SIMULATION_FACT_DOCUMENT_TYPE,
SIMULATION_FACT_SCHEMA_VERSION,
@ -35,8 +40,11 @@ from .wire import (
__all__ = [
"CultMeshNode",
"CultMesh",
"CultMeshGameSession",
"CultMeshGameSessionOptions",
"CultMeshPeerCard",
"CultMeshPeerCatalog",
"CultMeshPrediction",
"CultMeshSimulationFact",
"CultMeshSimulationFactCommit",
"CultMeshSimulationFactCommitter",

8
packages/cultcache-py/src/cultmesh_py/facade.py
View file

@ -3,6 +3,7 @@ from __future__ import annotations
from pathlib import Path
from .node import CultMeshNode, create_node
from .session import CultMeshGameSession, CultMeshGameSessionOptions
from .simulation import CultMeshSimulationFactCommitter
from .wire import (
CultMeshAuthorityLeaseCatalog,
@ -40,3 +41,10 @@ class CultMesh:
@staticmethod
def create_simulation_fact_committer(node: CultMeshNode) -> CultMeshSimulationFactCommitter:
return CultMeshSimulationFactCommitter(node)
@staticmethod
def create_game_session(
node: CultMeshNode,
options: CultMeshGameSessionOptions | None = None,
) -> CultMeshGameSession:
return CultMeshGameSession(node, options)

75
packages/cultcache-py/src/cultmesh_py/session.py Normal file
View file

@ -0,0 +1,75 @@
from __future__ import annotations
from dataclasses import dataclass
from typing import Any
from cultcache_py.documents import DocumentDefinition
from cultnet_py import CultNetSimulationConsensusOptions, CultNetSimulationObservationHub
from .node import CultMeshNode
from .simulation import (
CultMeshSimulationFact,
CultMeshSimulationFactCommit,
CultMeshSimulationFactCommitter,
simulation_fact_document,
)
from .wire import CultMeshAuthorityLeaseCatalog, CultMeshPeerCatalog, CultMeshVerseCatalog
@dataclass(frozen=True)
class CultMeshPrediction:
key: str
document: DocumentDefinition[Any]
value: Any
@dataclass
class CultMeshGameSessionOptions:
verse_catalog: CultMeshVerseCatalog | None = None
peer_catalog: CultMeshPeerCatalog | None = None
authority_leases: CultMeshAuthorityLeaseCatalog | None = None
consensus_options: CultNetSimulationConsensusOptions | None = None
@dataclass
class CultMeshGameSession:
node: CultMeshNode
options: CultMeshGameSessionOptions | None = None
def __post_init__(self) -> None:
options = self.options or CultMeshGameSessionOptions()
self.verse_catalog = options.verse_catalog or CultMeshVerseCatalog()
self.peer_catalog = options.peer_catalog or CultMeshPeerCatalog()
self.authority_leases = options.authority_leases or CultMeshAuthorityLeaseCatalog()
self.observation_hub = CultNetSimulationObservationHub(
options.consensus_options or CultNetSimulationConsensusOptions()
)
self.fact_committer = CultMeshSimulationFactCommitter(self.node)
self._committed_fact_ids: set[str] = set()
def predict(self, document: DocumentDefinition[Any], key: str, value: Any) -> CultMeshPrediction:
self.node.put(document, key, value)
return CultMeshPrediction(key=key, document=document, value=value)
def submit_observation(self, observation_or_message: dict[str, Any]) -> 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]:
return self.commit_quorum_candidates(self.submit_observation(observation_or_message))
def commit_quorum_candidates(self, candidates: list[dict[str, Any]]) -> list[CultMeshSimulationFactCommit]:
commits: list[CultMeshSimulationFactCommit] = []
for candidate in candidates:
if not bool(candidate.get("hasQuorum")):
continue
fact_id = CultMeshSimulationFact.compute_fact_id(candidate)
if fact_id in self._committed_fact_ids:
continue
key = CultMeshSimulationFact.create_record_key(candidate)
if self.node.get(simulation_fact_document, key) is not None:
self._committed_fact_ids.add(fact_id)
continue
commit = self.fact_committer.commit(candidate)
self._committed_fact_ids.add(fact_id)
commits.append(commit)
return commits

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

@ -25,12 +25,20 @@ from .replication import (
apply_shard_log_response,
schema_document_map,
)
from .simulation import (
CultNetSimulationConsensus,
CultNetSimulationConsensusOptions,
CultNetSimulationObservationHub,
)
__all__ = [
"CultNetMessage",
"CultNetDatabaseSubscription",
"CultNetRawClient",
"CultNetAppliedRecord",
"CultNetSimulationConsensus",
"CultNetSimulationConsensusOptions",
"CultNetSimulationObservationHub",
"apply_raw_document_record",
"apply_raw_snapshot",
"apply_shard_log_response",

137
packages/cultcache-py/src/cultnet_py/simulation.py Normal file
View file

@ -0,0 +1,137 @@
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any
@dataclass(frozen=True)
class CultNetSimulationConsensusOptions:
minimum_witnesses: int = 1
minimum_weight: float = 1.0
quorum_ratio: float = 0.5
@dataclass
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)])
groups: dict[tuple[str, int, int, str, str], list[dict[str, Any]]] = {}
for observation in clean:
groups.setdefault(_subject_key(observation), []).append(observation)
return [
self._build_candidate(group)
for _, group in sorted(groups.items(), key=lambda item: item[0])
]
def _build_candidate(self, observations: list[dict[str, Any]]) -> dict[str, Any]:
total_weight = sum(_weight_of(observation) for observation in observations)
claim_groups: dict[str, list[dict[str, Any]]] = {}
for observation in observations:
claim_groups.setdefault(str(observation.get("claimHash") or ""), []).append(observation)
claim_hash, best_observations = sorted(
claim_groups.items(),
key=lambda item: (-sum(_weight_of(observation) for observation in item[1]), item[0]),
)[0]
first = sorted(best_observations, key=lambda observation: str(observation.get("witnessRuntimeId") or ""))[0]
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": (
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,
}
@dataclass
class CultNetSimulationObservationHub:
options: CultNetSimulationConsensusOptions = field(default_factory=CultNetSimulationConsensusOptions)
observations: list[dict[str, Any]] = field(default_factory=list)
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")
self.observations.append(observation)
return [
candidate
for candidate in self._consensus.build_candidates(self.observations)
if _candidate_matches_observation(candidate, observation)
]
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]]] = {}
for observation in observations:
groups.setdefault(_dedupe_key(observation), []).append(observation)
return [
sorted(group, key=lambda observation: (-_weight_of(observation), str(observation.get("witnessRuntimeId") or "")))[0]
for _, group in sorted(groups.items(), key=lambda item: item[0])
]
def _is_valid_observation(observation: dict[str, Any]) -> bool:
return (
bool(str(observation.get("witnessRuntimeId") or "").strip())
and bool(str(observation.get("shardId") or "").strip())
and bool(str(observation.get("subjectId") or "").strip())
and bool(str(observation.get("claimKind") or "").strip())
and bool(str(observation.get("claimHash") or "").strip())
and _weight_of(observation) > 0.0
)
def _weight_of(observation: dict[str, Any]) -> float:
return max(float(observation.get("weight", 1.0) or 0.0), 0.0)
def _dedupe_key(observation: dict[str, Any]) -> tuple[str, str, int, int, str, str]:
return (
str(observation.get("witnessRuntimeId") or ""),
str(observation.get("shardId") or ""),
int(observation.get("shardEpoch") or 0),
int(observation.get("frame") or 0),
str(observation.get("subjectId") or ""),
str(observation.get("claimKind") or ""),
)
def _subject_key(observation: dict[str, Any]) -> tuple[str, int, int, str, str]:
return (
str(observation.get("shardId") or ""),
int(observation.get("shardEpoch") or 0),
int(observation.get("frame") or 0),
str(observation.get("subjectId") or ""),
str(observation.get("claimKind") or ""),
)
def _candidate_matches_observation(candidate: dict[str, Any], observation: dict[str, Any]) -> bool:
return (
candidate.get("shardId") == observation.get("shardId")
and candidate.get("shardEpoch") == observation.get("shardEpoch")
and candidate.get("frame") == observation.get("frame")
and candidate.get("subjectId") == observation.get("subjectId")
and candidate.get("claimKind") == observation.get("claimKind")
)

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

@ -21,6 +21,7 @@ from cultcache_py.interop import read_note, write_note
from cultnet_py import (
compute_simulation_claim_hash,
CultNetRawClient,
CultNetSimulationConsensusOptions,
apply_raw_snapshot,
apply_shard_log_response,
database_subscribe,
@ -44,6 +45,7 @@ from cultnet_py.interop_peer import wire_message_schema_descriptors
from cultmesh_py import create_node
from cultmesh_py import (
CultMesh,
CultMeshGameSessionOptions,
CultMeshPeerCard,
CultMeshPeerCatalog,
CultMeshSimulationFact,
@ -866,6 +868,99 @@ class CultCacheTests(unittest.TestCase):
self.assertEqual(stored.committed_at, "2026-06-13T00:00:00Z")
self.assertEqual(committed.fact.fact_id, stored.fact_id)
def test_cultnet_simulation_consensus_dedupes_witnesses_and_requires_quorum(self) -> None:
from cultnet_py import CultNetSimulationConsensus
consensus = CultNetSimulationConsensus(
CultNetSimulationConsensusOptions(minimum_witnesses=2, quorum_ratio=1.0)
)
claim_hash = compute_simulation_claim_hash("hit", "alice", "bob", "frame:100")
observations = [
{
"witnessRuntimeId": "watcher-1",
"shardId": "arena",
"shardEpoch": 4,
"frame": 100,
"subjectId": "bob",
"claimKind": "hit",
"claimHash": claim_hash,
"claimSummary": "alice shot bob first",
"weight": 1.0,
},
{
"witnessRuntimeId": "watcher-1",
"shardId": "arena",
"shardEpoch": 4,
"frame": 100,
"subjectId": "bob",
"claimKind": "hit",
"claimHash": "stale-duplicate",
"weight": 0.5,
},
{
"witnessRuntimeId": "watcher-2",
"shardId": "arena",
"shardEpoch": 4,
"frame": 100,
"subjectId": "bob",
"claimKind": "hit",
"claimHash": claim_hash,
"weight": 1.0,
},
]
candidates = consensus.build_candidates(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"])
def test_cultmesh_game_session_submits_observations_and_commits_quorum_once(self) -> None:
claim_hash = compute_simulation_claim_hash("hit", "alice", "bob", "frame:100")
first = simulation_observation(
message_id="obs-1",
witness_runtime_id="watcher-1",
shard_id="arena",
shard_epoch=4,
frame=100,
subject_id="bob",
claim_kind="hit",
claim_hash=claim_hash,
claim_summary="alice shot bob first",
).to_wire()
second = simulation_observation(
message_id="obs-2",
witness_runtime_id="watcher-2",
shard_id="arena",
shard_epoch=4,
frame=100,
subject_id="bob",
claim_kind="hit",
claim_hash=claim_hash,
claim_summary="alice shot bob first",
).to_wire()
with tempfile.TemporaryDirectory() as tmp:
node = CultMesh.create_node(Path(tmp) / "session.cc", runtime_id="mesh-session")
session = CultMesh.create_game_session(
node,
CultMeshGameSessionOptions(
consensus_options=CultNetSimulationConsensusOptions(minimum_witnesses=2, quorum_ratio=1.0)
),
)
self.assertEqual(session.submit_and_commit(first), [])
commits = session.submit_and_commit(second)
replay = session.commit_quorum_candidates(session.submit_observation(second))
self.assertEqual(len(commits), 1)
self.assertEqual(replay, [])
stored = node.get_required(simulation_fact_document, commits[0].key)
self.assertEqual(stored.claim_hash, claim_hash)
self.assertEqual(stored.witness_count, 2)
def test_cultmesh_verse_catalog_response_matches_schema_v0_wire_shape(self) -> None:
import msgpack # type: ignore