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..16edf6c0b6 --- /dev/null +++ b/scripts/agent-evals/ipc-client.test.ts @@ -0,0 +1,738 @@ +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, MAX_FRAME_BYTES, 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", () => {}) + // 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) => { + 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) + 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())) + }, + } +} + +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() + 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("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") + 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") + } + }) + + 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[] = [] + // 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", + 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. 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)) + void secondHalfGate.then(() => socket.write(encoded.slice(cut))) + }) + + 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") + + 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 }) + } + }) + + 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 { + // 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") + 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("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 { + // 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, + 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("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 new file mode 100644 index 0000000000..e7e6223120 --- /dev/null +++ b/scripts/agent-evals/ipc-client.ts @@ -0,0 +1,260 @@ +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 + +// 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 + +// 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 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 = "" + // 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 + private _closed = false + + 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() + } + + // 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) + } + + // 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 + + this._retry = setTimeout(() => this.connect(), RETRY_MS) + this._retry.unref?.() + } + + private onData(chunk: string) { + // A frame is complete only once the delimiter arrives; a partial frame + // 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 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) + offset = index + DELIMITER.length + } + + const tail = chunk.slice(offset) + + if (tail.length > 0) { + this._buffer += tail + this._bufferBytes += Buffer.byteLength(tail, "utf8") + } + + 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) { + clearTimeout(this._retry) + this._retry = undefined + } + + this.log("[client#onData] frame exceeded MAX_FRAME_BYTES, dropping connection") + + this._socket?.destroy() + } + + 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