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); + }, +);