From ce9d709daef08b2f553aaa3fd8aa4ea30f942f4f Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Tue, 16 Jun 2026 23:37:59 +0100 Subject: [PATCH] Add Python RUDP schema client transport --- docs/cultnet-transport-parity.md | 6 +- packages/cultcache-py/PARITY.md | 14 +++-- packages/cultcache-py/README.md | 10 ++++ .../cultcache-py/src/cultmesh_py/facade.py | 46 ++++++++++++++- .../cultcache-py/src/cultnet_py/__init__.py | 2 + .../cultcache-py/src/cultnet_py/client.py | 58 ++++++++++++++++++- .../cultcache-py/src/cultnet_py/transport.py | 16 +++++ packages/cultcache-py/tests/test_cultcache.py | 58 +++++++++++++++++++ 8 files changed, 202 insertions(+), 8 deletions(-) diff --git a/docs/cultnet-transport-parity.md b/docs/cultnet-transport-parity.md index b6e9d1b2..9894b58e 100644 --- a/docs/cultnet-transport-parity.md +++ b/docs/cultnet-transport-parity.md @@ -301,7 +301,11 @@ Current progress: `CultNetSchemaTransport` factory for request/response and database subscription flows. The default factory opens the existing TCP-framed schema transport, but the raw client no longer owns inline socket setup as its - request truth. + request truth. Python now also exposes `create_rudp_schema_transport(...)`, + which handshakes a socket-backed RUDP client and gives `CultNetRawClient` the + same reliable ordered `schema` request path over `cultnet.transport.rudp.v0`; + `CultMesh.create_client(endpoint="rudp://...")` is the branded ergonomic + helper for that path. - Python's branded `CultMesh` facade now creates schema catalogs, built-in schema catalogs, and shard catalogs by delegating to the existing `cultnet_py` owners, matching the TypeScript/Kotlin/Rust entrypoint pattern diff --git a/packages/cultcache-py/PARITY.md b/packages/cultcache-py/PARITY.md index 9e3ca7dc..2aa7a308 100644 --- a/packages/cultcache-py/PARITY.md +++ b/packages/cultcache-py/PARITY.md @@ -22,6 +22,10 @@ CultCache, CultNet, and CultMesh. It is evidence, not confetti. subscriptions now consume a caller-provided `CultNetSchemaTransport` factory, with `create_tcp_framed_schema_transport(...)` as the default, so request and subscription semantics are no longer owned by inline TCP socket setup. + `create_rudp_schema_transport(...)` opens the same raw-client request path + over a socket-backed RUDP reliable ordered `schema` channel, and the branded + `CultMesh.create_client(endpoint="rudp://...")` helper selects that transport + from a RUDP endpoint without making callers assemble a factory closure. - Python CultNet raw snapshot: Provides typed `CultNetRawSnapshotResponse` and `CultNetRawDocumentRecord` helpers plus `CultNetRawClient.fetch_snapshot_response(...)` for inspecting @@ -303,10 +307,12 @@ node --test packages\cultnet-ts\dist-test\test\interop\cultnet-interop.test.js port instead of cloning LiteNetLib. The raw client request/subscription layer now routes through a schema-transport factory, so the caller-facing semantics are transport-port-shaped even when the default factory opens TCP-framed - connections. The reconnect policy/controller primitive is portable and - advertised by RUDP profiles, and `CultNetRudpReconnectLoop` gives Python - callers the same RUDP retry-scheduling helper as TypeScript/Kotlin. - Automatic reconnect use by Python service bodies is not yet claimed. + connections. Python raw clients can now use a RUDP schema transport directly + through `create_rudp_schema_transport(...)` or the `CultMesh.create_client` + RUDP endpoint helper. The reconnect policy/controller primitive is portable + and advertised by RUDP profiles, and `CultNetRudpReconnectLoop` gives Python + callers the same RUDP retry-scheduling helper as TypeScript/Kotlin. Automatic + reconnect use by Python service bodies is not yet claimed. - Python does not implement the full C# daemon/server stack. Its current role is package runtime, raw interop peer, launchable local CultMesh endpoint, local CultMesh session facade, and typed state participant. diff --git a/packages/cultcache-py/README.md b/packages/cultcache-py/README.md index 0dd6ee81..9f53eaf2 100644 --- a/packages/cultcache-py/README.md +++ b/packages/cultcache-py/README.md @@ -169,6 +169,7 @@ from cultnet_py import ( simulation_observation, witness_artifact_bundle, ) +from cultmesh_py import CultMesh payload = hello(runtime_id="python-runtime").to_bytes() message = parse_message(decode_frame(encode_frame(payload))) @@ -186,6 +187,15 @@ shard_log = client.fetch_shard_log_response(shard_id="interop", shard_epoch=1, a # 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() + +# The same raw-client request surface can ride RUDP when the peer advertises a +# rudp:// endpoint and shares the connection id for that lane. +rudp_client = CultMesh.create_client( + endpoint="rudp://127.0.0.1:3076", + connection_id=0x43554C54, +) +rudp_catalog = rudp_client.fetch_schema_catalog(kinds=["wire_message"]) + claim_hash = compute_simulation_claim_hash("frame:42", "subject:player-1", "hit") observation = simulation_observation( message_id="obs-1", diff --git a/packages/cultcache-py/src/cultmesh_py/facade.py b/packages/cultcache-py/src/cultmesh_py/facade.py index 35b57042..fe8562d7 100644 --- a/packages/cultcache-py/src/cultmesh_py/facade.py +++ b/packages/cultcache-py/src/cultmesh_py/facade.py @@ -16,6 +16,7 @@ from cultnet_py import ( CultNetSchemaCatalog, CultNetShardCatalog, CultNetSimulationObservationHub, + create_rudp_schema_transport, wire_message_schema_catalog, ) @@ -310,8 +311,35 @@ class CultMesh: port: int = 3075, *, timeout_seconds: float = 4.0, + endpoint: str | CultMeshRudpEndpoint | None = None, + connection_id: int = 0x43554C54, + runtime_id: str = "cultmesh-python-rudp-client", + bind_host: str = "127.0.0.1", + bind_port: int = 0, ) -> CultNetRawClient: - return CultNetRawClient(host, port, timeout_seconds) + if endpoint is None and host.lower().startswith("rudp://"): + endpoint = host + if endpoint is None: + return CultNetRawClient(host, port, timeout_seconds) + parsed_endpoint = ( + CultMesh.parse_rudp_endpoint(endpoint) + if isinstance(endpoint, str) + else endpoint + ) + return CultNetRawClient( + parsed_endpoint.host, + parsed_endpoint.port, + timeout_seconds, + create_transport=lambda: create_rudp_schema_transport( + host=parsed_endpoint.host, + port=parsed_endpoint.port, + connection_id=connection_id, + timeout_seconds=timeout_seconds, + runtime_id=runtime_id, + bind_host=bind_host, + bind_port=bind_port, + ), + ) @staticmethod def connect_client( @@ -319,8 +347,22 @@ class CultMesh: port: int = 3075, *, timeout_seconds: float = 4.0, + endpoint: str | CultMeshRudpEndpoint | None = None, + connection_id: int = 0x43554C54, + runtime_id: str = "cultmesh-python-rudp-client", + bind_host: str = "127.0.0.1", + bind_port: int = 0, ) -> CultNetRawClient: - return CultNetRawClient(host, port, timeout_seconds) + return CultMesh.create_client( + host, + port, + timeout_seconds=timeout_seconds, + endpoint=endpoint, + connection_id=connection_id, + runtime_id=runtime_id, + bind_host=bind_host, + bind_port=bind_port, + ) def _bind_rudp_socket(bind_host: str, bind_port: int) -> socket_module.socket: diff --git a/packages/cultcache-py/src/cultnet_py/__init__.py b/packages/cultcache-py/src/cultnet_py/__init__.py index 5a0b567c..5a842dbe 100644 --- a/packages/cultcache-py/src/cultnet_py/__init__.py +++ b/packages/cultcache-py/src/cultnet_py/__init__.py @@ -4,6 +4,7 @@ from .client import ( CultNetRawClient, CultNetSchemaTransport, CultNetSchemaTransportFactory, + create_rudp_schema_transport, create_tcp_framed_schema_transport, ) from .framing import decode_frame, encode_frame, read_frame, write_frame @@ -170,6 +171,7 @@ __all__ = [ "compute_simulation_claim_hash", "compute_reconnect_delay_ms", "create_reconnect_policy", + "create_rudp_schema_transport", "create_tcp_framed_schema_transport", "create_rudp_transport_profile", "create_tcp_framed_transport_profile", diff --git a/packages/cultcache-py/src/cultnet_py/client.py b/packages/cultcache-py/src/cultnet_py/client.py index 8177219a..c30859d9 100644 --- a/packages/cultcache-py/src/cultnet_py/client.py +++ b/packages/cultcache-py/src/cultnet_py/client.py @@ -1,6 +1,7 @@ from __future__ import annotations import socket +import time from collections.abc import Callable from dataclasses import dataclass from typing import Any, Protocol @@ -21,7 +22,13 @@ from .shard_catalog import CultNetShardCatalog, CultNetShardDescriptor from .shard_log import CultNetShardLogResponse from .snapshot import CultNetRawSnapshotResponse from .subscription import CultNetDatabaseChange -from .transport import TcpFramedTransportConnection, create_tcp_framed_transport_profile +from .transport import ( + CultNetRudpSocketMode, + CultNetRudpSocketTransportConnection, + CultNetRudpSocketTransportOptions, + TcpFramedTransportConnection, + create_tcp_framed_transport_profile, +) class CultNetSchemaTransport(Protocol): @@ -377,6 +384,55 @@ def create_tcp_framed_schema_transport( ) +def create_rudp_schema_transport( + *, + host: str, + port: int, + connection_id: int, + timeout_seconds: float = 4.0, + runtime_id: str = "cultnet-python-rudp-client", + bind_host: str = "127.0.0.1", + bind_port: int = 0, + initial_sequence: int = 1, + resend_delay_ms: int = 25, + transport_id: str = "rudp", + max_payload_bytes: int | None = None, + max_fragment_bytes: int | None = None, + max_pending_reliable_packets: int | None = None, +) -> CultNetRudpSocketTransportConnection: + transport_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + transport_socket.bind((bind_host, bind_port)) + transport_socket.settimeout(min(timeout_seconds, 0.02)) + transport = CultNetRudpSocketTransportConnection( + CultNetRudpSocketTransportOptions( + runtime_id=runtime_id, + socket=transport_socket, + mode=CultNetRudpSocketMode.CLIENT, + remote_addr=(host, port), + connection_id=connection_id, + initial_sequence=initial_sequence, + resend_delay_ms=resend_delay_ms, + transport_id=transport_id, + max_payload_bytes=max_payload_bytes, + max_fragment_bytes=max_fragment_bytes, + max_pending_reliable_packets=max_pending_reliable_packets, + ) + ) + + try: + transport.connect(b"cultnet-schema") + deadline = time.monotonic() + timeout_seconds + while not transport.connected: + transport.receive_once() + transport.poll_resends() + if time.monotonic() >= deadline: + raise TimeoutError("Timed out waiting for RUDP schema transport handshake") + return transport + except Exception: + transport.close() + raise + + class _OwnedTcpFramedTransportConnection(TcpFramedTransportConnection): def __init__(self, stream: Any, connection: socket.socket, *, profile: dict[str, Any]) -> None: super().__init__(stream, profile=profile) diff --git a/packages/cultcache-py/src/cultnet_py/transport.py b/packages/cultcache-py/src/cultnet_py/transport.py index aabc0782..25a4c891 100644 --- a/packages/cultcache-py/src/cultnet_py/transport.py +++ b/packages/cultcache-py/src/cultnet_py/transport.py @@ -819,6 +819,12 @@ class CultNetRudpSocketTransportConnection: self.disconnect_reason: bytes | None = None self.pong_payloads: deque[bytes] = deque() + def __enter__(self) -> "CultNetRudpSocketTransportConnection": + return self + + def __exit__(self, exc_type: Any, exc: Any, traceback: Any) -> None: + self.close() + @property def connected(self) -> bool: return self.session.connected @@ -890,6 +896,16 @@ class CultNetRudpSocketTransportConnection: self._send_packet(self.session.create_ack()) return frame + def receive(self, timeout_seconds: float | None = None) -> CultNetTransportFrame: + deadline = None if timeout_seconds is None else time.monotonic() + timeout_seconds + while True: + frame = self.receive_once() + if frame is not None: + return frame + self.poll_resends() + if deadline is not None and time.monotonic() >= deadline: + raise TimeoutError("Timed out waiting for RUDP schema frame") + def disconnect(self, reason: bytes = b"") -> None: self._send_packet(self.session.create_disconnect(reason)) diff --git a/packages/cultcache-py/tests/test_cultcache.py b/packages/cultcache-py/tests/test_cultcache.py index e23010f1..ea1ba26e 100644 --- a/packages/cultcache-py/tests/test_cultcache.py +++ b/packages/cultcache-py/tests/test_cultcache.py @@ -668,6 +668,64 @@ class CultCacheTests(unittest.TestCase): self.assertEqual(sent_messages[0]["schemaVersion"], "cultnet.schema_catalog_request.v0") self.assertEqual(sent_messages[0]["messageId"], "transport-factory") + def test_cultmesh_client_uses_rudp_schema_transport_endpoint(self) -> None: + import msgpack # type: ignore + + server_socket = bind_udp_socket() + server_host, server_port = server_socket.getsockname()[:2] + connection_id = 0x43554C54 + server = CultNetRudpSocketTransportConnection( + CultNetRudpSocketTransportOptions( + runtime_id="python-rudp-schema-server", + socket=server_socket, + mode=CultNetRudpSocketMode.SERVER, + connection_id=connection_id, + initial_sequence=100, + resend_delay_ms=25, + ) + ) + received_messages: list[dict[str, object]] = [] + + def serve_once() -> None: + deadline = time.monotonic() + 2.0 + try: + while time.monotonic() < deadline: + frame = server.receive_once() + server.poll_resends() + if frame is None: + time.sleep(0.005) + continue + received = msgpack.unpackb(frame.payload, raw=False) + received_messages.append(received) + response = { + "schemaVersion": "cultnet.schema_catalog_response.v0", + "messageId": received["messageId"], + "schemas": [], + } + server.send("schema", msgpack.packb(response, use_bin_type=True)) + return + raise AssertionError("RUDP schema server did not receive a request") + finally: + server.close() + + server_thread = threading.Thread(target=serve_once, daemon=True) + server_thread.start() + client = CultMesh.create_client( + endpoint=f"rudp://{server_host}:{server_port}", + timeout_seconds=2.0, + connection_id=connection_id, + runtime_id="python-rudp-schema-client", + ) + + response = client.fetch_schema_catalog(message_id="rudp-transport-factory") + server_thread.join(timeout=2.0) + + self.assertFalse(server_thread.is_alive()) + self.assertEqual(response["schemaVersion"], "cultnet.schema_catalog_response.v0") + self.assertEqual(response["messageId"], "rudp-transport-factory") + self.assertEqual(received_messages[0]["schemaVersion"], "cultnet.schema_catalog_request.v0") + self.assertEqual(received_messages[0]["messageId"], "rudp-transport-factory") + def test_cultnet_rudp_packet_codec_uses_deterministic_reliable_ordered_fixture(self) -> None: encoded = encode_rudp_packet( CultNetRudpPacket(