mirror of
https://github.com/morluto/rea
synced 2026-10-25 14:01:13 -04:00
fix(lifecycle): retain IDA cleanup owners and drain JEB fixture work
Preserve primary copy failures, exact descriptor retry ownership and complete-session cleanup state. Make the authored JEB HTTP fixture join admitted callbacks after transport closure. Validated with controlled real-file faults, cancellation/finalization and retry cases, plus local HTTP and existing CLI/MCP lifecycle regressions. No real IDA or JEB engine was exercised.
This commit is contained in:
morluto
@@ -53,6 +53,14 @@ import {
|
||||
} from "./IdaWorkspace.js";
|
||||
import { parseIdaInput } from "./IdaInput.js";
|
||||
import { createEvidence } from "../domain/evidence.js";
|
||||
import { ArtifactReaderFailure } from "../artifacts/ArtifactReader.js";
|
||||
import {
|
||||
artifactFilePrimaryFacts,
|
||||
artifactFileCancellationCause,
|
||||
bindArtifactFileCleanup,
|
||||
} from "../artifacts/ArtifactFileOperation.js";
|
||||
import { ProviderCleanupError } from "../domain/providerCleanupError.js";
|
||||
import { analysisErrorWithCleanupFailure } from "../domain/analysisErrorCleanup.js";
|
||||
|
||||
const legacyTools = [
|
||||
"get_metadata",
|
||||
@@ -108,7 +116,6 @@ interface UpstreamSession {
|
||||
|
||||
/** A failed cleanup retains its private workspace and reports the remaining resource. */
|
||||
class IdaCleanupError extends ProviderAdapterError {
|
||||
override readonly cleanupIncomplete = true;
|
||||
constructor(
|
||||
path: string,
|
||||
cause: unknown,
|
||||
@@ -116,6 +123,11 @@ class IdaCleanupError extends ProviderAdapterError {
|
||||
) {
|
||||
super("ida", "close", {
|
||||
cause,
|
||||
cleanup: {
|
||||
reason:
|
||||
"IDA connection or private workspace cleanup remains unverified.",
|
||||
resources: [path],
|
||||
},
|
||||
capturedOutput:
|
||||
cause instanceof AnalysisError ? cause.capturedOutput : undefined,
|
||||
diagnostics: {
|
||||
@@ -149,7 +161,6 @@ class IdaCleanupError extends ProviderAdapterError {
|
||||
}),
|
||||
},
|
||||
});
|
||||
this.setCleanupResources([path]);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -194,6 +205,7 @@ export class IdaSessionClient implements AnalysisClient {
|
||||
#closed = false;
|
||||
#closing = false;
|
||||
#closePromise: Promise<Result<null, AnalysisError>> | undefined;
|
||||
#closeFailures: WeakRef<IdaCleanupError>[] = [];
|
||||
#workspace: IdaWorkspace | undefined;
|
||||
#database: string | undefined;
|
||||
#databaseCloseCompleted = false;
|
||||
@@ -449,6 +461,10 @@ export class IdaSessionClient implements AnalysisClient {
|
||||
options: ExecutionOptions | undefined,
|
||||
context: IdaOperationFailureContext,
|
||||
) {
|
||||
const artifactFailure =
|
||||
cause instanceof ArtifactReaderFailure ? cause : undefined;
|
||||
if (artifactFailure !== undefined)
|
||||
cause = artifactFilePrimaryFacts(artifactFailure).cause;
|
||||
let failure: AnalysisError;
|
||||
if (cause instanceof IdaIdentityRefusal) {
|
||||
failure = new AnalysisOutputError(operation, cause.reason, {
|
||||
@@ -457,6 +473,10 @@ export class IdaSessionClient implements AnalysisClient {
|
||||
remediationAction: cause.recovery,
|
||||
diagnostics: cause.diagnostics,
|
||||
});
|
||||
} else if (artifactFailure?.reason === "cancelled") {
|
||||
failure = new AnalysisCancelledError(operation, {
|
||||
cause: artifactFileCancellationCause(artifactFailure),
|
||||
});
|
||||
} else if (cause instanceof AnalysisCancelledError) {
|
||||
failure = new AnalysisCancelledError(operation, {
|
||||
cause:
|
||||
@@ -523,6 +543,20 @@ export class IdaSessionClient implements AnalysisClient {
|
||||
"Check the configured upstream MCP registration and selected target; see docs/ida-provider.md.",
|
||||
},
|
||||
});
|
||||
if (artifactFailure?.cleanup !== undefined) {
|
||||
const cleanup = new ProviderCleanupError(
|
||||
"ida",
|
||||
artifactFailure.cleanup.resources,
|
||||
{ reason: artifactFailure.cleanup.reason },
|
||||
{ operation, cause: artifactFailure },
|
||||
);
|
||||
bindArtifactFileCleanup(artifactFailure, cleanup);
|
||||
failure = analysisErrorWithCleanupFailure(
|
||||
failure,
|
||||
cleanup,
|
||||
operation,
|
||||
).bindCleanupDependencies([cleanup]);
|
||||
}
|
||||
// Only operation calls made after target identity/readiness validation can
|
||||
// become partial Evidence. The failing call itself is absent from `raw`
|
||||
// when its transport or MCP envelope could not be decoded.
|
||||
@@ -1005,16 +1039,31 @@ export class IdaSessionClient implements AnalysisClient {
|
||||
await this.connection.close();
|
||||
await this.#workspace?.remove();
|
||||
this.#closed = true;
|
||||
for (const failure of this.#closeFailures)
|
||||
failure.deref()?.resolveCleanup();
|
||||
this.#closeFailures = [];
|
||||
return ok(null);
|
||||
} catch (cause: unknown) {
|
||||
this.connection.detach();
|
||||
return err(
|
||||
new IdaCleanupError(
|
||||
this.#workspace?.directory ?? "IDA MCP connection",
|
||||
cause,
|
||||
this.#unreleasedSessions(),
|
||||
),
|
||||
const path = this.#workspace?.directory ?? "IDA MCP connection";
|
||||
const failure = new IdaCleanupError(
|
||||
path,
|
||||
cause,
|
||||
this.#unreleasedSessions(),
|
||||
);
|
||||
const client = this;
|
||||
failure.bindCleanupResourceOwner({
|
||||
get cleanupResources() {
|
||||
return client.#closed ? [] : [path];
|
||||
},
|
||||
});
|
||||
this.#closeFailures = [
|
||||
...this.#closeFailures.filter(
|
||||
(reference) => reference.deref() !== undefined,
|
||||
),
|
||||
new WeakRef(failure),
|
||||
];
|
||||
return err(failure);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+67
-39
@@ -1,6 +1,6 @@
|
||||
import { createHash, randomUUID } from "node:crypto";
|
||||
import { createReadStream } from "node:fs";
|
||||
import { open, realpath, stat } from "node:fs/promises";
|
||||
import { open, realpath, stat, type FileHandle } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import { basename, join, normalize } from "node:path";
|
||||
import type { BinaryTarget } from "../domain/binaryTargetTypes.js";
|
||||
@@ -8,6 +8,11 @@ import { AnalysisProtocolError } from "../domain/analysisErrorCore.js";
|
||||
import { ConfigurationError } from "../domain/configurationErrors.js";
|
||||
import { AnalysisError } from "../domain/analysisErrorBase.js";
|
||||
import { PrivateRuntimeRoot } from "../process/PrivateRuntimeRoot.js";
|
||||
import { ArtifactResourceScope } from "../artifacts/ArtifactResourceScope.js";
|
||||
import {
|
||||
assertArtifactReadActive,
|
||||
withArtifactFileHandle,
|
||||
} from "../artifacts/ArtifactFileOperation.js";
|
||||
|
||||
/** Read a file digest with bounded memory rather than retaining binary bytes. */
|
||||
export const idaFileDigest = async (path: string): Promise<string> => {
|
||||
@@ -36,6 +41,7 @@ export const resolveIdaWorkspaceParent = async (
|
||||
export class IdaWorkspace {
|
||||
readonly sessionId = `rea-${randomUUID()}`;
|
||||
#inputSha256: string | undefined;
|
||||
readonly #files = new ArtifactResourceScope();
|
||||
private constructor(
|
||||
private readonly runtime: PrivateRuntimeRoot,
|
||||
readonly inputPath: string,
|
||||
@@ -91,54 +97,76 @@ export class IdaWorkspace {
|
||||
* the upstream engine analyzes the architecture admission reported; the
|
||||
* whole-source digest still anchors the copy against admission-time changes.
|
||||
*/
|
||||
async copyInput(): Promise<void> {
|
||||
copyInput(signal?: AbortSignal): Promise<void> {
|
||||
return this.#files.run(async () => {
|
||||
assertArtifactReadActive(signal);
|
||||
const source = await open(this.target.path, "r");
|
||||
await withArtifactFileHandle(
|
||||
this.target.path,
|
||||
source,
|
||||
this.#files,
|
||||
signal,
|
||||
async (source) => {
|
||||
assertArtifactReadActive(signal);
|
||||
const input = await open(this.inputPath, "wx");
|
||||
await withArtifactFileHandle(
|
||||
this.inputPath,
|
||||
input,
|
||||
this.#files,
|
||||
signal,
|
||||
(input) => this.#copyBytes(source, input, signal),
|
||||
);
|
||||
},
|
||||
);
|
||||
});
|
||||
}
|
||||
|
||||
async #copyBytes(
|
||||
source: FileHandle,
|
||||
input: FileHandle,
|
||||
signal: AbortSignal | undefined,
|
||||
): Promise<void> {
|
||||
const slice =
|
||||
this.target.format === "mach-o-universal"
|
||||
? this.target.universalSlice
|
||||
: undefined;
|
||||
const source = await open(this.target.path, "r");
|
||||
try {
|
||||
const input = await open(this.inputPath, "wx");
|
||||
try {
|
||||
const sourceHash = createHash("sha256");
|
||||
const inputHash = createHash("sha256");
|
||||
const chunk = Buffer.allocUnsafe(64 * 1024);
|
||||
let position = 0;
|
||||
for (;;) {
|
||||
const observed = await source.read(chunk, 0, chunk.length, position);
|
||||
if (observed.bytesRead === 0) break;
|
||||
const bytes = chunk.subarray(0, observed.bytesRead);
|
||||
sourceHash.update(bytes);
|
||||
const start =
|
||||
slice === undefined ? position : Math.max(position, slice.offset);
|
||||
const end =
|
||||
slice === undefined
|
||||
? position + bytes.length
|
||||
: Math.min(position + bytes.length, slice.offset + slice.size);
|
||||
// A chunk after the slice has end < position, and a negative subarray
|
||||
// bound would count from the chunk's end instead of being empty.
|
||||
if (end > start) {
|
||||
const admitted = bytes.subarray(start - position, end - position);
|
||||
inputHash.update(admitted);
|
||||
await input.write(admitted);
|
||||
}
|
||||
position += observed.bytesRead;
|
||||
}
|
||||
if (sourceHash.digest("hex") !== this.target.sha256)
|
||||
throw new AnalysisProtocolError(
|
||||
"IDA target changed between admission and the private input copy; reopen the target.",
|
||||
);
|
||||
this.#inputSha256 = inputHash.digest("hex");
|
||||
} finally {
|
||||
await input.close();
|
||||
const sourceHash = createHash("sha256");
|
||||
const inputHash = createHash("sha256");
|
||||
const chunk = Buffer.allocUnsafe(64 * 1024);
|
||||
let position = 0;
|
||||
for (;;) {
|
||||
assertArtifactReadActive(signal);
|
||||
const observed = await source.read(chunk, 0, chunk.length, position);
|
||||
if (observed.bytesRead === 0) break;
|
||||
assertArtifactReadActive(signal);
|
||||
const bytes = chunk.subarray(0, observed.bytesRead);
|
||||
sourceHash.update(bytes);
|
||||
const start =
|
||||
slice === undefined ? position : Math.max(position, slice.offset);
|
||||
const end =
|
||||
slice === undefined
|
||||
? position + bytes.length
|
||||
: Math.min(position + bytes.length, slice.offset + slice.size);
|
||||
// A chunk after the slice has end < position, and a negative subarray
|
||||
// bound would count from the chunk's end instead of being empty.
|
||||
if (end > start) {
|
||||
const admitted = bytes.subarray(start - position, end - position);
|
||||
inputHash.update(admitted);
|
||||
await input.write(admitted);
|
||||
assertArtifactReadActive(signal);
|
||||
}
|
||||
} finally {
|
||||
await source.close();
|
||||
position += observed.bytesRead;
|
||||
}
|
||||
if (sourceHash.digest("hex") !== this.target.sha256)
|
||||
throw new AnalysisProtocolError(
|
||||
"IDA target changed between admission and the private input copy; reopen the target.",
|
||||
);
|
||||
this.#inputSha256 = inputHash.digest("hex");
|
||||
}
|
||||
|
||||
/** Delete only this object's privately allocated directory after worker cleanup. */
|
||||
async remove(): Promise<void> {
|
||||
await this.#files.close();
|
||||
await this.runtime.close();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,183 @@
|
||||
import { channel } from "node:diagnostics_channel";
|
||||
import { readFile, writeFile } from "node:fs/promises";
|
||||
import { IncomingMessage, request, Server, ServerResponse } from "node:http";
|
||||
import { join } from "node:path";
|
||||
import { setImmediate } from "node:timers/promises";
|
||||
import { expect, it, vi } from "vitest";
|
||||
import { z } from "zod";
|
||||
|
||||
import { startJebMcpFixture } from "../../fixtures/jebMcp.js";
|
||||
import { createDeferred } from "../../fixtures/binarySession.js";
|
||||
import { createTestTempDirectory } from "../../fixtures/temporaryDirectory.js";
|
||||
|
||||
const physicalRequestSchema = z.object({
|
||||
request: z.instanceof(IncomingMessage),
|
||||
response: z.instanceof(ServerResponse),
|
||||
server: z.instanceof(Server),
|
||||
});
|
||||
const observeRequest = (port: number) => {
|
||||
const observed = createDeferred<z.infer<typeof physicalRequestSchema>>();
|
||||
const diagnostic = channel("http.server.request.start");
|
||||
const observer = (message: unknown) => {
|
||||
const parsed = physicalRequestSchema.safeParse(message);
|
||||
if (parsed.success && parsed.data.request.socket.localPort === port)
|
||||
observed.resolve(parsed.data);
|
||||
};
|
||||
diagnostic.subscribe(observer);
|
||||
return {
|
||||
promise: observed.promise,
|
||||
close: () => diagnostic.unsubscribe(observer),
|
||||
};
|
||||
};
|
||||
const closed = (emitter: NodeJS.EventEmitter): Promise<void> =>
|
||||
new Promise((resolve) => emitter.once("close", () => resolve()));
|
||||
const bounded = async <Value>(operation: Promise<Value>): Promise<Value> => {
|
||||
let timer: ReturnType<typeof setTimeout> | undefined;
|
||||
try {
|
||||
return await Promise.race([
|
||||
operation,
|
||||
new Promise<never>((_resolve, reject) => {
|
||||
timer = setTimeout(
|
||||
() =>
|
||||
reject(new Error("JEB fixture lifecycle barrier was not reached")),
|
||||
2_000,
|
||||
);
|
||||
}),
|
||||
]);
|
||||
} finally {
|
||||
clearTimeout(timer);
|
||||
}
|
||||
};
|
||||
|
||||
for (const scenario of ["tool", "catalog", "late-render-failure"] as const) {
|
||||
it(`JEB fixture close joins admitted ${scenario} work after transport disconnect`, async () => {
|
||||
const root = await createTestTempDirectory("rea-jeb-fixture-close-");
|
||||
const artifact = join(root, "completed callback.txt");
|
||||
const entered = createDeferred<void>();
|
||||
const release = createDeferred<void>();
|
||||
const finished = createDeferred<void>();
|
||||
let admitted = false;
|
||||
let callbackFinished = false;
|
||||
const complete = async () => {
|
||||
admitted = true;
|
||||
entered.resolve();
|
||||
try {
|
||||
await release.promise;
|
||||
await writeFile(artifact, "admitted JEB callback completed");
|
||||
return { success: true };
|
||||
} finally {
|
||||
callbackFinished = true;
|
||||
finished.resolve();
|
||||
}
|
||||
};
|
||||
const fixture = await startJebMcpFixture(
|
||||
complete,
|
||||
scenario === "catalog" ? complete : undefined,
|
||||
);
|
||||
const observed = observeRequest(Number(new URL(fixture.endpoint).port));
|
||||
const client = request(fixture.endpoint, { method: "POST", agent: false });
|
||||
const clientClosed = createDeferred<void>();
|
||||
client.on("error", () => {});
|
||||
client.once("socket", (socket) =>
|
||||
socket.once("close", () => clientClosed.resolve()),
|
||||
);
|
||||
const renderFailure = new Error("authored response rendering failed");
|
||||
let restoreRender: (() => void) | undefined;
|
||||
const settle = (operation: Promise<void>) =>
|
||||
operation.then(
|
||||
() => ({ ok: true as const }),
|
||||
(cause: unknown) => ({ ok: false as const, cause }),
|
||||
);
|
||||
let closing: ReturnType<typeof settle> | undefined;
|
||||
try {
|
||||
client.end(
|
||||
JSON.stringify({
|
||||
jsonrpc: "2.0",
|
||||
id: 1,
|
||||
method: scenario === "catalog" ? "tools/list" : "tools/call",
|
||||
params: { name: "get_client_information" },
|
||||
}),
|
||||
);
|
||||
await bounded(entered.promise);
|
||||
const physical = await bounded(observed.promise);
|
||||
if (scenario === "late-render-failure") {
|
||||
const fault = vi
|
||||
.spyOn(physical.response, "writeHead")
|
||||
.mockImplementation(() => {
|
||||
throw renderFailure;
|
||||
});
|
||||
restoreRender = () => fault.mockRestore();
|
||||
}
|
||||
const responseClosed = closed(physical.response);
|
||||
const socketClosed = closed(physical.request.socket);
|
||||
client.destroy();
|
||||
await bounded(
|
||||
Promise.all([clientClosed.promise, responseClosed, socketClosed]),
|
||||
);
|
||||
const transportClosed = closed(physical.server).then(async () => {
|
||||
await setImmediate();
|
||||
return { kind: "transport-closed" };
|
||||
});
|
||||
closing = settle(fixture.close());
|
||||
const first = await bounded(Promise.race([closing, transportClosed]));
|
||||
expect(first).toEqual({ kind: "transport-closed" });
|
||||
expect(physical.server.listening).toBe(false);
|
||||
expect(callbackFinished).toBe(false);
|
||||
release.resolve();
|
||||
const outcome = await bounded(closing);
|
||||
if (scenario === "late-render-failure") {
|
||||
if (outcome.ok)
|
||||
throw new Error("Escaping response failure was discarded");
|
||||
if (!(outcome.cause instanceof AggregateError)) throw outcome.cause;
|
||||
expect(outcome.cause.errors).toHaveLength(1);
|
||||
expect(outcome.cause.errors[0]).toBe(renderFailure);
|
||||
} else expect(outcome).toEqual({ ok: true });
|
||||
expect(callbackFinished).toBe(true);
|
||||
expect(await readFile(artifact, "utf8")).toBe(
|
||||
"admitted JEB callback completed",
|
||||
);
|
||||
} finally {
|
||||
release.resolve();
|
||||
client.destroy();
|
||||
observed.close();
|
||||
try {
|
||||
if (admitted) await bounded(finished.promise);
|
||||
await bounded(closing ?? settle(fixture.close()));
|
||||
} finally {
|
||||
restoreRender?.();
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
it("JEB fixture preserves intentionally handled 500 replies and ordinary callback completion", async () => {
|
||||
const refusal = new Error("intentional protocol refusal");
|
||||
const fixture = await startJebMcpFixture((tool) => {
|
||||
if (tool === "refuse") throw refusal;
|
||||
return { success: true, selected: tool };
|
||||
});
|
||||
try {
|
||||
const send = (name: string) =>
|
||||
fetch(fixture.endpoint, {
|
||||
method: "POST",
|
||||
body: JSON.stringify({
|
||||
jsonrpc: "2.0",
|
||||
id: 1,
|
||||
method: "tools/call",
|
||||
params: { name },
|
||||
}),
|
||||
signal: AbortSignal.timeout(2_000),
|
||||
});
|
||||
const ordinary = await send("ordinary");
|
||||
expect(ordinary.status).toBe(200);
|
||||
expect(await ordinary.json()).toMatchObject({
|
||||
result: { structuredContent: { success: true, selected: "ordinary" } },
|
||||
});
|
||||
const refused = await send("refuse");
|
||||
expect(refused.status).toBe(500);
|
||||
expect(await refused.text()).toBe(String(refusal));
|
||||
expect(fixture.calls).toEqual(["ordinary", "refuse"]);
|
||||
} finally {
|
||||
await fixture.close();
|
||||
}
|
||||
});
|
||||
Vendored
+52
-55
@@ -1,4 +1,4 @@
|
||||
import { createServer } from "node:http";
|
||||
import { createVerifierHttpServer } from "../../scripts/lib/verifier-http-server.mjs";
|
||||
import { z } from "zod";
|
||||
import type { JsonValue } from "../../src/domain/jsonValue.js";
|
||||
import { JEB_ENGINE_TOOLS } from "../../src/jeb/JebMcpConnection.js";
|
||||
@@ -9,59 +9,63 @@ const messageSchema = z.object({
|
||||
params: z.object({ name: z.string().optional() }).passthrough().optional(),
|
||||
});
|
||||
|
||||
/** Real HTTP MCP transport with caller-selected JEB response envelopes. */
|
||||
/** Real HTTP MCP transport; close drains admitted catalog/tool callbacks, preserving handled 500 replies. */
|
||||
export const startJebMcpFixture = async (
|
||||
respond: (tool: string) => JsonValue | Promise<JsonValue>,
|
||||
catalogReply?: JsonValue | (() => JsonValue | Promise<JsonValue>),
|
||||
) => {
|
||||
const calls: string[] = [];
|
||||
const server = createServer(async (request, response) => {
|
||||
if (request.method !== "POST") {
|
||||
request.resume();
|
||||
response.writeHead(request.method === "DELETE" ? 200 : 405).end();
|
||||
return;
|
||||
}
|
||||
try {
|
||||
let body = "";
|
||||
for await (const chunk of request) body += String(chunk);
|
||||
const message = messageSchema.parse(JSON.parse(body));
|
||||
if (message.id === undefined) {
|
||||
response.writeHead(202).end();
|
||||
const { server, close } = createVerifierHttpServer(
|
||||
async (request, response) => {
|
||||
if (request.method !== "POST") {
|
||||
request.resume();
|
||||
response.writeHead(request.method === "DELETE" ? 200 : 405).end();
|
||||
return;
|
||||
}
|
||||
let result: unknown;
|
||||
if (message.method === "initialize")
|
||||
result = {
|
||||
protocolVersion: "2025-11-25",
|
||||
capabilities: { tools: {} },
|
||||
serverInfo: { name: "jeb-response-fixture", version: "1" },
|
||||
};
|
||||
else if (message.method === "tools/list")
|
||||
result = (typeof catalogReply === "function"
|
||||
? await catalogReply()
|
||||
: catalogReply) ?? {
|
||||
tools: Object.values(JEB_ENGINE_TOOLS).map((name) => ({
|
||||
name,
|
||||
inputSchema: { type: "object", properties: {} },
|
||||
})),
|
||||
};
|
||||
else if (
|
||||
message.method === "tools/call" &&
|
||||
message.params?.name !== undefined
|
||||
) {
|
||||
calls.push(message.params.name);
|
||||
const payload = await respond(message.params.name);
|
||||
result = {
|
||||
structuredContent: payload,
|
||||
content: [{ type: "text", text: JSON.stringify(payload) }],
|
||||
};
|
||||
} else throw new Error(`Unexpected fixture request: ${message.method}`);
|
||||
response.writeHead(200, { "Content-Type": "application/json" });
|
||||
response.end(JSON.stringify({ jsonrpc: "2.0", id: message.id, result }));
|
||||
} catch (cause: unknown) {
|
||||
response.writeHead(500).end(String(cause));
|
||||
}
|
||||
});
|
||||
try {
|
||||
let body = "";
|
||||
for await (const chunk of request) body += String(chunk);
|
||||
const message = messageSchema.parse(JSON.parse(body));
|
||||
if (message.id === undefined) {
|
||||
response.writeHead(202).end();
|
||||
return;
|
||||
}
|
||||
let result: unknown;
|
||||
if (message.method === "initialize")
|
||||
result = {
|
||||
protocolVersion: "2025-11-25",
|
||||
capabilities: { tools: {} },
|
||||
serverInfo: { name: "jeb-response-fixture", version: "1" },
|
||||
};
|
||||
else if (message.method === "tools/list")
|
||||
result = (typeof catalogReply === "function"
|
||||
? await catalogReply()
|
||||
: catalogReply) ?? {
|
||||
tools: Object.values(JEB_ENGINE_TOOLS).map((name) => ({
|
||||
name,
|
||||
inputSchema: { type: "object", properties: {} },
|
||||
})),
|
||||
};
|
||||
else if (
|
||||
message.method === "tools/call" &&
|
||||
message.params?.name !== undefined
|
||||
) {
|
||||
calls.push(message.params.name);
|
||||
const payload = await respond(message.params.name);
|
||||
result = {
|
||||
structuredContent: payload,
|
||||
content: [{ type: "text", text: JSON.stringify(payload) }],
|
||||
};
|
||||
} else throw new Error(`Unexpected fixture request: ${message.method}`);
|
||||
response.writeHead(200, { "Content-Type": "application/json" });
|
||||
response.end(
|
||||
JSON.stringify({ jsonrpc: "2.0", id: message.id, result }),
|
||||
);
|
||||
} catch (cause: unknown) {
|
||||
response.writeHead(500).end(String(cause));
|
||||
}
|
||||
},
|
||||
);
|
||||
await new Promise<void>((resolve) => server.listen(0, "127.0.0.1", resolve));
|
||||
const address = server.address();
|
||||
if (address === null || typeof address === "string")
|
||||
@@ -69,14 +73,7 @@ export const startJebMcpFixture = async (
|
||||
return {
|
||||
endpoint: `http://127.0.0.1:${address.port}/mcp`,
|
||||
calls,
|
||||
close: async () => {
|
||||
server.closeAllConnections();
|
||||
await new Promise<void>((resolve, reject) =>
|
||||
server.close((cause) =>
|
||||
cause === undefined ? resolve() : reject(cause),
|
||||
),
|
||||
);
|
||||
},
|
||||
close,
|
||||
};
|
||||
};
|
||||
|
||||
|
||||
@@ -0,0 +1,445 @@
|
||||
import { createHash } from "node:crypto";
|
||||
import { access, readFile, writeFile, type FileHandle } from "node:fs/promises";
|
||||
import { join } from "node:path";
|
||||
import { afterEach, expect, it, vi } from "vitest";
|
||||
|
||||
import { ArtifactFileCleanupFailure } from "../../src/artifacts/ArtifactFileOperation.js";
|
||||
import { ArtifactReaderFailure } from "../../src/artifacts/ArtifactReader.js";
|
||||
import { parseBinaryTarget } from "../../src/application/BinaryTargetResolver.js";
|
||||
import { elf } from "../../src/domain/binaryTarget.fixture.js";
|
||||
import { AnalysisProtocolError } from "../../src/domain/analysisErrorCore.js";
|
||||
import { IdaWorkspace } from "../../src/ida/IdaWorkspace.js";
|
||||
import { IdaSessionClient } from "../../src/ida/IdaSessionClient.js";
|
||||
import { PrivateRuntimeRoot } from "../../src/process/PrivateRuntimeRoot.js";
|
||||
import { RecordingIdaMcp } from "../fixtures/idaMcp.js";
|
||||
import { createDeferred } from "../fixtures/binarySession.js";
|
||||
import { awaitAcknowledgement } from "../fixtures/lifecycleAcknowledgement.js";
|
||||
import { createTestTempDirectory } from "../fixtures/temporaryDirectory.js";
|
||||
|
||||
interface ObservedHandle {
|
||||
readonly path: string;
|
||||
readonly handle: FileHandle;
|
||||
readonly close: () => Promise<void>;
|
||||
closeAttempts: number;
|
||||
}
|
||||
interface CopyReadPort {
|
||||
read(
|
||||
buffer: Buffer,
|
||||
offset: number,
|
||||
length: number,
|
||||
position: number,
|
||||
): Promise<{ bytesRead: number; buffer: Buffer }>;
|
||||
}
|
||||
const control = vi.hoisted(() => ({
|
||||
paths: new Set<string>(),
|
||||
handles: new Map<FileHandle, ObservedHandle>(),
|
||||
denied: new Map<string, unknown>(),
|
||||
install: undefined as ((record: ObservedHandle) => void) | undefined,
|
||||
}));
|
||||
vi.mock("node:fs/promises", async (importOriginal) => {
|
||||
const original = await importOriginal<typeof import("node:fs/promises")>();
|
||||
return {
|
||||
...original,
|
||||
open: async (...args: Parameters<typeof original.open>) => {
|
||||
const handle = await original.open(...args);
|
||||
const path = String(args[0]);
|
||||
if (control.paths.has(path)) {
|
||||
const record = {
|
||||
handle,
|
||||
path,
|
||||
close: handle.close.bind(handle),
|
||||
closeAttempts: 0,
|
||||
};
|
||||
control.handles.set(handle, record);
|
||||
vi.spyOn(handle, "close").mockImplementation(async () => {
|
||||
record.closeAttempts++;
|
||||
if (control.denied.has(path)) throw control.denied.get(path);
|
||||
await record.close();
|
||||
});
|
||||
control.install?.(record);
|
||||
}
|
||||
return handle;
|
||||
},
|
||||
};
|
||||
});
|
||||
const workspaces = new Set<IdaWorkspace>();
|
||||
const clients = new Set<IdaSessionClient>();
|
||||
afterEach(async () => {
|
||||
control.install = undefined;
|
||||
control.denied.clear();
|
||||
vi.restoreAllMocks();
|
||||
for (const client of clients) {
|
||||
const closed = await client.close();
|
||||
if (!closed.ok) throw closed.error;
|
||||
}
|
||||
for (const workspace of workspaces) await workspace.remove();
|
||||
for (const record of control.handles.values())
|
||||
if (record.handle.fd >= 0) await record.close();
|
||||
control.handles.clear();
|
||||
control.paths.clear();
|
||||
workspaces.clear();
|
||||
clients.clear();
|
||||
});
|
||||
|
||||
const fixture = async () => {
|
||||
const root = await createTestTempDirectory("rea-ida-copy-owner-");
|
||||
const path = join(root, "authored.elf");
|
||||
const bytes = elf(2, 1, 62);
|
||||
await writeFile(path, bytes);
|
||||
const resolved = await parseBinaryTarget(path);
|
||||
if (!resolved.ok) throw resolved.error;
|
||||
const workspace = await IdaWorkspace.allocate(resolved.value, root);
|
||||
workspaces.add(workspace);
|
||||
control.paths.add(resolved.value.path);
|
||||
control.paths.add(workspace.inputPath);
|
||||
return { root, workspace, target: resolved.value, bytes };
|
||||
};
|
||||
const failed = (operation: Promise<unknown>): Promise<unknown> =>
|
||||
operation.then(
|
||||
() => {
|
||||
throw new Error("Expected the controlled operation to fail");
|
||||
},
|
||||
(cause: unknown) => cause,
|
||||
);
|
||||
const observed = (path: string): ObservedHandle => {
|
||||
const record = [...control.handles.values()].find(
|
||||
(record) => record.path === path,
|
||||
);
|
||||
if (record === undefined)
|
||||
throw new Error(`Selected descriptor was not observed: ${path}`);
|
||||
return record;
|
||||
};
|
||||
|
||||
it("retains a source read failure and the exact refusing descriptor until workspace retry", async () => {
|
||||
const { workspace, target } = await fixture();
|
||||
const readFailure = new Error("selected source read EIO");
|
||||
const closeFailure = new Error("selected source close refused");
|
||||
control.denied.set(target.path, closeFailure);
|
||||
control.install = ({ handle, path }) => {
|
||||
if (path === target.path)
|
||||
vi.spyOn(handle, "read").mockRejectedValue(readFailure);
|
||||
};
|
||||
const failure = await failed(workspace.copyInput());
|
||||
if (!(failure instanceof ArtifactFileCleanupFailure)) throw failure;
|
||||
const outcome = failure.fileCleanupFailure.outcome;
|
||||
if (outcome.kind !== "failed")
|
||||
throw new Error("Source read failure became completed copy");
|
||||
expect(outcome.cause).toBe(readFailure);
|
||||
expect(failure.fileCleanupFailure.cleanupCause).toBe(closeFailure);
|
||||
const source = observed(target.path);
|
||||
expect(failure.fileCleanupFailure.owner.handle).toBe(source.handle);
|
||||
expect(source.handle.fd).toBeGreaterThanOrEqual(0);
|
||||
await expect(workspace.remove()).rejects.toBeInstanceOf(
|
||||
ArtifactReaderFailure,
|
||||
);
|
||||
await access(workspace.directory);
|
||||
control.denied.delete(target.path);
|
||||
await workspace.remove();
|
||||
expect(source.handle.fd).toBe(-1);
|
||||
await expect(access(workspace.directory)).rejects.toMatchObject({
|
||||
code: "ENOENT",
|
||||
});
|
||||
});
|
||||
|
||||
it("keeps a private write failure primary and never retries a confirmed source close", async () => {
|
||||
const { workspace, target } = await fixture();
|
||||
const writeFailure = new Error("private write failed");
|
||||
control.denied.set(
|
||||
workspace.inputPath,
|
||||
new Error("private destination close refused"),
|
||||
);
|
||||
control.install = ({ handle, path }) => {
|
||||
if (path === workspace.inputPath)
|
||||
vi.spyOn(handle, "write").mockRejectedValue(writeFailure);
|
||||
};
|
||||
const failure = await failed(workspace.copyInput());
|
||||
if (!(failure instanceof ArtifactFileCleanupFailure)) throw failure;
|
||||
const outcome = failure.fileCleanupFailure.outcome;
|
||||
if (outcome.kind !== "failed")
|
||||
throw new Error("Private write failure became completed copy");
|
||||
expect(outcome.cause).toBe(writeFailure);
|
||||
const source = observed(target.path);
|
||||
const input = observed(workspace.inputPath);
|
||||
expect(source.handle.fd).toBe(-1);
|
||||
expect(input.handle.fd).toBeGreaterThanOrEqual(0);
|
||||
const confirmedSourceAttempts = source.closeAttempts;
|
||||
control.denied.delete(workspace.inputPath);
|
||||
await workspace.remove();
|
||||
expect(source.closeAttempts).toBe(confirmedSourceAttempts);
|
||||
expect(input.handle.fd).toBe(-1);
|
||||
});
|
||||
|
||||
it("retains completed digest facts and both close owners, retrying only unresolved descriptors", async () => {
|
||||
const { workspace, target, bytes } = await fixture();
|
||||
control.denied.set(target.path, new Error("source close refused"));
|
||||
control.denied.set(
|
||||
workspace.inputPath,
|
||||
new Error("destination close refused"),
|
||||
);
|
||||
const failure = await failed(workspace.copyInput());
|
||||
if (!(failure instanceof ArtifactFileCleanupFailure)) throw failure;
|
||||
expect(workspace.inputSha256).toBe(
|
||||
createHash("sha256").update(bytes).digest("hex"),
|
||||
);
|
||||
expect(failure.cleanup?.resources.slice().sort()).toEqual(
|
||||
[target.path, workspace.inputPath].sort(),
|
||||
);
|
||||
const source = observed(target.path);
|
||||
const input = observed(workspace.inputPath);
|
||||
expect(source.handle.fd).toBeGreaterThanOrEqual(0);
|
||||
expect(input.handle.fd).toBeGreaterThanOrEqual(0);
|
||||
control.denied.delete(workspace.inputPath);
|
||||
await expect(workspace.remove()).rejects.toBeInstanceOf(
|
||||
ArtifactReaderFailure,
|
||||
);
|
||||
expect(input.handle.fd).toBe(-1);
|
||||
expect(source.handle.fd).toBeGreaterThanOrEqual(0);
|
||||
const confirmedInputAttempts = input.closeAttempts;
|
||||
control.denied.delete(target.path);
|
||||
await workspace.remove();
|
||||
expect(input.closeAttempts).toBe(confirmedInputAttempts);
|
||||
expect(source.handle.fd).toBe(-1);
|
||||
});
|
||||
|
||||
it("preserves a digest mismatch through two rejected final closes", async () => {
|
||||
const { workspace, target } = await fixture();
|
||||
await writeFile(target.path, Buffer.from("changed after target admission"));
|
||||
control.denied.set(target.path, new Error("source close refused"));
|
||||
control.denied.set(
|
||||
workspace.inputPath,
|
||||
new Error("destination close refused"),
|
||||
);
|
||||
const failure = await failed(workspace.copyInput());
|
||||
if (!(failure instanceof ArtifactFileCleanupFailure)) throw failure;
|
||||
const sourceOutcome = failure.fileCleanupFailure.outcome;
|
||||
if (
|
||||
sourceOutcome.kind !== "failed" ||
|
||||
!(sourceOutcome.cause instanceof ArtifactFileCleanupFailure)
|
||||
)
|
||||
throw new Error(
|
||||
"Both descriptor failures did not retain the nested copy outcome",
|
||||
);
|
||||
const inputOutcome = sourceOutcome.cause.fileCleanupFailure.outcome;
|
||||
if (inputOutcome.kind !== "failed")
|
||||
throw new Error("Digest refusal became completed copy");
|
||||
expect(inputOutcome.cause).toBeInstanceOf(AnalysisProtocolError);
|
||||
expect(inputOutcome.cause).toMatchObject({
|
||||
message: expect.stringContaining("target changed"),
|
||||
});
|
||||
expect(workspace.inputSha256).toBeUndefined();
|
||||
});
|
||||
|
||||
it("honors exact cancellation during copying and closes both descriptors", async () => {
|
||||
const { workspace, target } = await fixture();
|
||||
const controller = new AbortController();
|
||||
const reason = { caller: "copy checkpoint" };
|
||||
control.install = ({ handle, path }) => {
|
||||
if (path !== target.path) return;
|
||||
const reading: CopyReadPort = handle;
|
||||
const read = reading.read.bind(handle);
|
||||
vi.spyOn(reading, "read").mockImplementation(async () => {
|
||||
const result = await read(Buffer.alloc(1), 0, 1, 0);
|
||||
controller.abort(reason);
|
||||
return result;
|
||||
});
|
||||
};
|
||||
const failure = await failed(workspace.copyInput(controller.signal));
|
||||
expect(failure).toBeInstanceOf(ArtifactReaderFailure);
|
||||
expect(failure).toMatchObject({ reason: "cancelled", cause: reason });
|
||||
expect(workspace.inputSha256).toBeUndefined();
|
||||
expect(observed(target.path).handle.fd).toBe(-1);
|
||||
expect(observed(workspace.inputPath).handle.fd).toBe(-1);
|
||||
});
|
||||
|
||||
it("retains verified copied bytes when cancellation arrives during final close", async () => {
|
||||
const { workspace, target, bytes } = await fixture();
|
||||
const controller = new AbortController();
|
||||
const reason = { caller: "final destination close" };
|
||||
control.install = ({ handle, path, close }) => {
|
||||
if (path !== workspace.inputPath) return;
|
||||
vi.spyOn(handle, "close").mockImplementation(async () => {
|
||||
await close();
|
||||
controller.abort(reason);
|
||||
});
|
||||
};
|
||||
const failure = await failed(workspace.copyInput(controller.signal));
|
||||
expect(failure).toMatchObject({ reason: "cancelled", cause: reason });
|
||||
expect(workspace.inputSha256).toBe(target.sha256);
|
||||
expect((await readFile(workspace.inputPath)).equals(bytes)).toBe(true);
|
||||
expect(observed(target.path).handle.fd).toBe(-1);
|
||||
expect(observed(workspace.inputPath).handle.fd).toBe(-1);
|
||||
});
|
||||
|
||||
it("projects retained workspace descriptors through the real IDA session and clears their original cleanup on retry", async () => {
|
||||
const { root, target } = await fixture();
|
||||
const producer = new RecordingIdaMcp(target, "headless");
|
||||
const client = new IdaSessionClient(
|
||||
{
|
||||
command: "fixture",
|
||||
args: [],
|
||||
env: {},
|
||||
mode: "headless",
|
||||
timeoutMs: 1000,
|
||||
workspaceRoot: root,
|
||||
},
|
||||
target,
|
||||
producer,
|
||||
);
|
||||
clients.add(client);
|
||||
const readFailure = new Error("source read primary");
|
||||
control.denied.set(target.path, new Error("source final close refused"));
|
||||
control.install = ({ handle, path }) => {
|
||||
if (path === target.path)
|
||||
vi.spyOn(handle, "read").mockRejectedValue(readFailure);
|
||||
};
|
||||
const result = await client.execute("procedure_pseudo_code", {
|
||||
procedure: "main",
|
||||
});
|
||||
if (result.ok)
|
||||
throw new Error("Failed copy reached a successful IDA operation");
|
||||
expect(result.error.cleanupIncomplete).toBe(true);
|
||||
expect(result.error.cleanupPrimary?.cause).toBe(readFailure);
|
||||
expect(result.error.cleanupResources).toContain(target.path);
|
||||
expect(producer.calls.some(({ name }) => name === "idb_open")).toBe(false);
|
||||
const blocked = await client.close();
|
||||
if (blocked.ok) throw new Error("Client dropped its retained descriptor");
|
||||
expect(blocked.error.cleanupIncomplete).toBe(true);
|
||||
expect(observed(target.path).handle.fd).toBeGreaterThanOrEqual(0);
|
||||
const workspacePath = blocked.error.cleanupResources[0];
|
||||
if (workspacePath === undefined) throw new Error("Client lost its workspace");
|
||||
const removeFailure = new Error("selected private workspace removal refused");
|
||||
const remove = PrivateRuntimeRoot.prototype.close;
|
||||
const closeRoot = vi
|
||||
.spyOn(PrivateRuntimeRoot.prototype, "close")
|
||||
.mockImplementation(function (this: PrivateRuntimeRoot) {
|
||||
if (this.path === workspacePath) return Promise.reject(removeFailure);
|
||||
return remove.call(this);
|
||||
});
|
||||
control.denied.delete(target.path);
|
||||
const retainedWorkspace = await client.close();
|
||||
if (retainedWorkspace.ok)
|
||||
throw new Error("Client dropped its retained workspace");
|
||||
expect(retainedWorkspace.error.cause).toBe(removeFailure);
|
||||
expect(observed(target.path).handle.fd).toBe(-1);
|
||||
expect(result.error.cleanupIncomplete).toBe(false);
|
||||
expect(blocked.error.cleanupIncomplete).toBe(true);
|
||||
expect(retainedWorkspace.error.cleanupIncomplete).toBe(true);
|
||||
expect(blocked.error.cleanupResources).toEqual([workspacePath]);
|
||||
await access(workspacePath);
|
||||
closeRoot.mockRestore();
|
||||
const closed = await client.close();
|
||||
if (!closed.ok) throw closed.error;
|
||||
expect(result.error.cleanupIncomplete).toBe(false);
|
||||
expect(blocked.error.cleanupIncomplete).toBe(false);
|
||||
expect(retainedWorkspace.error.cleanupIncomplete).toBe(false);
|
||||
expect(blocked.error.cleanupResources).toEqual([]);
|
||||
expect(retainedWorkspace.error.cleanupResources).toEqual([]);
|
||||
expect(observed(target.path).handle.fd).toBe(-1);
|
||||
await expect(access(workspacePath)).rejects.toMatchObject({ code: "ENOENT" });
|
||||
});
|
||||
|
||||
it("keeps final-close cancellation, completed digest facts and rejected source cleanup together", async () => {
|
||||
const { workspace, target } = await fixture();
|
||||
const controller = new AbortController();
|
||||
const reason = { caller: "source finalization" };
|
||||
const closeFailure = new Error("source close rejected during cancellation");
|
||||
let restore: (() => void) | undefined;
|
||||
control.install = ({ handle, path }) => {
|
||||
if (path !== target.path) return;
|
||||
const close = vi.spyOn(handle, "close").mockImplementation(async () => {
|
||||
controller.abort(reason);
|
||||
throw closeFailure;
|
||||
});
|
||||
restore = () => close.mockRestore();
|
||||
};
|
||||
const failure = await failed(workspace.copyInput(controller.signal));
|
||||
if (!(failure instanceof ArtifactFileCleanupFailure)) throw failure;
|
||||
expect(failure.fileCleanupFailure.outcome.kind).toBe("completed");
|
||||
expect(failure.fileCleanupFailure.completionCancellation?.reason).toBe(
|
||||
reason,
|
||||
);
|
||||
expect(failure.fileCleanupFailure.cleanupCause).toBe(closeFailure);
|
||||
expect(workspace.inputSha256).toBe(target.sha256);
|
||||
expect(observed(target.path).handle.fd).toBeGreaterThanOrEqual(0);
|
||||
restore?.();
|
||||
await workspace.remove();
|
||||
expect(observed(target.path).handle.fd).toBe(-1);
|
||||
});
|
||||
|
||||
it("keeps shared IDA startup owned when the caller leaves during source finalization", async () => {
|
||||
const { root, target } = await fixture();
|
||||
const producer = new RecordingIdaMcp(target, "headless");
|
||||
const client = new IdaSessionClient(
|
||||
{
|
||||
command: "fixture",
|
||||
args: [],
|
||||
env: {},
|
||||
mode: "headless",
|
||||
timeoutMs: 1000,
|
||||
workspaceRoot: root,
|
||||
},
|
||||
target,
|
||||
producer,
|
||||
);
|
||||
clients.add(client);
|
||||
const controller = new AbortController();
|
||||
const reason = { caller: "shared startup finalization" };
|
||||
control.install = ({ handle, path, close }) => {
|
||||
if (path !== target.path) return;
|
||||
vi.spyOn(handle, "close").mockImplementation(async () => {
|
||||
await close();
|
||||
controller.abort(reason);
|
||||
});
|
||||
};
|
||||
const result = await client.execute(
|
||||
"procedure_pseudo_code",
|
||||
{ procedure: "main" },
|
||||
{ signal: controller.signal },
|
||||
);
|
||||
if (result.ok)
|
||||
throw new Error("Caller cancellation became a successful IDA operation");
|
||||
expect(result.error).toMatchObject({
|
||||
_tag: "AnalysisCancelledError",
|
||||
cause: reason,
|
||||
});
|
||||
const closed = await client.close();
|
||||
if (!closed.ok) throw closed.error;
|
||||
expect(observed(target.path).handle.fd).toBe(-1);
|
||||
});
|
||||
|
||||
it("joins an admitted copy before removing its root and refuses new copy admission", async () => {
|
||||
const { workspace, target } = await fixture();
|
||||
const entered = createDeferred<void>();
|
||||
const release = createDeferred<void>();
|
||||
control.install = ({ handle, path }) => {
|
||||
if (path !== target.path) return;
|
||||
const reading: CopyReadPort = handle;
|
||||
const read = reading.read.bind(handle);
|
||||
vi.spyOn(reading, "read").mockImplementation(async (...args) => {
|
||||
entered.resolve();
|
||||
await release.promise;
|
||||
return read(...args);
|
||||
});
|
||||
};
|
||||
const copying = workspace.copyInput();
|
||||
let removing: Promise<void> | undefined;
|
||||
try {
|
||||
await awaitAcknowledgement(entered.promise, copying, "selected copy read");
|
||||
let removed = false;
|
||||
removing = workspace.remove().then(() => {
|
||||
removed = true;
|
||||
});
|
||||
await expect(workspace.copyInput()).rejects.toMatchObject({
|
||||
message: "Artifact resource scope is closed",
|
||||
});
|
||||
expect(removed).toBe(false);
|
||||
await access(workspace.directory);
|
||||
release.resolve();
|
||||
await copying;
|
||||
await removing;
|
||||
expect(workspace.inputSha256).toBe(target.sha256);
|
||||
} finally {
|
||||
release.resolve();
|
||||
await Promise.allSettled([copying, removing]);
|
||||
}
|
||||
});
|
||||
Reference in New Issue
Block a user