From a6c6a76d37c729e17cccf2dd20fe8e8177d98b14 Mon Sep 17 00:00:00 2001 From: Ysopwnable <161784452+N0zoM1z0@users.noreply.github.com> Date: Fri, 16 Oct 2026 22:49:06 +0800 Subject: [PATCH] fix(cutter): cancel discovery probes and queued work Propagate CLI/MCP cancellation through shared Cutter discovery, stop queued probes, and drain admitted workers. Deliver terminal bridge results only after socket closure. Preserve confirmed command Evidence and caller reasons when cancellation arrives during finalization. Validated 66 focused tests, full CI static/docs gates, actual CLI SIGINT, compiled stdio cancellation/recovery, and official Linux Cutter 2.5.0 Qt5 CLI/MCP ELF workflows. Native UI cancellation and descriptor filesystem cancellation remain unverified. Fixes #3946. --- src/cli/cutterCommands.ts | 7 +- src/cutter/CutterBridgeClient.ts | 57 +++++++--- src/cutter/CutterBridgeService.ts | 49 ++++++++- src/server/registerCutterTools.ts | 31 +++--- tests/boundary/cli/cutterCommands.test.ts | 72 +++++++++++- .../mcp/cutterCancellationMcp.test.ts | 102 +++++++++++++++++ .../providers/CutterBridgeClient.test.ts | 103 +++++++++++++++++- .../providers/CutterBridgeService.test.ts | 38 +++++++ 8 files changed, 421 insertions(+), 38 deletions(-) diff --git a/src/cli/cutterCommands.ts b/src/cli/cutterCommands.ts index d521d7c394..0b62412584 100644 --- a/src/cli/cutterCommands.ts +++ b/src/cli/cutterCommands.ts @@ -23,8 +23,11 @@ export const registerCutterCommands = ( "List live Cutter instances connected through the REA Python plugin bridge", args: z.object({}), run: () => - logCliCommand(logger, CLI_COMMANDS.listCutterSessions, () => - cutter.listSessions(), + withCommandCancellation((signal) => + logCliCommand(logger, CLI_COMMANDS.listCutterSessions, async () => { + const result = await cutter.listSessions({ signal }); + return result.ok ? result.value : projectAnalysisError(result.error); + }), ), }); cli.command(CLI_COMMANDS.cutterCommand, { diff --git a/src/cutter/CutterBridgeClient.ts b/src/cutter/CutterBridgeClient.ts index e725a5b2ab..f5c98f3d03 100644 --- a/src/cutter/CutterBridgeClient.ts +++ b/src/cutter/CutterBridgeClient.ts @@ -167,22 +167,39 @@ class CutterBridgeResponseFrames { const mapDiscovery = async ( inputs: readonly Input[], operation: (input: Input) => Promise, + signal?: AbortSignal, ): Promise => { + signal?.throwIfAborted(); const results = new Array(inputs.length); let next = 0; - await Promise.all( + let stopped = false; + const workers = await Promise.allSettled( Array.from( { length: Math.min(inputs.length, DISCOVERY_CONCURRENCY) }, async () => { - while (next < inputs.length) { - const index = next++; - results[index] = await operation(inputs[index]!); + try { + while (!stopped && next < inputs.length) { + signal?.throwIfAborted(); + const index = next++; + results[index] = await operation(inputs[index]!); + } + } catch (cause) { + stopped = true; + throw cause; } }, ), ); + // Do not release this discovery's work until all admitted probes settle. + for (const worker of workers) + if (worker.status === "rejected") throw worker.reason; + signal?.throwIfAborted(); return results; }; + +export interface CutterDiscoveryOptions { + readonly signal?: AbortSignal; +} export type CutterBridgeExecutionState = | "not_started" | "complete" @@ -286,8 +303,13 @@ export class CutterBridgeClient { ); } - async listSessions(): Promise { + async listSessions( + options: CutterDiscoveryOptions = {}, + ): Promise { + const { signal } = options; + signal?.throwIfAborted(); const loaded = await this.#readDescriptors(); + signal?.throwIfAborted(); if (loaded.error !== undefined) return { sessions: [], @@ -306,6 +328,7 @@ export class CutterBridgeClient { kind: "status", session_id: descriptor.session_id, }), + signal, ), ); if ( @@ -328,9 +351,11 @@ export class CutterBridgeClient { identity_status: "partial" as const, }; } catch { + signal?.throwIfAborted(); return undefined; } }, + signal, ); const sessions = descriptors.filter( (value): value is CutterBridgeSession => value !== undefined, @@ -783,18 +808,19 @@ export class CutterBridgeClient { port: descriptor.port, }); const frames = new CutterBridgeResponseFrames(); - let settled = false; + let completion: (() => void) | undefined; let requestSent = false; let authenticated = false; const clientNonce = randomBytes(32).toString("hex"); let serverNonce = ""; let connectionOrdinal: string | undefined; const finish = (complete: () => void): void => { - if (settled) return; - settled = true; + if (completion !== undefined) return; + completion = complete; clearTimeout(timer); signal?.removeEventListener("abort", onAbort); - complete(); + // Retain this request until its owned socket has actually closed. + connection.destroy(); }; const rejectRequest = (message: string): void => finish(() => @@ -802,20 +828,20 @@ export class CutterBridgeClient { ); const timer = setTimeout(() => { rejectRequest("Cutter bridge request timed out"); - connection.destroy(); }, 25_000); const onAbort = (): void => { rejectRequest("Cutter bridge request was cancelled"); - connection.destroy(); }; signal?.addEventListener("abort", onAbort, { once: true }); if (signal?.aborted) onAbort(); connection.once("connect", () => { + if (completion !== undefined) return; connection.write( `${JSON.stringify({ kind: "rea-auth-v2", nonce: clientNonce })}\n`, ); }); connection.on("data", (chunk: Buffer) => { + if (completion !== undefined) return; try { frames.push(chunk, (line) => { const frame: unknown = JSON.parse(line.toString("utf8")); @@ -869,24 +895,23 @@ export class CutterBridgeClient { ) throw new Error("response proof failed"); const response: unknown = JSON.parse(envelope.response); - connection.end(); finish(() => resolve(response)); return false; } }); } catch { rejectRequest("Cutter bridge peer authentication or response failed"); - connection.destroy(); } }); connection.once("error", (cause) => rejectRequest(`Cutter bridge is unavailable: ${cause.message}`), ); - connection.once("close", () => + connection.once("close", () => { rejectRequest( "Cutter bridge closed before returning a complete response", - ), - ); + ); + completion?.(); + }); }); } } diff --git a/src/cutter/CutterBridgeService.ts b/src/cutter/CutterBridgeService.ts index 2d05249f0e..3836ea7001 100644 --- a/src/cutter/CutterBridgeService.ts +++ b/src/cutter/CutterBridgeService.ts @@ -13,11 +13,14 @@ import { CutterBridgeClient, type CutterBridgeDiscovery, type CutterBridgeExecution, + type CutterDiscoveryOptions, } from "./CutterBridgeClient.js"; import { CUTTER_PROVIDER_IDENTITY } from "./CutterBridgeClient.js"; interface CutterBridgePort { - listSessions(): Promise; + listSessions( + options?: CutterDiscoveryOptions, + ): Promise; confirmCommandCompletion?( sessionId: string, requestId: string, @@ -74,8 +77,35 @@ export class CutterBridgeService { return ok(null); } - listSessions(): Promise { - return this.client.listSessions(); + async listSessions( + options: CutterDiscoveryOptions = {}, + ): Promise> { + const { signal } = options; + try { + signal?.throwIfAborted(); + const discovery = await this.client.listSessions(options); + signal?.throwIfAborted(); + return ok(discovery); + } catch (cause) { + if (signal?.aborted) + return err( + new AnalysisCancelledError("list_cutter_sessions", { + cause: signal.reason, + }), + ); + return err( + cause instanceof AnalysisError + ? cause + : new AnalysisCapabilityUnavailableError( + "cutter", + "list_cutter_sessions", + cause instanceof Error + ? cause.message + : "Cutter discovery failed", + { cause }, + ), + ); + } } async execute( @@ -85,7 +115,11 @@ export class CutterBridgeService { const parsed = cutterCommandInputSchema.safeParse(rawInput); if (!parsed.success) return err(new AnalysisInputError("cutter_command")); if (options.signal?.aborted) - return err(new AnalysisCancelledError("cutter_command")); + return err( + new AnalysisCancelledError("cutter_command", { + cause: options.signal.reason, + }), + ); const input = parsed.data; const cleanup = await this.retryCleanup(); if (!cleanup.ok) return err(cleanup.error); @@ -167,6 +201,13 @@ export class CutterBridgeService { }); return err(failure); } + if (options.signal?.aborted) + return err( + new AnalysisCancelledError("cutter_command", { + cause: options.signal.reason, + partialObservation: evidence, + }), + ); return ok(evidence); } catch (cause) { if (cause instanceof AnalysisError) return err(cause); diff --git a/src/server/registerCutterTools.ts b/src/server/registerCutterTools.ts index ab23feca91..527a4db747 100644 --- a/src/server/registerCutterTools.ts +++ b/src/server/registerCutterTools.ts @@ -4,7 +4,6 @@ import type { Logger } from "pino"; import type { CutterBridgeService } from "../cutter/CutterBridgeService.js"; import type { EvidenceWriter } from "../application/investigation/InvestigationRecordPort.js"; import { toolContract } from "../contracts/toolContracts.js"; -import { ok } from "../domain/result.js"; import { evidenceSchema } from "../domain/evidence.js"; import { runWithExternalMutationAdmission, @@ -25,19 +24,23 @@ export const registerCutterTools = ( ): void => { const list = toolContract("list_cutter_sessions"); const command = toolContract("cutter_command"); - server.registerTool(list.name, toolRegistrationOptions(list), async () => { - const result = await logToolExecution(logger, list.name, async () => - ok(await service.listSessions()), - ); - if (!result.ok) return delivery.toCallToolResult(result, list); - return delivery.toCallToolResult( - { - ok: true, - value: { ...result.value, sessions: [...result.value.sessions] }, - }, - list, - ); - }); + server.registerTool( + list.name, + toolRegistrationOptions(list), + async (_input, context) => { + const result = await logToolExecution(logger, list.name, () => + service.listSessions({ signal: context.mcpReq.signal }), + ); + if (!result.ok) return delivery.toCallToolResult(result, list); + return delivery.toCallToolResult( + { + ok: true, + value: { ...result.value, sessions: [...result.value.sessions] }, + }, + list, + ); + }, + ); server.registerTool( command.name, toolRegistrationOptions(command), diff --git a/tests/boundary/cli/cutterCommands.test.ts b/tests/boundary/cli/cutterCommands.test.ts index 4acc851a63..dd0931e0a4 100644 --- a/tests/boundary/cli/cutterCommands.test.ts +++ b/tests/boundary/cli/cutterCommands.test.ts @@ -1,5 +1,5 @@ import { createServer } from "node:net"; -import { createHmac } from "node:crypto"; +import { createHmac, randomUUID } from "node:crypto"; import { mkdtemp, rm, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; @@ -7,6 +7,10 @@ import { join } from "node:path"; import { afterEach, describe, expect, it } from "vitest"; import { serveCli } from "./cliServeHarness.js"; +import { + spawnOwnedProviderProcess, + ProviderProcessSupervisor, +} from "../../../src/process/ProviderProcess.js"; const SESSION_ID = "27d3e3f1-f1e5-49ae-91ec-95af1f343a5a"; const listen = async ( @@ -47,6 +51,72 @@ describe("Cutter CLI commands", () => { }); }); + it.skipIf(process.platform === "win32")( + "awaits discovery cancellation on SIGINT and returns its typed outcome with exit130", + async () => { + directory = await mkdtemp(join(tmpdir(), "rea-cutter-cli-cancel-")); + const sockets = new Set(); + let accepted = 0; + server = createServer((socket) => { + accepted += 1; + sockets.add(socket); + socket.on("error", () => undefined); + socket.on("close", () => sockets.delete(socket)); + socket.resume(); + }); + const port = await listen(server); + for (let index = 0; index < 9; index += 1) { + const id = randomUUID(); + await writeFile( + join(directory, `cutter-${process.pid}-${id}.json`), + JSON.stringify({ + session_id: id, + pid: process.pid, + host: "127.0.0.1", + port, + token: "cli-discovery-test-transport-token", + document_generation: 0, + current_file: null, + cutter_version: null, + identity_status: "partial", + }), + { mode: 0o600 }, + ); + } + const launched = await spawnOwnedProviderProcess({ + command: process.execPath, + arguments: ["scripts/rea.mjs", "list-cutter-sessions", "--json"], + runId: randomUUID(), + hostEnvironment: { + ...process.env, + REA_CUTTER_BRIDGE_DIR: directory, + REA_LOG_LEVEL: "silent", + }, + }); + const supervisor = new ProviderProcessSupervisor({ + ...launched, + ownsProcessLifetime: true, + }); + try { + await expect.poll(() => accepted, { timeout: 5_000 }).toBe(8); + launched.process.kill("SIGINT"); + expect(await supervisor.waitForExit(5_000)).toBe(true); + expect(await supervisor.waitForOutputClose(1_000)).toBe(true); + expect(launched.process.exitCode).toBe(130); + expect(launched.process.signalCode).toBeNull(); + expect(JSON.parse(supervisor.snapshot().stdout.text)).toMatchObject({ + code: "cancelled", + }); + await expect.poll(() => sockets.size).toBe(0); + expect(accepted).toBe(8); + } finally { + for (const socket of sockets) socket.destroy(); + const stopped = await supervisor.stop(); + expect(stopped.status).not.toBe("incomplete"); + } + }, + ); + it("reports an unavailable selected session for a command", async () => { directory = await mkdtemp(join(tmpdir(), "rea-cutter-cli-test-")); diff --git a/tests/boundary/mcp/cutterCancellationMcp.test.ts b/tests/boundary/mcp/cutterCancellationMcp.test.ts index 5a4a5167e6..82174957f3 100644 --- a/tests/boundary/mcp/cutterCancellationMcp.test.ts +++ b/tests/boundary/mcp/cutterCancellationMcp.test.ts @@ -1,11 +1,113 @@ import { Client, InMemoryTransport } from "@modelcontextprotocol/client"; import { McpServer } from "@modelcontextprotocol/server"; import { expect, it } from "vitest"; +import { chmod, rm, writeFile } from "node:fs/promises"; +import { createServer, type Socket } from "node:net"; +import { join } from "node:path"; +import { randomUUID } from "node:crypto"; +import { CutterBridgeClient } from "../../../src/cutter/CutterBridgeClient.js"; import { CutterBridgeService } from "../../../src/cutter/CutterBridgeService.js"; +import { toolContract } from "../../../src/contracts/toolContracts.js"; import { registerCutterTools } from "../../../src/server/registerCutterTools.js"; import { createToolResultDelivery } from "../../../src/server/toolResult.js"; import { silentLogger } from "../../../src/logger.js"; +import { createTestTempDirectory } from "../../fixtures/temporaryDirectory.js"; + +it.skipIf(process.platform === "win32")( + "closes cancelled discovery probes, stops queued probes and reuses the MCP connection", + async () => { + const directory = await createTestTempDirectory("rea-cutter-discovery-"); + await chmod(directory, 0o700); + const sockets = new Set(); + let accepted = 0; + const peer = createServer((socket) => { + accepted += 1; + sockets.add(socket); + socket.on("error", () => undefined); + socket.once("close", () => sockets.delete(socket)); + // Drain authentication bytes so a paused readable cannot hide peer EOF. + socket.resume(); + }); + const descriptors: string[] = []; + const service = new CutterBridgeService( + new CutterBridgeClient({ REA_CUTTER_BRIDGE_DIR: directory }), + ); + const server = new McpServer({ name: "cutter-discovery", version: "1" }); + registerCutterTools( + server, + service, + silentLogger, + createToolResultDelivery(undefined), + ); + const client = new Client({ name: "cutter-discovery", version: "1" }); + const [clientTransport, serverTransport] = + InMemoryTransport.createLinkedPair(); + const controller = new AbortController(); + try { + await new Promise((resolve) => + peer.listen(0, "127.0.0.1", resolve), + ); + const address = peer.address(); + if (address === null || typeof address === "string") + throw new Error("Expected private loopback listener"); + for (let index = 0; index < 9; index += 1) { + const sessionId = randomUUID(); + const path = join(directory, `cutter-${process.pid}-${sessionId}.json`); + descriptors.push(path); + await writeFile( + path, + JSON.stringify({ + session_id: sessionId, + pid: process.pid, + host: "127.0.0.1", + port: address.port, + token: "discovery-test-transport-token", + document_generation: 0, + current_file: null, + cutter_version: null, + identity_status: "partial", + }), + { mode: 0o600 }, + ); + } + await server.connect(serverTransport); + await client.connect(clientTransport); + const request = client.callTool( + { name: "list_cutter_sessions", arguments: {} }, + { signal: controller.signal }, + ); + await expect.poll(() => accepted, { timeout: 5_000 }).toBe(8); + controller.abort(); + await expect(request).rejects.toThrow(/abort/iu); + // Inspect the native peers; rejecting the SDK promise alone proves no cleanup. + await expect.poll(() => sockets.size, { timeout: 1_000 }).toBe(0); + expect(accepted).toBe(8); + for (const path of descriptors) await rm(path); + const reply = await client.callTool({ + name: "list_cutter_sessions", + arguments: {}, + }); + expect(reply.isError).not.toBe(true); + expect( + toolContract("list_cutter_sessions").outputSchema.parse( + reply.structuredContent, + ), + ).toMatchObject({ + sessions: [], + discovery_status: "no_live_bridge_found", + bridge_directory_security: "private_verified", + }); + expect(accepted).toBe(8); + } finally { + for (const socket of sockets) socket.destroy(); + await client.close(); + await server.close(); + await new Promise((resolve) => peer.close(() => resolve())); + await rm(directory, { recursive: true, force: true }); + } + }, +); it("propagates MCP cancellation into an active Cutter command and reports uncertain completion", async () => { let markDispatched: (() => void) | undefined; diff --git a/tests/boundary/providers/CutterBridgeClient.test.ts b/tests/boundary/providers/CutterBridgeClient.test.ts index dc3986b077..69a1775ad5 100644 --- a/tests/boundary/providers/CutterBridgeClient.test.ts +++ b/tests/boundary/providers/CutterBridgeClient.test.ts @@ -274,6 +274,105 @@ const verifyCutterCancellationAfterDispatch = async (): Promise => { } }; +it.skipIf(process.platform === "win32").each(["discovery", "command"] as const)( + "retains closed sockets and completed facts when %s is cancelled", + async (operation) => { + const directory = await mkdtemp(join(tmpdir(), "rea-cutter-close-order-")); + const sessionId = "27d3e3f1-f1e5-49ae-91ec-95af1f343a5a"; + const token = "close-order-test-token"; + const controller = new AbortController(); + const reason = new Error("opaque caller cancellation after response"); + let commandReplied = false; + let accepted!: () => void; + const connected = new Promise((resolve) => { + accepted = resolve; + }); + const peer = + operation === "discovery" + ? createServer((socket) => { + socket.resume(); + accepted(); + }) + : authenticatedServer(token, sessionId, (request, respond) => { + if (request.kind === "ack") { + respond({ ok: true, acknowledged: true }); + return; + } + if (request.kind === "command") commandReplied = true; + respond({ + ok: true, + output: "completed command output", + execution_state: "complete", + }); + }); + let closed = false; + await vi.resetModules(); + vi.doMock("node:net", async (importOriginal) => { + const net = await importOriginal(); + return { + ...net, + createConnection: (options: import("node:net").NetConnectOpts) => { + const socket = net.createConnection(options); + socket.once("close", () => { + closed = true; + if (commandReplied) controller.abort(reason); + }); + return socket; + }, + }; + }); + try { + const port = await listen(peer); + await writeDescriptor(directory, sessionId, port, token); + const { CutterBridgeClient: ObservedClient } = + await import("../../../src/cutter/CutterBridgeClient.js"); + const service = new CutterBridgeService( + new ObservedClient({ REA_CUTTER_BRIDGE_DIR: directory }), + ); + if (operation === "discovery") { + const discovery = service.listSessions({ signal: controller.signal }); + await connected; + controller.abort(); + expect(await discovery).toMatchObject({ + ok: false, + error: { _tag: "AnalysisCancelledError" }, + }); + } else { + const result = await service.execute( + { + session_id: sessionId, + expected_generation: 0, + command: "pd 4", + }, + { signal: controller.signal }, + ); + expect(result).toMatchObject({ + ok: false, + error: { + _tag: "AnalysisCancelledError", + cause: reason, + cleanupIncomplete: false, + partialObservation: { + normalized_result: { + execution_state: "complete", + output: "completed command output", + }, + }, + }, + }); + expect(await service.retryCleanup()).toEqual({ ok: true, value: null }); + } + expect(closed).toBe(true); + } finally { + controller.abort(); + vi.doUnmock("node:net"); + await vi.resetModules(); + await new Promise((resolve) => peer.close(() => resolve())); + await rm(directory, { recursive: true, force: true }); + } + }, +); + describe("CutterBridgeClient: scenario group 1", () => { let directory: string | undefined; let server: ReturnType | undefined; @@ -1381,7 +1480,9 @@ describe("CutterBridgeClient: scenario group 9", () => { throw new Error("unexpected command dispatch"); }, }); - await expect(service.listSessions()).resolves.toBe(discovery); + const result = await service.listSessions(); + expect(result.ok).toBe(true); + if (result.ok) expect(result.value).toBe(discovery); }); it("accepts only the exact completed-command receipt from status", async () => { diff --git a/tests/boundary/providers/CutterBridgeService.test.ts b/tests/boundary/providers/CutterBridgeService.test.ts index dd430246d6..45a7d1ad76 100644 --- a/tests/boundary/providers/CutterBridgeService.test.ts +++ b/tests/boundary/providers/CutterBridgeService.test.ts @@ -404,3 +404,41 @@ it("shares one cleanup attempt when commands retry the same receipt together", a expect(confirms).toBe(1); expect(dispatches).toBe(3); }); + +it.each([ + { phase: "before", reason: { code: "EACCES" } }, + { + phase: "during", + reason: { cleanupIncomplete: true, resource: "caller resource" }, + }, +])( + "models opaque discovery cancellation $phase the port call", + async ({ phase, reason }) => { + const controller = new AbortController(); + let calls = 0; + const service = new CutterBridgeService({ + listSessions: async () => { + calls += 1; + controller.abort(reason); + return { + sessions: [], + bridge_directory: "/private/bridge", + discovery_status: "no_live_bridge_found", + bridge_directory_security: "private_verified", + }; + }, + execute: async () => { + throw new Error("Unexpected command dispatch"); + }, + }); + if (phase === "before") controller.abort(reason); + const result = await service.listSessions({ signal: controller.signal }); + expect(result).toMatchObject({ + ok: false, + error: { _tag: "AnalysisCancelledError", cleanupIncomplete: false }, + }); + if (result.ok) throw new Error("Expected cancellation"); + expect(result.error.cause).toBe(reason); + expect(calls).toBe(phase === "before" ? 0 : 1); + }, +);