fix(codex): preserve JSONL framing across WebSocket writes (#110384)

* fix(codex): preserve UTF-8 websocket frames

* test(codex): satisfy websocket transport lint

* fix(codex): preserve websocket JSONL framing

extensions/codex/src/app-server/transport-websocket.test.ts

@@ -6,9 +6,11 @@ import path from "node:path";
 import { afterEach, describe, expect, it } from "vitest";
 import { WebSocketServer, type RawData } from "ws";
 import { CodexAppServerClient } from "./client.js";
+import { createWebSocketTransport } from "./transport-websocket.js";
 
 describe("Codex app-server websocket transport", () => {
   const clients: CodexAppServerClient[] = [];
+  const transports: Array<ReturnType<typeof createWebSocketTransport>> = [];
   const servers: WebSocketServer[] = [];
   const httpServers: http.Server[] = [];
   const tempDirs: string[] = [];
@@ -18,6 +20,10 @@ describe("Codex app-server websocket transport", () => {
       client.close();
     }
     clients.length = 0;
+    for (const transport of transports.splice(0)) {
+      transport.kill?.();
+      transport.stdin.destroy?.();
+    }
     await Promise.all(
       servers.splice(0).map(
         (server) =>
@@ -75,6 +81,79 @@ describe("Codex app-server websocket transport", () => {
     expect(authHeaders).toEqual(["Bearer secret"]);
   });
 
+  it("preserves UTF-8 JSON-RPC bytes split across writable chunks", async () => {
+    const server = new WebSocketServer({ host: "127.0.0.1", port: 0 });
+    servers.push(server);
+    let resolveConnection: (() => void) | undefined;
+    const connected = new Promise<void>((resolve) => {
+      resolveConnection = resolve;
+    });
+    let resolveMessage: ((message: string) => void) | undefined;
+    const message = new Promise<string>((resolve) => {
+      resolveMessage = resolve;
+    });
+    server.once("connection", (socket) => {
+      socket.once("message", (data) => resolveMessage?.(rawDataToText(data)));
+      resolveConnection?.();
+    });
+    await new Promise<void>((resolve) => {
+      server.once("listening", resolve);
+    });
+    const address = server.address();
+    if (!address || typeof address === "string") {
+      throw new Error("expected websocket test server port");
+    }
+    const transport = createWebSocketTransport({
+      transport: "websocket",
+      command: "codex",
+      args: [],
+      url: `ws://127.0.0.1:${address.port}`,
+      headers: {},
+    });
+    transports.push(transport);
+    await connected;
+    const frame = Buffer.from('{"jsonrpc":"2.0","method":"😀"}\n');
+    const emojiStart = frame.indexOf(Buffer.from("😀"));
+    transport.stdin.write(frame.subarray(0, emojiStart + 2));
+    transport.stdin.write(frame.subarray(emojiStart + 2));
+    await expect(message).resolves.toBe('{"jsonrpc":"2.0","method":"😀"}');
+  });
+
+  it("flushes an unterminated JSON-RPC frame when stdin finishes", async () => {
+    const server = new WebSocketServer({ host: "127.0.0.1", port: 0 });
+    servers.push(server);
+    let resolveMessage: ((message: string) => void) | undefined;
+    const message = new Promise<string>((resolve) => {
+      resolveMessage = resolve;
+    });
+    server.once("connection", (socket) => {
+      socket.once("message", (data) => resolveMessage?.(rawDataToText(data)));
+      socket.send("{}");
+    });
+    await new Promise<void>((resolve) => {
+      server.once("listening", resolve);
+    });
+    const address = server.address();
+    if (!address || typeof address === "string") {
+      throw new Error("expected websocket test server port");
+    }
+    const transport = createWebSocketTransport({
+      transport: "websocket",
+      command: "codex",
+      args: [],
+      url: `ws://127.0.0.1:${address.port}`,
+      headers: {},
+    });
+    transports.push(transport);
+    const clientReady = new Promise<void>((resolve) => {
+      transport.stdout.once("data", () => resolve());
+    });
+    await clientReady;
+    transport.stdin.write('{"jsonrpc":"2.0","method":"final"}');
+    transport.stdin.end?.();
+    await expect(message).resolves.toBe('{"jsonrpc":"2.0","method":"final"}');
+  }, 5_000);
+
   it("can speak JSON-RPC over the canonical unix control socket", async () => {
     const tempDir = await mkdtemp(path.join(os.tmpdir(), "openclaw-codex-unix-"));
     tempDirs.push(tempDir);

extensions/codex/src/app-server/transport-websocket.ts

@@ -6,6 +6,7 @@ import { EventEmitter } from "node:events";
 import net from "node:net";
 import path from "node:path";
 import { PassThrough, Writable } from "node:stream";
+import { StringDecoder } from "node:string_decoder";
 import WebSocket, { type RawData } from "ws";
 import { resolveCodexAppServerUserHomeDir, type CodexAppServerStartOptions } from "./config.js";
 import type { CodexAppServerTransport } from "./transport.js";
@@ -39,6 +40,8 @@ export function createWebSocketTransport(
       })
     : new WebSocket(options.url, websocketOptions);
   const pendingFrames: string[] = [];
+  const stdinDecoder = new StringDecoder("utf8");
+  let pendingLine = "";
   let killed = false;
 
   const sendFrame = (frame: string) => {
@@ -72,11 +75,22 @@ export function createWebSocketTransport(
 
   const stdin = new Writable({
     write(chunk, _encoding, callback) {
-      for (const frame of chunk.toString("utf8").split("\n")) {
+      pendingLine += stdinDecoder.write(chunk);
+      const lines = pendingLine.split("\n");
+      pendingLine = lines.pop() ?? "";
+      for (const frame of lines) {
         sendFrame(frame);
       }
       callback();
     },
+    final(callback) {
+      pendingLine += stdinDecoder.end();
+      if (pendingLine) {
+        sendFrame(pendingLine);
+      }
+      pendingLine = "";
+      callback();
+    },
   });
   const closeSocket = () => {
     if (socket.readyState === WebSocket.CLOSED || socket.readyState === WebSocket.CLOSING) {

extensions/codex/src/app-server/transport.ts

@@ -5,7 +5,7 @@
 /** Child-process-like transport shape consumed by the Codex app-server client. */
 export type CodexAppServerTransport = {
   stdin: {
-    write: (data: string, callback?: (error?: Error | null) => void) => unknown;
+    write: (data: string | Uint8Array, callback?: (error?: Error | null) => void) => unknown;
     end?: () => unknown;
     destroy?: () => unknown;
     unref?: () => unknown;