fix(cancellation): retain caller reasons at admission checkpoints

This commit is contained in:
morluto committed 2026-10-24 17:21:08 +08:00
1 parent 3fb26985c4
commit 9dfa44f776
8 files changed
+188 -29

No files matched your search

@@ -21,7 +21,9 @@ export const inspectAndroidPackage = async (
): Promise<Result<Evidence, AnalysisError>> => {
const operation = "inspect_android_package";
if (options?.signal?.aborted === true)
return err(new AnalysisCancelledError(operation));
return err(
new AnalysisCancelledError(operation, { cause: options.signal.reason }),
);
const parsed = androidPackageInspectionInputSchema.safeParse(rawInput);
if (!parsed.success)
return err(
+22 -6
View File
@@ -736,10 +736,14 @@ export class BinarySession
),
);
if (isAborted(signal))
return err(new AnalysisCancelledError(operationName));
return err(
new AnalysisCancelledError(operationName, { cause: signal?.reason }),
);
await this.#drainCalls(admittedThrough);
if (isAborted(signal))
return err(new AnalysisCancelledError(operationName));
return err(
new AnalysisCancelledError(operationName, { cause: signal?.reason }),
);
if (this.#externalMutationCleanupFailure?.cleanupIncomplete === true) {
if (cleanupRecovery === undefined)
@@ -1450,14 +1454,18 @@ export class BinarySession
transition: Promise<void>,
): Promise<Result<undefined, AnalysisCancelledError>> {
if (signal?.aborted === true)
return err(new AnalysisCancelledError(operation));
return err(
new AnalysisCancelledError(operation, { cause: signal.reason }),
);
if (signal === undefined) {
await transition;
return ok(undefined);
}
return new Promise((resolve) => {
const onAbort = (): void => {
resolve(err(new AnalysisCancelledError(operation)));
resolve(
err(new AnalysisCancelledError(operation, { cause: signal.reason })),
);
};
signal.addEventListener("abort", onAbort, { once: true });
transition.then(
@@ -1465,7 +1473,11 @@ export class BinarySession
signal.removeEventListener("abort", onAbort);
resolve(
signal.aborted
? err(new AnalysisCancelledError(operation))
? err(
new AnalysisCancelledError(operation, {
cause: signal.reason,
}),
)
: ok(undefined),
);
},
@@ -1473,7 +1485,11 @@ export class BinarySession
signal.removeEventListener("abort", onAbort);
resolve(
signal.aborted
? err(new AnalysisCancelledError(operation))
? err(
new AnalysisCancelledError(operation, {
cause: signal.reason,
}),
)
: ok(undefined),
);
},
@@ -110,6 +110,25 @@ const analysis: AnalysisOperationPort = {
},
};
describe("native value trace seed admission", () => {
it("preserves the exact abort reason before attempting a seed", async () => {
const reason = { request: "cancel before native trace" };
const attempted: unknown[] = [];
const result = await traceNativeValues(
{
execute: async (operation, parameters) => {
attempted.push([operation, parameters]);
return analysis.execute(operation, parameters);
},
},
{ procedure: "0x1000" },
AbortSignal.abort(reason),
);
if (result.ok) throw new Error("Expected pre-seed cancellation");
expect(result.error._tag).toBe("AnalysisCancelledError");
expect(result.error.cause).toBe(reason);
expect(attempted).toEqual([]);
});
it.each([
new AnalysisInputError("analyze_function", {
userMessage: "Unknown selector: missing",
+5 -1
View File
@@ -73,7 +73,11 @@ export const traceNativeValues = async (
};
while (pending.length > 0) {
if (signal?.aborted)
return err(new AnalysisCancelledError("trace_native_values"));
return err(
new AnalysisCancelledError("trace_native_values", {
cause: signal.reason,
}),
);
const next = pending.shift();
if (
next === undefined ||
+5 -1
View File
@@ -125,7 +125,11 @@ const registerImportEvidenceTool = ({
);
if (signal.aborted)
return server.delivery.toCallToolResult(
err(new AnalysisCancelledError("import_evidence_bundle")),
err(
new AnalysisCancelledError("import_evidence_bundle", {
cause: signal.reason,
}),
),
importContract,
);
const imported = session.importEvidenceBundle(loaded.value);
@@ -264,14 +264,16 @@ it("rejects unavailable retention, invalid selectors and cancellation before pro
retention,
),
).toMatchObject({ ok: false, error: { _tag: "AnalysisInputError" } });
expect(
await inspectAndroidPackage(
service,
{ path: "/missing.apk", detail: "summary" },
retention,
{ signal: AbortSignal.abort() },
),
).toMatchObject({ ok: false, error: { _tag: "AnalysisCancelledError" } });
const reason = { request: "cancel before Android admission" };
const cancelled = await inspectAndroidPackage(
service,
{ path: "/missing.apk", detail: "summary" },
retention,
{ signal: AbortSignal.abort(reason) },
);
if (cancelled.ok) throw new Error("Expected Android admission cancellation");
expect(cancelled.error._tag).toBe("AnalysisCancelledError");
expect(cancelled.error.cause).toBe(reason);
expect(execute).not.toHaveBeenCalled();
const session = createTestBinarySession(() => {
throw new Error("Unexpected deep provider");
@@ -1,16 +1,83 @@
import { Client, InMemoryTransport } from "@modelcontextprotocol/client";
import type { ServerContext, ToolCallback } from "@modelcontextprotocol/server";
import { writeFile } from "node:fs/promises";
import { join } from "node:path";
import { expect, it, onTestFinished, vi } from "vitest";
import { AnalysisCancelledError } from "../../../src/domain/analysisErrorCore.js";
import { createEvidence } from "../../../src/domain/evidence.js";
import { createEvidenceBundle } from "../../../src/domain/evidenceBundle.js";
import { toolContract } from "../../../src/contracts/toolContracts.js";
import { createServer } from "../../../src/server/createServer.js";
import { EvidenceMcpServer } from "../../../src/server/EvidenceMcpServer.js";
import { ToolResultDelivery } from "../../../src/server/toolResult.js";
import { OwnedFileHandle } from "../../../src/filesystem/OwnedFileHandle.js";
import { createTestBinarySession } from "../../fixtures/binarySession.js";
import { createTestTempDirectory } from "../../fixtures/temporaryDirectory.js";
it("preserves the exact reason when import is cancelled after reading the bundle", async () => {
const root = await createTestTempDirectory(
"rea-mcp-import-post-read-cancel-",
);
const path = join(root, "bundle.json");
const session = createTestBinarySession(() => {
throw new Error("Evidence import must not start a binary provider");
});
const evidence = createEvidence(
undefined,
{ id: "fixture", name: "Fixture", version: "1" },
{
operation: "imported_observation",
parameters: {},
result: { retained: true },
},
);
await writeFile(path, JSON.stringify(createEvidenceBundle([evidence])));
const registration = vi.spyOn(EvidenceMcpServer.prototype, "registerTool");
const server = createServer({ kind: "session", session });
onTestFinished(async () => {
vi.restoreAllMocks();
await server.close();
await session.close();
});
const contract = toolContract("import_evidence_bundle");
const registered = registration.mock.calls.find(
([name]) => name === contract.name,
);
if (registered === undefined)
throw new Error("Missing bundle import handler");
const handler = registered[2] as unknown as ToolCallback<
typeof contract.inputSchema
>;
const reason = { request: "cancel after authenticated bundle read" };
const controller = new AbortController();
const before = session.exportEvidenceBundle();
const snapshot = session.immutableEvidenceBundle.bind(session);
const snapshotRead = vi
.spyOn(session, "immutableEvidenceBundle")
.mockImplementationOnce(() => {
const bundle = snapshot();
controller.abort(reason);
return bundle;
});
const imported = vi.spyOn(session, "importEvidenceBundle");
const delivery = vi.spyOn(ToolResultDelivery.prototype, "toCallToolResult");
const response = await handler(contract.inputSchema.parse({ path }), {
mcpReq: { signal: controller.signal },
} as ServerContext);
const delivered = delivery.mock.calls.find(
([, deliveredContract]) => deliveredContract.name === contract.name,
)?.[0];
if (delivered === undefined || delivered.ok)
throw new Error("Expected import cancellation at delivery");
expect(response.isError).toBe(true);
expect(delivered.error._tag).toBe("AnalysisCancelledError");
expect(delivered.error.cause).toBe(reason);
expect(snapshotRead).toHaveBeenCalledOnce();
expect(imported).not.toHaveBeenCalled();
expect(session.exportEvidenceBundle()).toEqual(before);
});
it("keeps the complete session unchanged after SDK import cancellation and permits a retry", async () => {
const root = await createTestTempDirectory("rea-mcp-import-cancel-");
const path = join(root, "bundle.json");
@@ -174,6 +174,25 @@ describe("detached provider and target metadata", () => {
});
});
it("preserves the exact abort reason before external mutation admission", async () => {
const session = createTestBinarySession(() => {
throw new Error("Cancelled mutation must not start a provider");
});
const reason = { request: "cancel before external mutation" };
const operation = vi.fn(async () => resultOk(null));
const result = await session.withAdmittedExternalMutation(
undefined,
AbortSignal.abort(reason),
operation,
(_value, error) => error,
);
if (result.ok) throw new Error("Expected external mutation cancellation");
expect(result.error._tag).toBe("AnalysisCancelledError");
expect(result.error.cause).toBe(reason);
expect(operation).not.toHaveBeenCalled();
await session.close();
});
it("drains session calls and retires analysis after an admitted external mutation", async () => {
const [target] = await createBinarySessionTargets();
const alias = join(dirname(target), "first-alias.hop");
@@ -1097,6 +1116,24 @@ describe("provider health and availability", () => {
});
});
it("preserves the exact abort reason before composed admission", async () => {
const session = createTestBinarySession(() => {
throw new Error("Cancelled admission must not start a provider");
});
const reason = { request: "cancel before composed admission" };
const operation = vi.fn(async () => null);
const result = await session.withAdmittedAnalysis(
"binary_overview",
AbortSignal.abort(reason),
operation,
);
if (result.ok) throw new Error("Expected composed admission cancellation");
expect(result.error._tag).toBe("AnalysisCancelledError");
expect(result.error.cause).toBe(reason);
expect(operation).not.toHaveBeenCalled();
await session.close();
});
describe("typed unavailability without dispatching", () => {
it("returns typed unavailability without dispatching a partial provider", async () => {
const [first] = await createBinarySessionTargets();
@@ -1228,18 +1265,26 @@ describe("typed unavailability without dispatching", () => {
return null;
},
);
controller.abort();
await expect(admitted).resolves.toMatchObject({
ok: false,
error: {
_tag: "AnalysisCancelledError",
operation: "binary_overview",
},
});
expect(invoked).toBe(false);
health.resolve(ok(null));
expect((await opening).ok).toBe(true);
await session.close();
const reason = { request: "cancel queued composed admission" };
controller.abort(reason);
try {
const result = await admitted;
expect(result).toMatchObject({
ok: false,
error: {
_tag: "AnalysisCancelledError",
operation: "binary_overview",
},
});
if (result.ok) throw new Error("Expected queued admission cancellation");
expect(result.error.cause).toBe(reason);
expect(invoked).toBe(false);
} finally {
health.resolve(ok(null));
const opened = await opening;
await session.close();
expect(opened.ok).toBe(true);
}
});
});