From 409a42e0f7b06af243349d1372e0e706ad2005b5 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Tue, 16 Jun 2026 01:42:02 +0100 Subject: [PATCH] Prove Python schema messages over RUDP --- docs/cultnet-transport-parity.md | 9 +- docs/runtime-parity-scope.md | 10 +- .../test/interop/cultnet-interop.test.ts | 302 +++++++++++++++++- 3 files changed, 311 insertions(+), 10 deletions(-) diff --git a/docs/cultnet-transport-parity.md b/docs/cultnet-transport-parity.md index 06bdb5e2..2e50471e 100644 --- a/docs/cultnet-transport-parity.md +++ b/docs/cultnet-transport-parity.md @@ -181,6 +181,11 @@ Current progress: UDP peers, C#/Kotlin/Python/Rust can dial a TypeScript UDP peer, and all sides exchange reliable ordered `schema` frames through the shared handshake and packet language. +- The same harness now proves schema-v0 MessagePack messages over RUDP in both + directions for C#, Kotlin, and Python peers against TypeScript. Those tests + send `cultnet.schema_catalog_request.v0` payloads through reliable ordered + `schema` frames and validate `cultnet.hello.v0` responses with advertised + RUDP transport profiles where the responding runtime owns them. - The same harness now proves reliable ordered `schema` delivery survives one dropped data packet in both directions between TypeScript and C#/Kotlin/Python/Rust. The drops happen below the transport connection through a UDP bridge, so the @@ -234,8 +239,8 @@ Current progress: creating a second authority path. - The remaining parity work is to deepen Python's server-side use of its port and move each runtime's existing TCP/LiteNetLib/WebSocket bodies behind those - ports, then broaden RUDP interop into reconnect policy and the wider - schema-message matrix. + ports, then broaden RUDP interop into reconnect policy, Rust schema-v0 + MessagePack-over-RUDP coverage, and the wider schema-message matrix. ## RUDP Packet Contract V0 diff --git a/docs/runtime-parity-scope.md b/docs/runtime-parity-scope.md index b3367cfa..3089e457 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, shared transport profiles, shared RUDP socket transport, 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, 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, local TCP CultMesh server, session/prediction/simulation fact helpers, authority leases, streams, RUDP codec/session/socket transport, 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, transport profiles, TCP-framed transport port, shared RUDP socket transport, local CultMesh node/catalog/authority/stream ergonomics | Full C# game session/server stack | | Rust | Low-level cache/transport/runtime substrate | CultCache/CultNet interop lane, transport profiles, TCP-framed port, shared RUDP codec/session/socket transport | Full CultMesh ergonomic facade or production game-session server | | Kotlin | JVM/Android client substrate | Typed MessagePack cache, schema/shard catalogs, authority leases, stream catalog, branded `CultMesh` facade, RUDP client/server sugar, schema-v0 MessagePack over RUDP evidence | C# SoA store, full C# server daemon, or Unity hot-loop memory layout | @@ -42,8 +42,7 @@ CultMesh. wire harness. It covers schema-v0 interop and RUDP exchange across C#, TypeScript, Rust, Python, and Kotlin, including loss, reordering, fragmentation, disconnect reason propagation, ping/pong, and schema-v0 - MessagePack over RUDP for the runtimes currently wired into that message - lane. + MessagePack over RUDP for C#, Python, and Kotlin peers against TypeScript. - `packages/cultcache-py/PARITY.md` is the Python-specific audit ledger. Its "Still Not Claimed" section is part of the contract, not an apology. - `docs/cultnet-transport-parity.md` is the transport ownership map. It tracks @@ -57,9 +56,8 @@ CultMesh. - Broaden service adoption so local servers and daemon paths use the shared transport port consistently instead of keeping raw TCP/LiteNetLib/WebSocket ownership in individual bodies. -- Extend RUDP interop evidence through reconnect policy and the wider - schema-message matrix, not only the current schema-frame and selected - schema-v0 message paths. +- Extend RUDP interop evidence through reconnect policy, Rust schema-v0 + MessagePack-over-RUDP coverage, and the wider schema-message matrix. - Keep Kotlin ergonomic parity moving with Android/JVM-shaped APIs instead of backfilling C# server responsibilities it should not own. - Keep performance claims evidence-bound: C# owns SoA; Python has benchmark diff --git a/packages/cultnet-ts/test/interop/cultnet-interop.test.ts b/packages/cultnet-ts/test/interop/cultnet-interop.test.ts index 8c08ccb9..1442f323 100644 --- a/packages/cultnet-ts/test/interop/cultnet-interop.test.ts +++ b/packages/cultnet-ts/test/interop/cultnet-interop.test.ts @@ -237,6 +237,135 @@ transport.close() raise SystemExit("timed out waiting for TypeScript RUDP frame") `; +const pythonRudpMessagePeerScript = String.raw` +import msgpack +import socket +import time + +from cultnet_py import ( + CultNetRudpSocketMode, + CultNetRudpSocketTransportConnection, + CultNetRudpSocketTransportOptions, + create_rudp_transport_profile, +) + +sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) +sock.bind(("127.0.0.1", 0)) +sock.settimeout(0.02) +transport = CultNetRudpSocketTransportConnection(CultNetRudpSocketTransportOptions( + runtime_id="python-rudp-message-interop", + socket=sock, + mode=CultNetRudpSocketMode.SERVER, + connection_id=0x55667789, + initial_sequence=100, + resend_delay_ms=25, +)) +print(f"PORT {sock.getsockname()[1]}", flush=True) + +deadline = time.time() + 5 +while time.time() < deadline: + frame = transport.receive_once() + transport.poll_resends() + if frame is None: + time.sleep(0.005) + continue + if frame.channel_id != "schema": + transport.close() + raise SystemExit(f"unexpected RUDP channel: {frame.channel_id!r}") + message = msgpack.unpackb(frame.payload, raw=False) + if message.get("schemaVersion") != "cultnet.schema_catalog_request.v0": + transport.close() + raise SystemExit(f"unexpected schema message: {message!r}") + response = { + "schemaVersion": "cultnet.hello.v0", + "runtimeId": "python-rudp-message-interop", + "runtimeKind": "python", + "supportedDocumentTypes": [], + "supportedMutationContracts": [], + "supportedMessageVersions": ["cultnet.hello.v0", "cultnet.schema_catalog_request.v0"], + "transportProfiles": [create_rudp_transport_profile( + "python-rudp-message-interop", + host="127.0.0.1", + port=sock.getsockname()[1], + )], + "supportsSchemaCatalog": True, + } + transport.send("schema", msgpack.packb(response, use_bin_type=True)) + print("OK", flush=True) + ack_deadline = time.time() + 0.25 + while time.time() < ack_deadline: + transport.receive_once() + transport.poll_resends() + time.sleep(0.005) + transport.close() + raise SystemExit(0) + +transport.close() +raise SystemExit("timed out waiting for RUDP schema-v0 message") +`; + +const pythonRudpMessageClientScript = String.raw` +import msgpack +import socket +import sys +import time + +from cultnet_py import ( + CultNetRudpSocketMode, + CultNetRudpSocketTransportConnection, + CultNetRudpSocketTransportOptions, +) + +remote_port = int(sys.argv[1]) +sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) +sock.bind(("127.0.0.1", 0)) +sock.settimeout(0.02) +transport = CultNetRudpSocketTransportConnection(CultNetRudpSocketTransportOptions( + runtime_id="python-rudp-message-client-interop", + socket=sock, + mode=CultNetRudpSocketMode.CLIENT, + remote_addr=("127.0.0.1", remote_port), + connection_id=0x66554434, + initial_sequence=1, + resend_delay_ms=25, +)) +transport.connect(b"python-message-join") + +sent = False +deadline = time.time() + 5 +while time.time() < deadline: + frame = transport.receive_once() + transport.poll_resends() + if transport.connected and not sent: + transport.send("schema", msgpack.packb({ + "schemaVersion": "cultnet.schema_catalog_request.v0", + "messageId": "python-ts-schema-message", + "includeSchemaJson": False, + "schemaIds": [], + "kinds": ["wire_message"], + }, use_bin_type=True)) + sent = True + if frame is None: + time.sleep(0.005) + continue + if frame.channel_id != "schema": + transport.close() + raise SystemExit(f"unexpected RUDP channel: {frame.channel_id!r}") + message = msgpack.unpackb(frame.payload, raw=False) + if message.get("schemaVersion") != "cultnet.hello.v0": + transport.close() + raise SystemExit(f"unexpected schema message: {message!r}") + if message.get("runtimeId") != "ts-python-rudp-message-server": + transport.close() + raise SystemExit(f"unexpected runtime id: {message!r}") + print("OK", flush=True) + transport.close() + raise SystemExit(0) + +transport.close() +raise SystemExit("timed out waiting for TypeScript RUDP schema-v0 message") +`; + test("CultNet TS/Rust/C#/Python peers discover each other and exchange raw state over the shared schema-v0 lane", async (t) => { await buildInteropPeers(); cleanInteropStores([ @@ -834,6 +963,112 @@ test("CultNet Python and TypeScript exchange schema frames when Python dials a T } }); +test("CultNet TypeScript and Python exchange schema-v0 MessagePack messages over RUDP", async () => { + const pythonPeer = spawnPythonRudpMessagePeer(); + const clientSocket = await bindUdpSocket(); + let client: CultNetRudpSocketTransportConnection | undefined; + + try { + const pythonPort = await pythonPeer.ready; + client = new CultNetRudpSocketTransportConnection({ + runtimeId: "ts-python-rudp-message-interop", + socket: clientSocket, + mode: "client", + remoteHost: "127.0.0.1", + remotePort: pythonPort, + connectionId: 0x55667789, + initialSequence: 1, + resendDelayMs: 25, + resendPollMs: 5, + }); + + const receivedFrame = once(client, "frame"); + client.connect(Buffer.from("ts-python-message-join")); + await waitFor(() => client?.connected === true, "TypeScript RUDP client to complete Python message handshake"); + client.send("schema", Buffer.from(encode({ + schemaVersion: "cultnet.schema_catalog_request.v0", + messageId: "ts-python-schema-message", + includeSchemaJson: false, + schemaIds: [], + kinds: ["wire_message"], + }))); + + const [frame] = await withTimeout(receivedFrame, 2_000, "Python RUDP schema-v0 response frame"); + assert.equal(frame.channelId, "schema"); + const message = decode(frame.payload) as { + schemaVersion?: string; + runtimeId?: string; + supportsSchemaCatalog?: boolean; + transportProfiles?: Array<{ transports?: Array<{ protocol?: string; channels?: Array<{ channelId?: string }> }> }>; + }; + assert.equal(message.schemaVersion, "cultnet.hello.v0"); + assert.equal(message.runtimeId, "python-rudp-message-interop"); + assert.equal(message.supportsSchemaCatalog, true); + assert.equal(message.transportProfiles?.[0]?.transports?.[0]?.protocol, "rudp"); + assert.equal(message.transportProfiles?.[0]?.transports?.[0]?.channels?.[0]?.channelId, "schema"); + + const [exitCode] = await withTimeout(once(pythonPeer.child, "exit"), 2_000, "Python RUDP message peer exit"); + assert.equal(exitCode, 0, pythonPeer.stderr.join("")); + } finally { + if (client) { + client.close(); + } else { + clientSocket.close(); + } + if (pythonPeer.child.exitCode === null && !pythonPeer.child.killed) { + pythonPeer.child.kill("SIGTERM"); + await once(pythonPeer.child, "exit").catch(() => undefined); + } + } +}); + +test("CultNet Python and TypeScript exchange schema-v0 MessagePack messages over RUDP when Python dials", async () => { + const serverSocket = await bindUdpSocket(); + const server = new CultNetRudpSocketTransportConnection({ + runtimeId: "ts-python-rudp-message-server", + socket: serverSocket, + mode: "server", + connectionId: 0x66554434, + initialSequence: 100, + resendDelayMs: 25, + resendPollMs: 5, + }); + const pythonClient = spawnPythonRudpMessageClient(udpSocketPort(serverSocket)); + + try { + const receivedFrame = once(server, "frame"); + const [frame] = await withTimeout(receivedFrame, 2_000, "Python RUDP schema-v0 client frame"); + assert.equal(frame.channelId, "schema"); + const message = decode(frame.payload) as { + schemaVersion?: string; + messageId?: string; + kinds?: string[]; + }; + assert.equal(message.schemaVersion, "cultnet.schema_catalog_request.v0"); + assert.equal(message.messageId, "python-ts-schema-message"); + assert.deepEqual(message.kinds, ["wire_message"]); + server.send("schema", Buffer.from(encode({ + schemaVersion: "cultnet.hello.v0", + runtimeId: "ts-python-rudp-message-server", + runtimeKind: "typescript", + supportedDocumentTypes: [], + supportedMutationContracts: [], + supportedMessageVersions: ["cultnet.hello.v0", "cultnet.schema_catalog_request.v0"], + transportProfiles: [], + supportsSchemaCatalog: true, + }))); + + const [exitCode] = await withTimeout(once(pythonClient.child, "exit"), 2_000, "Python RUDP message client exit"); + assert.equal(exitCode, 0, pythonClient.stderr.join("")); + } finally { + server.close(); + if (pythonClient.child.exitCode === null && !pythonClient.child.killed) { + pythonClient.child.kill("SIGTERM"); + await once(pythonClient.child, "exit").catch(() => undefined); + } + } +}); + test("CultNet TypeScript and Rust exchange schema frames over the shared RUDP socket transport", async () => { await buildRustInteropPeer(); const rustPeer = spawnRustRudpServer(); @@ -1046,7 +1281,7 @@ test("CultNet TypeScript and C# exchange schema-v0 MessagePack messages over RUD } }); -test("CultNet C# and TypeScript exchange schema-v0 MessagePack messages when C# dials", async () => { +test("CultNet C# and TypeScript exchange schema-v0 MessagePack messages over RUDP when C# dials", async () => { await buildCSharpInteropPeer(); const serverSocket = await bindUdpSocket(); const server = new CultNetRudpSocketTransportConnection({ @@ -1230,7 +1465,7 @@ test("CultNet TypeScript and Kotlin exchange schema-v0 MessagePack messages over } }); -test("CultNet Kotlin and TypeScript exchange schema-v0 MessagePack messages when Kotlin dials", async () => { +test("CultNet Kotlin and TypeScript exchange schema-v0 MessagePack messages over RUDP when Kotlin dials", async () => { await buildKotlinInteropPeer(); const serverSocket = await bindUdpSocket(); const server = new CultNetRudpSocketTransportConnection({ @@ -1873,6 +2108,54 @@ function spawnPythonRudpPeer(options: RudpServerSpawnOptions = {}): RunningPytho return { child, ready, stderr }; } +function spawnPythonRudpMessagePeer(): RunningPythonRudpPeer { + const child = spawn(pythonCommand, ["-c", pythonRudpMessagePeerScript], { + cwd: cultcachePyRoot, + env: { ...process.env, PYTHONPATH: cultcachePySrc }, + stdio: ["ignore", "pipe", "pipe"], + }); + const stderr: string[] = []; + let stdoutBuffer = ""; + + const ready = new Promise((resolveReady, rejectReady) => { + child.stdout.setEncoding("utf8"); + child.stdout.on("data", (chunk: string) => { + stdoutBuffer += chunk; + while (true) { + const newline = stdoutBuffer.indexOf("\n"); + if (newline === -1) { + break; + } + + const line = stdoutBuffer.slice(0, newline).trim(); + stdoutBuffer = stdoutBuffer.slice(newline + 1); + if (!line) { + continue; + } + + const portMatch = /^PORT\s+(\d+)$/.exec(line); + if (portMatch) { + resolveReady(Number(portMatch[1])); + continue; + } + if (line !== "OK") { + rejectReady(new Error(`Python RUDP message peer emitted unexpected stdout: ${line}`)); + } + } + }); + child.stderr.setEncoding("utf8"); + child.stderr.on("data", (chunk: string) => { + stderr.push(chunk); + }); + child.once("exit", (code, signal) => { + rejectReady(new Error(`Python RUDP message peer exited before publishing a port (code=${code}, signal=${signal}).\n${stderr.join("")}`)); + }); + child.once("error", rejectReady); + }); + + return { child, ready, stderr }; +} + function spawnRustRudpServer(options: RudpServerSpawnOptions = {}): RunningPythonRudpPeer { const child = spawn(rustBinaryPath, [ "rudp-serve-once", @@ -2174,6 +2457,21 @@ function spawnPythonRudpClient(remotePort: number): Omit { + const child = spawn(pythonCommand, ["-c", pythonRudpMessageClientScript, String(remotePort)], { + cwd: cultcachePyRoot, + env: { ...process.env, PYTHONPATH: cultcachePySrc }, + stdio: ["ignore", "pipe", "pipe"], + }); + const stderr: string[] = []; + child.stderr.setEncoding("utf8"); + child.stderr.on("data", (chunk: string) => { + stderr.push(chunk); + }); + + return { child, stderr }; +} + function spawnKotlinRudpClient(remotePort: number): Omit { const child = spawn(kotlinJavaCommand, [ "-cp", kotlinClasspath,