From e7801483055352171d6b033a61563cbb8c1987c6 Mon Sep 17 00:00:00 2001 From: morluto <76467478+morluto@users.noreply.github.com> Date: Thu, 22 Oct 2026 22:07:11 +0800 Subject: [PATCH] fix(cutter): preserve producer failure and cancellation facts --- docs/reverse-engineering-tools.md | 6 +- src/cutter/CutterBridgeClient.ts | 408 ++++++++++-------- src/cutter/CutterBridgeService.ts | 45 +- .../mcp/cutterCancellationMcp.test.ts | 122 ++++++ .../providers/CutterBridgeClient.test.ts | 39 +- .../providers/CutterBridgeService.test.ts | 132 ++++-- .../CutterCommandCancellation.test.ts | 72 ++++ .../providers/cutterBridgeProducer.test.ts | 11 +- 8 files changed, 599 insertions(+), 236 deletions(-) diff --git a/docs/reverse-engineering-tools.md b/docs/reverse-engineering-tools.md index 0bd27ef2bb..ab2e0c83ed 100644 --- a/docs/reverse-engineering-tools.md +++ b/docs/reverse-engineering-tools.md @@ -88,7 +88,11 @@ identify `cutter_command` and retain `status` in `details.diagnostics.auxiliary_operation`; malformed proof still blocks dispatch. Descriptor failures likewise retain the requesting operation through cleanup. Command admission reports typed cancellation and treats the caller's abort reason -as opaque metadata, without inferring cleanup ownership from it. +as opaque metadata, without inferring cleanup ownership from it. A generation +mismatch is an input correction only when the authenticated reply confirms +`not_started`. Missing or ambiguous execution state retains the command's +completion uncertainty until terminal proof arrives, even if the reply names a +generation error. ## Configuration and coverage diff --git a/src/cutter/CutterBridgeClient.ts b/src/cutter/CutterBridgeClient.ts index c02c3b5eb2..1085a829e7 100644 --- a/src/cutter/CutterBridgeClient.ts +++ b/src/cutter/CutterBridgeClient.ts @@ -20,6 +20,7 @@ import { z } from "zod"; import { AnalysisCapabilityUnavailableError, + AnalysisInputError, AnalysisOutputError, AnalysisResourceConstraintError, } from "../domain/analysisErrorCore.js"; @@ -180,9 +181,12 @@ class CutterBridgeResponseFrames { } } -const assertCommandActive = (signal: AbortSignal | undefined): void => { +const assertBridgeActive = ( + signal: AbortSignal | undefined, + operation: "cutter_command" | "list_cutter_sessions", +): void => { if (signal?.aborted) - throw new AnalysisCancelledError("cutter_command", { + throw new AnalysisCancelledError(operation, { cause: signal.reason, }); }; @@ -190,9 +194,9 @@ const assertCommandActive = (signal: AbortSignal | undefined): void => { const mapDiscovery = async ( inputs: readonly Input[], operation: (input: Input) => Promise, - signal?: AbortSignal, + assertActive: () => void, ): Promise => { - signal?.throwIfAborted(); + assertActive(); const results = new Array(inputs.length); let next = 0; let stopped = false; @@ -202,7 +206,7 @@ const mapDiscovery = async ( async () => { try { while (!stopped && next < inputs.length) { - signal?.throwIfAborted(); + assertActive(); const index = next++; results[index] = await operation(inputs[index]!); } @@ -218,7 +222,7 @@ const mapDiscovery = async ( ); if (failures.length > 1) throw new CutterDiscoveryFailure(failures); if (failures.length === 1) throw failures[0]; - signal?.throwIfAborted(); + assertActive(); return results; }; @@ -248,6 +252,7 @@ class CutterBridgeRequestFailure extends Error { constructor( message: string, readonly requestSent: boolean, + readonly cancellation?: { readonly reason: unknown }, ) { super(message); } @@ -333,7 +338,7 @@ export class CutterBridgeClient { try { await this.#descriptorResources.retryCleanup(); } catch (cause: unknown) { - throw descriptorFailure(cause, "close_cutter_descriptors"); + throw discoveryFailure(cause, "close_cutter_descriptors"); } } @@ -341,7 +346,7 @@ export class CutterBridgeClient { try { await this.#descriptorResources.close(); } catch (cause: unknown) { - throw descriptorFailure(cause, "close_cutter_descriptors"); + throw discoveryFailure(cause, "close_cutter_descriptors"); } } @@ -360,10 +365,7 @@ export class CutterBridgeClient { const descriptors = await mapDiscovery( loaded.descriptors, async (descriptor) => { - if (options.signal?.aborted) - throw new AnalysisCancelledError("list_cutter_sessions", { - cause: options.signal.reason, - }); + assertBridgeActive(signal, "list_cutter_sessions"); try { const status = responseSchema.safeParse( await this.#request( @@ -394,17 +396,21 @@ export class CutterBridgeClient { status.data.cutter_version ?? descriptor.cutter_version, identity_status: "partial" as const, }; - } catch { - signal?.throwIfAborted(); + } catch (cause) { + if ( + cause instanceof CutterBridgeRequestFailure && + cause.cancellation !== undefined + ) + throw new AnalysisCancelledError("list_cutter_sessions", { + cause: cause.cancellation.reason, + }); return undefined; } }, - signal, - ); - if (options.signal?.aborted) - throw new AnalysisCancelledError("list_cutter_sessions", { - cause: options.signal.reason, - }); + () => assertBridgeActive(signal, "list_cutter_sessions"), + ).catch((cause: unknown) => { + throw discoveryFailure(cause, "list_cutter_sessions"); + }); const sessions = descriptors.filter( (value): value is CutterBridgeSession => value !== undefined, ); @@ -424,100 +430,140 @@ export class CutterBridgeClient { readonly json: boolean; readonly signal?: AbortSignal; }): Promise { - assertCommandActive(input.signal); - const descriptor = await this.#find(input.sessionId, input.signal); - assertCommandActive(input.signal); - const requestId = randomUUID(); - const commandRequest = serializeRequest(descriptor, { - kind: "command", - request_id: requestId, - expected_generation: input.expectedGeneration, - command: input.command, - json: input.json, - }); - await this.#recoverCompletedCommands(descriptor, input.signal); - await this.#retryPendingAcknowledgements(descriptor); - assertCommandActive(input.signal); - let completionToken: string | undefined; - let rawResponse: unknown; try { - rawResponse = await this.#request( - descriptor, - commandRequest, - input.signal, - (ordinal) => { - completionToken = ordinal; - }, - ); + assertBridgeActive(input.signal, "cutter_command"); + const descriptor = await this.#find(input.sessionId, input.signal); + assertBridgeActive(input.signal, "cutter_command"); + const requestId = randomUUID(); + const commandRequest = serializeRequest(descriptor, { + kind: "command", + request_id: requestId, + expected_generation: input.expectedGeneration, + command: input.command, + json: input.json, + }); + await this.#recoverCompletedCommands(descriptor, input.signal); + await this.#retryPendingAcknowledgements(descriptor); + assertBridgeActive(input.signal, "cutter_command"); + let completionToken: string | undefined; + let rawResponse: unknown; + try { + rawResponse = await this.#request( + descriptor, + commandRequest, + input.signal, + (ordinal) => { + completionToken = ordinal; + }, + ); + } catch (cause) { + if ( + !(cause instanceof CutterBridgeRequestFailure) || + !cause.requestSent + ) + throw cause; + return { + requestId, + output: null, + currentFile: descriptor.current_file, + documentGeneration: input.expectedGeneration, + identityStatus: "partial", + cutterVersion: descriptor.cutter_version, + executionState: "unknown", + error: "transport-response-missing", + message: `${cause.message}; the command may have completed; do not retry automatically`, + outputTruncated: false, + ...(completionToken === undefined ? {} : { completionToken }), + }; + } + const parsed = responseSchema.safeParse(rawResponse); + if (!parsed.success) + return { + requestId, + output: null, + currentFile: descriptor.current_file, + documentGeneration: input.expectedGeneration, + identityStatus: "partial", + cutterVersion: descriptor.cutter_version, + executionState: "unknown", + error: "malformed-bridge-response", + message: + "Cutter may have completed the command, but its response was invalid; do not retry automatically", + outputTruncated: false, + ...(completionToken === undefined ? {} : { completionToken }), + }; + const response = parsed.data; + const executionState: CutterBridgeExecutionState = response.ok + ? response.execution_state === undefined || + response.execution_state === "complete" + ? "complete" + : response.execution_state === "not_started" + ? "not_started" + : "unknown" + : response.execution_state === "failed" + ? "failed" + : response.execution_state === "not_started" + ? "not_started" + : "unknown"; + if (executionState !== "unknown") + await this.#acknowledgeCompletion(descriptor, requestId); + if ( + executionState === "not_started" && + response.error === "active-document-generation-changed" + ) + throw new AnalysisInputError( + "cutter_command", + { + diagnostics: { + session_id: input.sessionId, + expected_generation: input.expectedGeneration, + ...(response.document_generation === undefined + ? {} + : { observed_generation: response.document_generation }), + execution_state: executionState, + }, + }, + [ + { + path: ["expected_generation"], + reason: "invalid_value", + message: + "Cutter's active document generation changed. List sessions again and select the current generation.", + }, + ], + ); + return { + requestId, + output: response.output ?? null, + currentFile: + response.current_file === undefined + ? descriptor.current_file + : response.current_file, + documentGeneration: + response.document_generation ?? input.expectedGeneration, + identityStatus: "partial", + cutterVersion: response.cutter_version ?? descriptor.cutter_version, + executionState, + error: response.ok ? null : (response.error ?? "cutter-command-failed"), + message: response.message ?? null, + outputTruncated: response.output_truncated ?? false, + ...(completionToken === undefined ? {} : { completionToken }), + }; } catch (cause) { - if (!(cause instanceof CutterBridgeRequestFailure) || !cause.requestSent) - throw cause; - return { - requestId, - output: null, - currentFile: descriptor.current_file, - documentGeneration: input.expectedGeneration, - identityStatus: "partial", - cutterVersion: descriptor.cutter_version, - executionState: "unknown", - error: "transport-response-missing", - message: `${cause.message}; the command may have completed; do not retry automatically`, - outputTruncated: false, - ...(completionToken === undefined ? {} : { completionToken }), - }; - } - const parsed = responseSchema.safeParse(rawResponse); - if (!parsed.success) - return { - requestId, - output: null, - currentFile: descriptor.current_file, - documentGeneration: input.expectedGeneration, - identityStatus: "partial", - cutterVersion: descriptor.cutter_version, - executionState: "unknown", - error: "malformed-bridge-response", - message: - "Cutter may have completed the command, but its response was invalid; do not retry automatically", - outputTruncated: false, - ...(completionToken === undefined ? {} : { completionToken }), - }; - const response = parsed.data; - const executionState: CutterBridgeExecutionState = response.ok - ? response.execution_state === undefined || - response.execution_state === "complete" - ? "complete" - : response.execution_state === "not_started" - ? "not_started" - : "unknown" - : response.execution_state === "failed" - ? "failed" - : response.execution_state === "not_started" - ? "not_started" - : "unknown"; - if (executionState !== "unknown") - await this.#acknowledgeCompletion(descriptor, requestId); - if (response.error === "active-document-generation-changed") - throw new Error( - "Cutter's active document generation changed. List sessions again and select the current generation.", + // Sent command failures already became an unknown execution above. + // A transport failure escaping here precedes command admission. + if (!(cause instanceof CutterBridgeRequestFailure)) throw cause; + if (cause.cancellation !== undefined) + throw new AnalysisCancelledError("cutter_command", { + cause: cause.cancellation.reason, + }); + throw new AnalysisCapabilityUnavailableError( + "cutter", + "cutter_command", + cause.message, + { cause }, ); - return { - requestId, - output: response.output ?? null, - currentFile: - response.current_file === undefined - ? descriptor.current_file - : response.current_file, - documentGeneration: - response.document_generation ?? input.expectedGeneration, - identityStatus: "partial", - cutterVersion: response.cutter_version ?? descriptor.cutter_version, - executionState, - error: response.ok ? null : (response.error ?? "cutter-command-failed"), - message: response.message ?? null, - outputTruncated: response.output_truncated ?? false, - ...(completionToken === undefined ? {} : { completionToken }), - }; + } } /** Confirm completion by exact receipt or authenticated connection watermark. */ @@ -688,7 +734,14 @@ export class CutterBridgeClient { (candidate) => candidate.session_id === sessionId, ); if (descriptor === undefined) - throw new Error("Cutter bridge session is unavailable or stale"); + throw new AnalysisInputError("cutter_command", undefined, [ + { + path: ["session_id"], + reason: "invalid_value", + message: + "Cutter bridge session is unavailable or stale; list sessions again and select an available session.", + }, + ]); return descriptor; } @@ -708,7 +761,7 @@ export class CutterBridgeClient { this.#loadDescriptors(signal), ); } catch (cause: unknown) { - throw descriptorFailure(cause, operation); + throw discoveryFailure(cause, operation); } } @@ -812,61 +865,65 @@ export class CutterBridgeClient { error: "bridge_directory_unavailable", }; } - const loaded = await mapDiscovery(entries, async (entry) => { - assertArtifactReadActive(signal); - try { - let bytes: Buffer; - if (this.#platform === "win32") { - bytes = windowsPrivateReader!.readDescriptor( - this.#directory, - entry, - MAX_DESCRIPTOR_BYTES, - ); - } else { - const boundedBytes = await withArtifactRegularFile( - join(this.#directory, entry), - this.#descriptorResources, - signal, - async (file) => { - const info = await file.stat(); - if ( - info.size > MAX_DESCRIPTOR_BYTES || - info.uid !== currentUid || - (info.mode & 0o077) !== 0 - ) - return undefined; - assertArtifactReadActive(signal); - const bytes = await readBoundedFileBytes( - file, - MAX_DESCRIPTOR_BYTES, - ); - assertArtifactReadActive(signal); - return bytes; - }, - ); - if (boundedBytes === undefined) return undefined; - bytes = boundedBytes; - } + const loaded = await mapDiscovery( + entries, + async (entry) => { assertArtifactReadActive(signal); - if (bytes.byteLength > MAX_DESCRIPTOR_BYTES) return undefined; - const parsed = descriptorSchema.safeParse( - JSON.parse(bytes.toString("utf8")) as unknown, - ); - if ( - !parsed.success || - entry !== `cutter-${parsed.data.pid}-${parsed.data.session_id}.json` - ) + try { + let bytes: Buffer; + if (this.#platform === "win32") { + bytes = windowsPrivateReader!.readDescriptor( + this.#directory, + entry, + MAX_DESCRIPTOR_BYTES, + ); + } else { + const boundedBytes = await withArtifactRegularFile( + join(this.#directory, entry), + this.#descriptorResources, + signal, + async (file) => { + const info = await file.stat(); + if ( + info.size > MAX_DESCRIPTOR_BYTES || + info.uid !== currentUid || + (info.mode & 0o077) !== 0 + ) + return undefined; + assertArtifactReadActive(signal); + const bytes = await readBoundedFileBytes( + file, + MAX_DESCRIPTOR_BYTES, + ); + assertArtifactReadActive(signal); + return bytes; + }, + ); + if (boundedBytes === undefined) return undefined; + bytes = boundedBytes; + } + assertArtifactReadActive(signal); + if (bytes.byteLength > MAX_DESCRIPTOR_BYTES) return undefined; + const parsed = descriptorSchema.safeParse( + JSON.parse(bytes.toString("utf8")) as unknown, + ); + if ( + !parsed.success || + entry !== `cutter-${parsed.data.pid}-${parsed.data.session_id}.json` + ) + return undefined; + return parsed.data; + } catch (cause: unknown) { + if ( + cause instanceof ArtifactReaderFailure && + (cause.cleanup !== undefined || cause.reason === "cancelled") + ) + throw cause; return undefined; - return parsed.data; - } catch (cause: unknown) { - if ( - cause instanceof ArtifactReaderFailure && - (cause.cleanup !== undefined || cause.reason === "cancelled") - ) - throw cause; - return undefined; - } - }); + } + }, + () => assertArtifactReadActive(signal), + ); assertArtifactReadActive(signal); return { descriptors: loaded.filter( @@ -903,9 +960,14 @@ export class CutterBridgeClient { // Retain this request until its owned socket has actually closed. connection.destroy(); }; - const rejectRequest = (message: string): void => + const rejectRequest = ( + message: string, + cancellation?: { readonly reason: unknown }, + ): void => finish(() => - reject(new CutterBridgeRequestFailure(message, requestSent)), + reject( + new CutterBridgeRequestFailure(message, requestSent, cancellation), + ), ); const timer = setTimeout(() => { rejectRequest( @@ -913,7 +975,9 @@ export class CutterBridgeClient { ); }, REQUEST_TIMEOUT_MS); const onAbort = (): void => { - rejectRequest("Cutter bridge request was cancelled"); + rejectRequest("Cutter bridge request was cancelled", { + reason: signal?.reason, + }); }; signal?.addEventListener("abort", onAbort, { once: true }); if (signal?.aborted) onAbort(); @@ -1024,13 +1088,13 @@ class CutterDiscoveryFailure extends Error { } } -const descriptorFailure = ( +const discoveryFailure = ( cause: unknown, operation: string, ): AnalysisError | unknown => { if (cause instanceof CutterDiscoveryFailure) { const failures = cause.failures.map((failure) => { - const parsed = descriptorFailure(failure, operation); + const parsed = discoveryFailure(failure, operation); return parsed instanceof AnalysisError ? parsed : new ProviderAdapterError("cutter", operation, { cause: parsed }); @@ -1059,7 +1123,7 @@ const descriptorFailure = ( : undefined; const primary = outcome?.kind === "failed" - ? descriptorFailure(outcome.cause, operation) + ? discoveryFailure(outcome.cause, operation) : outcome?.kind === "completed" && cause instanceof ArtifactFileCleanupFailure && cause.fileCleanupFailure.completionCancellation !== undefined diff --git a/src/cutter/CutterBridgeService.ts b/src/cutter/CutterBridgeService.ts index f267b41404..4509f31f28 100644 --- a/src/cutter/CutterBridgeService.ts +++ b/src/cutter/CutterBridgeService.ts @@ -85,23 +85,23 @@ export class CutterBridgeService { options: CutterDiscoveryOptions = {}, ): Promise> { const { signal } = options; + if (signal?.aborted) + return err( + new AnalysisCancelledError("list_cutter_sessions", { + cause: signal.reason, + }), + ); try { - signal?.throwIfAborted(); const discovery = await this.client.listSessions(options); - signal?.throwIfAborted(); - return ok(discovery); - } catch (cause) { - if ( - cause instanceof AnalysisError && - !(signal?.aborted && cause === signal.reason) - ) - return err(cause); if (signal?.aborted) return err( new AnalysisCancelledError("list_cutter_sessions", { cause: signal.reason, }), ); + return ok(discovery); + } catch (cause) { + if (cause instanceof AnalysisError) return err(cause); return err( new AnalysisCapabilityUnavailableError( "cutter", @@ -244,26 +244,15 @@ export class CutterBridgeService { return ok(evidence); } catch (cause) { if (cause instanceof AnalysisError) return err(cause); - const message = - cause instanceof Error - ? cause.message - : "Unknown Cutter bridge failure"; - if (options.signal?.aborted && !message.includes("may have completed")) - return err( - new AnalysisCancelledError("cutter_command", { - cause: options.signal.reason, - }), - ); return err( - message.includes("generation changed") || - message.includes("unavailable or stale") - ? new AnalysisInputError("cutter_command", { cause }) - : new AnalysisCapabilityUnavailableError( - "cutter", - "cutter_command", - message, - { cause }, - ), + new AnalysisCapabilityUnavailableError( + "cutter", + "cutter_command", + cause instanceof Error + ? cause.message + : "Unknown Cutter bridge failure", + { cause }, + ), ); } } diff --git a/tests/boundary/mcp/cutterCancellationMcp.test.ts b/tests/boundary/mcp/cutterCancellationMcp.test.ts index 35d32f6b80..4a2a5ca589 100644 --- a/tests/boundary/mcp/cutterCancellationMcp.test.ts +++ b/tests/boundary/mcp/cutterCancellationMcp.test.ts @@ -6,6 +6,11 @@ import { createServer, type Socket } from "node:net"; import { join } from "node:path"; import { randomUUID } from "node:crypto"; +import { + authenticatedServer, + writeDescriptor, +} from "../../fixtures/cutterBridgeProtocol.js"; +import { parseMcpToolError } from "../../fixtures/mcpToolError.js"; import { CutterBridgeClient } from "../../../src/cutter/CutterBridgeClient.js"; import { CutterBridgeService } from "../../../src/cutter/CutterBridgeService.js"; import { toolContract } from "../../../src/contracts/toolContracts.js"; @@ -370,3 +375,120 @@ it("cancels active discovery sockets, stops queued probes and recovers on the sa await new Promise((resolve) => peer.close(() => resolve())); } }); + +it.skipIf(process.platform === "win32")( + "retains a generation-error command until authenticated terminal proof", + async () => { + const directory = await createTestTempDirectory( + "rea-cutter-generation-proof-", + ); + await chmod(directory, 0o700); + const sessionId = "27d3e3f1-f1e5-49ae-91ec-95af1f343a5a"; + const token = "synthetic-cutter-bridge-token-for-test"; + let terminal = false; + let commands = 0; + let pendingRequestId = ""; + const peer = authenticatedServer( + token, + sessionId, + (request, respond) => { + if (request.kind === "status") + respond({ + ok: true, + session_id: sessionId, + completed_commands: + terminal && pendingRequestId !== "" + ? [ + { + request_id: pendingRequestId, + execution_state: "not_started", + }, + ] + : [], + completed_connection: terminal + ? "000000000000ffff" + : "0000000000000000", + }); + else if (request.kind === "ack") + respond({ ok: true, acknowledged: true }); + else { + commands++; + pendingRequestId = String(request.request_id); + respond( + commands === 1 + ? { ok: false, error: "active-document-generation-changed" } + : { ok: true, execution_state: "complete", output: "observed" }, + ); + } + }, + true, + ); + 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("Missing listener"); + await writeDescriptor(directory, sessionId, address.port, token); + const service = new CutterBridgeService( + new CutterBridgeClient({ REA_CUTTER_BRIDGE_DIR: directory }), + ); + const server = new McpServer({ + name: "cutter-generation-proof", + version: "1", + }); + registerCutterTools( + server, + service, + silentLogger, + createToolResultDelivery(undefined), + ); + const client = new Client({ + name: "cutter-generation-proof", + version: "1", + }); + const [ct, st] = InMemoryTransport.createLinkedPair(); + try { + await server.connect(st); + await client.connect(ct); + const invoke = () => + client.callTool({ + name: "cutter_command", + arguments: { + session_id: sessionId, + expected_generation: 0, + command: "pd 1", + }, + }); + const first = await invoke(); + expect(parseMcpToolError(first).error).toMatchObject({ + code: "cleanup_incomplete", + details: { + partial_observation: { + normalized_result: { + execution_state: "unknown", + error: "active-document-generation-changed", + }, + }, + }, + }); + const blocked = await invoke(); + expect(parseMcpToolError(blocked).error.code).toBe("cleanup_incomplete"); + expect(commands).toBe(1); + terminal = true; + const recovered = await invoke(); + expect(recovered.isError).not.toBe(true); + const evidence = toolContract("cutter_command").outputSchema.parse( + recovered.structuredContent, + ); + expect(evidence.normalized_result).toMatchObject({ + execution_state: "complete", + output: "observed", + }); + expect(commands).toBe(2); + } finally { + terminal = true; + await Promise.allSettled([client.close(), server.close()]); + await service.close(); + await new Promise((resolve) => peer.close(() => resolve())); + } + }, +); diff --git a/tests/boundary/providers/CutterBridgeClient.test.ts b/tests/boundary/providers/CutterBridgeClient.test.ts index 55ac06adf5..6c5bd40892 100644 --- a/tests/boundary/providers/CutterBridgeClient.test.ts +++ b/tests/boundary/providers/CutterBridgeClient.test.ts @@ -19,6 +19,7 @@ import { join } from "node:path"; import { afterEach, describe, expect, it, vi } from "vitest"; import { CutterBridgeClient } from "../../../src/cutter/CutterBridgeClient.js"; +import { AnalysisInputError } from "../../../src/domain/analysisErrorCore.js"; import { CutterBridgeService } from "../../../src/cutter/CutterBridgeService.js"; const listen = async ( @@ -197,7 +198,9 @@ it.skipIf(process.platform === "win32").each(["discovery", "command"] as const)( await writeDescriptor(directory, sessionId, port, token); const { CutterBridgeClient: ObservedClient } = await import("../../../src/cutter/CutterBridgeClient.js"); - const service = new CutterBridgeService( + const { CutterBridgeService: ObservedService } = + await import("../../../src/cutter/CutterBridgeService.js"); + const service = new ObservedService( new ObservedClient({ REA_CUTTER_BRIDGE_DIR: directory }), ); if (operation === "discovery") { @@ -696,12 +699,32 @@ describe("CutterBridgeClient: scenario group 4", () => { documentGeneration: 9, cutterVersion: "new", }); + for (const state of [undefined, "running", "unknown", "failed"]) { + response = { + ok: false, + execution_state: state, + error: "active-document-generation-changed", + }; + await expect(run()).resolves.toMatchObject({ + executionState: state === "failed" ? "failed" : "unknown", + error: "active-document-generation-changed", + }); + } response = { ok: false, execution_state: "not_started", error: "active-document-generation-changed", }; - await expect(run()).rejects.toThrow("List sessions again"); + await expect(run()).rejects.toMatchObject({ + _tag: "AnalysisInputError", + issues: [ + { + path: ["expected_generation"], + reason: "invalid_value", + message: expect.stringContaining("List sessions again"), + }, + ], + }); response = { ok: true, unexpected: true }; // Valid schema responses can omit all optional fields; the fallback result is nullable. await expect(run()).resolves.toMatchObject({ @@ -921,7 +944,7 @@ describe("CutterBridgeClient: scenario group 6", () => { command: "pd 1", json: false, }), - ).rejects.toThrow("unavailable or stale"); + ).rejects.toBeInstanceOf(AnalysisInputError); const unopened = createServer(); const port = await listen(unopened); @@ -997,12 +1020,11 @@ describe("CutterBridgeClient: scenario group 7", () => { ); const controller = new AbortController(); const signal = controller.signal; + const reason = new Error("cancel immediately after listener registration"); const add = signal.addEventListener.bind(signal); vi.spyOn(signal, "addEventListener").mockImplementation((...args) => { add(...args); - controller.abort( - new Error("cancel immediately after listener registration"), - ); + controller.abort(reason); }); await expect( createClient({ REA_CUTTER_BRIDGE_DIR: directory }).execute({ @@ -1012,7 +1034,10 @@ describe("CutterBridgeClient: scenario group 7", () => { json: false, signal, }), - ).rejects.toThrow("Cutter bridge request was cancelled"); + ).rejects.toMatchObject({ + _tag: "AnalysisCancelledError", + cause: reason, + }); }); it("supports platforms without O_NOFOLLOW using the documented read-only flags", async () => { diff --git a/tests/boundary/providers/CutterBridgeService.test.ts b/tests/boundary/providers/CutterBridgeService.test.ts index 7ec6e120f0..f7a7771761 100644 --- a/tests/boundary/providers/CutterBridgeService.test.ts +++ b/tests/boundary/providers/CutterBridgeService.test.ts @@ -1,6 +1,10 @@ import { expect, it, vi } from "vitest"; import { CutterBridgeService } from "../../../src/cutter/CutterBridgeService.js"; +import { + AnalysisInputError, + AnalysisCancelledError, +} from "../../../src/domain/analysisErrorCore.js"; import { ProviderCleanupError } from "../../../src/domain/providerCleanupError.js"; const cutterInput = { @@ -274,7 +278,7 @@ it("rejects malformed Cutter commands and pre-aborted calls", async () => { }); it.each([ - [new Error("session generation changed"), "AnalysisInputError"], + [new AnalysisInputError("cutter_command"), "AnalysisInputError"], [new Error("bridge disconnected"), "AnalysisCapabilityUnavailableError"], ["plain failure", "AnalysisCapabilityUnavailableError"], ])("maps a Cutter bridge failure to %s", async (failure, tag) => { @@ -294,9 +298,12 @@ it.each([ expect(result).toMatchObject({ ok: false, error: { _tag: tag } }); }); -it("preserves cancellation unless the bridge reports an ambiguous completion", async () => { - const createService = (message: string, controller: AbortController) => - new CutterBridgeService({ +it.each(["transport cancelled", "command may have completed"])( + "preserves a port failure through a later unrelated cancellation: %s", + async (message) => { + const controller = new AbortController(); + const primary = new Error(message); + const service = new CutterBridgeService({ listSessions: async () => ({ sessions: [], bridge_directory: "/private/bridge", @@ -304,31 +311,19 @@ it("preserves cancellation unless the bridge reports an ambiguous completion", a bridge_directory_security: "private_verified", }), execute: async () => { - controller.abort(); - throw new Error(message); + controller.abort(new Error("later cancellation")); + throw primary; }, }); - const cancelledController = new AbortController(); - - const cancelled = await createService( - "transport cancelled", - cancelledController, - ).execute(cutterInput, { signal: cancelledController.signal }); - expect(cancelled).toMatchObject({ - ok: false, - error: { _tag: "AnalysisCancelledError" }, - }); - - const uncertainController = new AbortController(); - const uncertain = await createService( - "command may have completed", - uncertainController, - ).execute(cutterInput, { signal: uncertainController.signal }); - expect(uncertain).toMatchObject({ - ok: false, - error: { _tag: "AnalysisCapabilityUnavailableError" }, - }); -}); + const result = await service.execute(cutterInput, { + signal: controller.signal, + }); + expect(result).toMatchObject({ + ok: false, + error: { _tag: "AnalysisCapabilityUnavailableError", cause: primary }, + }); + }, +); it("shares one cleanup attempt when commands retry the same receipt together", async () => { const requestId = "4b18ce4d-bf31-456e-ae31-e8c547b2f9db"; @@ -514,3 +509,86 @@ it.each(["descriptor-first", "command-first"])( expect(mixed.cleanupIncomplete).toBe(false); }, ); + +it.each([ + new Error("generation changed while recording transport diagnostics"), + new Error("unavailable or stale is diagnostic text"), +])( + "does not infer a session-selection failure from diagnostic prose: %s", + async (cause) => { + const service = new CutterBridgeService({ + listSessions: async () => ({ + sessions: [], + bridge_directory: "/private/bridge", + discovery_status: "no_live_bridge_found", + bridge_directory_security: "private_verified", + }), + execute: async () => { + throw cause; + }, + }); + const result = await service.execute(cutterInput); + expect(result).toMatchObject({ + ok: false, + error: { _tag: "AnalysisCapabilityUnavailableError", cause }, + }); + }, +); + +it.each([ + new Error("command may have completed is an opaque abort reason"), + new AnalysisInputError("caller-owned"), + new ProviderCleanupError("caller-owned", ["caller resource"], { + reason: "opaque value", + }), +])( + "retains typed cancellation without interpreting its opaque reason: %s", + async (reason) => { + const controller = new AbortController(); + const service = new CutterBridgeService({ + listSessions: async () => ({ + sessions: [], + bridge_directory: "/private/bridge", + discovery_status: "no_live_bridge_found", + bridge_directory_security: "private_verified", + }), + execute: async () => { + controller.abort(reason); + throw new AnalysisCancelledError("cutter_command", { cause: reason }); + }, + }); + const result = await service.execute(cutterInput, { + signal: controller.signal, + }); + expect(result).toMatchObject({ + ok: false, + error: { _tag: "AnalysisCancelledError", cleanupIncomplete: false }, + }); + if (result.ok) throw new Error("Missing cancellation"); + expect(result.error.cause).toBe(reason); + }, +); + +it.each([ + new AnalysisInputError("discovery producer failure"), + new ProviderCleanupError("cutter", ["owned descriptor"], { + reason: "physical descriptor close failure", + }), +])( + "preserves a discovery failure also selected as abort reason: $_tag", + async (failure) => { + const controller = new AbortController(); + const service = new CutterBridgeService({ + listSessions: async () => { + controller.abort(failure); + throw failure; + }, + execute: async () => { + throw new Error("Unexpected command dispatch"); + }, + }); + const result = await service.listSessions({ signal: controller.signal }); + if (result.ok) throw new Error("Expected discovery failure"); + expect(result.error).toBe(failure); + }, +); diff --git a/tests/boundary/providers/CutterCommandCancellation.test.ts b/tests/boundary/providers/CutterCommandCancellation.test.ts index 45d9e40feb..5cbcd7d4eb 100644 --- a/tests/boundary/providers/CutterCommandCancellation.test.ts +++ b/tests/boundary/providers/CutterCommandCancellation.test.ts @@ -1,3 +1,8 @@ +import { chmod } from "node:fs/promises"; +import { + authenticatedServer, + writeDescriptor, +} from "../../fixtures/cutterBridgeProtocol.js"; import { expect, it, onTestFinished } from "vitest"; import { AnalysisInputError, @@ -77,3 +82,70 @@ it.each([ expect(result.error).toBe(physical); }, ); + +it.skipIf(process.platform === "win32").each([ + new Error("caller stopped before command admission"), + new AnalysisInputError("caller-owned"), + new ProviderCleanupError("caller-owned", ["opaque caller resource"], { + reason: "opaque caller metadata", + }), +])( + "records cancellation during status recovery before command dispatch: %s", + async (reason) => { + const directory = await createTestTempDirectory( + "rea-cutter-status-cancel-", + ); + await chmod(directory, 0o700); + const sessionId = "27d3e3f1-f1e5-49ae-91ec-95af1f343a5a"; + const token = "synthetic-cutter-bridge-token-for-test"; + const controller = new AbortController(); + let commands = 0; + const peer = authenticatedServer( + token, + sessionId, + (request, respond) => { + if (request.kind === "status") { + controller.abort(reason); + return; + } + commands++; + respond({ + ok: true, + session_id: sessionId, + completed_commands: [], + completed_connection: "0000000000000000", + }); + }, + true, + ); + 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("Missing listener"); + await writeDescriptor(directory, sessionId, address.port, token); + const client = new CutterBridgeClient({ REA_CUTTER_BRIDGE_DIR: directory }); + try { + const failure = await client + .execute({ + sessionId, + expectedGeneration: 0, + command: "pd 1", + json: false, + signal: controller.signal, + }) + .catch((cause: unknown) => cause); + expect(failure).toMatchObject({ + _tag: "AnalysisCancelledError", + cause: reason, + cleanupIncomplete: false, + }); + if (!(failure instanceof Error)) + throw new Error("Expected typed cancellation"); + expect(failure.cause).toBe(reason); + expect(commands).toBe(0); + } finally { + await client.close(); + await new Promise((resolve) => peer.close(() => resolve())); + } + }, +); diff --git a/tests/boundary/providers/cutterBridgeProducer.test.ts b/tests/boundary/providers/cutterBridgeProducer.test.ts index 50a7e9aa19..a9a06c283a 100644 --- a/tests/boundary/providers/cutterBridgeProducer.test.ts +++ b/tests/boundary/providers/cutterBridgeProducer.test.ts @@ -120,7 +120,16 @@ it("round-trips authenticated status and command bytes across the Python produce command: "iI", json: false, }), - ).rejects.toThrow("List sessions again"); + ).rejects.toMatchObject({ + _tag: "AnalysisInputError", + issues: [ + { + path: ["expected_generation"], + reason: "invalid_value", + message: expect.stringContaining("List sessions again"), + }, + ], + }); } const result = await client.execute({