From 6c10e6fb7133578fb64c53ca858e281c4656e785 Mon Sep 17 00:00:00 2001 From: myk1yt Date: Sat, 31 Oct 2026 04:21:26 +0900 Subject: [PATCH 1/6] chore(ipc): drop the unused @roo-code/ipc client half What: - Delete packages/ipc/src/ipc-client.ts (147 lines) and its IpcClientEvents type in packages/types/src/ipc.ts. - Move the one live consumer, scripts/agent-evals/launcher.ts, onto a local EvalIpcClient in scripts/agent-evals/ipc-client.ts that speaks the same node-ipc framing the extension's IpcServer already uses. - Update packages/ipc exports, package.json description and README to the server-only surface. Why: The D4 premise was "unpublished, verified 404 on npm, so nothing consumes it". That holds for npm (404 re-confirmed) but missed an in-repo consumer: scripts/agent-evals/launcher.ts imported IpcClient. scripts/** is knip-ignored, which is why no gate flagged it. The launcher is not vestigial, it is exercised by the CI `compile` job (test:agent-evals:unit, agent-evals:check-types) and it is the runtime behind the main-only solheim-provider-smoke workflow, so the honest route is to keep the behaviour and remove the library class. Behaviour preserved: ack handshake, isReady gating, task event delivery, disconnect reporting, and the 500ms connect retry the launcher relies on because it connects before the extension host creates the socket. Impact: - No IpcClient symbol remains anywhere in the repo. - IpcServer and src/extension/api.ts are untouched. - The launcher no longer depends on a workspace package for its transport. - +6 unit tests and +1 interop test that drives the real IpcServer. --- packages/ipc/README.md | 40 ++-- packages/ipc/package.json | 2 +- packages/ipc/src/index.ts | 1 - packages/ipc/src/ipc-client.ts | 147 ------------- packages/types/src/ipc.ts | 12 - .../agent-evals/ipc-client.interop.test.ts | 75 +++++++ scripts/agent-evals/ipc-client.test.ts | 205 ++++++++++++++++++ scripts/agent-evals/ipc-client.ts | 191 ++++++++++++++++ scripts/agent-evals/launcher.ts | 4 +- 9 files changed, 494 insertions(+), 183 deletions(-) delete mode 100644 packages/ipc/src/ipc-client.ts create mode 100644 scripts/agent-evals/ipc-client.interop.test.ts create mode 100644 scripts/agent-evals/ipc-client.test.ts create mode 100644 scripts/agent-evals/ipc-client.ts diff --git a/packages/ipc/README.md b/packages/ipc/README.md index a6163c8a10..d897ed4530 100644 --- a/packages/ipc/README.md +++ b/packages/ipc/README.md @@ -48,27 +48,27 @@ Resumes a task from history. ## Usage Example +The package exports only the server half. A client connects to the socket +`IpcServer` listens on and speaks the same `node-ipc` framing: a js-message +envelope (`{"type":"message","data":…}`) terminated by a form feed. The +extension exposes the server through `src/extension/api.ts` when +`ROO_CODE_IPC_SOCKET_PATH` is set; `scripts/agent-evals/ipc-client.ts` is the +in-repo reference client. + ```typescript -import { IpcClient } from "@roo-code/ipc" - -const client = new IpcClient("/path/to/socket") - -// Resume a task -client.sendCommand({ - commandName: "ResumeTask", - data: "task-123", -}) - -// Start a new task -client.sendCommand({ - commandName: "StartNewTask", - data: { - configuration: {/* RooCode settings */}, - text: "Hello, world!", - images: [], - newTab: false, - }, -}) +import { IpcMessageType, IpcOrigin, TaskCommandName } from "@roo-code/types" + +socket.write( + JSON.stringify({ + type: "message", + data: { + type: IpcMessageType.TaskCommand, + origin: IpcOrigin.Client, + clientId, + data: { commandName: TaskCommandName.ResumeTask, data: "task-123" }, + }, + }) + "\f", +) ``` ## Events diff --git a/packages/ipc/package.json b/packages/ipc/package.json index 2f8c103e2b..e34b39cf02 100644 --- a/packages/ipc/package.json +++ b/packages/ipc/package.json @@ -1,6 +1,6 @@ { "name": "@roo-code/ipc", - "description": "IPC server and client for remote Roo Code access.", + "description": "IPC server for remote Roo Code access.", "version": "0.0.1", "type": "module", "exports": "./src/index.ts", diff --git a/packages/ipc/src/index.ts b/packages/ipc/src/index.ts index f50b4458be..9c10115848 100644 --- a/packages/ipc/src/index.ts +++ b/packages/ipc/src/index.ts @@ -1,2 +1 @@ -export * from "./ipc-client.js" export * from "./ipc-server.js" diff --git a/packages/ipc/src/ipc-client.ts b/packages/ipc/src/ipc-client.ts deleted file mode 100644 index 8926e424df..0000000000 --- a/packages/ipc/src/ipc-client.ts +++ /dev/null @@ -1,147 +0,0 @@ -import EventEmitter from "node:events" -import * as crypto from "node:crypto" - -import ipc from "node-ipc" - -import { - type TaskCommand, - type IpcClientEvents, - type IpcMessage, - IpcOrigin, - IpcMessageType, - TaskCommandName, - ipcMessageSchema, -} from "@roo-code/types" - -export class IpcClient extends EventEmitter { - private readonly _socketPath: string - private readonly _id: string - private readonly _log: (...args: unknown[]) => void - private _isConnected = false - private _clientId?: string - - constructor(socketPath: string, log = console.log) { - super() - - this._socketPath = socketPath - this._id = `roo-code-ipc-${crypto.randomBytes(6).toString("hex")}` - this._log = log - - ipc.config.silent = true - - ipc.connectTo(this._id, this.socketPath, () => { - ipc.of[this._id]?.on("connect", () => this.onConnect()) - ipc.of[this._id]?.on("disconnect", () => this.onDisconnect()) - ipc.of[this._id]?.on("message", (data) => this.onMessage(data)) - }) - } - - private onConnect() { - if (this._isConnected) { - return - } - - this.log("[client#onConnect]") - this._isConnected = true - this.emit(IpcMessageType.Connect) - } - - private onDisconnect() { - if (!this._isConnected) { - return - } - - this.log("[client#onDisconnect]") - this._isConnected = false - this.emit(IpcMessageType.Disconnect) - } - - private onMessage(data: unknown) { - if (typeof data !== "object") { - this.log(`[client#onMessage] invalid data -> ${JSON.stringify(data)}`) - return - } - - const result = ipcMessageSchema.safeParse(data) - - if (!result.success) { - this.log( - `[client#onMessage] invalid payload -> ${JSON.stringify(result.error.issues)} -> ${JSON.stringify(data)}`, - ) - return - } - - const payload = result.data - - if (payload.origin === IpcOrigin.Server) { - switch (payload.type) { - case IpcMessageType.Ack: - this._clientId = payload.data.clientId - this.emit(IpcMessageType.Ack, payload.data) - break - case IpcMessageType.TaskEvent: - this.emit(IpcMessageType.TaskEvent, payload.data) - break - } - } - } - - private log(...args: unknown[]) { - this._log(...args) - } - - public sendCommand(command: TaskCommand) { - const message: IpcMessage = { - type: IpcMessageType.TaskCommand, - origin: IpcOrigin.Client, - clientId: this._clientId!, - data: command, - } - - this.sendMessage(message) - } - - public sendTaskMessage(text?: string, images?: string[]) { - this.sendCommand({ - commandName: TaskCommandName.SendMessage, - data: { text, images }, - }) - } - - public deleteQueuedMessage(messageId: string) { - this.sendCommand({ - commandName: TaskCommandName.DeleteQueuedMessage, - data: messageId, - }) - } - - public sendMessage(message: IpcMessage) { - ipc.of[this._id]?.emit("message", message) - } - - public disconnect() { - try { - ipc.disconnect(this._id) - } catch (error) { - this.log( - `[client#disconnect] error disconnecting -> ${error instanceof Error ? error.message : String(error)}`, - ) - } - } - - public get socketPath() { - return this._socketPath - } - - public get clientId() { - return this._clientId - } - - public get isConnected() { - return this._isConnected - } - - public get isReady() { - return this._isConnected && this._clientId !== undefined - } -} diff --git a/packages/types/src/ipc.ts b/packages/types/src/ipc.ts index e7df762069..b2928be99c 100644 --- a/packages/types/src/ipc.ts +++ b/packages/types/src/ipc.ts @@ -128,18 +128,6 @@ export const ipcMessageSchema = z.discriminatedUnion("type", [ export type IpcMessage = z.infer -/** - * IpcClientEvents - */ - -export type IpcClientEvents = { - [IpcMessageType.Connect]: [] - [IpcMessageType.Disconnect]: [] - [IpcMessageType.Ack]: [data: Ack] - [IpcMessageType.TaskCommand]: [data: TaskCommand] - [IpcMessageType.TaskEvent]: [data: TaskEvent] -} - /** * IpcServerEvents */ diff --git a/scripts/agent-evals/ipc-client.interop.test.ts b/scripts/agent-evals/ipc-client.interop.test.ts new file mode 100644 index 0000000000..14970f47db --- /dev/null +++ b/scripts/agent-evals/ipc-client.interop.test.ts @@ -0,0 +1,75 @@ +import assert from "node:assert/strict" +import fs from "node:fs/promises" +import { createRequire } from "node:module" +import os from "node:os" +import path from "node:path" +import { describe, it } from "node:test" +import { fileURLToPath, pathToFileURL } from "node:url" + +import { IpcMessageType, IpcOrigin, RooCodeEventName, TaskCommandName, type TaskEvent } from "@roo-code/types" +import { IpcServer } from "@roo-code/ipc" +import { EvalIpcClient } from "./ipc-client.ts" + +const repoRoot = path.resolve(path.dirname(fileURLToPath(import.meta.url)), "../..") + +// IpcServer keeps node-ipc's listening handle open and exposes no stop(), so the +// suite has to close it or the runner never exits. scripts/ cannot resolve +// node-ipc itself, and the ESM entry is the singleton IpcServer actually uses, +// so resolve it through packages/ipc and import that exact module URL. +const stopListening = async (): Promise => { + const requireFromIpc = createRequire(path.join(repoRoot, "packages/ipc/package.json")) + const esmEntry = requireFromIpc.resolve("node-ipc").replace(/node-ipc\.cjs$/, "node-ipc.js") + const ipc = (await import(pathToFileURL(esmEntry).href)).default as { server: { stop?: () => void } | false } + ipc.server && ipc.server.stop?.() +} + +const socketDir = async (): Promise => fs.mkdtemp(path.join(os.tmpdir(), "agent-evals-interop-")) + +const socketPathIn = (dir: string, name: string): string => + process.platform === "win32" ? `agent-evals-interop-${process.pid}-${name}` : path.join(dir, `${name}.sock`) + +const waitFor = async (check: () => boolean, timeoutMs = 5_000): Promise => { + const deadline = Date.now() + timeoutMs + while (!check() && Date.now() < deadline) await new Promise((resolve) => setTimeout(resolve, 20)) +} + +// The extension runs the real IpcServer, so the launcher client has to speak its +// exact framing. This suite is the interop gate between the two halves. +describe("eval ipc client against the real extension server", () => { + it("handshakes with IpcServer and round-trips a command and a task event", async () => { + const dir = await socketDir() + const socketPath = socketPathIn(dir, "interop") + const server = new IpcServer(socketPath, () => {}) + const commands: string[] = [] + const events: TaskEvent[] = [] + + server.on(IpcMessageType.TaskCommand, (clientId, command) => { + commands.push(command.commandName) + server.send(clientId, { + type: IpcMessageType.TaskEvent, + origin: IpcOrigin.Server, + data: { eventName: RooCodeEventName.TaskStarted, payload: ["task-1"], taskId: 1 } satisfies TaskEvent, + }) + }) + server.listen() + + const client = new EvalIpcClient(socketPath, () => {}) + client.on(IpcMessageType.TaskEvent, (event) => events.push(event as TaskEvent)) + + try { + await waitFor(() => client.isReady) + assert.equal(client.isReady, true, "client must receive the server ack") + + client.sendCommand({ commandName: TaskCommandName.CloseTask }) + await waitFor(() => events.length > 0) + + assert.deepEqual(commands, [TaskCommandName.CloseTask]) + assert.equal(events.length, 1) + assert.equal(events[0]?.eventName, RooCodeEventName.TaskStarted) + } finally { + client.disconnect() + await stopListening() + await fs.rm(dir, { recursive: true, force: true }) + } + }) +}) diff --git a/scripts/agent-evals/ipc-client.test.ts b/scripts/agent-evals/ipc-client.test.ts new file mode 100644 index 0000000000..21d76ee955 --- /dev/null +++ b/scripts/agent-evals/ipc-client.test.ts @@ -0,0 +1,205 @@ +import assert from "node:assert/strict" +import fs from "node:fs/promises" +import net from "node:net" +import os from "node:os" +import path from "node:path" +import { describe, it } from "node:test" + +import { IpcMessageType, IpcOrigin, RooCodeEventName, TaskCommandName, type TaskEvent } from "@roo-code/types" +import { EvalIpcClient, resolveSocketPath } from "./ipc-client.ts" + +const DELIMITER = "\f" + +const socketDir = async (): Promise => fs.mkdtemp(path.join(os.tmpdir(), "agent-evals-ipc-")) + +// node-ipc listens on a named pipe on Windows, not on a filesystem path. +const socketPathIn = (dir: string, name: string): string => + process.platform === "win32" ? `agent-evals-test-${process.pid}-${name}` : path.join(dir, `${name}.sock`) + +const waitFor = async (check: () => boolean, timeoutMs = 5_000): Promise => { + const deadline = Date.now() + timeoutMs + while (!check() && Date.now() < deadline) await new Promise((resolve) => setTimeout(resolve, 20)) +} + +// Stands in for the extension-side IpcServer with the same node-ipc framing but +// a closeable socket, so tests can end the connection deterministically. +const startPeer = async (socketPath: string) => { + const received: unknown[] = [] + const sockets = new Set() + const server = net.createServer((socket) => { + sockets.add(socket) + socket.setEncoding("utf8") + socket.on("close", () => sockets.delete(socket)) + socket.on("error", () => {}) + socket.on("data", (chunk: string) => { + for (const frame of chunk.split(DELIMITER)) { + if (frame.length === 0) continue + const message = JSON.parse(frame) as { data: { clientId: string } } + received.push(message.data) + socket.write( + JSON.stringify({ + type: "message", + data: { + type: IpcMessageType.TaskEvent, + origin: IpcOrigin.Server, + data: { + eventName: RooCodeEventName.TaskStarted, + payload: ["task-1"], + taskId: 1, + } satisfies TaskEvent, + }, + }) + DELIMITER, + ) + } + }) + socket.write( + JSON.stringify({ + type: "message", + data: { + type: IpcMessageType.Ack, + origin: IpcOrigin.Server, + data: { clientId: "client-1", pid: process.pid, ppid: process.ppid }, + }, + }) + DELIMITER, + ) + }) + + await new Promise((resolve, reject) => { + server.once("error", reject) + server.listen(resolveSocketPath(socketPath), resolve) + }) + + return { + received, + async close() { + for (const socket of sockets) socket.destroy() + await new Promise((resolve) => server.close(() => resolve())) + }, + } +} + +describe("eval ipc client", () => { + it("completes the ack handshake and reports readiness", async () => { + const dir = await socketDir() + const socketPath = socketPathIn(dir, "handshake") + const peer = await startPeer(socketPath) + const client = new EvalIpcClient(socketPath, () => {}) + + try { + await waitFor(() => client.isReady) + assert.equal(client.isConnected, true) + assert.equal(client.clientId, "client-1") + assert.equal(client.isReady, true) + } finally { + client.disconnect() + await peer.close() + await fs.rm(dir, { recursive: true, force: true }) + } + }) + + it("sends a command in the frame shape the extension parses and receives task events", async () => { + const dir = await socketDir() + const socketPath = socketPathIn(dir, "roundtrip") + const peer = await startPeer(socketPath) + const events: TaskEvent[] = [] + const client = new EvalIpcClient(socketPath, () => {}) + client.on(IpcMessageType.TaskEvent, (event) => events.push(event as TaskEvent)) + + try { + await waitFor(() => client.isReady) + client.sendCommand({ commandName: TaskCommandName.CloseTask }) + await waitFor(() => events.length > 0) + + assert.equal(peer.received.length, 1) + assert.deepEqual(peer.received[0], { + type: IpcMessageType.TaskCommand, + origin: IpcOrigin.Client, + clientId: "client-1", + data: { commandName: TaskCommandName.CloseTask }, + }) + assert.equal(events.length, 1) + assert.equal(events[0]?.eventName, RooCodeEventName.TaskStarted) + } finally { + client.disconnect() + await peer.close() + await fs.rm(dir, { recursive: true, force: true }) + } + }) + + it("retries until the socket exists instead of failing when the host is not up yet", async () => { + const dir = await socketDir() + const socketPath = socketPathIn(dir, "late") + const client = new EvalIpcClient(socketPath, () => {}) + let peer: Awaited> | undefined + + try { + // The launcher connects before the extension host creates the socket. + await new Promise((resolve) => setTimeout(resolve, 150)) + assert.equal(client.isReady, false) + + peer = await startPeer(socketPath) + await waitFor(() => client.isReady) + assert.equal(client.isReady, true) + } finally { + client.disconnect() + await peer?.close() + await fs.rm(dir, { recursive: true, force: true }) + } + }) + + it("reports a disconnect when the peer goes away", async () => { + const dir = await socketDir() + const socketPath = socketPathIn(dir, "drop") + const peer = await startPeer(socketPath) + const client = new EvalIpcClient(socketPath, () => {}) + let disconnected = false + client.on(IpcMessageType.Disconnect, () => { + disconnected = true + }) + + try { + await waitFor(() => client.isReady) + await peer.close() + await waitFor(() => disconnected) + assert.equal(client.isConnected, false) + } finally { + client.disconnect() + await fs.rm(dir, { recursive: true, force: true }) + } + }) + + it("ignores payloads that do not match the IPC message schema", async () => { + const dir = await socketDir() + const socketPath = socketPathIn(dir, "invalid") + const logs: unknown[][] = [] + const server = net.createServer((socket) => { + socket.on("error", () => {}) + socket.write(JSON.stringify({ type: "message", data: { type: IpcMessageType.Ack } }) + DELIMITER) + socket.write("not-json" + DELIMITER) + }) + await new Promise((resolve, reject) => { + server.once("error", reject) + server.listen(resolveSocketPath(socketPath), resolve) + }) + const client = new EvalIpcClient(socketPath, (...args) => logs.push(args)) + + try { + await waitFor(() => logs.length >= 2) + assert.equal(client.isReady, false) + assert.ok(!JSON.stringify(logs).includes(IpcMessageType.Ack)) + } finally { + client.disconnect() + await new Promise((resolve) => server.close(() => resolve())) + await fs.rm(dir, { recursive: true, force: true }) + } + }) + + it("maps a filesystem socket path onto the Windows named pipe node-ipc listens on", () => { + const mapped = resolveSocketPath("/tmp/agent-evals/ipc.sock") + if (process.platform === "win32") { + assert.equal(mapped, "\\\\.\\pipe\\tmp-agent-evals-ipc.sock") + } else { + assert.equal(mapped, "/tmp/agent-evals/ipc.sock") + } + }) +}) diff --git a/scripts/agent-evals/ipc-client.ts b/scripts/agent-evals/ipc-client.ts new file mode 100644 index 0000000000..dbc481d7ec --- /dev/null +++ b/scripts/agent-evals/ipc-client.ts @@ -0,0 +1,191 @@ +import { EventEmitter } from "node:events" +import net from "node:net" + +import { + IpcMessageType, + IpcOrigin, + ipcMessageSchema, + type Ack, + type IpcMessage, + type TaskCommand, +} from "@roo-code/types" + +// The extension side frames IPC messages with node-ipc: a js-message envelope +// terminated by a form feed. The launcher has to match that framing byte for +// byte, so this client owns it instead of depending on @roo-code/ipc, whose +// client half has no other consumer. +const DELIMITER = "\f" + +const RETRY_MS = 500 + +const envelope = (data: unknown): string => JSON.stringify({ type: "message", data }) + DELIMITER + +// node-ipc rewrites non-pipe socket paths on Windows before listening, so a +// client must apply the same rewrite to reach the socket the server created. +export const resolveSocketPath = (socketPath: string): string => { + if (process.platform !== "win32" || socketPath.startsWith("\\\\.\\pipe\\")) return socketPath + return `\\\\.\\pipe\\${socketPath.replace(/^\//, "").replace(/\//g, "-")}` +} + +export interface EvalIpcClientEvents { + [IpcMessageType.Connect]: [] + [IpcMessageType.Disconnect]: [] + [IpcMessageType.Ack]: [data: Ack] + [IpcMessageType.TaskEvent]: [data: unknown] +} + +export class EvalIpcClient extends EventEmitter { + private readonly _socketPath: string + private readonly _log: (...args: unknown[]) => void + private _socket?: net.Socket + private _buffer = "" + private _retry?: NodeJS.Timeout + private _isConnected = false + private _clientId?: string + private _closed = false + + constructor(socketPath: string, log: (...args: unknown[]) => void = console.log) { + super() + + this._socketPath = socketPath + this._log = log + + this.connect() + } + + // The launcher connects immediately after spawning the host, before the + // extension has created the socket, so a refused connection is expected and + // must be retried until the host comes up. + private connect() { + if (this._closed) return + + const socket = (this._socket = net.connect(resolveSocketPath(this._socketPath))) + socket.setEncoding("utf8") + socket.on("connect", () => this.onConnect()) + socket.on("close", () => this.onClose()) + socket.on("error", (error) => this.log(`[client#error] ${error.message}`)) + socket.on("data", (chunk: string) => this.onData(chunk)) + } + + private onConnect() { + if (this._isConnected) return + + this._isConnected = true + this.emit(IpcMessageType.Connect) + } + + private onClose() { + // A close before the first connect is a refused attempt, not a lost + // session; only a connected client reports a disconnect. + if (this._isConnected) { + this._isConnected = false + this.emit(IpcMessageType.Disconnect) + } + + if (this._closed) return + + this._retry = setTimeout(() => this.connect(), RETRY_MS) + this._retry.unref?.() + } + + private onData(chunk: string) { + this._buffer += chunk + + // A frame is complete only once the delimiter arrives; a partial frame + // stays buffered so large task events are never truncated. + const frames = this._buffer.split(DELIMITER) + this._buffer = frames.pop() ?? "" + + for (const frame of frames) { + if (frame.length > 0) this.onFrame(frame) + } + } + + private onFrame(frame: string) { + let parsed: unknown + + try { + parsed = JSON.parse(frame) + } catch { + this.log("[client#onFrame] unparseable frame") + return + } + + const result = ipcMessageSchema.safeParse((parsed as { data?: unknown } | null)?.data) + + if (!result.success) { + // Payload values may carry credentials, so only the count is logged. + this.log("[client#onFrame] invalid payload", { issueCount: result.error.issues.length }) + return + } + + const message = result.data + + if (message.origin !== IpcOrigin.Server) return + + switch (message.type) { + case IpcMessageType.Ack: + this._clientId = message.data.clientId + this.emit(IpcMessageType.Ack, message.data) + break + case IpcMessageType.TaskEvent: + this.emit(IpcMessageType.TaskEvent, message.data) + break + } + } + + private log(...args: unknown[]) { + this._log(...args) + } + + public sendCommand(command: TaskCommand) { + if (this._clientId === undefined) { + this.log("[client#sendCommand] no client id yet") + return + } + + this.sendMessage({ + type: IpcMessageType.TaskCommand, + origin: IpcOrigin.Client, + clientId: this._clientId, + data: command, + }) + } + + public sendMessage(message: IpcMessage) { + if (this._closed || !this._socket || this._socket.destroyed) return + this._socket.write(envelope(message)) + } + + public disconnect() { + this._closed = true + + if (this._retry) { + clearTimeout(this._retry) + this._retry = undefined + } + + try { + this._socket?.end() + this._socket?.destroy() + } catch (error) { + this.log(`[client#disconnect] ${error instanceof Error ? error.message : String(error)}`) + } + } + + public get socketPath() { + return this._socketPath + } + + public get clientId() { + return this._clientId + } + + public get isConnected() { + return this._isConnected + } + + public get isReady() { + return this._isConnected && this._clientId !== undefined + } +} diff --git a/scripts/agent-evals/launcher.ts b/scripts/agent-evals/launcher.ts index 1edd2917d2..15f03986cc 100644 --- a/scripts/agent-evals/launcher.ts +++ b/scripts/agent-evals/launcher.ts @@ -4,8 +4,8 @@ import os from "node:os" import path from "node:path" import { promisify } from "node:util" import { downloadAndUnzipVSCode } from "@vscode/test-electron" -import { IpcClient } from "@roo-code/ipc" import { IpcMessageType } from "@roo-code/types" +import { EvalIpcClient } from "./ipc-client.ts" import { buildChildEnvironment, buildLaunchArgs } from "./isolation.ts" import type { AgentClient, AgentHost, AgentRuntime } from "./runner.ts" import { identityUnavailable, measured, type MeasuredString, type TraceRunIdentity } from "./trace.ts" @@ -127,7 +127,7 @@ export function createHostRuntime(options: HostRuntimeOptions): AgentRuntime { }, connect(): AgentClient { const socket = path.join(temp ?? "", "ipc.sock") - const client = new IpcClient(socket, () => {}) + const client = new EvalIpcClient(socket, () => {}) return { get isReady() { return client.isReady From 6547456205ffddb7a8346187345deda6ad86f839 Mon Sep 17 00:00:00 2001 From: myk1yt Date: Sat, 31 Oct 2026 04:47:29 +0900 Subject: [PATCH 2/6] fix(ipc): reset the frame buffer on close and bound pre-delimiter growth What: - Clear _buffer in onClose() so a partial frame from a dead connection cannot prepend to the next session's stream after the 500ms auto-reconnect. - Add MAX_FRAME_BYTES (64 MiB) on pre-delimiter accumulation; past it the client logs one bounded line and drops the connection. - Add five tests: reconnect carryover, frame split across chunks, a 4 MB delimiter-free flood followed by a valid frame, the ceiling drop, and a pin on the production ceiling value. Why: Review of #2265 found the carryover by probe: 34 bytes of a partial frame survived close and reconnection, which corrupts the first frame of the new session. The rest of that frame can never arrive on a new socket, so keeping it guarantees a bad parse later. The ceiling is hardening, not a fix: the node-ipc client this replaces had no bound at all, and a legitimate task event is orders of magnitude smaller than 64 MiB, so it cannot truncate a real frame in practice. Impact: - Two witnesses fail without their fix (verified by reverting each one). - The split-chunk and flood witnesses pin existing correct behaviour and pass without any change; they are coverage, not fixes. - No new dependency, no suppressions, no change to the wire format. --- scripts/agent-evals/ipc-client.test.ts | 237 ++++++++++++++++++++++++- scripts/agent-evals/ipc-client.ts | 42 ++++- 2 files changed, 277 insertions(+), 2 deletions(-) diff --git a/scripts/agent-evals/ipc-client.test.ts b/scripts/agent-evals/ipc-client.test.ts index 21d76ee955..4565b7ac67 100644 --- a/scripts/agent-evals/ipc-client.test.ts +++ b/scripts/agent-evals/ipc-client.test.ts @@ -6,7 +6,7 @@ import path from "node:path" import { describe, it } from "node:test" import { IpcMessageType, IpcOrigin, RooCodeEventName, TaskCommandName, type TaskEvent } from "@roo-code/types" -import { EvalIpcClient, resolveSocketPath } from "./ipc-client.ts" +import { EvalIpcClient, MAX_FRAME_BYTES, resolveSocketPath } from "./ipc-client.ts" const DELIMITER = "\f" @@ -202,4 +202,239 @@ describe("eval ipc client", () => { assert.equal(mapped, "/tmp/agent-evals/ipc.sock") } }) + + it("drops a partial frame when the socket closes so it cannot corrupt the next session", async () => { + const dir = await socketDir() + const socketPath = socketPathIn(dir, "carryover") + const events: TaskEvent[] = [] + const encoder = (data: unknown) => JSON.stringify({ type: "message", data }) + DELIMITER + const ack = encoder({ + type: IpcMessageType.Ack, + origin: IpcOrigin.Server, + data: { clientId: "client-1", pid: process.pid, ppid: process.ppid }, + }) + const taskEvent = encoder({ + type: IpcMessageType.TaskEvent, + origin: IpcOrigin.Server, + data: { eventName: RooCodeEventName.TaskStarted, payload: ["task-1"], taskId: 1 } satisfies TaskEvent, + }) + + let connections = 0 + const server = net.createServer((socket) => { + socket.on("error", () => {}) + connections++ + + if (connections === 1) { + // Only half a frame, no ack, then the peer dies. The rest can + // never arrive, so these bytes are unrecoverable garbage. + socket.write(taskEvent.slice(0, 34)) + socket.destroy() + return + } + + // A clean session. If the stale 34 bytes survived, they prepend to + // this ack, the ack frame fails to parse, no clientId is ever set, + // and the client can never become ready. + socket.write(ack + taskEvent) + }) + + await new Promise((resolve, reject) => { + server.once("error", reject) + server.listen(resolveSocketPath(socketPath), resolve) + }) + + const client = new EvalIpcClient(socketPath, () => {}) + client.on(IpcMessageType.TaskEvent, (event) => events.push(event as TaskEvent)) + + try { + await waitFor(() => connections >= 2 && client.isReady, 10_000) + assert.equal(connections, 2, "client must reconnect after the peer closes") + assert.equal(client.isReady, true, "the ack of the new session must parse") + assert.equal(client.clientId, "client-1") + assert.equal(events.length, 1, "the task event of the new session must parse") + assert.equal(events[0]?.eventName, RooCodeEventName.TaskStarted) + } finally { + client.disconnect() + await new Promise((resolve) => server.close(() => resolve())) + await fs.rm(dir, { recursive: true, force: true }) + } + }) + + it("waits for the delimiter when one frame arrives split across chunks", async () => { + const dir = await socketDir() + const socketPath = socketPathIn(dir, "split") + const events: TaskEvent[] = [] + const encoded = + JSON.stringify({ + type: "message", + data: { + type: IpcMessageType.TaskEvent, + origin: IpcOrigin.Server, + data: { + eventName: RooCodeEventName.TaskStarted, + payload: ["task-1"], + taskId: 1, + } satisfies TaskEvent, + }, + }) + DELIMITER + + const server = net.createServer((socket) => { + socket.on("error", () => {}) + socket.write( + JSON.stringify({ + type: "message", + data: { + type: IpcMessageType.Ack, + origin: IpcOrigin.Server, + data: { clientId: "client-1", pid: process.pid, ppid: process.ppid }, + }, + }) + DELIMITER, + ) + + // Split mid-frame, well before the delimiter, with a real pause. + const cut = Math.floor(encoded.length / 2) + socket.write(encoded.slice(0, cut)) + setTimeout(() => socket.write(encoded.slice(cut)), 120) + }) + + await new Promise((resolve, reject) => { + server.once("error", reject) + server.listen(resolveSocketPath(socketPath), resolve) + }) + + const client = new EvalIpcClient(socketPath, () => {}) + client.on(IpcMessageType.TaskEvent, (event) => events.push(event as TaskEvent)) + + try { + await waitFor(() => client.isReady) + // The first half alone must not produce a frame. + await new Promise((resolve) => setTimeout(resolve, 60)) + assert.equal(events.length, 0, "a frame without its delimiter must not be dispatched") + + await waitFor(() => events.length > 0) + assert.equal(events.length, 1) + assert.equal(events[0]?.eventName, RooCodeEventName.TaskStarted) + } finally { + client.disconnect() + await new Promise((resolve) => server.close(() => resolve())) + await fs.rm(dir, { recursive: true, force: true }) + } + }) + + it("stays alive through a delimiter-free flood and still parses the frame that follows", async () => { + const dir = await socketDir() + const socketPath = socketPathIn(dir, "flood") + const events: TaskEvent[] = [] + const logs: unknown[][] = [] + const encoded = + JSON.stringify({ + type: "message", + data: { + type: IpcMessageType.TaskEvent, + origin: IpcOrigin.Server, + data: { + eventName: RooCodeEventName.TaskStarted, + payload: ["task-1"], + taskId: 1, + } satisfies TaskEvent, + }, + }) + DELIMITER + + const server = net.createServer((socket) => { + socket.on("error", () => {}) + socket.write( + JSON.stringify({ + type: "message", + data: { + type: IpcMessageType.Ack, + origin: IpcOrigin.Server, + data: { clientId: "client-1", pid: process.pid, ppid: process.ppid }, + }, + }) + DELIMITER, + ) + // 4 MB of garbage terminated as its own frame, then a valid frame. + // The garbage frame is dropped; the client must stay alive and parse + // the frame that follows. + socket.write("A".repeat(4 * 1024 * 1024) + DELIMITER) + socket.write(encoded) + }) + + await new Promise((resolve, reject) => { + server.once("error", reject) + server.listen(resolveSocketPath(socketPath), resolve) + }) + + const client = new EvalIpcClient(socketPath, (...args) => logs.push(args)) + client.on(IpcMessageType.TaskEvent, (event) => events.push(event as TaskEvent)) + + try { + await waitFor(() => events.length > 0, 10_000) + assert.equal(events.length, 1, "the process must survive the flood and parse the next frame") + assert.equal(events[0]?.eventName, RooCodeEventName.TaskStarted) + // The flood is below the ceiling, so the connection is not dropped. + assert.equal( + logs.some((entry) => String(entry[0]).includes("MAX_FRAME_BYTES")), + false, + ) + assert.equal(client.isConnected, true) + } finally { + client.disconnect() + await new Promise((resolve) => server.close(() => resolve())) + await fs.rm(dir, { recursive: true, force: true }) + } + }) + + it("drops the connection with one bounded log line past the frame ceiling", async () => { + const dir = await socketDir() + const socketPath = socketPathIn(dir, "oversized") + const logs: unknown[][] = [] + const events: TaskEvent[] = [] + const secret = "sk-should-never-be-logged" + const ceiling = 4096 + + const server = net.createServer((socket) => { + socket.on("error", () => {}) + socket.write( + JSON.stringify({ + type: "message", + data: { + type: IpcMessageType.Ack, + origin: IpcOrigin.Server, + data: { clientId: "client-1", pid: process.pid, ppid: process.ppid }, + }, + }) + DELIMITER, + ) + // Past the ceiling with no delimiter, credential-shaped content included. + socket.write(secret + "A".repeat(ceiling + 1024)) + }) + + await new Promise((resolve, reject) => { + server.once("error", reject) + server.listen(resolveSocketPath(socketPath), resolve) + }) + + // A small ceiling keeps this test fast; the production default is pinned + // by the assertion below and by the constant's own test. + const client = new EvalIpcClient(socketPath, (...args) => logs.push(args), { maxFrameBytes: ceiling }) + client.on(IpcMessageType.TaskEvent, (event) => events.push(event as TaskEvent)) + + try { + await waitFor(() => logs.some((entry) => String(entry[0]).includes("MAX_FRAME_BYTES")), 10_000) + + const oversized = logs.filter((entry) => String(entry[0]).includes("MAX_FRAME_BYTES")) + assert.equal(oversized.length, 1, "exactly one bounded log line") + assert.equal(oversized[0]?.length, 1, "no payload or size detail accompanies the line") + assert.ok(!JSON.stringify(logs).includes(secret), "buffer content must never be logged") + assert.equal(events.length, 0) + assert.equal(client.isConnected, false, "the connection is dropped") + } finally { + client.disconnect() + await new Promise((resolve) => server.close(() => resolve())) + await fs.rm(dir, { recursive: true, force: true }) + } + }) + + it("keeps the production frame ceiling far above any legitimate task event", () => { + assert.equal(MAX_FRAME_BYTES, 64 * 1024 * 1024) + }) }) diff --git a/scripts/agent-evals/ipc-client.ts b/scripts/agent-evals/ipc-client.ts index dbc481d7ec..a8ce16a97b 100644 --- a/scripts/agent-evals/ipc-client.ts +++ b/scripts/agent-evals/ipc-client.ts @@ -18,6 +18,13 @@ const DELIMITER = "\f" const RETRY_MS = 500 +// Defensive ceiling on pre-delimiter accumulation. A legitimate task event is +// orders of magnitude smaller than this, and the node-ipc client this replaces +// had no ceiling at all, so the bound is pure hardening: it cannot truncate a +// real frame in practice. Measured with String#length, which under-counts +// multi-byte UTF-8 by at most 2x; that slack is irrelevant at this size. +export const MAX_FRAME_BYTES = 64 * 1024 * 1024 + const envelope = (data: unknown): string => JSON.stringify({ type: "message", data }) + DELIMITER // node-ipc rewrites non-pipe socket paths on Windows before listening, so a @@ -34,9 +41,15 @@ export interface EvalIpcClientEvents { [IpcMessageType.TaskEvent]: [data: unknown] } +export interface EvalIpcClientOptions { + // Test seam only; production always uses MAX_FRAME_BYTES. + maxFrameBytes?: number +} + export class EvalIpcClient extends EventEmitter { private readonly _socketPath: string private readonly _log: (...args: unknown[]) => void + private readonly _maxFrameBytes: number private _socket?: net.Socket private _buffer = "" private _retry?: NodeJS.Timeout @@ -44,11 +57,16 @@ export class EvalIpcClient extends EventEmitter { private _clientId?: string private _closed = false - constructor(socketPath: string, log: (...args: unknown[]) => void = console.log) { + constructor( + socketPath: string, + log: (...args: unknown[]) => void = console.log, + options: EvalIpcClientOptions = {}, + ) { super() this._socketPath = socketPath this._log = log + this._maxFrameBytes = options.maxFrameBytes ?? MAX_FRAME_BYTES this.connect() } @@ -82,6 +100,10 @@ export class EvalIpcClient extends EventEmitter { this.emit(IpcMessageType.Disconnect) } + // The rest of a partial frame can never arrive on a new socket, so + // keeping it would corrupt the first frame of the next session. + this._buffer = "" + if (this._closed) return this._retry = setTimeout(() => this.connect(), RETRY_MS) @@ -99,6 +121,24 @@ export class EvalIpcClient extends EventEmitter { for (const frame of frames) { if (frame.length > 0) this.onFrame(frame) } + + if (this._buffer.length > this._maxFrameBytes) this.onOversizedFrame() + } + + // Bounded log line only: the pending bytes may carry credentials, so neither + // their content nor their exact size is written out. + private onOversizedFrame() { + this._buffer = "" + this._closed = true + + if (this._retry) { + clearTimeout(this._retry) + this._retry = undefined + } + + this.log("[client#onData] frame exceeded MAX_FRAME_BYTES, dropping connection") + + this._socket?.destroy() } private onFrame(frame: string) { From be2acdc8c92815bf3e083d1c7ef6ed76704e8767 Mon Sep 17 00:00:00 2001 From: myk1yt Date: Sat, 31 Oct 2026 07:21:38 +0900 Subject: [PATCH 3/6] fix(ipc): address CodeRabbit round-2 findings on the eval client What: - Buffer the test peer's partial frames: accumulate chunks, dispatch only delimiter-terminated segments, and retain the tail for the next event. - Replace the split-frame test's 120 ms timer with a promise gate the test releases after the zero-event assertion, so the assertion cannot race a scheduling delay. - Enforce MAX_FRAME_BYTES in UTF-8 bytes with incremental Buffer.byteLength accounting on the appended tail, and correct the comment's 2x claim. - Add two witnesses: a peer frame split across two chunks, and a 3-byte character frame whose code-unit count is under a 4096 ceiling but whose 6000 wire bytes are over it. Why: CodeRabbit round 2 raised three findings on #2265. All three premises were verified in the pre-fix source before changing anything: the peer split each chunk on the delimiter with no carry buffer and threw on a split frame; the split-frame test gated the second half on a 120 ms timer that readiness plus 60 ms could outrun; and the ceiling compared against String#length, which counts UTF-16 code units, so 3-byte characters let the pending frame reach ~192 MiB of wire bytes instead of 64 MiB. Impact: - Three witnesses fail on the pre-fix head and pass after the fix (each verified by reverting its own change). - The byte ceiling is O(chunk): each chunk is scanned once and only its trailing partial segment is measured, so a delimiter-free flood does not re-measure the whole pending buffer per event. - No behaviour change for ASCII frames; the ceiling now bounds wire bytes. --- scripts/agent-evals/ipc-client.test.ts | 111 ++++++++++++++++++++++++- scripts/agent-evals/ipc-client.ts | 43 +++++++--- 2 files changed, 138 insertions(+), 16 deletions(-) diff --git a/scripts/agent-evals/ipc-client.test.ts b/scripts/agent-evals/ipc-client.test.ts index 4565b7ac67..79e848d081 100644 --- a/scripts/agent-evals/ipc-client.test.ts +++ b/scripts/agent-evals/ipc-client.test.ts @@ -31,8 +31,15 @@ const startPeer = async (socketPath: string) => { socket.setEncoding("utf8") socket.on("close", () => sockets.delete(socket)) socket.on("error", () => {}) + // The client frames with one write, but TCP does not preserve write + // boundaries, so a frame can arrive split. Buffer the tail of an + // incomplete frame and dispatch only delimiter-terminated segments. + let pending = "" socket.on("data", (chunk: string) => { - for (const frame of chunk.split(DELIMITER)) { + pending += chunk + const frames = pending.split(DELIMITER) + pending = frames.pop() ?? "" + for (const frame of frames) { if (frame.length === 0) continue const message = JSON.parse(frame) as { data: { clientId: string } } received.push(message.data) @@ -126,6 +133,46 @@ describe("eval ipc client", () => { } }) + it("assembles a peer frame that arrives split across two chunks", async () => { + const dir = await socketDir() + const socketPath = socketPathIn(dir, "peer-split") + const peer = await startPeer(socketPath) + const raw = net.connect(resolveSocketPath(socketPath)) + raw.setEncoding("utf8") + const frame = + JSON.stringify({ + type: "message", + data: { + type: IpcMessageType.TaskCommand, + origin: IpcOrigin.Client, + clientId: "client-1", + data: { commandName: TaskCommandName.CloseTask }, + }, + }) + DELIMITER + + try { + await new Promise((resolve) => raw.once("connect", () => resolve())) + // Two separate writes with a pause, so the peer sees two data events. + const cut = Math.floor(frame.length / 2) + raw.write(frame.slice(0, cut)) + await new Promise((resolve) => setTimeout(resolve, 60)) + raw.write(frame.slice(cut)) + + await waitFor(() => peer.received.length > 0) + assert.equal(peer.received.length, 1, "the peer must assemble the frame from both chunks") + assert.deepEqual(peer.received[0], { + type: IpcMessageType.TaskCommand, + origin: IpcOrigin.Client, + clientId: "client-1", + data: { commandName: TaskCommandName.CloseTask }, + }) + } finally { + raw.destroy() + await peer.close() + await fs.rm(dir, { recursive: true, force: true }) + } + }) + it("retries until the socket exists instead of failing when the host is not up yet", async () => { const dir = await socketDir() const socketPath = socketPathIn(dir, "late") @@ -264,6 +311,12 @@ describe("eval ipc client", () => { const dir = await socketDir() const socketPath = socketPathIn(dir, "split") const events: TaskEvent[] = [] + // The second half of the frame is held here until the test has asserted + // that the first half alone dispatches nothing. + let releaseSecondHalf = () => {} + const secondHalfGate = new Promise((resolve) => { + releaseSecondHalf = resolve + }) const encoded = JSON.stringify({ type: "message", @@ -291,10 +344,13 @@ describe("eval ipc client", () => { }) + DELIMITER, ) - // Split mid-frame, well before the delimiter, with a real pause. + // Split mid-frame, well before the delimiter. The second half waits on + // a gate the test releases after the zero-event assertion, so the + // assertion cannot race a timer: no amount of scheduling delay can + // complete the frame before the test allows it. const cut = Math.floor(encoded.length / 2) socket.write(encoded.slice(0, cut)) - setTimeout(() => socket.write(encoded.slice(cut)), 120) + void secondHalfGate.then(() => socket.write(encoded.slice(cut))) }) await new Promise((resolve, reject) => { @@ -311,10 +367,12 @@ describe("eval ipc client", () => { await new Promise((resolve) => setTimeout(resolve, 60)) assert.equal(events.length, 0, "a frame without its delimiter must not be dispatched") + releaseSecondHalf() await waitFor(() => events.length > 0) assert.equal(events.length, 1) assert.equal(events[0]?.eventName, RooCodeEventName.TaskStarted) } finally { + releaseSecondHalf() client.disconnect() await new Promise((resolve) => server.close(() => resolve())) await fs.rm(dir, { recursive: true, force: true }) @@ -434,6 +492,53 @@ describe("eval ipc client", () => { } }) + it("measures the frame ceiling in UTF-8 bytes, not UTF-16 code units", async () => { + const dir = await socketDir() + const socketPath = socketPathIn(dir, "multibyte") + const logs: unknown[][] = [] + const ceiling = 4096 + // 3-byte UTF-8 characters: 2000 of them are 2000 code units (under the + // ceiling) but 6000 wire bytes (over it), so only a byte ceiling drops. + const frame = "가".repeat(2000) + + const server = net.createServer((socket) => { + socket.on("error", () => {}) + socket.write( + JSON.stringify({ + type: "message", + data: { + type: IpcMessageType.Ack, + origin: IpcOrigin.Server, + data: { clientId: "client-1", pid: process.pid, ppid: process.ppid }, + }, + }) + DELIMITER, + ) + socket.write(frame) + }) + + await new Promise((resolve, reject) => { + server.once("error", reject) + server.listen(resolveSocketPath(socketPath), resolve) + }) + + const client = new EvalIpcClient(socketPath, (...args) => logs.push(args), { maxFrameBytes: ceiling }) + + try { + await waitFor(() => logs.some((entry) => String(entry[0]).includes("MAX_FRAME_BYTES")), 10_000) + + assert.equal( + logs.filter((entry) => String(entry[0]).includes("MAX_FRAME_BYTES")).length, + 1, + "the byte ceiling must drop a frame whose code-unit count is under the limit", + ) + assert.equal(client.isConnected, false, "the connection is dropped") + } finally { + client.disconnect() + await new Promise((resolve) => server.close(() => resolve())) + await fs.rm(dir, { recursive: true, force: true }) + } + }) + it("keeps the production frame ceiling far above any legitimate task event", () => { assert.equal(MAX_FRAME_BYTES, 64 * 1024 * 1024) }) diff --git a/scripts/agent-evals/ipc-client.ts b/scripts/agent-evals/ipc-client.ts index a8ce16a97b..b7e16e34d3 100644 --- a/scripts/agent-evals/ipc-client.ts +++ b/scripts/agent-evals/ipc-client.ts @@ -18,11 +18,12 @@ const DELIMITER = "\f" const RETRY_MS = 500 -// Defensive ceiling on pre-delimiter accumulation. A legitimate task event is -// orders of magnitude smaller than this, and the node-ipc client this replaces -// had no ceiling at all, so the bound is pure hardening: it cannot truncate a -// real frame in practice. Measured with String#length, which under-counts -// multi-byte UTF-8 by at most 2x; that slack is irrelevant at this size. +// Defensive ceiling on pre-delimiter accumulation, enforced in UTF-8 bytes of +// the pending frame rather than UTF-16 code units, so the wire size a peer can +// force the client to hold is bounded regardless of the character mix. A +// legitimate task event is orders of magnitude smaller than this, and the +// node-ipc client this replaces had no ceiling at all, so the bound is pure +// hardening: it cannot truncate a real frame in practice. export const MAX_FRAME_BYTES = 64 * 1024 * 1024 const envelope = (data: unknown): string => JSON.stringify({ type: "message", data }) + DELIMITER @@ -52,6 +53,9 @@ export class EvalIpcClient extends EventEmitter { private readonly _maxFrameBytes: number private _socket?: net.Socket private _buffer = "" + // UTF-8 size of _buffer, maintained incrementally so the ceiling check stays + // O(chunk) instead of re-measuring the whole pending frame on every event. + private _bufferBytes = 0 private _retry?: NodeJS.Timeout private _isConnected = false private _clientId?: string @@ -103,6 +107,7 @@ export class EvalIpcClient extends EventEmitter { // The rest of a partial frame can never arrive on a new socket, so // keeping it would corrupt the first frame of the next session. this._buffer = "" + this._bufferBytes = 0 if (this._closed) return @@ -111,24 +116,36 @@ export class EvalIpcClient extends EventEmitter { } private onData(chunk: string) { - this._buffer += chunk - // A frame is complete only once the delimiter arrives; a partial frame - // stays buffered so large task events are never truncated. - const frames = this._buffer.split(DELIMITER) - this._buffer = frames.pop() ?? "" - - for (const frame of frames) { + // stays buffered so large task events are never truncated. Each chunk is + // scanned once and only its trailing partial segment is measured, so the + // byte accounting stays O(chunk) even on a delimiter-free flood. + let offset = 0 + let index: number + + while ((index = chunk.indexOf(DELIMITER, offset)) !== -1) { + const frame = this._buffer + chunk.slice(offset, index) + this._buffer = "" + this._bufferBytes = 0 if (frame.length > 0) this.onFrame(frame) + offset = index + DELIMITER.length + } + + const tail = chunk.slice(offset) + + if (tail.length > 0) { + this._buffer += tail + this._bufferBytes += Buffer.byteLength(tail, "utf8") } - if (this._buffer.length > this._maxFrameBytes) this.onOversizedFrame() + if (this._bufferBytes > this._maxFrameBytes) this.onOversizedFrame() } // Bounded log line only: the pending bytes may carry credentials, so neither // their content nor their exact size is written out. private onOversizedFrame() { this._buffer = "" + this._bufferBytes = 0 this._closed = true if (this._retry) { From bbbef701f35a325687b47b27048f96dc606ef856 Mon Sep 17 00:00:00 2001 From: myk1yt Date: Sat, 31 Oct 2026 09:00:45 +0900 Subject: [PATCH 4/6] test(ipc): wait for the dropped connection, not just the ceiling log Both oversized-frame tests waited only for the MAX_FRAME_BYTES log line and then asserted `isConnected === false`. The handler logs before it destroys the socket, and `_isConnected` is cleared in the socket close callback, which runs after a timer tick. The assertion therefore depended on that callback winning a race with the 20ms waitFor poll. The wait predicate now covers the disconnect as well, so the assertion reads the settled state. Confirmed against a probe that records `isConnected` inside the log callback: it is `true` there and `false` one macrotask later. Red proof: with the close deferred by 50ms, both tests fail with "the connection is dropped" on the old wait and pass on the new one. --- scripts/agent-evals/ipc-client.test.ts | 15 +++++++++++++-- 1 file changed, 13 insertions(+), 2 deletions(-) diff --git a/scripts/agent-evals/ipc-client.test.ts b/scripts/agent-evals/ipc-client.test.ts index 79e848d081..b37d36bcdc 100644 --- a/scripts/agent-evals/ipc-client.test.ts +++ b/scripts/agent-evals/ipc-client.test.ts @@ -477,7 +477,13 @@ describe("eval ipc client", () => { client.on(IpcMessageType.TaskEvent, (event) => events.push(event as TaskEvent)) try { - await waitFor(() => logs.some((entry) => String(entry[0]).includes("MAX_FRAME_BYTES")), 10_000) + // The handler logs and then destroys the socket, but `isConnected` is only + // cleared in the close callback, which runs after a timer tick. Waiting on + // the log alone would race that callback, so the predicate covers both. + await waitFor( + () => !client.isConnected && logs.some((entry) => String(entry[0]).includes("MAX_FRAME_BYTES")), + 10_000, + ) const oversized = logs.filter((entry) => String(entry[0]).includes("MAX_FRAME_BYTES")) assert.equal(oversized.length, 1, "exactly one bounded log line") @@ -524,7 +530,12 @@ describe("eval ipc client", () => { const client = new EvalIpcClient(socketPath, (...args) => logs.push(args), { maxFrameBytes: ceiling }) try { - await waitFor(() => logs.some((entry) => String(entry[0]).includes("MAX_FRAME_BYTES")), 10_000) + // Same close-callback race as the oversized-frame test above: wait for the + // disconnect, not just for the log line. + await waitFor( + () => !client.isConnected && logs.some((entry) => String(entry[0]).includes("MAX_FRAME_BYTES")), + 10_000, + ) assert.equal( logs.filter((entry) => String(entry[0]).includes("MAX_FRAME_BYTES")).length, From 8b5bf9efafa5d604c57fca0fb5a977d8bb1b30fc Mon Sep 17 00:00:00 2001 From: myk1yt Date: Sat, 31 Oct 2026 14:39:54 +0900 Subject: [PATCH 5/6] fix(ipc): enforce the frame ceiling on delimited frames What: check the assembled frame's UTF-8 byte size inside the delimiter loop, before the buffer reset, and return after onOversizedFrame instead of parsing. Why: the ceiling only ran on the trailing partial segment, so an oversized frame whose delimiter arrived in the same chunk was parsed and dispatched instead of dropping the connection. Impact: a peer can no longer bypass MAX_FRAME_BYTES by sending the delimiter with the overrun; the connection drops with a single bounded log line. The O(chunk) byte accounting is unchanged, and the exact-ceiling boundary still passes. --- scripts/agent-evals/ipc-client.test.ts | 183 +++++++++++++++++++++++++ scripts/agent-evals/ipc-client.ts | 14 +- 2 files changed, 196 insertions(+), 1 deletion(-) diff --git a/scripts/agent-evals/ipc-client.test.ts b/scripts/agent-evals/ipc-client.test.ts index b37d36bcdc..fe91c3d94d 100644 --- a/scripts/agent-evals/ipc-client.test.ts +++ b/scripts/agent-evals/ipc-client.test.ts @@ -85,6 +85,56 @@ const startPeer = async (socketPath: string) => { } } +const ACK_FRAME = + JSON.stringify({ + type: "message", + data: { + type: IpcMessageType.Ack, + origin: IpcOrigin.Server, + data: { clientId: "client-1", pid: process.pid, ppid: process.ppid }, + }, + }) + DELIMITER + +// A schema-valid TaskEvent frame padded to an exact UTF-8 byte length by growing +// the payload string, so a test can sit precisely on the ceiling boundary. The +// padding is ASCII, so byte length and code-unit length coincide. +const taskEventFrameOfBytes = (bytes: number): string => { + const base = JSON.stringify({ + type: "message", + data: { + type: IpcMessageType.TaskEvent, + origin: IpcOrigin.Server, + data: { eventName: RooCodeEventName.TaskStarted, payload: [""], taskId: 1 } satisfies TaskEvent, + }, + }) + const paddingBytes = bytes - Buffer.byteLength(base, "utf8") + + if (paddingBytes < 0) + throw new Error(`frame base is ${Buffer.byteLength(base, "utf8")} bytes, cannot shrink to ${bytes}`) + + return base.replace('[""]', () => JSON.stringify(["A".repeat(paddingBytes)])) +} + +// Completes the handshake and then writes `payload` in a single call, so the test +// controls exactly which bytes share a chunk. +const startOneShotPeer = async (socketPath: string, payload: string): Promise<() => Promise> => { + const server = net.createServer((socket) => { + socket.on("error", () => {}) + socket.write(ACK_FRAME) + socket.write(payload) + }) + + await new Promise((resolve, reject) => { + server.once("error", reject) + server.listen(resolveSocketPath(socketPath), resolve) + }) + + return () => new Promise((resolve) => server.close(() => resolve())) +} + +const ceilingLogs = (logs: unknown[][]): unknown[][] => + logs.filter((entry) => String(entry[0]).includes("MAX_FRAME_BYTES")) + describe("eval ipc client", () => { it("completes the ack handshake and reports readiness", async () => { const dir = await socketDir() @@ -550,6 +600,139 @@ describe("eval ipc client", () => { } }) + it("drops an oversized frame whose delimiter arrives in the same chunk", async () => { + const dir = await socketDir() + const socketPath = socketPathIn(dir, "oversized-inline") + const logs: unknown[][] = [] + const events: TaskEvent[] = [] + const ceiling = 1024 + // A well-formed frame that is one byte past the ceiling. Without a check + // inside the delimiter loop it parses and dispatches instead of dropping. + const oversized = taskEventFrameOfBytes(ceiling + 1) + DELIMITER + const closePeer = await startOneShotPeer(socketPath, oversized) + + const client = new EvalIpcClient(socketPath, (...args) => logs.push(args), { maxFrameBytes: ceiling }) + client.on(IpcMessageType.TaskEvent, (event) => events.push(event as TaskEvent)) + + try { + await waitFor(() => !client.isConnected && ceilingLogs(logs).length > 0, 10_000) + + assert.equal(ceilingLogs(logs).length, 1, "exactly one bounded log line") + assert.equal(events.length, 0, "a frame past the ceiling must never be dispatched") + assert.equal(client.isConnected, false, "the connection is dropped") + } finally { + client.disconnect() + await closePeer() + await fs.rm(dir, { recursive: true, force: true }) + } + }) + + it("stops scanning the chunk after an oversized frame instead of parsing what follows", async () => { + const dir = await socketDir() + const socketPath = socketPathIn(dir, "oversized-then-valid") + const logs: unknown[][] = [] + const events: TaskEvent[] = [] + const ceiling = 1024 + // The oversized frame and a valid frame share one chunk. The handler must + // return, not break: a break leaves the offset unmoved, so the rest of the + // chunk becomes the tail, re-populates the buffer and logs a second time. + const chunk = taskEventFrameOfBytes(ceiling + 1) + DELIMITER + taskEventFrameOfBytes(256) + DELIMITER + const closePeer = await startOneShotPeer(socketPath, chunk) + + const client = new EvalIpcClient(socketPath, (...args) => logs.push(args), { maxFrameBytes: ceiling }) + client.on(IpcMessageType.TaskEvent, (event) => events.push(event as TaskEvent)) + + try { + await waitFor(() => !client.isConnected && ceilingLogs(logs).length > 0, 10_000) + // Give any stray second frame a chance to surface before asserting. + await new Promise((resolve) => setTimeout(resolve, 60)) + + assert.equal(ceilingLogs(logs).length, 1, "the tail must not re-trip the ceiling") + assert.equal(events.length, 0, "the frame after the oversized one must not be processed") + assert.equal(client.isConnected, false, "the connection is dropped") + } finally { + client.disconnect() + await closePeer() + await fs.rm(dir, { recursive: true, force: true }) + } + }) + + it("passes a frame at exactly the ceiling and drops one byte past it", async () => { + const ceiling = 512 + const run = async (name: string, bytes: number) => { + const dir = await socketDir() + const socketPath = socketPathIn(dir, name) + const logs: unknown[][] = [] + const events: TaskEvent[] = [] + const closePeer = await startOneShotPeer(socketPath, taskEventFrameOfBytes(bytes) + DELIMITER) + const client = new EvalIpcClient(socketPath, (...args) => logs.push(args), { maxFrameBytes: ceiling }) + + client.on(IpcMessageType.TaskEvent, (event) => events.push(event as TaskEvent)) + + try { + await waitFor(() => events.length > 0 || ceilingLogs(logs).length > 0, 10_000) + return { events: events.length, drops: ceilingLogs(logs).length, connected: client.isConnected } + } finally { + client.disconnect() + await closePeer() + await fs.rm(dir, { recursive: true, force: true }) + } + } + + const atCeiling = await run("ceiling-exact", ceiling) + assert.equal(atCeiling.drops, 0, "a frame at exactly the ceiling is legal") + assert.equal(atCeiling.events, 1, "the frame at the ceiling is dispatched") + assert.equal(atCeiling.connected, true, "the connection survives a frame at the ceiling") + + const pastCeiling = await run("ceiling-plus-one", ceiling + 1) + assert.equal(pastCeiling.drops, 1, "one byte past the ceiling drops the connection") + assert.equal(pastCeiling.events, 0, "the frame past the ceiling is never dispatched") + assert.equal(pastCeiling.connected, false, "the connection is dropped") + }) + + it("counts a delimited frame's multibyte content in bytes, not characters", async () => { + const dir = await socketDir() + const socketPath = socketPathIn(dir, "multibyte-inline") + const logs: unknown[][] = [] + const events: TaskEvent[] = [] + const ceiling = 2048 + // 1000 three-byte characters: 1000 code units (under the ceiling) but 3000 + // wire bytes (over it), and the delimiter shares the chunk with them. + const frame = + JSON.stringify({ + type: "message", + data: { + type: IpcMessageType.TaskEvent, + origin: IpcOrigin.Server, + data: { + eventName: RooCodeEventName.TaskStarted, + payload: ["가".repeat(1000)], + taskId: 1, + } satisfies TaskEvent, + }, + }) + DELIMITER + const closePeer = await startOneShotPeer(socketPath, frame) + + const client = new EvalIpcClient(socketPath, (...args) => logs.push(args), { maxFrameBytes: ceiling }) + client.on(IpcMessageType.TaskEvent, (event) => events.push(event as TaskEvent)) + + try { + await waitFor(() => !client.isConnected && ceilingLogs(logs).length > 0, 10_000) + + assert.equal( + ceilingLogs(logs).length, + 1, + "the byte ceiling must drop a frame whose code-unit count is under it", + ) + assert.equal(events.length, 0, "the multibyte frame past the byte ceiling is never dispatched") + assert.equal(client.isConnected, false, "the connection is dropped") + } finally { + client.disconnect() + await closePeer() + await fs.rm(dir, { recursive: true, force: true }) + } + }) + it("keeps the production frame ceiling far above any legitimate task event", () => { assert.equal(MAX_FRAME_BYTES, 64 * 1024 * 1024) }) diff --git a/scripts/agent-evals/ipc-client.ts b/scripts/agent-evals/ipc-client.ts index b7e16e34d3..e7e6223120 100644 --- a/scripts/agent-evals/ipc-client.ts +++ b/scripts/agent-evals/ipc-client.ts @@ -124,7 +124,19 @@ export class EvalIpcClient extends EventEmitter { let index: number while ((index = chunk.indexOf(DELIMITER, offset)) !== -1) { - const frame = this._buffer + chunk.slice(offset, index) + const segment = chunk.slice(offset, index) + const frameBytes = this._bufferBytes + Buffer.byteLength(segment, "utf8") + + // A delimiter in the same chunk as the overrun must not buy the frame a + // pass: the ceiling applies to the assembled frame, not just the tail. + // Returning (not breaking) stops the scan, so the closed connection + // cannot re-populate the buffer and log a second time. + if (frameBytes > this._maxFrameBytes) { + this.onOversizedFrame() + return + } + + const frame = this._buffer + segment this._buffer = "" this._bufferBytes = 0 if (frame.length > 0) this.onFrame(frame) From f5ffe8c60824ccc67c7a354e893b2abec5c26d35 Mon Sep 17 00:00:00 2001 From: myk1yt Date: Sat, 31 Oct 2026 17:03:15 +0900 Subject: [PATCH 6/6] test(ipc): destroy the partial-frame peer only after the write flushes What: in the partial-frame carryover test, the connection-1 peer now destroys the socket from the write callback instead of calling socket.destroy() on the line after socket.write(). Why: socket.destroy() can discard a pending write, so the 34-byte partial frame was not structurally guaranteed to reach the client before the socket died. The test then could pass on the second session alone, which is exactly the coverage gap CodeRabbit raised on PR #2265 (thread PRRT_kwDOSK-bwM6yOYMK). The callback form makes the red-on-revert proof structural instead of empirical. Impact: test-only, one executable line. No production code, no protocol change. The 34-byte write is in practice already delivered (panel probes: 40/40 and 20/20), so this removes a latent soundness hole rather than a live flake. --- scripts/agent-evals/ipc-client.test.ts | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/scripts/agent-evals/ipc-client.test.ts b/scripts/agent-evals/ipc-client.test.ts index fe91c3d94d..16edf6c0b6 100644 --- a/scripts/agent-evals/ipc-client.test.ts +++ b/scripts/agent-evals/ipc-client.test.ts @@ -324,8 +324,7 @@ describe("eval ipc client", () => { if (connections === 1) { // Only half a frame, no ack, then the peer dies. The rest can // never arrive, so these bytes are unrecoverable garbage. - socket.write(taskEvent.slice(0, 34)) - socket.destroy() + socket.write(taskEvent.slice(0, 34), () => socket.destroy()) return }