diff --git a/packages/cultcache-py/README.md b/packages/cultcache-py/README.md index dcc7cadb..713d4b6e 100644 --- a/packages/cultcache-py/README.md +++ b/packages/cultcache-py/README.md @@ -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. diff --git a/packages/cultcache-py/src/cultmesh_py/__init__.py b/packages/cultcache-py/src/cultmesh_py/__init__.py index 0ecebfc1..8e39aaff 100644 --- a/packages/cultcache-py/src/cultmesh_py/__init__.py +++ b/packages/cultcache-py/src/cultmesh_py/__init__.py @@ -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", diff --git a/packages/cultcache-py/src/cultmesh_py/facade.py b/packages/cultcache-py/src/cultmesh_py/facade.py index d989193a..889eea13 100644 --- a/packages/cultcache-py/src/cultmesh_py/facade.py +++ b/packages/cultcache-py/src/cultmesh_py/facade.py @@ -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) diff --git a/packages/cultcache-py/src/cultmesh_py/session.py b/packages/cultcache-py/src/cultmesh_py/session.py new file mode 100644 index 00000000..a0ca3bbb --- /dev/null +++ b/packages/cultcache-py/src/cultmesh_py/session.py @@ -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 diff --git a/packages/cultcache-py/src/cultnet_py/__init__.py b/packages/cultcache-py/src/cultnet_py/__init__.py index 08ca4690..d80b7103 100644 --- a/packages/cultcache-py/src/cultnet_py/__init__.py +++ b/packages/cultcache-py/src/cultnet_py/__init__.py @@ -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", diff --git a/packages/cultcache-py/src/cultnet_py/simulation.py b/packages/cultcache-py/src/cultnet_py/simulation.py new file mode 100644 index 00000000..8344803e --- /dev/null +++ b/packages/cultcache-py/src/cultnet_py/simulation.py @@ -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") + ) diff --git a/packages/cultcache-py/tests/test_cultcache.py b/packages/cultcache-py/tests/test_cultcache.py index 251ff75c..c90f0406 100644 --- a/packages/cultcache-py/tests/test_cultcache.py +++ b/packages/cultcache-py/tests/test_cultcache.py @@ -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