From 6dc91157e835be905f10adf83dadafc8e5604996 Mon Sep 17 00:00:00 2001 From: Arul Sharma <31745423+arul28@users.noreply.github.com> Date: Fri, 28 Aug 2026 22:03:15 -0400 Subject: [PATCH 1/3] Fix Cursor hook.sock recycle and forked-worker screenshot IPC. Token recycle used to unlink the replacement worker's policy gate; per-instance sockets wait for the dying process. Cursor and Pi now send attachment paths over child.send instead of screenshot base64, and the worker re-opens those files through the attachment sandbox. Co-authored-by: Cursor --- .../main/services/chat/agentChatService.ts | 109 +++++-- .../main/services/chat/cursorSdkPool.test.ts | 281 +++++++++++++++++- .../src/main/services/chat/cursorSdkPool.ts | 97 +++++- .../main/services/chat/cursorSdkProtocol.ts | 17 +- .../src/main/services/chat/cursorSdkWorker.ts | 37 ++- .../main/services/chat/piSdkProtocol.test.ts | 48 +++ .../src/main/services/chat/piSdkProtocol.ts | 31 +- .../src/main/services/chat/piSdkWorker.ts | 22 +- .../chat/workerAttachmentImages.test.ts | 70 +++++ .../services/chat/workerAttachmentImages.ts | 102 +++++++ .../desktop/src/main/services/shared/utils.ts | 9 +- docs/features/chat/README.md | 20 +- 12 files changed, 762 insertions(+), 81 deletions(-) create mode 100644 apps/desktop/src/main/services/chat/workerAttachmentImages.test.ts create mode 100644 apps/desktop/src/main/services/chat/workerAttachmentImages.ts diff --git a/apps/desktop/src/main/services/chat/agentChatService.ts b/apps/desktop/src/main/services/chat/agentChatService.ts index 1d0eecfb0..bcee2c6a2 100644 --- a/apps/desktop/src/main/services/chat/agentChatService.ts +++ b/apps/desktop/src/main/services/chat/agentChatService.ts @@ -713,6 +713,7 @@ import { type CursorSdkHookRequest, type CursorSdkPermissionPolicy, } from "./cursorSdkProtocol"; +import { workerPathImagesFromAttachments, type WorkerIpcImage } from "./workerAttachmentImages"; import { resolveCursorCloudCreateCloudExtras } from "./cursorCloudCreateOptions"; import type { DroidSdkAskUserRequest, @@ -23810,14 +23811,10 @@ export function createAgentChatService(args: { const guidance = buildAdeGuidanceForLane(managed.laneWorktreePath, managed.session); if (guidance.trim()) prompt = `${guidance}\n\n${prompt}`; } - const promptBlocks = await buildAgentPromptBlocks(prompt, args.resolvedAttachments ?? []); - const promptText = promptBlocks - .filter((block): block is { type: "text"; text: string } => block.type === "text") - .map((block) => block.text) - .join("\n\n"); - const images = promptBlocks - .filter((block): block is { type: "image"; data: string; mimeType: string } => block.type === "image") - .map(({ data, mimeType }) => ({ data, mimeType })); + const { promptText, images } = await buildPiWorkerPrompt( + prompt, + args.resolvedAttachments ?? [], + ); args.onDispatched?.(); const accepted = runtime.sdk.sendPrompt({ prompt: promptText, @@ -36722,6 +36719,69 @@ export function createAgentChatService(args: { return blocks; }; + const promptTextWithoutInlineImages = async ( + promptText: string, + resolvedAttachments: ResolvedAgentChatFileRef[], + ): Promise => { + const promptBlocks = await buildAgentPromptBlocks( + promptText, + resolvedAttachments.filter((attachment) => attachment.type !== "image" && attachment.type !== "image-url"), + ); + return promptBlocks + .filter((block): block is { type: "text"; text: string } => block.type === "text") + .map((block) => block.text) + .join("\n\n"); + }; + + const pathImagesFromResolved = ( + resolvedAttachments: ResolvedAgentChatFileRef[], + ): Array<{ path: string; mimeType: string; rootPath: string }> => ( + workerPathImagesFromAttachments( + resolvedAttachments.flatMap((attachment) => { + if (attachment.type !== "image") return []; + const rootPath = attachment._rootPath.trim(); + if (!rootPath) return []; + return [{ + path: attachment.path, + resolvedPath: attachment._resolvedPath, + rootPath, + }]; + }), + ) + ); + + const buildCursorWorkerPrompt = async ( + promptText: string, + resolvedAttachments: ResolvedAgentChatFileRef[], + ): Promise<{ promptText: string; images: WorkerIpcImage[] }> => ({ + promptText: await promptTextWithoutInlineImages(promptText, resolvedAttachments), + images: [ + ...pathImagesFromResolved(resolvedAttachments), + ...resolvedAttachments.flatMap((attachment) => { + if (attachment.type !== "image-url") return []; + const url = attachment.url?.trim(); + if (!url) return []; + return [{ url }]; + }), + ], + }); + + const buildPiWorkerPrompt = async ( + promptText: string, + resolvedAttachments: ResolvedAgentChatFileRef[], + ): Promise<{ promptText: string; images: Array<{ path: string; mimeType: string; rootPath: string }> }> => { + const text = await promptTextWithoutInlineImages(promptText, resolvedAttachments); + const urlHints = resolvedAttachments.flatMap((attachment) => { + if (attachment.type !== "image-url") return []; + const url = attachment.url?.trim(); + return url ? [`Image URL: ${url}`] : []; + }); + return { + promptText: [...(text ? [text] : []), ...urlHints].join("\n\n"), + images: pathImagesFromResolved(resolvedAttachments), + }; + }; + const mapChatDecisionToDroidPermission = ( decision: AgentChatApprovalDecision | undefined, request: DroidSdkPermissionRequest, @@ -37562,14 +37622,7 @@ export function createAgentChatService(args: { } } - const promptBlocks = await buildAgentPromptBlocks(composed, args.resolvedAttachments); - const promptText = promptBlocks - .filter((block): block is { type: "text"; text: string } => block.type === "text") - .map((block) => block.text) - .join("\n\n"); - const images = promptBlocks - .filter((block): block is { type: "image"; data: string; mimeType: string } => block.type === "image") - .map((block) => ({ data: block.data, mimeType: block.mimeType })); + const { promptText, images } = await buildCursorWorkerPrompt(composed, args.resolvedAttachments); const modelParams = resolveCursorSdkModelParamsForSession(managed.session, runtime.modelSdkId); persistChatState(managed); @@ -37583,6 +37636,7 @@ export function createAgentChatService(args: { approvalPolicy: approvalPolicyLabel(policy.approvalPolicy), fullAuto: policy.fullAuto, transport: "sdk", + imageCount: images.length, }); if (args.onDispatched) { @@ -37630,7 +37684,7 @@ export function createAgentChatService(args: { const { watch: silenceWatch, guard: silenceGuard } = armCursorSdkSilenceWatch(runtime, turnId); const sendPromise = runtime.sdk.sendPrompt({ promptText, - images, + ...(images.length ? { images } : {}), modelSdkId: runtime.modelSdkId, ...(modelParams?.length ? { modelParams } : {}), // Set only when ADE deliberately abandoned the previous run (recovery @@ -38325,11 +38379,7 @@ export function createAgentChatService(args: { cloudComposed = `${injected}\n\n${cloudComposed}`; } } - const promptBlocks = await buildAgentPromptBlocks(cloudComposed, args.resolvedAttachments); - const promptText = promptBlocks - .filter((block): block is { type: "text"; text: string } => block.type === "text") - .map((block) => block.text) - .join("\n\n"); + const { promptText, images } = await buildCursorWorkerPrompt(cloudComposed, args.resolvedAttachments); const cloudLogModelParams = runtime.modelSdkId ? resolveCursorSdkModelParamsForSession(managed.session, runtime.modelSdkId) @@ -38342,6 +38392,7 @@ export function createAgentChatService(args: { hasAgentId: Boolean(managed.session.cursorCloudAgentId), ...(runtime.modelSdkId ? { modelSdkId: runtime.modelSdkId } : {}), ...(cloudLogModelParams?.length ? { modelParams: cursorModelParamsForLog(cloudLogModelParams) } : {}), + imageCount: images.length, }); if (args.onDispatched) { @@ -38366,6 +38417,7 @@ export function createAgentChatService(args: { apiKey, agentId: managed.session.cursorCloudAgentId, promptText, + ...(images.length ? { images } : {}), idempotencyKey: cursorCloudIdempotencyKey(managed, turnId, "followup"), mode: sdkMode, ...(runtime.modelSdkId ? { modelSdkId: runtime.modelSdkId } : {}), @@ -38411,6 +38463,7 @@ export function createAgentChatService(args: { const payload: CursorSdkCloudSendStreamPayload = { apiKey, promptText, + ...(images.length ? { images } : {}), repoUrl, idempotencyKey: cursorCloudIdempotencyKey(managed, turnId, "create"), mode: sdkMode, @@ -40556,14 +40609,10 @@ export function createAgentChatService(args: { allowActiveSession: true, }); if (!preparedSteer) return { steerId, queued: false }; - const promptBlocks = await buildAgentPromptBlocks(preparedSteer.submittedText, preparedSteer.resolvedAttachments); - const promptText = promptBlocks - .filter((block): block is { type: "text"; text: string } => block.type === "text") - .map((block) => block.text) - .join("\n\n"); - const images = promptBlocks - .filter((block): block is { type: "image"; data: string; mimeType: string } => block.type === "image") - .map(({ data, mimeType }) => ({ data, mimeType })); + const { promptText, images } = await buildPiWorkerPrompt( + preparedSteer.submittedText, + preparedSteer.resolvedAttachments, + ); await runtime.sdk.steer(promptText, images); preparedSteer.onDispatched?.(); options?.onAcceptedDispatch?.(); diff --git a/apps/desktop/src/main/services/chat/cursorSdkPool.test.ts b/apps/desktop/src/main/services/chat/cursorSdkPool.test.ts index d02d25afa..1de3de00d 100644 --- a/apps/desktop/src/main/services/chat/cursorSdkPool.test.ts +++ b/apps/desktop/src/main/services/chat/cursorSdkPool.test.ts @@ -37,11 +37,16 @@ vi.mock("node:child_process", () => ({ class FakeSdkChild extends EventEmitter { stdout = new EventEmitter(); stderr = new EventEmitter(); + pid = 4242; exitCode: number | null = null; killed = false; + connected = true; disposeCount = 0; + sent: unknown[] = []; + private exited = false; - send(message: { type?: string; requestId?: string }): boolean { + send(message: { type?: string; requestId?: string; payload?: unknown }): boolean { + this.sent.push(message); if (message.type === "init" && message.requestId) { queueMicrotask(() => { this.emit("message", { @@ -52,15 +57,53 @@ class FakeSdkChild extends EventEmitter { }); }); } + if (message.type === "send" && message.requestId) { + queueMicrotask(() => { + this.emit("message", { + type: "response", + requestId: message.requestId, + ok: true, + result: {}, + }); + }); + } if (message.type === "dispose") { this.disposeCount += 1; + queueMicrotask(() => this.finishExit(0, null)); } return true; } + finishExit(code: number | null, signal: NodeJS.Signals | null): void { + if (this.exited) return; + this.exited = true; + this.exitCode = code; + this.connected = false; + this.emit("exit", code, signal); + } + kill(signal?: NodeJS.Signals): boolean { this.killed = true; - this.emit("exit", null, signal ?? "SIGTERM"); + this.finishExit(null, signal ?? "SIGTERM"); + return true; + } +} + +class DelayedExitChild extends FakeSdkChild { + override send(message: { type?: string; requestId?: string }): boolean { + if (message.type === "init" && message.requestId) { + queueMicrotask(() => { + this.emit("message", { + type: "response", + requestId: message.requestId, + ok: true, + result: { agentId: "agent-1" }, + }); + }); + } + if (message.type === "dispose") { + this.disposeCount += 1; + } return true; } } @@ -168,6 +211,11 @@ function makeTempDir(prefix: string): string { return dir; } +function forkedSocketPath(callIndex: number): string | undefined { + const options = forkMock.mock.calls[callIndex]?.[2] as { env?: NodeJS.ProcessEnv } | undefined; + return options?.env?.ADE_CURSOR_SDK_SOCKET; +} + describe("Cursor SDK pool paths", () => { it("uses the real user home while keeping ADE runtime state under the project cache", () => { const projectRoot = path.join(os.tmpdir(), "ade-project"); @@ -175,6 +223,7 @@ describe("Cursor SDK pool paths", () => { const paths = buildCursorSdkPaths({ projectRoot, poolKey: "lane:/repo:session", + instanceId: "worker-a", userHomeDir, }); @@ -189,6 +238,26 @@ describe("Cursor SDK pool paths", () => { } }); + it("gives each worker instance its own hook socket while sharing the pool state root", () => { + const projectRoot = path.join(os.tmpdir(), "ade-project"); + const args = { + projectRoot, + poolKey: "lane:/repo:session", + userHomeDir: path.join(os.tmpdir(), "real-home"), + }; + const first = buildCursorSdkPaths({ ...args, instanceId: "worker-a" }); + const second = buildCursorSdkPaths({ ...args, instanceId: "worker-b" }); + expect(first.socketPath).not.toBe(second.socketPath); + expect(first.stateRoot).toBe(second.stateRoot); + if (process.platform === "win32") { + expect(first.socketPath.startsWith("\\\\.\\pipe\\ade-cursor-sdk-")).toBe(true); + expect(second.socketPath.startsWith("\\\\.\\pipe\\ade-cursor-sdk-")).toBe(true); + } else { + expect(path.dirname(first.socketPath)).not.toBe(path.dirname(second.socketPath)); + expect(path.basename(first.socketPath)).toBe("hook.sock"); + } + }); + it("retries one-shot SDK state removal until the worker releases its handles", async () => { // Cleanup runs while the worker is still shutting down. On Windows the // SDK's open `state/index.db` makes the first `rmSync` fail with EBUSY and @@ -236,11 +305,13 @@ describe("Cursor SDK pool paths", () => { const first = buildCursorSdkPaths({ projectRoot, poolKey: "session-1:composer-2.5:full-auto", + instanceId: "shared", stateKey: "session-1:lane-1:state", }); const second = buildCursorSdkPaths({ projectRoot, poolKey: "session-1:claude-sonnet-5:edit", + instanceId: "shared", stateKey: "session-1:lane-1:state", }); @@ -249,6 +320,61 @@ describe("Cursor SDK pool paths", () => { expect(second.socketPath).not.toBe(first.socketPath); }); + it("gives each worker instance its own hook socket while keeping durable state stable", () => { + const projectRoot = path.join(os.tmpdir(), "ade-project"); + const shared = { + projectRoot, + poolKey: "session-1:composer-2.5:full-auto", + stateKey: "session-1:lane-1:state", + }; + const first = buildCursorSdkPaths({ ...shared, instanceId: "worker-a" }); + const second = buildCursorSdkPaths({ ...shared, instanceId: "worker-b" }); + + expect(second.stateRoot).toBe(first.stateRoot); + expect(second.cacheRoot).toBe(first.cacheRoot); + expect(second.socketPath).not.toBe(first.socketPath); + if (process.platform !== "win32") { + expect(path.basename(first.socketPath)).toBe("hook.sock"); + expect(path.basename(second.socketPath)).toBe("hook.sock"); + expect(path.dirname(second.socketPath)).not.toBe(path.dirname(first.socketPath)); + } + }); + + it("refuses an empty worker instance id", () => { + expect(() => buildCursorSdkPaths({ + projectRoot: path.join(os.tmpdir(), "ade-project"), + poolKey: "session-1:composer-2.5:full-auto", + instanceId: " ", + })).toThrow(/instance id is required/); + }); + + it("does not delete a sibling worker's hook socket directory during cleanup", () => { + if (process.platform === "win32") return; + const cacheRoot = makeTempDir("ade-cursor-cleanup-socket-"); + const stateRoot = path.join(cacheRoot, "state"); + fs.mkdirSync(stateRoot, { recursive: true }); + const poolRoot = makeTempDir("ade-cursor-sdk-pool-"); + const firstInstance = path.join(poolRoot, "worker-a"); + const secondInstance = path.join(poolRoot, "worker-b"); + fs.mkdirSync(firstInstance, { recursive: true }); + fs.mkdirSync(secondInstance, { recursive: true }); + const firstSock = path.join(firstInstance, "hook.sock"); + const secondSock = path.join(secondInstance, "hook.sock"); + fs.writeFileSync(firstSock, ""); + fs.writeFileSync(secondSock, ""); + + cleanupCursorSdkRuntimePaths({ + cacheRoot, + stateRoot, + socketPath: firstSock, + cleanupStateRoot: true, + }); + + expect(fs.existsSync(firstInstance)).toBe(false); + expect(fs.existsSync(secondSock)).toBe(true); + expect(fs.existsSync(poolRoot)).toBe(true); + }); + it("builds a worker environment with real HOME parity and no ADE brain ownership metadata", () => { const cliRoot = makeTempDir("ade-cli-current-"); const cliBinDir = path.join(cliRoot, "bin"); @@ -451,10 +577,125 @@ describe("Cursor SDK pool paths", () => { const third = await acquireCursorSdkConnection(args); expect(third.pooled).not.toBe(first.pooled); expect(forkMock).toHaveBeenCalledTimes(2); + expect(forkedSocketPath(1)).toBeTruthy(); + expect(forkedSocketPath(1)).not.toBe(forkedSocketPath(0)); releaseCursorSdkConnection(poolKey, third.generation); }); + it("does not fork a replacement until the poisoned worker has exited", async () => { + const firstChild = new DelayedExitChild(); + const nextChild = new FakeSdkChild(); + forkMock.mockReturnValueOnce(firstChild); + const poolKey = `test-replace-wait:${Date.now()}:${Math.random()}`; + const args = { + poolKey, + projectRoot: path.join(os.tmpdir(), "ade-project"), + workspacePath: path.join(os.tmpdir(), "ade-workspace"), + modelSdkId: "cursor-model", + sessionId: "session-1", + policy: { ...TEST_POLICY }, + }; + + const first = await acquireCursorSdkConnection(args); + expect(poisonCursorSdkConnection(poolKey, first.generation)).toBe(true); + expect(firstChild.disposeCount).toBe(1); + + forkMock.mockReturnValue(nextChild); + let replaced = false; + const pending = acquireCursorSdkConnection(args).then((result) => { + replaced = true; + return result; + }); + await new Promise((resolve) => setTimeout(resolve, 30)); + expect(forkMock).toHaveBeenCalledTimes(1); + expect(replaced).toBe(false); + + firstChild.finishExit(0, null); + const second = await pending; + expect(replaced).toBe(true); + expect(second.pooled).not.toBe(first.pooled); + expect(forkMock).toHaveBeenCalledTimes(2); + expect(forkedSocketPath(1)).not.toBe(forkedSocketPath(0)); + + releaseCursorSdkConnection(poolKey, second.generation); + }); + + it("does not treat a dispatched kill plus IPC error as the worker exiting", async () => { + const firstChild = new DelayedExitChild(); + const nextChild = new FakeSdkChild(); + forkMock.mockReturnValueOnce(firstChild); + const poolKey = `test-replace-epipe:${Date.now()}:${Math.random()}`; + const args = { + poolKey, + projectRoot: path.join(os.tmpdir(), "ade-project"), + workspacePath: path.join(os.tmpdir(), "ade-workspace"), + modelSdkId: "cursor-model", + sessionId: "session-1", + policy: { ...TEST_POLICY }, + }; + + const first = await acquireCursorSdkConnection(args); + expect(poisonCursorSdkConnection(poolKey, first.generation)).toBe(true); + firstChild.killed = true; + firstChild.emit("error", Object.assign(new Error("write EPIPE"), { code: "EPIPE" })); + + forkMock.mockReturnValue(nextChild); + let replaced = false; + const pending = acquireCursorSdkConnection(args).then((result) => { + replaced = true; + return result; + }); + await new Promise((resolve) => setTimeout(resolve, 30)); + expect(forkMock).toHaveBeenCalledTimes(1); + expect(replaced).toBe(false); + + firstChild.finishExit(null, "SIGTERM"); + const second = await pending; + expect(replaced).toBe(true); + expect(second.pooled).not.toBe(first.pooled); + expect(forkMock).toHaveBeenCalledTimes(2); + + releaseCursorSdkConnection(poolKey, second.generation); + }); + + it("waits for exit when a live worker's IPC channel errors before dispose", async () => { + const firstChild = new DelayedExitChild(); + const nextChild = new FakeSdkChild(); + forkMock.mockReturnValueOnce(firstChild); + const poolKey = `test-live-epipe:${Date.now()}:${Math.random()}`; + const args = { + poolKey, + projectRoot: path.join(os.tmpdir(), "ade-project"), + workspacePath: path.join(os.tmpdir(), "ade-workspace"), + modelSdkId: "cursor-model", + sessionId: "session-1", + policy: { ...TEST_POLICY }, + }; + + const first = await acquireCursorSdkConnection(args); + firstChild.emit("error", Object.assign(new Error("write EPIPE"), { code: "EPIPE" })); + + forkMock.mockReturnValue(nextChild); + let replaced = false; + const pending = acquireCursorSdkConnection(args).then((result) => { + replaced = true; + return result; + }); + await new Promise((resolve) => setTimeout(resolve, 30)); + expect(forkMock).toHaveBeenCalledTimes(1); + expect(replaced).toBe(false); + expect(firstChild.disposeCount).toBe(1); + + firstChild.finishExit(null, "SIGTERM"); + const second = await pending; + expect(replaced).toBe(true); + expect(second.pooled).not.toBe(first.pooled); + expect(forkMock).toHaveBeenCalledTimes(2); + + releaseCursorSdkConnection(poolKey, second.generation); + }); + it("reuses a oneshot worker during idle instead of colliding on cleanup", async () => { const firstChild = new FakeSdkChild(); const secondChild = new FakeSdkChild(); @@ -513,6 +754,42 @@ describe("Cursor SDK pool paths", () => { releaseCursorSdkConnection(poolKey, acquired.generation); }); + it("sends screenshot paths over worker IPC instead of inline bytes", async () => { + const child = new FakeSdkChild(); + forkMock.mockReturnValue(child); + const poolKey = `test-image-paths:${Date.now()}:${Math.random()}`; + const acquired = await acquireCursorSdkConnection({ + poolKey, + projectRoot: path.join(os.tmpdir(), "ade-project"), + workspacePath: path.join(os.tmpdir(), "ade-workspace"), + modelSdkId: "cursor-model", + sessionId: "session-1", + policy: { ...TEST_POLICY }, + }); + + await acquired.pooled.sendPrompt({ + promptText: "compare these screens", + images: [ + { path: "/repo/.ade/attachments/a.png", mimeType: "image/png", rootPath: "/repo" }, + { path: "/repo/.ade/attachments/b.png", mimeType: "image/png", rootPath: "/repo" }, + ], + }); + + const sendReq = child.sent.find((message) => ( + message + && typeof message === "object" + && "type" in message + && message.type === "send" + )) as { payload?: { images?: Array<{ path?: string; data?: string }> } } | undefined; + expect(sendReq?.payload?.images).toEqual([ + { path: "/repo/.ade/attachments/a.png", mimeType: "image/png", rootPath: "/repo" }, + { path: "/repo/.ade/attachments/b.png", mimeType: "image/png", rootPath: "/repo" }, + ]); + expect(sendReq?.payload?.images?.some((image) => image.data)).toBeFalsy(); + + releaseCursorSdkConnection(poolKey, acquired.generation); + }); + it("does not reuse a worker whose IPC channel has closed", async () => { const firstChild = new FakeSdkChild(); const secondChild = new FakeSdkChild(); diff --git a/apps/desktop/src/main/services/chat/cursorSdkPool.ts b/apps/desktop/src/main/services/chat/cursorSdkPool.ts index 5be39d8f2..79e933af5 100644 --- a/apps/desktop/src/main/services/chat/cursorSdkPool.ts +++ b/apps/desktop/src/main/services/chat/cursorSdkPool.ts @@ -77,6 +77,8 @@ export type CursorSdkPooled = { updatePolicy: (policy: CursorSdkPermissionPolicy) => Promise; cancel: () => Promise; dispose: () => void; + /** Resolves only after the worker process has actually exited. */ + waitForExit: () => Promise; }; let cursorSdkGenCounter = 0; @@ -93,6 +95,8 @@ type CursorSdkPoolEntry = { const pools = new Map(); const pendingInits = new Map>(); +/** Poisoned/released workers still shutting down, keyed by pool key. */ +const departingWorkers = new Map>(); const STALE_INIT_RETRY_LIMIT = 2; /** * How long the worker gets to answer the IPC `dispose` request before the pool @@ -100,6 +104,8 @@ const STALE_INIT_RETRY_LIMIT = 2; * closing the SDK agent, and on Windows it is the only orderly path there is. */ const CURSOR_SDK_DISPOSE_GRACE_MS = 3_000; +/** Cap how long a replacement waits for the previous worker of the same pool key. */ +const CURSOR_SDK_REPLACE_WAIT_MS = CURSOR_SDK_DISPOSE_GRACE_MS + 500; const CURSOR_SDK_WORKER_ENV_DENYLIST = [ "CURSOR_API_KEY", "CURSOR_AUTH_TOKEN", @@ -140,13 +146,20 @@ function resolveWorkerPath(): string { return candidates[0]!; } -function socketPathFor(poolKey: string): string { +function socketPathFor(poolKey: string, instanceId: string): string { + const trimmedInstance = instanceId.trim(); + if (!trimmedInstance) { + throw new Error("Cursor SDK worker instance id is required."); + } const name = hashKey(poolKey); + const instance = hashKey(trimmedInstance); if (process.platform === "win32") { - return `\\\\.\\pipe\\ade-cursor-sdk-${name}`; + return `\\\\.\\pipe\\ade-cursor-sdk-${name}-${instance}`; } const userPart = typeof process.getuid === "function" ? String(process.getuid()) : hashKey(os.homedir()); - return path.join(os.tmpdir(), `ade-cursor-sdk-${userPart}`, name, "hook.sock"); + // Per-instance directory so a dying worker's close()/unlink cannot delete + // the replacement's hook socket (same pool key, overlapping shutdown). + return path.join(os.tmpdir(), `ade-cursor-sdk-${userPart}`, name, instance, "hook.sock"); } export function sanitizeCursorSdkWorkerBaseEnv(base: NodeJS.ProcessEnv): NodeJS.ProcessEnv { @@ -376,10 +389,18 @@ function ensurePrivateDirectory(dir: string): void { function ensurePrivateSocketPath(socketPath: string): void { if (process.platform === "win32") return; - const rootDir = path.dirname(path.dirname(socketPath)); - const socketDir = path.dirname(socketPath); - ensurePrivateDirectory(rootDir); - ensurePrivateDirectory(socketDir); + const dirs: string[] = []; + let dir = path.dirname(socketPath); + for (let i = 0; i < 6; i += 1) { + dirs.push(dir); + if (path.basename(dir).startsWith("ade-cursor-sdk-")) break; + const parent = path.dirname(dir); + if (parent === dir) break; + dir = parent; + } + for (let i = dirs.length - 1; i >= 0; i -= 1) { + ensurePrivateDirectory(dirs[i]!); + } } export function resolveCursorSdkUserHome(env: NodeJS.ProcessEnv = process.env): string { @@ -392,6 +413,7 @@ export function resolveCursorSdkUserHome(env: NodeJS.ProcessEnv = process.env): export function buildCursorSdkPaths(args: { projectRoot: string; poolKey: string; + instanceId: string; stateKey?: string; userHomeDir?: string; }): { userHomeDir: string; cacheRoot: string; stateRoot: string; socketPath: string } { @@ -401,7 +423,7 @@ export function buildCursorSdkPaths(args: { userHomeDir: args.userHomeDir?.trim() || resolveCursorSdkUserHome(), cacheRoot, stateRoot: path.join(cacheRoot, "state"), - socketPath: socketPathFor(args.poolKey), + socketPath: socketPathFor(args.poolKey, args.instanceId), }; } @@ -454,6 +476,7 @@ export async function acquireCursorSdkConnection(args: { return { pooled: existing.pooled, generation: existing.generation }; } if (existing) disposeCursorSdkPoolEntry(args.poolKey, existing); + await waitForDepartingCursorSdkWorker(args.poolKey); let initOwner = false; let init = pendingInits.get(args.poolKey); @@ -484,9 +507,11 @@ export async function acquireCursorSdkConnection(args: { async function createCursorSdkConnection(args: Parameters[0]): Promise { const workerPath = resolveWorkerPath(); + const instanceId = randomUUID(); const paths = buildCursorSdkPaths({ projectRoot: args.projectRoot, poolKey: args.poolKey, + instanceId, stateKey: args.stateKey, }); fs.mkdirSync(paths.stateRoot, { recursive: true }); @@ -524,6 +549,16 @@ async function createCursorSdkConnection(args: Parameters void; + let exitSettled = false; + const settleExit = (): void => { + if (exitSettled) return; + exitSettled = true; + resolveExit(); + }; + const exitPromise = new Promise((resolve) => { + resolveExit = resolve; + }); const rememberStderr = (text: string): void => { const trimmed = text.trim(); if (!trimmed) return; @@ -641,6 +676,7 @@ async function createCursorSdkConnection(args: Parameters exitPromise, }; child.on("message", (raw: unknown) => { @@ -784,7 +820,19 @@ async function createCursorSdkConnection(args: Parameters { rejectPending(normalizeIpcSendError(error)); - cleanupPoolEntry(pooled); + if (child.pid == null) { + cleanupPoolEntry(pooled); + settleExit(); + return; + } + // A live worker with a broken IPC channel is still holding state/index.db. + // Evict through dispose so the next acquire waits for a real `exit`. + for (const [poolKey, entry] of pools) { + if (entry.pooled === pooled) { + disposeCursorSdkPoolEntry(poolKey, entry); + return; + } + } }); child.on("exit", (code, signal) => { @@ -795,6 +843,7 @@ async function createCursorSdkConnection(args: Parameters((resolve) => { + const timer = setTimeout(resolve, CURSOR_SDK_REPLACE_WAIT_MS); + timer.unref(); + }), + ]).catch(() => {}); if (args.cleanupStateRoot) { cleanupCursorSdkRuntimePaths({ cacheRoot: paths.cacheRoot, @@ -877,6 +933,9 @@ export function cleanupCursorSdkRuntimePaths(entry: { const targets = new Set(); targets.add(entry.cacheRoot ?? entry.stateRoot); if (process.platform !== "win32" && entry.socketPath) { + // Per-instance socket directory (`...///hook.sock`). Do not + // walk up to the pool directory — a replacement worker may already be + // listening there. targets.add(path.dirname(entry.socketPath)); } for (const target of targets) { @@ -902,10 +961,30 @@ function clearCursorSdkIdleTimer(entry: CursorSdkPoolEntry): void { function disposeCursorSdkPoolEntry(poolKey: string, entry: CursorSdkPoolEntry): void { clearCursorSdkIdleTimer(entry); pools.delete(poolKey); + trackDepartingCursorSdkWorker(poolKey, entry.pooled.waitForExit()); entry.pooled.dispose(); cleanupCursorSdkRuntimePaths(entry); } +function trackDepartingCursorSdkWorker(poolKey: string, wait: Promise): void { + const tracked = wait.finally(() => { + if (departingWorkers.get(poolKey) === tracked) departingWorkers.delete(poolKey); + }); + departingWorkers.set(poolKey, tracked); +} + +async function waitForDepartingCursorSdkWorker(poolKey: string): Promise { + const prior = departingWorkers.get(poolKey); + if (!prior) return; + await Promise.race([ + prior, + new Promise((resolve) => { + const timer = setTimeout(resolve, CURSOR_SDK_REPLACE_WAIT_MS); + timer.unref(); + }), + ]); +} + /** * Force-dispose a pooled worker regardless of its refcount, so the next * `acquireCursorSdkConnection` for this key forks a brand-new one. diff --git a/apps/desktop/src/main/services/chat/cursorSdkProtocol.ts b/apps/desktop/src/main/services/chat/cursorSdkProtocol.ts index 23ca5d651..abbd002cd 100644 --- a/apps/desktop/src/main/services/chat/cursorSdkProtocol.ts +++ b/apps/desktop/src/main/services/chat/cursorSdkProtocol.ts @@ -98,10 +98,17 @@ export type CursorSdkWorkerInit = { mcpServers?: Record; }; -export type CursorSdkUserImage = { - data: string; - mimeType: string; -}; +/** + * Worker-IPC image reference. Prefer `path` or `url` — never put multi-megabyte + * screenshot bytes on this object. The worker materializes `{ data, mimeType }` + * for `@cursor/sdk` locally. `data` remains for tests and tiny inline cases. + * Path images include `rootPath` so the worker re-opens through the same + * attachment sandbox the main process used to use. + */ +export type CursorSdkUserImage = + | { path: string; mimeType: string; rootPath: string } + | { data: string; mimeType: string } + | { url: string }; export type CursorSdkSendPrompt = { promptText: string; @@ -129,6 +136,7 @@ export type CursorSdkCloudRepoOverride = { export type CursorSdkCloudSendStreamPayload = { apiKey?: string | null; promptText: string; + images?: CursorSdkUserImage[]; modelSdkId?: string | null; modelParams?: CursorSdkModelParameterValue[]; idempotencyKey?: string | null; @@ -153,6 +161,7 @@ export type CursorSdkCloudFollowupPayload = { apiKey?: string | null; agentId: string; promptText: string; + images?: CursorSdkUserImage[]; modelSdkId?: string | null; modelParams?: CursorSdkModelParameterValue[]; idempotencyKey?: string | null; diff --git a/apps/desktop/src/main/services/chat/cursorSdkWorker.ts b/apps/desktop/src/main/services/chat/cursorSdkWorker.ts index 1f97a7419..836a8ad89 100644 --- a/apps/desktop/src/main/services/chat/cursorSdkWorker.ts +++ b/apps/desktop/src/main/services/chat/cursorSdkWorker.ts @@ -17,6 +17,8 @@ import type { CursorSdkHookRequest, CursorSdkModelParameterValue, CursorSdkPermissionPolicy, + CursorSdkSendPrompt, + CursorSdkUserImage, CursorSdkWorkerInit, CursorSdkWorkerRequest, CursorSdkWorkerResponse, @@ -25,6 +27,7 @@ import { isCursorSdkBackoffErrorText, isCursorSdkTransportErrorText, } from "./cursorSdkProtocol"; +import { materializeWorkerImages } from "./workerAttachmentImages"; import { cursorSdkResultWithStreamFailure, isCursorSdkSandboxUnsupportedError, @@ -301,6 +304,8 @@ function cursorRunRequestId(run: unknown): string | undefined { function removeSocketIfNeeded(socketPath: string): void { if (process.platform === "win32") return; + // Instance-specific path: unlinking here cannot delete a replacement worker's + // hook socket, which lives under a different instance directory. try { fs.rmSync(socketPath, { force: true }); } catch { @@ -521,15 +526,15 @@ async function initWorker(init: CursorSdkWorkerInit): Promise<{ agentId: string; return { agentId: agent.agentId, modelSdkId: init.modelSdkId }; } -async function sendPrompt(payload: { - promptText: string; - images?: Array<{ data: string; mimeType: string }>; - modelSdkId?: string | null; - modelParams?: CursorSdkModelParameterValue[]; - forceExpireActiveRun?: boolean; - idempotencyKey?: string | null; - mode?: CursorSdkAgentMode; -}): Promise { +async function cursorSdkSendMessage( + promptText: string, + images: CursorSdkUserImage[] | undefined, +) { + const materialized = await materializeWorkerImages(images, { label: "Cursor SDK" }); + return materialized.length ? { text: promptText, images: materialized } : promptText; +} + +async function sendPrompt(payload: CursorSdkSendPrompt): Promise { if (!agent || !initState) throw new Error("Cursor SDK worker is not initialized."); try { await applyLocalAgentOptions(); @@ -542,9 +547,7 @@ async function sendPrompt(payload: { }); } if (!agent) throw new Error("Cursor SDK worker is not initialized."); - const message = payload.images?.length - ? { text: payload.promptText, images: payload.images } - : payload.promptText; + const message = await cursorSdkSendMessage(payload.promptText, payload.images); const mode = payload.mode ?? cursorSdkLocalAgentMode(initState.policy); const idempotencyKey = trimIdempotencyKey(payload.idempotencyKey); const sendOptions: SendOptionsWithAdeMode = { @@ -975,7 +978,10 @@ async function handleCloudRequest(req: CursorSdkWorkerRequest): Promise await validateCloudModelSelection(req.payload.apiKey?.trim() || undefined, modelSelection); const cloudAgent = await Agent.create(buildCloudCreateOptions(req.payload)); const sendOpts = buildSendOptions(modelSelection, req.payload.idempotencyKey, req.payload.mode); - const run = await cloudAgent.send(req.payload.promptText, sendOpts); + const run = await cloudAgent.send( + await cursorSdkSendMessage(req.payload.promptText, req.payload.images), + sendOpts, + ); const result = await streamCloudRun({ requestId: req.requestId, agentId: cloudAgent.agentId, @@ -998,7 +1004,10 @@ async function handleCloudRequest(req: CursorSdkWorkerRequest): Promise const modelSelection = buildCursorModelSelection(req.payload.modelSdkId, req.payload.modelParams); await validateCloudModelSelection(req.payload.apiKey?.trim() || undefined, modelSelection); const sendOpts = buildSendOptions(modelSelection, req.payload.idempotencyKey, req.payload.mode); - const run = await cloudAgent.send(req.payload.promptText, sendOpts); + const run = await cloudAgent.send( + await cursorSdkSendMessage(req.payload.promptText, req.payload.images), + sendOpts, + ); const result = await streamCloudRun({ requestId: req.requestId, agentId: cloudAgent.agentId, diff --git a/apps/desktop/src/main/services/chat/piSdkProtocol.test.ts b/apps/desktop/src/main/services/chat/piSdkProtocol.test.ts index 6d8c7a5c8..7e28e7a02 100644 --- a/apps/desktop/src/main/services/chat/piSdkProtocol.test.ts +++ b/apps/desktop/src/main/services/chat/piSdkProtocol.test.ts @@ -20,6 +20,54 @@ describe("Pi SDK protocol", () => { })).toContain("non-empty prompt"); }); + it("accepts path images on send without inlined screenshot bytes", () => { + expect(validatePiSdkWorkerRequest({ + protocolVersion: PI_SDK_PROTOCOL_VERSION, + type: "send", + requestId: "send-images", + payload: { + prompt: "look", + images: [{ path: "/repo/.ade/attachments/shot.png", mimeType: "image/png", rootPath: "/repo" }], + }, + })).toBeNull(); + expect(validatePiSdkWorkerRequest({ + protocolVersion: PI_SDK_PROTOCOL_VERSION, + type: "steer", + requestId: "steer-images", + payload: { + prompt: "look", + images: [{ data: "abc", mimeType: "image/jpeg" }], + }, + })).toBeNull(); + expect(validatePiSdkWorkerRequest({ + protocolVersion: PI_SDK_PROTOCOL_VERSION, + type: "send", + requestId: "send-url", + payload: { + prompt: "look", + images: [{ url: "https://example.com/ui.png", mimeType: "image/png" }], + }, + })).toMatch(/path or data/u); + expect(validatePiSdkWorkerRequest({ + protocolVersion: PI_SDK_PROTOCOL_VERSION, + type: "follow_up", + requestId: "follow-both", + payload: { + prompt: "look", + images: [{ path: "/shot.png", data: "abc", mimeType: "image/png" }], + }, + })).toMatch(/path or data/u); + expect(validatePiSdkWorkerRequest({ + protocolVersion: PI_SDK_PROTOCOL_VERSION, + type: "send", + requestId: "send-no-root", + payload: { + prompt: "look", + images: [{ path: "/repo/.ade/attachments/shot.png", mimeType: "image/png" }], + }, + })).toMatch(/path or data/u); + }); + it("accepts an init message without requiring Pi types", () => { expect(validatePiSdkWorkerRequest({ protocolVersion: PI_SDK_PROTOCOL_VERSION, diff --git a/apps/desktop/src/main/services/chat/piSdkProtocol.ts b/apps/desktop/src/main/services/chat/piSdkProtocol.ts index 807b4fcf5..942dc3ce1 100644 --- a/apps/desktop/src/main/services/chat/piSdkProtocol.ts +++ b/apps/desktop/src/main/services/chat/piSdkProtocol.ts @@ -78,10 +78,17 @@ export type PiSdkWorkerInit = PiSdkPackageLocation & { approvalTools?: string[]; }; -export type PiSdkImage = { - data: string; - mimeType: string; -}; +/** + * Worker-IPC image reference. Prefer `path` — never put multi-megabyte + * screenshot bytes on this object. The worker materializes `{ data, mimeType }` + * for Pi locally. `data` remains for tests and tiny inline cases. Pi's prompt + * API has no remote-URL image form, so `url` is rejected at the protocol gate. + * Path images include `rootPath` so the worker re-opens through the attachment + * sandbox. + */ +export type PiSdkImage = + | { path: string; mimeType: string; rootPath: string } + | { data: string; mimeType: string }; export type PiSdkPromptPayload = { prompt: string; @@ -341,6 +348,18 @@ function isSessionTarget(value: unknown): boolean { && (value.resume.sessionId == null || typeof value.resume.sessionId === "string")); } +function isPiSdkImage(image: unknown): boolean { + if (!isRecord(image)) return false; + if (typeof image.url === "string" && image.url.trim()) return false; + const data = typeof image.data === "string" ? image.data.trim() : ""; + const filePath = typeof image.path === "string" ? image.path.trim() : ""; + const mimeType = typeof image.mimeType === "string" ? image.mimeType.trim() : ""; + const rootPath = typeof image.rootPath === "string" ? image.rootPath.trim() : ""; + if (data) return !filePath && mimeType.length > 0; + if (filePath) return mimeType.length > 0 && rootPath.length > 0; + return false; +} + /** Returns a human-readable validation failure without throwing. */ export function validatePiSdkWorkerRequest(raw: unknown): string | null { if (!isRecord(raw)) return "Pi SDK worker message must be an object."; @@ -389,8 +408,8 @@ export function validatePiSdkWorkerRequest(raw: unknown): string | null { if (["send", "steer", "follow_up"].includes(type)) { if (!isRecord(payload) || !nonEmptyString(payload.prompt)) return `Pi SDK ${type} requires a non-empty prompt.`; if (payload.images != null && (!Array.isArray(payload.images) - || payload.images.some((image) => !isRecord(image) || typeof image.data !== "string" || typeof image.mimeType !== "string"))) { - return "Pi SDK send images must contain data and mimeType strings."; + || payload.images.some((image) => !isPiSdkImage(image)))) { + return "Pi SDK send images must each contain a path or data with mimeType."; } if (type === "send" && payload.streamingBehavior != null && payload.streamingBehavior !== "steer" && payload.streamingBehavior !== "followUp") { diff --git a/apps/desktop/src/main/services/chat/piSdkWorker.ts b/apps/desktop/src/main/services/chat/piSdkWorker.ts index 753210e8d..15fbaa63b 100644 --- a/apps/desktop/src/main/services/chat/piSdkWorker.ts +++ b/apps/desktop/src/main/services/chat/piSdkWorker.ts @@ -12,10 +12,12 @@ import { type PiSdkModelRef, type PiSdkReady, type PiSdkSessionTarget, + type PiSdkImage, type PiSdkWorkerInit, type PiSdkWorkerRequest, type PiSdkWorkerResponse, } from "./piSdkProtocol"; +import { materializeWorkerImages } from "./workerAttachmentImages"; import { piSessionHeaderMatchesCwd, readPiSessionHeader } from "./piSessionStore"; import { PI_ASK_USER_TOOL_NAME, @@ -721,10 +723,16 @@ function requireSession(): PiSession { return session; } -function imageContents(images: Array<{ data: string; mimeType: string }> | undefined): unknown[] | undefined { - return images?.length - ? images.map((image) => ({ type: "image", data: image.data, mimeType: image.mimeType })) - : undefined; +async function imageContents(images: PiSdkImage[] | undefined): Promise { + const materialized = await materializeWorkerImages(images, { label: "Pi SDK" }); + const contents: Array<{ type: "image"; data: string; mimeType: string }> = []; + for (const image of materialized) { + if (!("data" in image)) { + throw new Error("Pi SDK image URLs are not supported."); + } + contents.push({ type: "image", data: image.data, mimeType: image.mimeType }); + } + return contents.length ? contents : undefined; } async function sendPrompt(request: Extract): Promise { @@ -733,7 +741,7 @@ async function sendPrompt(request: Extract post({ protocolVersion: PI_SDK_PROTOCOL_VERSION, type: "lifecycle", event: "prompt_started", requestId: request.requestId }); try { const promptOptions: Record = {}; - const images = imageContents(request.payload.images); + const images = await imageContents(request.payload.images); if (images) promptOptions.images = images; if (request.payload.streamingBehavior) promptOptions.streamingBehavior = request.payload.streamingBehavior; await method(active, "prompt").call(active, request.payload.prompt, promptOptions); @@ -910,12 +918,12 @@ async function dispatch(request: PiSdkWorkerRequest): Promise { + for (const dir of tempDirs.splice(0)) { + fs.rmSync(dir, { recursive: true, force: true }); + } +}); + +function makeTempDir(): string { + const dir = fs.mkdtempSync(path.join(os.tmpdir(), "ade-worker-images-")); + tempDirs.push(dir); + return dir; +} + +describe("workerPathImagesFromAttachments", () => { + it("sends local screenshots as paths with a sandbox root, not inline bytes", () => { + expect(workerPathImagesFromAttachments([ + { path: "shot.png", resolvedPath: "/repo/.ade/attachments/shot.png", rootPath: "/repo" }, + ])).toEqual([ + { path: "/repo/.ade/attachments/shot.png", mimeType: "image/png", rootPath: "/repo" }, + ]); + }); +}); + +describe("materializeWorkerImages", () => { + it("reads path images inside the attachment root and keeps URLs remote", async () => { + const root = makeTempDir(); + const filePath = path.join(root, "shot.png"); + fs.writeFileSync(filePath, Buffer.from([0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a])); + + await expect(materializeWorkerImages([ + { path: filePath, mimeType: "image/png", rootPath: root }, + { url: "https://example.com/ui.png" }, + { data: "abc", mimeType: "image/jpeg" }, + ])).resolves.toEqual([ + { data: Buffer.from([0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a]).toString("base64"), mimeType: "image/png" }, + { url: "https://example.com/ui.png" }, + { data: "abc", mimeType: "image/jpeg" }, + ]); + }); + + it("rejects an oversized screenshot instead of stuffing it onto the IPC pipe", async () => { + const root = makeTempDir(); + const filePath = path.join(root, "huge.png"); + fs.writeFileSync(filePath, Buffer.alloc(8, 1)); + await expect(materializeWorkerImages( + [{ path: filePath, mimeType: "image/png", rootPath: root }], + { maxBytes: 4 }, + )).rejects.toThrow(/too large/); + }); + + it("refuses a path that escaped the attachment root", async () => { + const root = makeTempDir(); + const outside = makeTempDir(); + const filePath = path.join(outside, "secret.png"); + fs.writeFileSync(filePath, Buffer.from([0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a])); + await expect(materializeWorkerImages([ + { path: filePath, mimeType: "image/png", rootPath: root }, + ])).rejects.toThrow(/could not be read/); + }); +}); diff --git a/apps/desktop/src/main/services/chat/workerAttachmentImages.ts b/apps/desktop/src/main/services/chat/workerAttachmentImages.ts new file mode 100644 index 000000000..0f4c9ea89 --- /dev/null +++ b/apps/desktop/src/main/services/chat/workerAttachmentImages.ts @@ -0,0 +1,102 @@ +import path from "node:path"; +import { getImageAttachmentMediaType } from "../../../shared/types/chat"; +import { readFileWithinRootSecure } from "../shared/utils"; + +/** Match the composer temp-attachment cap so IPC and disk agree. */ +export const WORKER_MAX_IMAGE_FILE_BYTES = 10 * 1024 * 1024; + +export type WorkerPathImageSource = { + path: string; + resolvedPath?: string; + rootPath: string; +}; + +export type WorkerIpcImage = + | { path: string; mimeType: string; rootPath: string } + | { data: string; mimeType: string } + | { url: string }; + +export type WorkerMaterializedImage = + | { data: string; mimeType: string } + | { url: string }; + +/** + * Path-only worker-IPC images. Never inline bytes — stuffing screenshot + * base64 through `child.send` JSON can fill the pipe and stall the turn. + * Remote URLs are a Cursor-only send shape and stay at that call site. + */ +export function workerPathImagesFromAttachments( + attachments: readonly WorkerPathImageSource[], +): Array<{ path: string; mimeType: string; rootPath: string }> { + const images: Array<{ path: string; mimeType: string; rootPath: string }> = []; + for (const attachment of attachments) { + const filePath = attachment.resolvedPath?.trim() || attachment.path.trim(); + const rootPath = attachment.rootPath.trim(); + if (!filePath || !rootPath) continue; + images.push({ + path: filePath, + rootPath, + mimeType: getImageAttachmentMediaType(filePath) ?? "image/jpeg", + }); + } + return images; +} + +export async function materializeWorkerImages( + images: readonly WorkerIpcImage[] | undefined, + options?: { maxBytes?: number; label?: string }, +): Promise { + if (!images?.length) return []; + const maxBytes = options?.maxBytes ?? WORKER_MAX_IMAGE_FILE_BYTES; + const label = options?.label ?? "Chat worker"; + const out: WorkerMaterializedImage[] = []; + for (const image of images) { + out.push(materializeOneWorkerImage(image, maxBytes, label)); + } + return out; +} + +function materializeOneWorkerImage( + image: WorkerIpcImage, + maxBytes: number, + label: string, +): WorkerMaterializedImage { + if ("url" in image) { + const url = image.url.trim(); + if (!url) { + throw new Error(`${label} image is missing data, path, or url.`); + } + return { url }; + } + if ("data" in image) { + const inline = image.data.trim(); + const mimeType = image.mimeType.trim(); + if (!inline || !mimeType) { + throw new Error(`${label} image is missing mimeType.`); + } + return { data: inline, mimeType }; + } + if ("path" in image) { + const filePath = image.path.trim(); + const rootPath = image.rootPath.trim(); + if (!filePath || !rootPath) { + throw new Error(`${label} image is missing data, path, or url.`); + } + const mimeType = image.mimeType.trim() + || getImageAttachmentMediaType(filePath) + || "image/jpeg"; + const fileLabel = path.basename(filePath); + try { + const buf = readFileWithinRootSecure(rootPath, filePath, { maxBytes }); + return { data: buf.toString("base64"), mimeType }; + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + if (/too large/i.test(message)) { + throw new Error(`${label} image is too large: ${fileLabel}`); + } + throw new Error(`${label} image could not be read: ${fileLabel}`); + } + } + const _exhaustive: never = image; + throw new Error(`${label} image is missing data, path, or url.`); +} diff --git a/apps/desktop/src/main/services/shared/utils.ts b/apps/desktop/src/main/services/shared/utils.ts index e81af5773..09f16503d 100644 --- a/apps/desktop/src/main/services/shared/utils.ts +++ b/apps/desktop/src/main/services/shared/utils.ts @@ -557,7 +557,11 @@ export async function readAgentAccessibleFileBytes(args: { return readFileWithinRootSecure(root, absPath); } -export function readFileWithinRootSecure(root: string, candidate: string): Buffer { +export function readFileWithinRootSecure( + root: string, + candidate: string, + options?: { maxBytes?: number }, +): Buffer { let expectedPath: string; try { expectedPath = resolvePathWithinRoot(root, candidate, { allowMissing: false }); @@ -575,6 +579,9 @@ export function readFileWithinRootSecure(root: string, candidate: string): Buffe if (!openStat.isFile()) { throw new Error("Path is not a regular file"); } + if (options?.maxBytes != null && openStat.size > options.maxBytes) { + throw new Error(`File is too large (${openStat.size} bytes)`); + } const currentPath = resolvePathWithinRoot(root, expectedPath, { allowMissing: false }); const currentStat = fs.statSync(currentPath); if (openStat.dev !== currentStat.dev || openStat.ino !== currentStat.ino) { diff --git a/docs/features/chat/README.md b/docs/features/chat/README.md index 9aa065a98..13895feed 100644 --- a/docs/features/chat/README.md +++ b/docs/features/chat/README.md @@ -67,13 +67,14 @@ for its separate RPC, sync, storage, and UI contracts. | `apps/desktop/src/main/services/chat/codexMcpElicitation.ts` | Converts Codex app-server MCP elicitation JSON Schemas into pending-input questions and coerces accepted form answers back to boolean/number/array/object values. Persistent consent is gated by request metadata. | | `apps/desktop/src/main/utils/codexComputerUse.ts` | Resolves and strictly verifies the OpenAI-signed standalone Computer Use client after explicit user opt-in, then supplies the canonical `computer_use` MCP config to Work chat and tracked Codex CLI launch/resume paths. | | `apps/desktop/src/main/services/chat/sessionRecovery.ts` | Version-2 persisted-state reconstruction when sessions resume from disk. | -| `apps/desktop/src/main/services/chat/cursorSdkPool.ts` | Cursor SDK adapter: spawns and pools `cursorSdkWorker.ts` Node workers per session, sends turns, brokers permission/hook callbacks, maps SDK events to chat events, and handles teardown. Worker env construction sanitizes ADE/runtime ownership variables while preserving packaged `NODE_PATH` entries through the shared runtime helper so workers copied under `Resources/ade-cli/` can still resolve unpacked app dependencies such as `@cursor/sdk`. The connection envelope carries stable durable-state keys, model parameters, worker request ids, SDK request ids, and structured `CursorSdkErrorDetail` so service logs can correlate ADE chat sessions with Cursor SDK/backend failures. `poisonCursorSdkConnection(poolKey, generation?)` force-evicts a pooled worker regardless of refcount so the next acquire forks a brand-new one: process liveness (`isCursorSdkPooledAlive`) is not connection health, because a run that dies on a transport error leaves the worker process alive while the server-side Cursor agent thread is wedged, and a refcount cannot express that. Pool keys embed the session id, so the lease a refcount decrement would preserve belongs to the same session, never another chat. Cloud oneshot RPCs (`runCursorSdkCloudRequest`) share one `cloud-oneshot:` worker whose idle timer lives on the pool entry (60 s, cancelled on the next acquire) so overlapping list/conversation/watch polls cannot collide with Windows named-pipe / state-dir cleanup. | +| `apps/desktop/src/main/services/chat/cursorSdkPool.ts` | Cursor SDK adapter: spawns and pools `cursorSdkWorker.ts` Node workers per session, sends turns, brokers permission/hook callbacks, maps SDK events to chat events, and handles teardown. Worker env construction sanitizes ADE/runtime ownership variables while preserving packaged `NODE_PATH` entries through the shared runtime helper so workers copied under `Resources/ade-cli/` can still resolve unpacked app dependencies such as `@cursor/sdk`. The connection envelope carries stable durable-state keys, model parameters, worker request ids, SDK request ids, and structured `CursorSdkErrorDetail` so service logs can correlate ADE chat sessions with Cursor SDK/backend failures. Hook sockets are per worker instance (`…///hook.sock`, unique named pipes on Windows) so a recycle cannot unlink the replacement's policy gate. `poisonCursorSdkConnection(poolKey, generation?)` force-evicts a pooled worker regardless of refcount so the next acquire forks a brand-new one after waiting for the previous process to exit: process liveness (`isCursorSdkPooledAlive`) is not connection health, because a run that dies on a transport error leaves the worker process alive while the server-side Cursor agent thread is wedged, and a refcount cannot express that. Pool keys embed the session id, so the lease a refcount decrement would preserve belongs to the same session, never another chat. Cloud oneshot RPCs (`runCursorSdkCloudRequest`) share one `cloud-oneshot:` worker whose idle timer lives on the pool entry (60 s, cancelled on the next acquire) so overlapping list/conversation/watch polls cannot collide with Windows named-pipe / state-dir cleanup. | | `apps/desktop/src/main/services/chat/cursorCloudConversation.ts` | Cursor Cloud conversation unwrap, turn fingerprints, live-run status, and presence-gated inbound-sync helpers. `run.conversation()` is per-run, not full agent history; fingerprints plus prefix/suffix matching let hydrate skip turns ADE already has. `nextCursorCloudMirrorDelay` walks `3s → 8s → 20s → 45s` while a watched chat is quiet and resets to 3 s on new turns. `releaseCursorCloudAttachLease` drops a failed `cloud.run.attach` so watches can poll again. | | `apps/desktop/src/main/services/chat/cursorCloudMirrorWatch.ts` | Per-session watch refcount + backoff scheduler extracted from `agentChatService`. First watch hydrates immediately; later ticks poll only that session; last unwatch clears the timer. Clients call `ai.watchCursorCloudMirror` (`cursorCloudWatchMirror` in preload). Desktop watches while the selected cloud chat is visible, TUI while that session is active, iOS while the scene is active. The sync host registers `ai.watchCursorCloudMirror` and `ai.openCursorCloudChat` so a web/remote client watching a cloud chat on that machine is a real host command, not an adapter fallback. Cursor Cloud has no create-time webhook, so this poll is the inbound path for an **open cloud chat**. The account-level **fleet view** deliberately does not join this timer: its freshness comes from the Cursor Cloud ingress relay (`cursorCloudIngressService`) re-broadcasting each terminal FINISHED/ERROR delivery as the `ade.ai.cursorCloud.fleetEvent` project event (`main.ts` dispatch), so open fleet surfaces refresh when agents finish and otherwise wait for the manual refresh button. | | `apps/desktop/src/main/services/chat/cursorCloudFleetService.ts` | Project-scoped Cursor Cloud **fleet view** backend behind `ade.ai.cursorCloud.fleet` / `.pullIntoLane` / `.resolveLane` / `.stopRun` (registered as ADE actions on the `ai` domain and as sync remote commands, so iOS/web reach the same host implementation). An agent belongs to the open project's fleet when either an ADE chat session links to it (`cursorCloudAgentId`) or its repos include the project origin — compared through shared `cursorCloudRepoMatch.ts`, which normalizes SSH, HTTPS, and `.git`-suffixed spellings to one `host/owner/repo` key — and every entry reports which fact matched (`matchedBy: "session" \| "repo" \| "both"`). One page (default 100, cap 200) is deliberately the whole fleet rather than an unbounded crawl; only live rows are enriched with their latest run (concurrency 4), so finished rows cost nothing until a pull or expansion asks. Pull-into-lane resolves the target lane as linked session's lane → any local lane already on the pushed branch → a fresh lane imported from the remote branch, refuses dirty worktrees, fetches + merges `FETCH_HEAD`, aborts the merge and says exactly where things stand on conflict, scopes multi-repo agents to branches pushed to *this* project's repo (branches attributed to other repos refuse instead of falling back to a name-only fetch), and guards remote-reported refs against git argv injection (`safeBranchRef`). `resolveLaneForAgent` is the same resolution without touching git; `stopAgentRun` cancels an agent's latest run even when no ADE chat exists. | -| `apps/desktop/src/main/services/chat/cursorSdkWorker.ts` | Node worker that hosts the official `@cursor/sdk` and bridges it to the main process via the JSON line protocol in `cursorSdkProtocol.ts`. It creates the SDK local agent platform with the lane workspace/state root, configures local agents to use HTTP/1 by default (`ADE_CURSOR_SDK_USE_HTTP1_FOR_AGENT=0` disables it), enables SDK local agent retries, passes ADE mode/idempotency keys on sends, and tolerates stream-iteration failures long enough to call `run.wait()` and emit a structured terminal result. The SDK's `local.force` send option (expire the currently active persisted run before starting this message as a new follow-up) is wired to the explicit `forceExpireActiveRun` payload flag and is set **only** on ADE's automatic recovery re-send — a normal send that expired a genuinely running turn would discard its output. | +| `apps/desktop/src/main/services/chat/cursorSdkWorker.ts` | Node worker that hosts the official `@cursor/sdk` and bridges it to the main process via the JSON line protocol in `cursorSdkProtocol.ts`. It creates the SDK local agent platform with the lane workspace/state root, configures local agents to use HTTP/1 by default (`ADE_CURSOR_SDK_USE_HTTP1_FOR_AGENT=0` disables it), enables SDK local agent retries, passes ADE mode/idempotency keys on sends, and tolerates stream-iteration failures long enough to call `run.wait()` and emit a structured terminal result. The SDK's `local.force` send option (expire the currently active persisted run before starting this message as a new follow-up) is wired to the explicit `forceExpireActiveRun` payload flag and is set **only** on ADE's automatic recovery re-send — a normal send that expired a genuinely running turn would discard its output. User images are materialized here from attachment paths or URLs (`workerAttachmentImages.ts`) rather than as base64 on the JSON IPC pipe — several large screenshots on `child.send` can stall the turn so Cursor never sees the message. | | `apps/desktop/src/main/services/chat/cursorSdkErrors.ts` | Cursor SDK error normalization helpers shared by the worker: extracts `code`, `status`, `requestId`, `operation`, and `endpoint` from SDK errors/results, reads terminal run details through the public local store API, and classifies resource/backoff vs transport failures without reaching into private SDK run fields. Classification yields a bare `CursorSdkErrorKind`; there is no companion `retryable` bit, because what a caller does about a failure (recycle the thread, surface a rate limit, re-auth) is decided per call site rather than encoded in the classifier. | -| `apps/desktop/src/main/services/chat/cursorSdkProtocol.ts` | Shared types for the worker IPC: chat mode, approval policy, sandbox mode, hook decisions, hook requests, `CursorSdkModelParameterValue`, `CursorSdkWorkerInit`, local/cloud send payloads, SDK request ids, and `CursorSdkErrorDetail`. It exports Cursor-specific error classifiers for transport (`nghttp2`, dropped sockets, stream closures, plus the socket-side cousins `ECANCELED` / `EPIPE` / `write after end`, which poison the server-side agent thread the same way) and backoff/resource exhaustion (`resource_exhausted`, `rate_limited`, `NGHTTP2_ENHANCE_YOUR_CALM`, 429-style text) so UI/service paths present rate-limit and network failures consistently. `classifyCursorSdkErrorText` returns a bare `CursorSdkErrorKind` (`auth` / `rate_limit` / `network` / `busy` / `not_found` / `unknown`). The expired-short-lived-access-token signature lives here too, as one greppable literal (`CURSOR_SDK_STALE_ACCESS_TOKEN_TEXT`) plus `isCursorSdkStaleAccessTokenText` (matches the sentence's two halves independently, so a reflowed clause or a request-id suffix still matches, while a genuinely bad API key does not) and `readCursorSdkStaleTokenFailure`, which reads the worker's synthetic terminal `status: ERROR` event into a `CursorSdkStaleTokenFailure` (`turnId`, message, optional code and request id) in one pass, or returns `null` for any other error. `CursorSdkPermissionPolicy.fullAuto` is a permission-mode marker only — it separates full-auto sessions into their own worker pool and labels logs, and deliberately does **not** map onto the SDK's `local.force`; run expiry is the separate recovery-only `CursorSdkSendPrompt.forceExpireActiveRun`. | +| `apps/desktop/src/main/services/chat/cursorSdkProtocol.ts` | Shared types for the worker IPC: chat mode, approval policy, sandbox mode, hook decisions, hook requests, `CursorSdkModelParameterValue`, `CursorSdkWorkerInit`, local/cloud send payloads, SDK request ids, and `CursorSdkErrorDetail`. User images on those payloads are path/URL references (`CursorSdkUserImage`), not inlined screenshot bytes. It exports Cursor-specific error classifiers for transport (`nghttp2`, dropped sockets, stream closures, plus the socket-side cousins `ECANCELED` / `EPIPE` / `write after end`, which poison the server-side agent thread the same way) and backoff/resource exhaustion (`resource_exhausted`, `rate_limited`, `NGHTTP2_ENHANCE_YOUR_CALM`, 429-style text) so UI/service paths present rate-limit and network failures consistently. `classifyCursorSdkErrorText` returns a bare `CursorSdkErrorKind` (`auth` / `rate_limit` / `network` / `busy` / `not_found` / `unknown`). The expired-short-lived-access-token signature lives here too, as one greppable literal (`CURSOR_SDK_STALE_ACCESS_TOKEN_TEXT`) plus `isCursorSdkStaleAccessTokenText` (matches the sentence's two halves independently, so a reflowed clause or a request-id suffix still matches, while a genuinely bad API key does not) and `readCursorSdkStaleTokenFailure`, which reads the worker's synthetic terminal `status: ERROR` event into a `CursorSdkStaleTokenFailure` (`turnId`, message, optional code and request id) in one pass, or returns `null` for any other error. `CursorSdkPermissionPolicy.fullAuto` is a permission-mode marker only — it separates full-auto sessions into their own worker pool and labels logs, and deliberately does **not** map onto the SDK's `local.force`; run expiry is the separate recovery-only `CursorSdkSendPrompt.forceExpireActiveRun`. | +| `apps/desktop/src/main/services/chat/workerAttachmentImages.ts` | Shared forked-worker image IPC. Composer screenshots become `{ path, mimeType, rootPath }` (Cursor also appends `{ url }` for remote images); the worker re-opens the existing `.ade/attachments` file through `readFileWithinRootSecure` (10 MB cap, same as temp attachments) instead of stuffing screenshot base64 through `child.send`. | | `apps/desktop/src/main/services/chat/cursorSdkPolicy.ts` | Maps ADE permission modes onto Cursor SDK chat mode + approval policy + sandbox mode (`ade` / `cursor-native` / `off`) plus the `fullAuto` marker; decides which tool calls auto-approve and which require a user prompt. `fullAuto` names ADE's full-auto permission mode and only affects pool partitioning and log labels — it is not a Cursor SDK option. | | `apps/desktop/src/main/services/chat/cursorSdkSystemPrompt.ts` | Builds the system prompt the Cursor worker injects (lane context, ADE CLI guidance, persona overlays). | | `apps/desktop/src/main/services/chat/cursorSdkEventMapper.ts` | Translates `@cursor/sdk` stream events into the ADE `AgentChatEventEnvelope` shape consumed by the renderer. SDK `task` messages remain parent-run activity summaries; typed `Task` tool calls/results produce subagent start/result events keyed by tool call id, including the returned child agent id when available. Cursor MCP calls retain provider/tool identity in `event.mcp`; generated-image tools become compact image-generation rows. On a terminal `ERROR` status it reads the worker-injected `adeErrorCode` / `adeErrorDetail`, emits stable user-facing headlines for rate-limit and transport failures (a transport failure reads **Cursor's connection dropped mid-run.** rather than leaking `NGHTTP2_INTERNAL_ERROR` or `[internal] write ECANCELED` into the transcript), preserves exact Cursor request ids/details in `detail`, and sets `errorInfo.category` to `rate_limit`, `network`, `busy`, or `auth` when classification is known. Whenever the friendly headline replaces the raw code, that code is kept as the first `detail` line so the underlying failure is still recoverable from the transcript. | @@ -84,8 +85,8 @@ for its separate RPC, sync, storage, and UI contracts. | `apps/desktop/src/main/services/chat/droidSdkEventMapper.ts` | Per-session `DroidSdkEventMapperState` + `mapDroidSdkMessageToChatEvents` / `mapDroidSdkRunResultToDoneEvent`. Tracks streaming text/thinking/image item ids, maps tool calls and results, maps `mission_worker_started` / `mission_worker_completed` notifications to provider-neutral subagent lifecycle events keyed by worker session id, surfaces image content as compact generation rows, and reports token usage. Replaces the deleted `droidAcpPool.ts` + `droidAcpEventMapper` path. | | `apps/desktop/src/main/services/chat/droidModelsDiscovery.ts` | SDK-driven model probe (`listDroidModelsFromSdk`) plus the `/config.json` custom-proxy merge (`~/.factory` unless `FACTORY_HOME_OVERRIDE` is set — see [Provider config homes](agent-routing.md#provider-config-homes)). Normalizes the generic `opus` row to Opus 5 with its `high` default reasoning effort and Fast capability, while retired factory Claude ids still resolve forward (Sonnet 4.6 -> Sonnet 5, basic Opus 4.7 -> Opus 4.8) before descriptors reach desktop, mobile, or TUI model pickers. Exposes `discoverDroidSdkModelDescriptors` (alias for the legacy `discoverDroidCliModelDescriptors` while callers migrate). | | `apps/desktop/src/main/services/chat/piSdkPool.ts` | Pi adapter. Forks `piSdkWorker` per session key, exposes `acquirePiSdkConnection` / `releasePiSdkConnection`, and proxies prompts, model/thinking changes, compaction, inventory reads, `login` / `cancelLogin`, and `respondToUi`. Also routes the reverse-RPC UI channel onto `bridge.onUiRequest` / `onUiNotice` / `onUiCancel`; when no `onUiRequest` handler is installed the pool answers `{ ok: false }` immediately, so an unattended worker fails closed instead of hanging a turn. | -| `apps/desktop/src/main/services/chat/piSdkWorker.ts` | Node worker that hosts the user's own Pi installation (resolved at runtime, never statically imported). Owns the Pi agent session, the model runtime, `ModelRuntime.login`, tool assembly (`ask_user` plus approval-gated rebuilds of Pi's built-ins), extension binding, and the settings manager that pins `projectTrusted: false`. | -| `apps/desktop/src/main/services/chat/piSdkProtocol.ts` | Worker IPC types and validators, at protocol version 2. Adds the `ui_request` / `ui_notice` / `ui_cancel` / `ui_response` frames and `login` / `login_cancel` on top of version 1, plus the `extensions` / `askUserTool` / `approvalTools` init flags and the `extensions` / `extensionsError` / `ungateableTools` fields on `PiSdkReady`. Every frame is validated in both directions. | +| `apps/desktop/src/main/services/chat/piSdkWorker.ts` | Node worker that hosts the user's own Pi installation (resolved at runtime, never statically imported). Owns the Pi agent session, the model runtime, `ModelRuntime.login`, tool assembly (`ask_user` plus approval-gated rebuilds of Pi's built-ins), extension binding, and the settings manager that pins `projectTrusted: false`. User images are materialized here from attachment paths (`workerAttachmentImages.ts`) rather than as base64 on the JSON IPC pipe. | +| `apps/desktop/src/main/services/chat/piSdkProtocol.ts` | Worker IPC types and validators, at protocol version 2. Adds the `ui_request` / `ui_notice` / `ui_cancel` / `ui_response` frames and `login` / `login_cancel` on top of version 1, plus the `extensions` / `askUserTool` / `approvalTools` init flags and the `extensions` / `extensionsError` / `ungateableTools` fields on `PiSdkReady`. User images on send/steer/follow_up are path (or tiny inline `data`) references, not inlined screenshot bytes; remote `url` images are rejected because Pi's prompt API has no URL form. Every frame is validated in both directions. | | `apps/desktop/src/main/services/chat/piSdkUiBridge.ts` | Worker-side bridge from Pi's callback-shaped UI APIs to ADE cards, with no Pi imports of its own. `createPiUiBridge` is the never-rejecting request channel; `createPiAskUserTool` / `createPiApprovalGate` / `withPiApproval` build the `ask_user` tool and the per-call approval wrapper; `createPiAuthInteraction` implements Pi's `AuthInteraction`; `createPiExtensionUiContext` implements `ExtensionUIContext`. | | `apps/desktop/src/main/services/chat/piSdkEventMapper.ts` | Pi SDK event → `AgentChatEvent` translation, plus the card helpers: `piUiRequestToPendingInput` (blocking worker request → `PendingInputRequest` with `source: "pi"`), `piUiResponseFromAnswer` (card answer → worker reply, mapping `accept` / `accept_for_session` onto the gate's `allow` / `allow_session` values), `piUiNoticeToChatEvents`, and `piExtensionLoadNotice`. | | `apps/desktop/src/main/services/chat/piSessionStore.ts` | The one native Pi session store ADE chat, tracked Pi CLI terminals, and external-session discovery all resolve against. `piSessionStoreForEnvironment` returns a `{ root, storageDir }` pair — the root is the authorization boundary, `storageDir` is set only when the user configured one, because Pi nests per-cwd subdirectories only when it is told nothing. Also owns header reads (`readPiSessionHeader`, `piSessionHeaderMatchesCwd`), file authorization (`resolvePiSessionFile`, `classifyPiSessionFile`), the per-cwd listing, and `repositoryOverridesPiSessionDir`. A checkout's `.pi/settings.json` is deliberately never read. | @@ -1267,9 +1268,12 @@ first-event watchdog and one automatic recovery attempt. signature (never a bad key), and the recycle runs with `reason: "stale_token"` and `preserveAgentId: true` — the worker is replaced, but the same `cursorSdkAgentId` is resumed in the fresh one, so no rotation, no continuity - preamble, and no lost conversation. The raw error card is swallowed in the - bridge. Recovery is silent when nothing had streamed yet (the original prompt - is re-sent verbatim, logged as + preamble, and no lost conversation. The replacement worker listens on a new + per-instance `hook.sock` (and the next acquire waits for the poisoned process + to exit) so the dying worker's `server.close()`/unlink cannot delete the new + policy gate; tools keep working in the same chat. The raw error card is + swallowed in the bridge. Recovery is silent when nothing had streamed yet (the + original prompt is re-sent verbatim, logged as `agent_chat.cursor_sdk_stale_token_recovered`); when the token died mid-turn the re-send asks Cursor to continue from where it stopped and the transcript says only "Reconnected to Cursor and continued." If the retry hits the same From d8be2f4b3b4be5b7172334c34d49c8d0b5c956de Mon Sep 17 00:00:00 2001 From: Arul Sharma <31745423+arul28@users.noreply.github.com> Date: Sat, 29 Aug 2026 01:37:28 -0400 Subject: [PATCH 2/3] Fail Cursor SDK replace-wait timeout instead of overlapping workers. validate-docs rejected a Windows vacuous return; skipIf that Unix-socket test. Acquire now throws if the poisoned worker outlives REPLACE_WAIT_MS so the next fork cannot share state/index.db. Co-authored-by: Cursor --- .../main/services/chat/cursorSdkPool.test.ts | 49 ++++++++++++++++++- .../src/main/services/chat/cursorSdkPool.ts | 15 ++++-- docs/features/chat/README.md | 4 +- 3 files changed, 59 insertions(+), 9 deletions(-) diff --git a/apps/desktop/src/main/services/chat/cursorSdkPool.test.ts b/apps/desktop/src/main/services/chat/cursorSdkPool.test.ts index 1de3de00d..e42dbe5b8 100644 --- a/apps/desktop/src/main/services/chat/cursorSdkPool.test.ts +++ b/apps/desktop/src/main/services/chat/cursorSdkPool.test.ts @@ -8,6 +8,7 @@ import { buildCursorSdkPaths, buildCursorSdkWorkerEnv, cleanupCursorSdkRuntimePaths, + CURSOR_SDK_REPLACE_WAIT_MS, isCursorSdkPooledAlive, poisonCursorSdkConnection, releaseCursorSdkConnection, @@ -108,6 +109,14 @@ class DelayedExitChild extends FakeSdkChild { } } +/** Dispose/kill never reaps the pid — the replace wait must not fork over it. */ +class StuckExitChild extends DelayedExitChild { + override kill(): boolean { + this.killed = true; + return true; + } +} + class ExitingBeforeInitChild extends EventEmitter { stdout = new EventEmitter(); stderr = new EventEmitter(); @@ -348,8 +357,7 @@ describe("Cursor SDK pool paths", () => { })).toThrow(/instance id is required/); }); - it("does not delete a sibling worker's hook socket directory during cleanup", () => { - if (process.platform === "win32") return; + it.skipIf(process.platform === "win32")("does not delete a sibling worker's hook socket directory during cleanup", () => { const cacheRoot = makeTempDir("ade-cursor-cleanup-socket-"); const stateRoot = path.join(cacheRoot, "state"); fs.mkdirSync(stateRoot, { recursive: true }); @@ -621,6 +629,43 @@ describe("Cursor SDK pool paths", () => { releaseCursorSdkConnection(poolKey, second.generation); }); + it("fails acquire if the poisoned worker outlives the replace wait", async () => { + const firstChild = new StuckExitChild(); + const nextChild = new FakeSdkChild(); + forkMock.mockReturnValueOnce(firstChild); + const poolKey = `test-replace-timeout:${Date.now()}:${Math.random()}`; + const args = { + poolKey, + projectRoot: path.join(os.tmpdir(), "ade-project"), + workspacePath: path.join(os.tmpdir(), "ade-workspace"), + modelSdkId: "cursor-model", + sessionId: "session-1", + policy: { ...TEST_POLICY }, + }; + + const first = await acquireCursorSdkConnection(args); + vi.useFakeTimers(); + try { + expect(poisonCursorSdkConnection(poolKey, first.generation)).toBe(true); + forkMock.mockReturnValue(nextChild); + const pending = expect(acquireCursorSdkConnection(args)).rejects.toThrow( + /did not exit before replacement/, + ); + await vi.advanceTimersByTimeAsync(CURSOR_SDK_REPLACE_WAIT_MS); + await pending; + expect(forkMock).toHaveBeenCalledTimes(1); + } finally { + vi.useRealTimers(); + } + + firstChild.finishExit(0, null); + const second = await acquireCursorSdkConnection(args); + expect(second.pooled).not.toBe(first.pooled); + expect(forkMock).toHaveBeenCalledTimes(2); + + releaseCursorSdkConnection(poolKey, second.generation); + }); + it("does not treat a dispatched kill plus IPC error as the worker exiting", async () => { const firstChild = new DelayedExitChild(); const nextChild = new FakeSdkChild(); diff --git a/apps/desktop/src/main/services/chat/cursorSdkPool.ts b/apps/desktop/src/main/services/chat/cursorSdkPool.ts index 79e933af5..be4b8cf30 100644 --- a/apps/desktop/src/main/services/chat/cursorSdkPool.ts +++ b/apps/desktop/src/main/services/chat/cursorSdkPool.ts @@ -105,7 +105,7 @@ const STALE_INIT_RETRY_LIMIT = 2; */ const CURSOR_SDK_DISPOSE_GRACE_MS = 3_000; /** Cap how long a replacement waits for the previous worker of the same pool key. */ -const CURSOR_SDK_REPLACE_WAIT_MS = CURSOR_SDK_DISPOSE_GRACE_MS + 500; +export const CURSOR_SDK_REPLACE_WAIT_MS = CURSOR_SDK_DISPOSE_GRACE_MS + 500; const CURSOR_SDK_WORKER_ENV_DENYLIST = [ "CURSOR_API_KEY", "CURSOR_AUTH_TOKEN", @@ -976,13 +976,18 @@ function trackDepartingCursorSdkWorker(poolKey: string, wait: Promise): vo async function waitForDepartingCursorSdkWorker(poolKey: string): Promise { const prior = departingWorkers.get(poolKey); if (!prior) return; - await Promise.race([ - prior, - new Promise((resolve) => { - const timer = setTimeout(resolve, CURSOR_SDK_REPLACE_WAIT_MS); + let timer: ReturnType | null = null; + const outcome = await Promise.race([ + prior.then(() => "exited" as const), + new Promise<"timeout">((resolve) => { + timer = setTimeout(() => resolve("timeout"), CURSOR_SDK_REPLACE_WAIT_MS); timer.unref(); }), ]); + if (timer) clearTimeout(timer); + if (outcome === "timeout") { + throw new Error("Cursor SDK worker did not exit before replacement."); + } } /** diff --git a/docs/features/chat/README.md b/docs/features/chat/README.md index 13895feed..c97b22d44 100644 --- a/docs/features/chat/README.md +++ b/docs/features/chat/README.md @@ -67,7 +67,7 @@ for its separate RPC, sync, storage, and UI contracts. | `apps/desktop/src/main/services/chat/codexMcpElicitation.ts` | Converts Codex app-server MCP elicitation JSON Schemas into pending-input questions and coerces accepted form answers back to boolean/number/array/object values. Persistent consent is gated by request metadata. | | `apps/desktop/src/main/utils/codexComputerUse.ts` | Resolves and strictly verifies the OpenAI-signed standalone Computer Use client after explicit user opt-in, then supplies the canonical `computer_use` MCP config to Work chat and tracked Codex CLI launch/resume paths. | | `apps/desktop/src/main/services/chat/sessionRecovery.ts` | Version-2 persisted-state reconstruction when sessions resume from disk. | -| `apps/desktop/src/main/services/chat/cursorSdkPool.ts` | Cursor SDK adapter: spawns and pools `cursorSdkWorker.ts` Node workers per session, sends turns, brokers permission/hook callbacks, maps SDK events to chat events, and handles teardown. Worker env construction sanitizes ADE/runtime ownership variables while preserving packaged `NODE_PATH` entries through the shared runtime helper so workers copied under `Resources/ade-cli/` can still resolve unpacked app dependencies such as `@cursor/sdk`. The connection envelope carries stable durable-state keys, model parameters, worker request ids, SDK request ids, and structured `CursorSdkErrorDetail` so service logs can correlate ADE chat sessions with Cursor SDK/backend failures. Hook sockets are per worker instance (`…///hook.sock`, unique named pipes on Windows) so a recycle cannot unlink the replacement's policy gate. `poisonCursorSdkConnection(poolKey, generation?)` force-evicts a pooled worker regardless of refcount so the next acquire forks a brand-new one after waiting for the previous process to exit: process liveness (`isCursorSdkPooledAlive`) is not connection health, because a run that dies on a transport error leaves the worker process alive while the server-side Cursor agent thread is wedged, and a refcount cannot express that. Pool keys embed the session id, so the lease a refcount decrement would preserve belongs to the same session, never another chat. Cloud oneshot RPCs (`runCursorSdkCloudRequest`) share one `cloud-oneshot:` worker whose idle timer lives on the pool entry (60 s, cancelled on the next acquire) so overlapping list/conversation/watch polls cannot collide with Windows named-pipe / state-dir cleanup. | +| `apps/desktop/src/main/services/chat/cursorSdkPool.ts` | Cursor SDK adapter: spawns and pools `cursorSdkWorker.ts` Node workers per session, sends turns, brokers permission/hook callbacks, maps SDK events to chat events, and handles teardown. Worker env construction sanitizes ADE/runtime ownership variables while preserving packaged `NODE_PATH` entries through the shared runtime helper so workers copied under `Resources/ade-cli/` can still resolve unpacked app dependencies such as `@cursor/sdk`. The connection envelope carries stable durable-state keys, model parameters, worker request ids, SDK request ids, and structured `CursorSdkErrorDetail` so service logs can correlate ADE chat sessions with Cursor SDK/backend failures. Hook sockets are per worker instance (`…///hook.sock`, unique named pipes on Windows) so a recycle cannot unlink the replacement's policy gate. `poisonCursorSdkConnection(poolKey, generation?)` force-evicts a pooled worker regardless of refcount so the next acquire forks a brand-new one after waiting for the previous process to exit (and fails rather than overlapping if that wait times out): process liveness (`isCursorSdkPooledAlive`) is not connection health, because a run that dies on a transport error leaves the worker process alive while the server-side Cursor agent thread is wedged, and a refcount cannot express that. Pool keys embed the session id, so the lease a refcount decrement would preserve belongs to the same session, never another chat. Cloud oneshot RPCs (`runCursorSdkCloudRequest`) share one `cloud-oneshot:` worker whose idle timer lives on the pool entry (60 s, cancelled on the next acquire) so overlapping list/conversation/watch polls cannot collide with Windows named-pipe / state-dir cleanup. | | `apps/desktop/src/main/services/chat/cursorCloudConversation.ts` | Cursor Cloud conversation unwrap, turn fingerprints, live-run status, and presence-gated inbound-sync helpers. `run.conversation()` is per-run, not full agent history; fingerprints plus prefix/suffix matching let hydrate skip turns ADE already has. `nextCursorCloudMirrorDelay` walks `3s → 8s → 20s → 45s` while a watched chat is quiet and resets to 3 s on new turns. `releaseCursorCloudAttachLease` drops a failed `cloud.run.attach` so watches can poll again. | | `apps/desktop/src/main/services/chat/cursorCloudMirrorWatch.ts` | Per-session watch refcount + backoff scheduler extracted from `agentChatService`. First watch hydrates immediately; later ticks poll only that session; last unwatch clears the timer. Clients call `ai.watchCursorCloudMirror` (`cursorCloudWatchMirror` in preload). Desktop watches while the selected cloud chat is visible, TUI while that session is active, iOS while the scene is active. The sync host registers `ai.watchCursorCloudMirror` and `ai.openCursorCloudChat` so a web/remote client watching a cloud chat on that machine is a real host command, not an adapter fallback. Cursor Cloud has no create-time webhook, so this poll is the inbound path for an **open cloud chat**. The account-level **fleet view** deliberately does not join this timer: its freshness comes from the Cursor Cloud ingress relay (`cursorCloudIngressService`) re-broadcasting each terminal FINISHED/ERROR delivery as the `ade.ai.cursorCloud.fleetEvent` project event (`main.ts` dispatch), so open fleet surfaces refresh when agents finish and otherwise wait for the manual refresh button. | | `apps/desktop/src/main/services/chat/cursorCloudFleetService.ts` | Project-scoped Cursor Cloud **fleet view** backend behind `ade.ai.cursorCloud.fleet` / `.pullIntoLane` / `.resolveLane` / `.stopRun` (registered as ADE actions on the `ai` domain and as sync remote commands, so iOS/web reach the same host implementation). An agent belongs to the open project's fleet when either an ADE chat session links to it (`cursorCloudAgentId`) or its repos include the project origin — compared through shared `cursorCloudRepoMatch.ts`, which normalizes SSH, HTTPS, and `.git`-suffixed spellings to one `host/owner/repo` key — and every entry reports which fact matched (`matchedBy: "session" \| "repo" \| "both"`). One page (default 100, cap 200) is deliberately the whole fleet rather than an unbounded crawl; only live rows are enriched with their latest run (concurrency 4), so finished rows cost nothing until a pull or expansion asks. Pull-into-lane resolves the target lane as linked session's lane → any local lane already on the pushed branch → a fresh lane imported from the remote branch, refuses dirty worktrees, fetches + merges `FETCH_HEAD`, aborts the merge and says exactly where things stand on conflict, scopes multi-repo agents to branches pushed to *this* project's repo (branches attributed to other repos refuse instead of falling back to a name-only fetch), and guards remote-reported refs against git argv injection (`safeBranchRef`). `resolveLaneForAgent` is the same resolution without touching git; `stopAgentRun` cancels an agent's latest run even when no ADE chat exists. | @@ -1270,7 +1270,7 @@ first-event watchdog and one automatic recovery attempt. `cursorSdkAgentId` is resumed in the fresh one, so no rotation, no continuity preamble, and no lost conversation. The replacement worker listens on a new per-instance `hook.sock` (and the next acquire waits for the poisoned process - to exit) so the dying worker's `server.close()`/unlink cannot delete the new + to exit, failing instead of overlapping if that wait times out) so the dying worker's `server.close()`/unlink cannot delete the new policy gate; tools keep working in the same chat. The raw error card is swallowed in the bridge. Recovery is silent when nothing had streamed yet (the original prompt is re-sent verbatim, logged as From b9aff83b31795a0658e1fd00667395e19a93db76 Mon Sep 17 00:00:00 2001 From: Arul Sharma <31745423+arul28@users.noreply.github.com> Date: Sat, 29 Aug 2026 02:06:58 -0400 Subject: [PATCH 3/3] Flush Cursor silence-recycle cancel timeout in fake-timer tests. Watchdog trips schedule a 3s cancel race after the 90s jump; advancing only the watchdog left nested recovery waiting until pumpUntil timed out. Co-authored-by: Cursor --- .../services/chat/agentChatService.test.ts | 43 +++++++++++++------ .../main/services/chat/agentChatService.ts | 2 +- 2 files changed, 32 insertions(+), 13 deletions(-) diff --git a/apps/desktop/src/main/services/chat/agentChatService.test.ts b/apps/desktop/src/main/services/chat/agentChatService.test.ts index 3b286db92..790a886ae 100644 --- a/apps/desktop/src/main/services/chat/agentChatService.test.ts +++ b/apps/desktop/src/main/services/chat/agentChatService.test.ts @@ -921,6 +921,7 @@ import { writeSessionLinearIssueContextFile, createAgentChatService, CURSOR_SDK_FIRST_EVENT_WATCHDOG_MS, + CURSOR_SDK_RECYCLE_CANCEL_TIMEOUT_MS, } from "./agentChatService"; import { createChatRuntimeBudget } from "./chatRuntimeBudget"; import { readThreadPointerLedger } from "./threadPointerLedger"; @@ -2434,6 +2435,22 @@ describe("buildLinearSessionDirective", () => { /** Just past the real watchdog budget, derived rather than mirrored. */ const CURSOR_SILENCE_WATCHDOG_TRIP_MS = CURSOR_SDK_FIRST_EVENT_WATCHDOG_MS + 1; +/** + * Recycle races `cancel()` against a 3s timeout. Advancing only the 90s + * watchdog schedules that timer; it does not flush it. Nested timers created + * during the watchdog callback are not included in the same advance. + */ +const flushCursorSdkSilenceRecycle = async (): Promise => { + await Promise.resolve(); + await vi.advanceTimersByTimeAsync(CURSOR_SDK_RECYCLE_CANCEL_TIMEOUT_MS); + await Promise.resolve(); +}; + +const tripCursorSdkSilenceWatchAndRecycle = async (): Promise => { + await vi.advanceTimersByTimeAsync(CURSOR_SILENCE_WATCHDOG_TRIP_MS); + await flushCursorSdkSilenceRecycle(); +}; + /** * Real async setup work still needs event-loop turns while the clock is faked, * so pump the fake clock instead of assuming a fixed number of ticks. @@ -2450,6 +2467,8 @@ const pumpUntil = async (label: string, ready: () => boolean): Promise => await vi.advanceTimersByTimeAsync(tripWatchdog ? CURSOR_SILENCE_WATCHDOG_TRIP_MS : 1); + if (tripWatchdog) await flushCursorSdkSilenceRecycle(); + await Promise.resolve(); } if (!ready()) throw new Error(`pumpUntil timed out waiting for: ${label}`); }; @@ -17025,7 +17044,7 @@ describe("createAgentChatService", () => { event.event.type === "user_message" && event.event.deliveryState === "queued")); mockState.cursorSendPromptGate = null; - await vi.advanceTimersByTimeAsync(CURSOR_SILENCE_WATCHDOG_TRIP_MS); + await tripCursorSdkSilenceWatchAndRecycle(); // Re-send, then the carried steer delivered as its own turn. await pumpUntil("steer delivered after recovery", () => mockState.cursorSdkSendCalls.length >= 3); @@ -17440,9 +17459,9 @@ describe("createAgentChatService", () => { await pumpUntil("queued steer", () => events.some((event) => event.event.type === "user_message" && event.event.deliveryState === "queued")); - await vi.advanceTimersByTimeAsync(CURSOR_SILENCE_WATCHDOG_TRIP_MS); + await tripCursorSdkSilenceWatchAndRecycle(); await pumpUntil("recovery re-send", () => mockState.cursorSdkSendCalls.length >= 2); - await vi.advanceTimersByTimeAsync(CURSOR_SILENCE_WATCHDOG_TRIP_MS); + await tripCursorSdkSilenceWatchAndRecycle(); await pumpUntil("terminal failure", () => events.some((event) => event.event.type === "error")); // Settled, and settled once: the attempt body and the wrapper cover @@ -17537,7 +17556,7 @@ describe("createAgentChatService", () => { await pumpUntil("carried steer queued", () => events.filter((event) => event.event.type === "user_message" && event.event.deliveryState === "queued").length >= 1); - await vi.advanceTimersByTimeAsync(CURSOR_SILENCE_WATCHDOG_TRIP_MS); + await tripCursorSdkSilenceWatchAndRecycle(); await pumpUntil("recovery re-send", () => mockState.cursorSdkSendCalls.length >= 2); // Queued during attempt 2 — lands on the rebuilt runtime, which the @@ -17546,7 +17565,7 @@ describe("createAgentChatService", () => { await pumpUntil("second steer queued", () => events.filter((event) => event.event.type === "user_message" && event.event.deliveryState === "queued").length >= 2); - await vi.advanceTimersByTimeAsync(CURSOR_SILENCE_WATCHDOG_TRIP_MS); + await tripCursorSdkSilenceWatchAndRecycle(); await pumpUntil("terminal failure", () => events.some((event) => event.event.type === "error")); const cancelNotices = events.filter((event) => @@ -17603,11 +17622,11 @@ describe("createAgentChatService", () => { // Attempt 2 succeeds and delivers the carried steer. mockState.cursorSendPromptGate = null; - await vi.advanceTimersByTimeAsync(CURSOR_SILENCE_WATCHDOG_TRIP_MS); + await tripCursorSdkSilenceWatchAndRecycle(); await pumpUntil("steer delivered", () => mockState.cursorSdkSendCalls.length >= 3); // The delivered steer's turn is silent in turn; its recovery re-send // (send 4) succeeds and leaves a different runtime on the session. - await vi.advanceTimersByTimeAsync(CURSOR_SILENCE_WATCHDOG_TRIP_MS); + await tripCursorSdkSilenceWatchAndRecycle(); await pumpUntil("nested recovery", () => mockState.cursorSdkSendCalls.length >= 4); // Let the nested chain unwind fully, so the outer wrapper's finally // runs while the session is on a different runtime than it re-queued on. @@ -17698,9 +17717,9 @@ describe("createAgentChatService", () => { text: "Both attempts go silent.", }, { awaitDispatch: true }).catch(() => undefined); await pumpUntil("first cursor send", () => mockState.cursorSdkSendCalls.length >= 1); - await vi.advanceTimersByTimeAsync(CURSOR_SILENCE_WATCHDOG_TRIP_MS); + await tripCursorSdkSilenceWatchAndRecycle(); await pumpUntil("second cursor send", () => mockState.cursorSdkSendCalls.length >= 2); - await vi.advanceTimersByTimeAsync(CURSOR_SILENCE_WATCHDOG_TRIP_MS); + await tripCursorSdkSilenceWatchAndRecycle(); await pumpUntil("terminal failure", () => mockState.cursorSdkPoisonCalls.length >= 2); } finally { vi.useRealTimers(); @@ -17948,7 +17967,7 @@ describe("createAgentChatService", () => { // The retry must run against a healthy worker. mockState.cursorSendPromptGate = null; - await vi.advanceTimersByTimeAsync(CURSOR_SILENCE_WATCHDOG_TRIP_MS); + await tripCursorSdkSilenceWatchAndRecycle(); await pumpUntil("recovery re-send", () => mockState.cursorSdkSendCalls.length >= 3); expect(mockState.cursorSdkPoisonCalls).toHaveLength(1); @@ -17995,9 +18014,9 @@ describe("createAgentChatService", () => { }, { awaitDispatch: true }).catch(() => undefined); await pumpUntil("first cursor send", () => mockState.cursorSdkSendCalls.length >= 1); - await vi.advanceTimersByTimeAsync(CURSOR_SILENCE_WATCHDOG_TRIP_MS); + await tripCursorSdkSilenceWatchAndRecycle(); await pumpUntil("second cursor send", () => mockState.cursorSdkSendCalls.length >= 2); - await vi.advanceTimersByTimeAsync(CURSOR_SILENCE_WATCHDOG_TRIP_MS); + await tripCursorSdkSilenceWatchAndRecycle(); await pumpUntil("terminal error event", () => events.some((event) => event.event.type === "error")); const errorEvent = events.find((event) => diff --git a/apps/desktop/src/main/services/chat/agentChatService.ts b/apps/desktop/src/main/services/chat/agentChatService.ts index bcee2c6a2..26fec5dde 100644 --- a/apps/desktop/src/main/services/chat/agentChatService.ts +++ b/apps/desktop/src/main/services/chat/agentChatService.ts @@ -3392,7 +3392,7 @@ const CURSOR_SDK_AGENT_PROTOCOL_VERSION = 2; */ export const CURSOR_SDK_FIRST_EVENT_WATCHDOG_MS = 90_000; /** Upper bound on the best-effort cancel issued while recycling a wedged thread. */ -const CURSOR_SDK_RECYCLE_CANCEL_TIMEOUT_MS = 3_000; +export const CURSOR_SDK_RECYCLE_CANCEL_TIMEOUT_MS = 3_000; const CURSOR_SDK_SILENT_RUN_MESSAGE = "Cursor stopped responding. ADE opened a fresh Cursor thread — try sending again."; const CLAUDE_WARMUP_WAIT_TIMEOUT_MS = 20_000;