Route TS interop through transport port

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

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 {

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();
   }
 }

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<string, string>): Promise<void> {
 
   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<string, string>): Promise<void> {
   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<string, string>): Promise<void> {
     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<string, unknown>): void {
   process.stderr.write(`${JSON.stringify({ event, ...payload })}\n`);
 }
 
+function writeTrace(event: string, payload: Record<string, unknown>): void {
+  if (process.env.CULTNET_INTEROP_TRACE === "1") {
+    writeLog(event, payload);
+  }
+}
+
 function runtimeStorePath(runtimeId: string): string {
   return join(tmpdir(), `cultnet-ts-interop-${runtimeId}.msgpack`);
 }

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<void> | undefined;
 let csharpInteropPeerBuild: Promise<void> | undefined;
 let kotlinInteropPeerBuild: Promise<void> | 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<R
   let stdoutBuffer = "";
 
   const ready = new Promise<unknown>((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<R
         try {
           const parsed = JSON.parse(line) as { status?: string };
           if (parsed.status === "ready") {
-            resolve(parsed);
+            resolveReady(parsed);
             return;
           }
         } catch (error) {
-          reject(new Error(`${name} emitted non-JSON stdout while starting: ${line}`));
+          rejectReady(new Error(`${name} emitted non-JSON stdout while starting: ${line}`));
           return;
         }
       }
@@ -2125,9 +2160,9 @@ async function spawnServeProcess(name: string, command: ServeCommand): Promise<R
     });
 
     child.once("exit", (code, signal) => {
-      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<strin
 }
 
 async function stopProcess(processState: RunningServeProcess): Promise<void> {
-  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<number> {