mirror of
https://github.com/morluto/rea.git
synced 2026-10-26 17:35:11 +00:00
fix(cancellation): retain caller reasons at admission checkpoints
This commit is contained in:
1 parent
c101193eb3
commit
dabb22c09d
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(
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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 ||
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
Reference in new issue
Block a user