From fad79906be6cfb26b44b0e3c8ac8d6ba9e12ba3e Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Tue, 16 Jun 2026 09:51:00 +0100 Subject: [PATCH] Route TS interop through transport port --- docs/cultnet-transport-parity.md | 7 +- packages/cultnet-ts/src/schema-discovery.ts | 20 ++++-- packages/cultnet-ts/src/transport.ts | 1 + .../test/interop/cultnet-interop-peer.ts | 43 ++++++++++-- .../test/interop/cultnet-interop.test.ts | 68 ++++++++++++++++--- 5 files changed, 119 insertions(+), 20 deletions(-) diff --git a/docs/cultnet-transport-parity.md b/docs/cultnet-transport-parity.md index eb0bbbb8..985397a3 100644 --- a/docs/cultnet-transport-parity.md +++ b/docs/cultnet-transport-parity.md @@ -158,8 +158,11 @@ Current progress: - TypeScript has the first narrow transport connection port: `TcpFramedTransportConnection` owns length-prefixed frame delivery, exposes frame/close/error events, `send(channel, payload)`, `close`, and transfer - stats, and `CultNetPeer` can now run over either the legacy `Duplex` stream - path or the transport connection path. + stats, and destroys its owned stream on close. `CultNetPeer` can still run + over either the compatibility `Duplex` stream path or the transport + connection path, while the TypeScript interop peer now accepts and dials TCP + peers through `TcpFramedTransportConnection` so the parity harness exercises + the shared port instead of raw socket framing. - Python now has the same `tcp_framed` transport profile helper plus a synchronous `TcpFramedTransportConnection` with `send`, `receive`, `close`, and stats. `CultNetRawClient`, database subscriptions, and the local diff --git a/packages/cultnet-ts/src/schema-discovery.ts b/packages/cultnet-ts/src/schema-discovery.ts index 61a92aa7..103572c6 100644 --- a/packages/cultnet-ts/src/schema-discovery.ts +++ b/packages/cultnet-ts/src/schema-discovery.ts @@ -101,16 +101,26 @@ function toSchemaDescriptor( entry: CultNetRegisteredSchemaRecord, includeSchemaJson: boolean, ): CultNetSchemaDescriptor { - return { + const descriptor: CultNetSchemaDescriptor = { schemaId: entry.schemaId, kind: entry.kind, - schemaVersion: entry.schemaVersion, - documentType: entry.documentType, - title: entry.title, wireContracts: [...entry.wireContracts], contentHash: entry.contentHash, - schemaJson: includeSchemaJson ? entry.canonicalSchemaJson : undefined, }; + if (entry.schemaVersion !== undefined) { + descriptor.schemaVersion = entry.schemaVersion; + } + if (entry.documentType !== undefined) { + descriptor.documentType = entry.documentType; + } + if (entry.title !== undefined) { + descriptor.title = entry.title; + } + if (includeSchemaJson) { + descriptor.schemaJson = entry.canonicalSchemaJson; + } + + return descriptor; } function stableStringify(value: unknown): string { diff --git a/packages/cultnet-ts/src/transport.ts b/packages/cultnet-ts/src/transport.ts index 60c14445..53f64127 100644 --- a/packages/cultnet-ts/src/transport.ts +++ b/packages/cultnet-ts/src/transport.ts @@ -228,5 +228,6 @@ export class TcpFramedTransportConnection extends EventEmitter implements CultNe close(): void { this.#stream.end(); + this.#stream.destroy(); } } diff --git a/packages/cultnet-ts/test/interop/cultnet-interop-peer.ts b/packages/cultnet-ts/test/interop/cultnet-interop-peer.ts index 0d9571cb..8d031744 100644 --- a/packages/cultnet-ts/test/interop/cultnet-interop-peer.ts +++ b/packages/cultnet-ts/test/interop/cultnet-interop-peer.ts @@ -11,6 +11,7 @@ import { CultNetDocumentRegistry, CultNetPeer, CultNetSchemaRegistry, + TcpFramedTransportConnection, cultNetBuiltinSchemaRegistry, createTcpFramedTransportProfile, defineCultNetDocumentBinding, @@ -132,10 +133,19 @@ async function serve(args: Map): Promise { const tcpServer = createServer(); tcpServer.on("connection", (socket) => { - const peer = new CultNetPeer(socket, { wireContract: INTEROP_WIRE_CONTRACT }); + const transport = new TcpFramedTransportConnection( + socket, + createTcpFramedTransportProfile(runtimeId, { + transportId: "interop-tcp", + host: advertiseHost, + port: tcpPort, + }), + ); + const peer = new CultNetPeer(transport, { wireContract: INTEROP_WIRE_CONTRACT }); peer.on("invalidMessage", (error) => writeLog("invalidMessage", { runtimeId, error: error.message })); peer.on("error", (error) => writeLog("tcpError", { runtimeId, error: error.message })); peer.on("message", async (message) => { + writeTrace("serveMessage", { runtimeId, schemaVersion: message.schemaVersion }); try { await handleServerMessage({ peer, @@ -529,8 +539,18 @@ async function dial(args: Map): Promise { const documentRegistry = new CultNetDocumentRegistry(Object.values(documents)); const socket = await connectTo(targetHost, targetPort); - const peer = new CultNetPeer(socket, { wireContract: INTEROP_WIRE_CONTRACT }); + const transport = new TcpFramedTransportConnection( + socket, + createTcpFramedTransportProfile(runtimeId, { + transportId: "interop-tcp", + host: targetHost, + port: targetPort, + }), + ); + const peer = new CultNetPeer(transport, { wireContract: INTEROP_WIRE_CONTRACT }); + writeTrace("dialSendHello", { runtimeId }); + const remoteHelloWait = waitForMessage(peer, (message) => message.schemaVersion === "cultnet.hello.v0", timeoutMs); peer.sendHello({ schemaVersion: "cultnet.hello.v0", runtimeId, @@ -549,23 +569,30 @@ async function dial(args: Map): Promise { supportedMessageVersions: [INTEROP_SCHEMA_VERSION], supportsSchemaCatalog: true, }); - const remoteHello = await waitForMessage(peer, (message) => message.schemaVersion === "cultnet.hello.v0", timeoutMs); + const remoteHello = await remoteHelloWait; + writeTrace("dialReceivedHello", { runtimeId, remoteRuntimeId: remoteHello.runtimeId }); const catalogRequest: CultNetSchemaCatalogRequestMessage = { schemaVersion: "cultnet.schema_catalog_request.v0", messageId: `${runtimeId}-catalog`, includeSchemaJson: true, }; + const catalogResponseWait = waitForMessage(peer, (message) => message.schemaVersion === "cultnet.schema_catalog_response.v0", timeoutMs); + writeTrace("dialSendCatalogRequest", { runtimeId }); peer.sendSchemaCatalogRequest(catalogRequest); - const catalogResponse = await waitForMessage(peer, (message) => message.schemaVersion === "cultnet.schema_catalog_response.v0", timeoutMs); + const catalogResponse = await catalogResponseWait; + writeTrace("dialReceivedCatalogResponse", { runtimeId }); const snapshotRequest: CultNetSnapshotRequestMessage = { schemaVersion: "cultnet.snapshot_request.v0", messageId: `${runtimeId}-snapshot`, schemaIds: [interopSchema.schemaId], }; + const snapshotResponseWait = waitForMessage(peer, (message) => message.schemaVersion === "cultnet.snapshot_response_raw.v0", timeoutMs); + writeTrace("dialSendSnapshotRequest", { runtimeId }); peer.sendSnapshotRequest(snapshotRequest); - const snapshotResponse = await waitForMessage(peer, (message) => message.schemaVersion === "cultnet.snapshot_response_raw.v0", timeoutMs); + const snapshotResponse = await snapshotResponseWait; + writeTrace("dialReceivedSnapshotResponse", { runtimeId }); await documentRegistry.applyRawSnapshotResponse(cache, snapshotResponse as CultNetSnapshotResponseRawMessage); const note = cache.getRequired(documents.note.definition, `note:${(remoteHello as { runtimeId: string }).runtimeId}`) as InteropNote; @@ -774,6 +801,12 @@ function writeLog(event: string, payload: Record): void { process.stderr.write(`${JSON.stringify({ event, ...payload })}\n`); } +function writeTrace(event: string, payload: Record): void { + if (process.env.CULTNET_INTEROP_TRACE === "1") { + writeLog(event, payload); + } +} + function runtimeStorePath(runtimeId: string): string { return join(tmpdir(), `cultnet-ts-interop-${runtimeId}.msgpack`); } diff --git a/packages/cultnet-ts/test/interop/cultnet-interop.test.ts b/packages/cultnet-ts/test/interop/cultnet-interop.test.ts index 6510d0d9..8b6cae30 100644 --- a/packages/cultnet-ts/test/interop/cultnet-interop.test.ts +++ b/packages/cultnet-ts/test/interop/cultnet-interop.test.ts @@ -1,5 +1,5 @@ import assert from "node:assert/strict"; -import { execFile, spawn, type ChildProcessByStdio } from "node:child_process"; +import { execFile, spawn, spawnSync, type ChildProcessByStdio } from "node:child_process"; import { createHash } from "node:crypto"; import dgram, { type Socket } from "node:dgram"; import { once } from "node:events"; @@ -20,7 +20,7 @@ const execFileAsync = promisify(execFile); const cargoCommand = process.env.CARGO ?? (process.platform === "win32" ? join(homedir(), ".cargo", "bin", "cargo.exe") : "cargo"); const dotnetCommand = process.env.DOTNET ?? (process.platform === "win32" ? join("C:", "Program Files", "dotnet", "dotnet.exe") : "dotnet"); -const pythonCommand = process.env.PYTHON ?? "python"; +const pythonCommand = process.env.PYTHON ?? resolvePythonCommand(); const cultNetTsRoot = resolve(__dirname, "../../.."); const cultLibRoot = findAncestor(cultNetTsRoot, "CultLib.sln") ?? resolve(cultNetTsRoot, "..", "CultLib"); const cultcachePyRoot = resolve(cultLibRoot, "packages", "cultcache-py"); @@ -68,6 +68,27 @@ let rustInteropPeerBuild: Promise | undefined; let csharpInteropPeerBuild: Promise | undefined; let kotlinInteropPeerBuild: Promise | undefined; +function resolvePythonCommand(): string { + const codexPythonCommand = join(homedir(), ".cache", "codex-runtimes", "codex-primary-runtime", "dependencies", "python", "python.exe"); + const candidates = process.platform === "win32" + ? [ + ...(existsSync(codexPythonCommand) ? [codexPythonCommand] : []), + "py", + "python", + "python3", + ] + : ["python3", "python"]; + + for (const candidate of candidates) { + const result = spawnSync(candidate, ["--version"], { stdio: "ignore" }); + if (result.status === 0) { + return candidate; + } + } + + return process.platform === "win32" ? "python" : "python3"; +} + interface RudpServerInteropPeerSpec { name: string; connectionId: number; @@ -2091,6 +2112,20 @@ async function spawnServeProcess(name: string, command: ServeCommand): Promise((resolve, reject) => { + const timeout = setTimeout(() => { + reject(new Error(`${name} serve process did not become ready within 30000ms.\n${stderr.join("")}`)); + }, 30_000); + + const resolveReady = (value: unknown) => { + clearTimeout(timeout); + resolve(value); + }; + + const rejectReady = (error: Error) => { + clearTimeout(timeout); + reject(error); + }; + child.stdout.setEncoding("utf8"); child.stdout.on("data", (chunk: string) => { stdoutBuffer += chunk; @@ -2109,11 +2144,11 @@ async function spawnServeProcess(name: string, command: ServeCommand): Promise { - reject(new Error(`${name} serve process exited before becoming ready (code=${code}, signal=${signal}).\n${stderr.join("")}`)); + rejectReady(new Error(`${name} serve process exited before becoming ready (code=${code}, signal=${signal}).\n${stderr.join("")}`)); }); - child.once("error", reject); + child.once("error", rejectReady); }); return { name, child, ready, stderr }; @@ -3509,12 +3544,29 @@ async function putAndSnapshotPythonWitnessBundle(port: number, put: Record { - if (processState.child.killed || processState.child.exitCode !== null) { + if (processState.child.exitCode !== null) { return; } + const gracefulExit = once(processState.child, "exit").then(() => true); processState.child.kill("SIGTERM"); - await once(processState.child, "exit"); + const exited = await Promise.race([ + gracefulExit, + delay(2_000).then(() => false), + ]); + if (exited || processState.child.exitCode !== null) { + return; + } + + const forcedExit = once(processState.child, "exit").then(() => true); + processState.child.kill("SIGKILL"); + const killed = await Promise.race([ + forcedExit, + delay(2_000).then(() => false), + ]); + if (!killed && processState.child.exitCode === null) { + throw new Error(`${processState.name} serve process did not exit after SIGKILL.`); + } } async function getFreePort(): Promise {