From 27db7894b52ce5f8e1b86822b95cf37bdce0df39 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sun, 14 Jun 2026 03:18:38 +0100 Subject: [PATCH] Add Python CultMesh local server --- packages/cultcache-py/PARITY.md | 6 +- packages/cultcache-py/README.md | 2 + .../cultcache-py/src/cultcache_py/verify.py | 1 + .../cultcache-py/src/cultmesh_py/__init__.py | 2 + .../cultcache-py/src/cultmesh_py/facade.py | 20 ++ .../cultcache-py/src/cultmesh_py/server.py | 197 ++++++++++++++++++ packages/cultcache-py/tests/test_cultcache.py | 52 +++++ 7 files changed, 279 insertions(+), 1 deletion(-) create mode 100644 packages/cultcache-py/src/cultmesh_py/server.py diff --git a/packages/cultcache-py/PARITY.md b/packages/cultcache-py/PARITY.md index d2e2722b..8159629e 100644 --- a/packages/cultcache-py/PARITY.md +++ b/packages/cultcache-py/PARITY.md @@ -20,7 +20,11 @@ CultCache, CultNet, and CultMesh. It is evidence, not confetti. - Python CultMesh facade: Exposes C#-matching client entrypoints for Verse discovery, peer exchange, raw CultNet client construction, local nodes, catalogs, sessions, streams, - authority leases, and simulation fact commits. + authority leases, local TCP node serving, and simulation fact commits. +- Python CultMesh local server: + Serves a `CultMeshNode` plus Verse and peer catalogs over the same framed + MessagePack request/response lane used by `CultNetRawClient` and + `CultMeshDiscoveryClient`. - Python CultMesh reactive catalogs: Verse and peer catalogs expose sorted local views, `get(...)`, response application, and `watch(...)` callbacks for local discovery updates. Verse diff --git a/packages/cultcache-py/README.md b/packages/cultcache-py/README.md index 7bebbd70..bcb802a9 100644 --- a/packages/cultcache-py/README.md +++ b/packages/cultcache-py/README.md @@ -246,6 +246,7 @@ delete_message = node.database.delete_raw_message(note_doc, "note:2", shard_id=" peers = CultMesh.create_peer_catalog() response = peers.create_response(peer_exchange_request("pex-1", verse_id="local")) +server = CultMesh.serve_node(node, peer_catalog=peers, port=3075) client = CultMesh.create_verse_discovery_client("127.0.0.1", 3075) verses = client.fetch_verses(transport_version="cultmesh.v0") @@ -255,6 +256,7 @@ client.sync_peer_catalog(peers, verse_id="python-interop", roles=["read-replica" raw_client = CultNetRawClient("127.0.0.1", 3075) node.database.sync_snapshot(raw_client, schema_ids=[note_doc.catalog_entry().schema_id]) node.database.sync_shard_log(raw_client, shard_id="interop", shard_epoch=1) +server.stop() streams = CultMesh.create_stream_catalog() diff --git a/packages/cultcache-py/src/cultcache_py/verify.py b/packages/cultcache-py/src/cultcache_py/verify.py index 270fb24e..805bd0bc 100644 --- a/packages/cultcache-py/src/cultcache_py/verify.py +++ b/packages/cultcache-py/src/cultcache_py/verify.py @@ -68,6 +68,7 @@ EXPECTED_EXPORTS: dict[str, tuple[str, ...]] = { "CultMeshDiscoveryClient", "CultMeshGameSession", "CultMeshGameSessionOptions", + "CultMeshLocalServer", "CultMeshNode", "CultMeshPeerCard", "CultMeshPeerCatalog", diff --git a/packages/cultcache-py/src/cultmesh_py/__init__.py b/packages/cultcache-py/src/cultmesh_py/__init__.py index 932d921b..3d1c779e 100644 --- a/packages/cultcache-py/src/cultmesh_py/__init__.py +++ b/packages/cultcache-py/src/cultmesh_py/__init__.py @@ -1,6 +1,7 @@ from .client import CultMeshDiscoveryClient, CultMeshPeerExchangeClient, CultMeshVerseDiscoveryClient from .facade import CultMesh from .node import CultMeshDatabase, CultMeshDatabaseChange, CultMeshNode, create_node +from .server import CultMeshLocalServer from .session import ( CultMeshGameSession, CultMeshGameSessionOptions, @@ -45,6 +46,7 @@ __all__ = [ "CultMesh", "CultMeshGameSession", "CultMeshGameSessionOptions", + "CultMeshLocalServer", "CultMeshPeerCard", "CultMeshPeerCatalog", "CultMeshPrediction", diff --git a/packages/cultcache-py/src/cultmesh_py/facade.py b/packages/cultcache-py/src/cultmesh_py/facade.py index 18eb7d00..1e5aa062 100644 --- a/packages/cultcache-py/src/cultmesh_py/facade.py +++ b/packages/cultcache-py/src/cultmesh_py/facade.py @@ -6,6 +6,7 @@ from cultnet_py import CultNetRawClient from .client import CultMeshPeerExchangeClient, CultMeshVerseDiscoveryClient from .node import CultMeshNode, create_node +from .server import CultMeshLocalServer from .session import CultMeshGameSession, CultMeshGameSessionOptions from .simulation import CultMeshSimulationFactCommitter from .wire import ( @@ -70,6 +71,25 @@ class CultMesh: ) -> CultMeshGameSession: return CultMeshGameSession(node, options) + @staticmethod + def serve_node( + node: CultMeshNode, + *, + verse_catalog: CultMeshVerseCatalog | None = None, + peer_catalog: CultMeshPeerCatalog | None = None, + host: str = "127.0.0.1", + port: int = 0, + display_name: str | None = None, + ) -> CultMeshLocalServer: + return CultMeshLocalServer( + node=node, + verse_catalog=verse_catalog or CultMeshVerseCatalog(), + peer_catalog=peer_catalog or CultMeshPeerCatalog(), + host=host, + port=port, + display_name=display_name, + ).start() + @staticmethod def create_client( host: str = "localhost", diff --git a/packages/cultcache-py/src/cultmesh_py/server.py b/packages/cultcache-py/src/cultmesh_py/server.py new file mode 100644 index 00000000..5a0367f2 --- /dev/null +++ b/packages/cultcache-py/src/cultmesh_py/server.py @@ -0,0 +1,197 @@ +from __future__ import annotations + +import socket +import threading +from dataclasses import dataclass, field +from typing import Any + +import msgpack + +from cultnet_py import read_frame, write_frame + +from .node import CultMeshNode +from .wire import ( + PEER_EXCHANGE_REQUEST, + VERSE_CATALOG_REQUEST, + CultMeshPeerCatalog, + CultMeshVerseCatalog, +) + + +@dataclass +class CultMeshLocalServer: + node: CultMeshNode + verse_catalog: CultMeshVerseCatalog = field(default_factory=CultMeshVerseCatalog) + peer_catalog: CultMeshPeerCatalog = field(default_factory=CultMeshPeerCatalog) + host: str = "127.0.0.1" + port: int = 0 + display_name: str | None = None + runtime_kind: str = "python" + _socket: socket.socket | None = field(default=None, init=False, repr=False) + _stop: threading.Event = field(default_factory=threading.Event, init=False, repr=False) + _thread: threading.Thread | None = field(default=None, init=False, repr=False) + + def start(self) -> "CultMeshLocalServer": + if self._socket is not None: + return self + server = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + server.bind((self.host, self.port)) + server.listen() + server.settimeout(0.2) + self._socket = server + self.port = int(server.getsockname()[1]) + self._stop.clear() + self._thread = threading.Thread(target=self._accept_loop, daemon=True) + self._thread.start() + return self + + def stop(self) -> None: + self._stop.set() + if self._socket is not None: + try: + self._socket.close() + except OSError: + pass + self._socket = None + if self._thread is not None: + self._thread.join(timeout=2.0) + self._thread = None + + def __enter__(self) -> "CultMeshLocalServer": + return self.start() + + def __exit__(self, exc_type: Any, exc: Any, traceback: Any) -> None: + self.stop() + + def handle_message(self, message: dict[str, Any]) -> dict[str, Any] | None: + schema_version = message.get("schemaVersion") + if schema_version == "cultnet.hello.v0": + return self._hello_response() + if schema_version == "cultnet.schema_catalog_request.v0": + return self._schema_catalog_response(message) + if schema_version == "cultnet.snapshot_request.v0": + return self.node.database.create_snapshot_response( + message_id=str(message.get("messageId") or ""), + schema_ids=[str(value) for value in message.get("schemaIds") or []], + record_keys=[str(value) for value in message.get("recordKeys") or []], + shard_id=message.get("shardId"), + shard_epoch=message.get("shardEpoch"), + ) + if schema_version == "cultnet.shard_catalog_request.v0": + return self._shard_catalog_response(message) + if schema_version == "cultnet.shard_log_request.v0": + return self.node.database.create_shard_log_response( + message_id=str(message.get("messageId") or ""), + shard_id=str(message.get("shardId") or ""), + shard_epoch=message.get("shardEpoch"), + after_sequence=int(message.get("afterSequence") or 0), + limit=message.get("limit"), + ) + if schema_version == VERSE_CATALOG_REQUEST: + return self.verse_catalog.create_response(message) + if schema_version == PEER_EXCHANGE_REQUEST: + return self.peer_catalog.create_response(message) + return None + + def _accept_loop(self) -> None: + while not self._stop.is_set(): + server = self._socket + if server is None: + return + try: + client, _ = server.accept() + except TimeoutError: + continue + except OSError: + return + threading.Thread(target=self._handle_connection, args=(client,), daemon=True).start() + + def _handle_connection(self, client: socket.socket) -> None: + with client: + stream = client.makefile("rwb") + while not self._stop.is_set(): + try: + message = msgpack.unpackb(read_frame(stream), raw=False) + except EOFError: + return + if not isinstance(message, dict): + continue + response = self.handle_message(message) + if response is None: + continue + write_frame(stream, msgpack.packb(response, use_bin_type=True)) + stream.flush() + + def _hello_response(self) -> dict[str, Any]: + return { + "schemaVersion": "cultnet.hello.v0", + "runtimeId": self.node.runtime_id, + "runtimeKind": self.runtime_kind, + "displayName": self.display_name or self.node.runtime_id, + "supportedDocumentTypes": [document.type for document in self.node.documents], + "supportedMessageVersions": [ + "cultnet.hello.v0", + "cultnet.schema_catalog_request.v0", + "cultnet.snapshot_request.v0", + "cultnet.shard_catalog_request.v0", + "cultnet.shard_log_request.v0", + VERSE_CATALOG_REQUEST, + PEER_EXCHANGE_REQUEST, + ], + "supportsSchemaCatalog": True, + } + + def _schema_catalog_response(self, request: dict[str, Any]) -> dict[str, Any]: + requested_schema_ids = {str(value) for value in request.get("schemaIds") or []} + requested_kinds = {str(value) for value in request.get("kinds") or []} + include_schema_json = request.get("includeSchemaJson") is True + schemas = [] + for document in self.node.documents: + entry = document.catalog_entry() + if requested_schema_ids and entry.schema_id not in requested_schema_ids: + continue + if requested_kinds and "document_payload" not in requested_kinds: + continue + descriptor = { + "schemaId": entry.schema_id, + "kind": "document_payload", + "schemaVersion": entry.schema_version, + "documentType": document.type, + "title": entry.schema_name, + "wireContracts": ["cultnet.document_put_raw.v0", "cultnet.snapshot_response_raw.v0"], + "contentHash": entry.content_hash, + } + if include_schema_json: + descriptor["schemaJson"] = entry.canonical_schema_json + schemas.append(descriptor) + return { + "schemaVersion": "cultnet.schema_catalog_response.v0", + "messageId": request.get("messageId", ""), + "schemas": schemas, + } + + def _shard_catalog_response(self, request: dict[str, Any]) -> dict[str, Any]: + requested_schema_ids = {str(value) for value in request.get("schemaIds") or []} + schema_ids = [document.catalog_entry().schema_id for document in self.node.documents] + if requested_schema_ids: + schema_ids = [schema_id for schema_id in schema_ids if schema_id in requested_schema_ids] + shards = [] + if schema_ids: + shards.append({ + "shardId": "primary", + "ownerRuntimeId": self.node.runtime_id, + "epoch": 0, + "isPrimary": True, + "schemaIds": schema_ids, + "keyPrefix": "", + "primaryEndpoints": [f"cultnet://{self.host}:{self.port}"], + "replicaEndpoints": [], + "readReplicaEndpoints": [f"cultnet://{self.host}:{self.port}"], + "region": "local", + }) + return { + "schemaVersion": "cultnet.shard_catalog_response.v0", + "messageId": request.get("messageId", ""), + "shards": shards, + } diff --git a/packages/cultcache-py/tests/test_cultcache.py b/packages/cultcache-py/tests/test_cultcache.py index 2bc1b995..508b28bc 100644 --- a/packages/cultcache-py/tests/test_cultcache.py +++ b/packages/cultcache-py/tests/test_cultcache.py @@ -1624,6 +1624,58 @@ class CultCacheTests(unittest.TestCase): self.assertEqual(local_peers.find("aetheria-main", role="read-replica")[0].peer_id, "peer-a") self.assertEqual(synced_peers[0].roles, ("read-replica",)) + def test_cultmesh_local_server_serves_node_and_catalogs_over_clients(self) -> None: + document = define_database_entry_type( + "mesh.server_note", + [("body", 0)], + schema_id="mesh.server_note.v1", + ) + node = CultMesh.create_node(runtime_id="mesh-server") + node.database.register_document(document) + node.database.put_raw_message(document, "note:1", {"body": "served"}, shard_id="primary", shard_epoch=1) + verses = CultMeshVerseCatalog() + verses.upsert( + CultMeshVerseDescriptor( + verse_id="server-verse", + display_name="Server Verse", + authority_model="local", + compatibility=CultMeshVerseCompatibility("cultmesh.v0", "rules"), + ) + ) + peers = CultMeshPeerCatalog() + peers.upsert( + CultMeshPeerCard( + peer_id="mesh-server", + verse_id="server-verse", + endpoints=("cultnet://127.0.0.1:0",), + roles=("read-replica",), + ) + ) + + server = CultMesh.serve_node(node, verse_catalog=verses, peer_catalog=peers, display_name="Mesh Server") + try: + raw_client = CultMesh.create_client("127.0.0.1", server.port, timeout_seconds=2.0) + hello_response = raw_client.request(hello(runtime_id="probe"), expected_schema_version="cultnet.hello.v0") + schema_response = raw_client.fetch_schema_catalog(schema_ids=["mesh.server_note.v1"], include_schema_json=True) + snapshot_response = raw_client.fetch_snapshot(schema_ids=["mesh.server_note.v1"]) + shard_catalog = raw_client.fetch_shard_catalog(schema_ids=["mesh.server_note.v1"]) + shard_log = raw_client.fetch_shard_log(shard_id="primary", shard_epoch=1) + discovery_client = CultMesh.create_verse_discovery_client("127.0.0.1", server.port, timeout_seconds=2.0) + fetched_verses = discovery_client.fetch_verses(transport_version="cultmesh.v0") + fetched_peers = discovery_client.fetch_peers(verse_id="server-verse", roles=["read-replica"]) + finally: + server.stop() + + self.assertEqual(hello_response["runtimeId"], "mesh-server") + self.assertEqual(hello_response["displayName"], "Mesh Server") + self.assertEqual(schema_response["schemas"][0]["schemaId"], "mesh.server_note.v1") + self.assertIn("schemaJson", schema_response["schemas"][0]) + self.assertEqual(snapshot_response["documents"][0]["recordKey"], "note:1") + self.assertEqual(shard_catalog["shards"][0]["schemaIds"], ["mesh.server_note.v1"]) + self.assertEqual(shard_log["entries"][0]["changeKind"], "added") + self.assertEqual(fetched_verses[0].verse_id, "server-verse") + self.assertEqual(fetched_peers[0].peer_id, "mesh-server") + def test_cultmesh_authority_lease_requires_live_matching_lease(self) -> None: peer = CultMeshPeerCard( peer_id="voidbot-local",