fix(cutter): preserve producer failure and cancellation facts

This commit is contained in:
morluto committed 2026-10-22 22:07:11 +08:00
1 parent edccb430c4
commit e780148305
8 files changed
+599 -236

No files matched your search

+5 -1
View File
@@ -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
+236 -172
View File
@@ -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 <Input, Output>(
inputs: readonly Input[],
operation: (input: Input) => Promise<Output>,
signal?: AbortSignal,
assertActive: () => void,
): Promise<Output[]> => {
signal?.throwIfAborted();
assertActive();
const results = new Array<Output>(inputs.length);
let next = 0;
let stopped = false;
@@ -202,7 +206,7 @@ const mapDiscovery = async <Input, Output>(
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 <Input, Output>(
);
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<CutterBridgeExecution> {
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
+17 -28
View File
@@ -85,23 +85,23 @@ export class CutterBridgeService {
options: CutterDiscoveryOptions = {},
): Promise<Result<CutterBridgeDiscovery, AnalysisError>> {
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 },
),
);
}
}
@@ -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<void>((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<void>((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<void>((resolve) => peer.close(() => resolve()));
}
},
);
@@ -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 () => {
@@ -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);
},
);
@@ -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<void>((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<void>((resolve) => peer.close(() => resolve()));
}
},
);
@@ -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({