From e65c0d0fc60452cdcd5b464ae4ca8ce8ad44b1af Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Tue, 16 Jun 2026 16:13:56 +0100 Subject: [PATCH] Add Python schema transport client port --- docs/cultnet-transport-parity.md | 5 + docs/runtime-parity-scope.md | 2 +- packages/cultcache-py/PARITY.md | 17 ++-- .../cultcache-py/src/cultnet_py/__init__.py | 12 ++- .../cultcache-py/src/cultnet_py/client.py | 98 +++++++++++++------ .../cultcache-py/src/cultnet_py/transport.py | 6 ++ packages/cultcache-py/tests/test_cultcache.py | 42 ++++++++ 7 files changed, 144 insertions(+), 38 deletions(-) diff --git a/docs/cultnet-transport-parity.md b/docs/cultnet-transport-parity.md index 138afe7f..66c07c37 100644 --- a/docs/cultnet-transport-parity.md +++ b/docs/cultnet-transport-parity.md @@ -291,6 +291,11 @@ Current progress: callers: the caller-owned receive loop reports closure, the shared controller owns retry delay/exhaustion state, and the caller-provided factory opens the next `CultNetRudpSocketTransportConnection`. +- Python raw CultNet clients now consume a caller-provided + `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. - Rust now has a poll-driven `CultNetRudpReconnectLoop` for substrate callers: service loops report closure, ask `reconnect_if_due(now_ms)` inside their own reactor, and the shared controller owns retry delay/exhaustion state while the diff --git a/docs/runtime-parity-scope.md b/docs/runtime-parity-scope.md index e7b57afe..976c0023 100644 --- a/docs/runtime-parity-scope.md +++ b/docs/runtime-parity-scope.md @@ -31,7 +31,7 @@ CultMesh. | Runtime | Role | Claimed parity | Not claimed | | --- | --- | --- | --- | | C# | Reference implementation and production game/server substrate | CultCache typed documents and SoA, CultNet schema-v0, LiteNetLib production path with adapter profile, shared transport profiles, shared RUDP socket transport, RUDP reconnect policy/controller and reconnect-loop helper, LiteNetLib client reconnect-controller adoption, schema-client transport port for service request/response clients, CultMesh node/session/catalog/stream/simulation surfaces | No need to abandon LiteNetLib before the CultNet adapter fully owns production service adoption | -| Python | Package runtime, local daemon endpoint, raw interop peer, typed state participant | CultCache/CultNet/CultMesh package surface, schema catalogs, shard catalogs/logs, local TCP CultMesh server, session/prediction/simulation fact helpers, authority leases, streams, RUDP codec/session/socket transport, RUDP reconnect policy/controller and reconnect-loop helper, schema-v0 MessagePack over RUDP evidence, benchmark evidence against C# cache paths | Universal C# throughput, full C# daemon/server stack, and RUDP as the only public service pipe | +| Python | Package runtime, local daemon endpoint, raw interop peer, typed state participant | CultCache/CultNet/CultMesh package surface, schema catalogs, shard catalogs/logs, schema-transport factory for raw client request/subscription flows, local TCP CultMesh server, session/prediction/simulation fact helpers, authority leases, streams, RUDP codec/session/socket transport, RUDP reconnect policy/controller and reconnect-loop helper, schema-v0 MessagePack over RUDP evidence, benchmark evidence against C# cache paths | Universal C# throughput, full C# daemon/server stack, and RUDP as the only public service pipe | | TypeScript | Node/browser-adjacent projection and interop runtime | CultCache parity tests, CultNet schema-v0 peer, advertised schema/shard catalog wire contracts plus request/sync helpers, transport profiles, TCP-framed transport port and peer helper, shared RUDP socket transport, RUDP reconnect policy/controller, RUDP reconnect-loop helper, local CultMesh node/catalog/authority/stream ergonomics | Full C# game session/server stack | | Rust | Low-level cache/transport/runtime substrate | CultCache/CultNet interop lane, schema and shard catalog helpers, transport profiles, TCP-framed port, shared RUDP codec/session/socket transport, RUDP reconnect policy/controller and reconnect-loop helper, schema-v0 MessagePack over RUDP evidence, small CultMesh facade for peer cards, authority leases, RUDP endpoint parsing, and peer-card RUDP client construction | Full CultMesh game-session/server facade, stream catalog, or production game-session server | | Kotlin | JVM/Android client substrate | Typed MessagePack cache, built-in schema catalog advertisement, schema/shard catalogs with shared schema-message transport fetch/sync helpers, authority leases, stream catalog, branded `CultMesh` facade, WebSocket transport profile/connection helper, RUDP client/server sugar, RUDP reconnect policy/controller and reconnect-loop helper, schema-v0 MessagePack over RUDP evidence | C# SoA store, full C# server daemon, or Unity hot-loop memory layout | diff --git a/packages/cultcache-py/PARITY.md b/packages/cultcache-py/PARITY.md index d0c1a9e7..d6c61a31 100644 --- a/packages/cultcache-py/PARITY.md +++ b/packages/cultcache-py/PARITY.md @@ -18,7 +18,10 @@ CultCache, CultNet, and CultMesh. It is evidence, not confetti. database subscription changes. Request-response reads and database subscription streams raise typed `CultNetPeerError` values for `cultnet.error.v0` peer responses while preserving the raw error document for - callers that need the wire evidence. + callers that need the wire evidence. `CultNetRawClient` and database + 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. - Python CultNet raw snapshot: Provides typed `CultNetRawSnapshotResponse` and `CultNetRawDocumentRecord` helpers plus `CultNetRawClient.fetch_snapshot_response(...)` for inspecting @@ -289,11 +292,13 @@ node --test packages\cultnet-ts\dist-test\test\interop\cultnet-interop.test.js raw client, and interop peer still use the TCP-framed schema lane for many service and harness flows, while `docs/cultnet-transport-parity.md` tracks the remaining work to put broader server-side bodies behind the shared transport - port instead of cloning LiteNetLib. 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. + 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. - 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/src/cultnet_py/__init__.py b/packages/cultcache-py/src/cultnet_py/__init__.py index db23a0d6..5a0b567c 100644 --- a/packages/cultcache-py/src/cultnet_py/__init__.py +++ b/packages/cultcache-py/src/cultnet_py/__init__.py @@ -1,4 +1,11 @@ -from .client import CultNetDatabaseSubscription, CultNetPeerError, CultNetRawClient +from .client import ( + CultNetDatabaseSubscription, + CultNetPeerError, + CultNetRawClient, + CultNetSchemaTransport, + CultNetSchemaTransportFactory, + create_tcp_framed_schema_transport, +) from .framing import decode_frame, encode_frame, read_frame, write_frame from .messages import ( CultNetMessage, @@ -105,6 +112,8 @@ __all__ = [ "CultNetDatabaseChange", "CultNetPeerError", "CultNetRawClient", + "CultNetSchemaTransport", + "CultNetSchemaTransportFactory", "CultNetClientSecurityOptions", "CultNetSecret", "CultNetServerSecurityOptions", @@ -161,6 +170,7 @@ __all__ = [ "compute_simulation_claim_hash", "compute_reconnect_delay_ms", "create_reconnect_policy", + "create_tcp_framed_schema_transport", "create_rudp_transport_profile", "create_tcp_framed_transport_profile", "database_subscribe", diff --git a/packages/cultcache-py/src/cultnet_py/client.py b/packages/cultcache-py/src/cultnet_py/client.py index 9fafcc5c..8177219a 100644 --- a/packages/cultcache-py/src/cultnet_py/client.py +++ b/packages/cultcache-py/src/cultnet_py/client.py @@ -1,8 +1,9 @@ from __future__ import annotations import socket +from collections.abc import Callable from dataclasses import dataclass -from typing import Any +from typing import Any, Protocol import msgpack @@ -23,6 +24,17 @@ from .subscription import CultNetDatabaseChange from .transport import TcpFramedTransportConnection, create_tcp_framed_transport_profile +class CultNetSchemaTransport(Protocol): + def __enter__(self) -> "CultNetSchemaTransport": ... + def __exit__(self, exc_type: Any, exc: Any, traceback: Any) -> None: ... + def send(self, channel_id: str, payload: bytes) -> None: ... + def receive(self) -> Any: ... + def close(self) -> None: ... + + +CultNetSchemaTransportFactory = Callable[[], CultNetSchemaTransport] + + class CultNetPeerError(RuntimeError): def __init__(self, response: dict[str, Any]) -> None: self.response = dict(response) @@ -35,27 +47,16 @@ class CultNetRawClient: host: str port: int timeout_seconds: float = 4.0 + create_transport: CultNetSchemaTransportFactory | None = None def send(self, message: CultNetMessage | dict[str, Any]) -> None: wire = message.to_wire() if isinstance(message, CultNetMessage) else message - with socket.create_connection((self.host, self.port), timeout=self.timeout_seconds) as connection: - connection.settimeout(self.timeout_seconds) - stream = connection.makefile("rwb") - transport = TcpFramedTransportConnection( - stream, - profile=create_tcp_framed_transport_profile("cultnet-python-client", host=self.host, port=self.port), - ) + with self.open_transport() as transport: transport.send("schema", msgpack.packb(wire, use_bin_type=True)) def request(self, message: CultNetMessage | dict[str, Any], *, expected_schema_version: str) -> dict[str, Any]: wire = message.to_wire() if isinstance(message, CultNetMessage) else message - with socket.create_connection((self.host, self.port), timeout=self.timeout_seconds) as connection: - connection.settimeout(self.timeout_seconds) - stream = connection.makefile("rwb") - transport = TcpFramedTransportConnection( - stream, - profile=create_tcp_framed_transport_profile("cultnet-python-client", host=self.host, port=self.port), - ) + with self.open_transport() as transport: transport.send("schema", msgpack.packb(wire, use_bin_type=True)) response = msgpack.unpackb(transport.receive().payload, raw=False) if not isinstance(response, dict): @@ -66,6 +67,16 @@ class CultNetRawClient: raise ValueError(f"Expected {expected_schema_version}, received {schema_version!r}") return response + def open_transport(self) -> CultNetSchemaTransport: + if self.create_transport is not None: + return self.create_transport() + return create_tcp_framed_schema_transport( + host=self.host, + port=self.port, + timeout_seconds=self.timeout_seconds, + runtime_id="cultnet-python-client", + ) + def fetch_schema_catalog( self, *, @@ -245,6 +256,7 @@ class CultNetRawClient: host=self.host, port=self.port, timeout_seconds=self.timeout_seconds, + create_transport=self.create_transport, subscription_id=subscription_id, message_id=message_id, schema_ids=schema_ids, @@ -263,17 +275,19 @@ class CultNetDatabaseSubscription: schema_ids: list[str] | None = None record_keys: list[str] | None = None include_snapshot: bool = True - _connection: socket.socket | None = None - _stream: Any | None = None - _transport: TcpFramedTransportConnection | None = None + create_transport: CultNetSchemaTransportFactory | None = None + _transport: CultNetSchemaTransport | None = None def __enter__(self) -> "CultNetDatabaseSubscription": - self._connection = socket.create_connection((self.host, self.port), timeout=self.timeout_seconds) - self._connection.settimeout(self.timeout_seconds) - self._stream = self._connection.makefile("rwb") - self._transport = TcpFramedTransportConnection( - self._stream, - profile=create_tcp_framed_transport_profile("cultnet-python-subscription", host=self.host, port=self.port), + self._transport = ( + self.create_transport() + if self.create_transport is not None + else create_tcp_framed_schema_transport( + host=self.host, + port=self.port, + timeout_seconds=self.timeout_seconds, + runtime_id="cultnet-python-subscription", + ) ) self.send(database_subscribe( message_id=self.message_id, @@ -286,7 +300,7 @@ class CultNetDatabaseSubscription: def __exit__(self, exc_type: Any, exc: Any, traceback: Any) -> None: try: - if self._stream is not None: + if self._transport is not None: self.send(database_unsubscribe( message_id=f"{self.message_id}-unsubscribe", subscription_id=self.subscription_id, @@ -295,11 +309,6 @@ class CultNetDatabaseSubscription: if self._transport is not None: self._transport.close() self._transport = None - if self._stream is not None: - self._stream = None - if self._connection is not None: - self._connection.close() - self._connection = None def send(self, message: CultNetMessage | dict[str, Any]) -> None: if self._transport is None: @@ -349,3 +358,32 @@ class CultNetDatabaseSubscription: def _raise_peer_error(response: dict[str, Any]) -> None: if response.get("schemaVersion") == "cultnet.error.v0": raise CultNetPeerError(response) + + +def create_tcp_framed_schema_transport( + *, + host: str, + port: int, + timeout_seconds: float = 4.0, + runtime_id: str = "cultnet-python-client", +) -> TcpFramedTransportConnection: + connection = socket.create_connection((host, port), timeout=timeout_seconds) + connection.settimeout(timeout_seconds) + stream = connection.makefile("rwb") + return _OwnedTcpFramedTransportConnection( + stream, + connection, + profile=create_tcp_framed_transport_profile(runtime_id, host=host, port=port), + ) + + +class _OwnedTcpFramedTransportConnection(TcpFramedTransportConnection): + def __init__(self, stream: Any, connection: socket.socket, *, profile: dict[str, Any]) -> None: + super().__init__(stream, profile=profile) + self._connection = connection + + def close(self) -> None: + try: + super().close() + finally: + self._connection.close() diff --git a/packages/cultcache-py/src/cultnet_py/transport.py b/packages/cultcache-py/src/cultnet_py/transport.py index 7393cb3f..aabc0782 100644 --- a/packages/cultcache-py/src/cultnet_py/transport.py +++ b/packages/cultcache-py/src/cultnet_py/transport.py @@ -1026,6 +1026,12 @@ class TcpFramedTransportConnection: self._frames_received = 0 self._frames_sent = 0 + def __enter__(self) -> "TcpFramedTransportConnection": + return self + + def __exit__(self, exc_type: Any, exc: Any, traceback: Any) -> None: + self.close() + @property def stats(self) -> CultNetTransportStats: return CultNetTransportStats( diff --git a/packages/cultcache-py/tests/test_cultcache.py b/packages/cultcache-py/tests/test_cultcache.py index d7460f14..5bda8491 100644 --- a/packages/cultcache-py/tests/test_cultcache.py +++ b/packages/cultcache-py/tests/test_cultcache.py @@ -626,6 +626,48 @@ class CultCacheTests(unittest.TestCase): self.assertEqual(receiver.stats.bytes_received, len(payload) + 4) self.assertEqual(receiver.profile["transports"][0]["protocol"], "tcp_framed") + def test_cultnet_raw_client_uses_schema_transport_factory(self) -> None: + import msgpack # type: ignore + + sent_messages: list[dict[str, object]] = [] + test_case = self + + class FakeSchemaTransport: + profile = create_tcp_framed_transport_profile("fake-client") + + def __enter__(self) -> "FakeSchemaTransport": + return self + + def __exit__(self, exc_type: object, exc: object, traceback: object) -> None: + self.close() + + def send(self, channel_id: str, payload: bytes) -> None: + test_case.assertEqual(channel_id, "schema") + sent_messages.append(msgpack.unpackb(payload, raw=False)) + + def receive(self) -> object: + response = { + "schemaVersion": "cultnet.schema_catalog_response.v0", + "messageId": "transport-factory", + "schemas": [], + } + return type("Frame", (), {"channel_id": "schema", "payload": msgpack.packb(response, use_bin_type=True)})() + + def close(self) -> None: + pass + + client = CultNetRawClient( + "unused.example.test", + 1, + create_transport=FakeSchemaTransport, + ) + + response = client.fetch_schema_catalog(message_id="transport-factory") + + self.assertEqual(response["schemaVersion"], "cultnet.schema_catalog_response.v0") + self.assertEqual(sent_messages[0]["schemaVersion"], "cultnet.schema_catalog_request.v0") + self.assertEqual(sent_messages[0]["messageId"], "transport-factory") + def test_cultnet_rudp_packet_codec_uses_deterministic_reliable_ordered_fixture(self) -> None: encoded = encode_rudp_packet( CultNetRudpPacket(