From e97f47b0b734e14c17b4d706eaa2f143a99acb6f Mon Sep 17 00:00:00 2001 From: morluto <76467478+morluto@users.noreply.github.com> Date: Sun, 25 Oct 2026 09:45:45 +0800 Subject: [PATCH] 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. --- src/ida/IdaSessionClient.ts | 65 ++- src/ida/IdaWorkspace.ts | 106 +++-- .../providers/jebFixtureCompletion.test.ts | 183 +++++++ tests/fixtures/jebMcp.ts | 107 ++--- .../idaWorkspaceCleanup.test.ts | 445 ++++++++++++++++++ 5 files changed, 804 insertions(+), 102 deletions(-) create mode 100644 tests/boundary/providers/jebFixtureCompletion.test.ts create mode 100644 tests/process-global/idaWorkspaceCleanup.test.ts diff --git a/src/ida/IdaSessionClient.ts b/src/ida/IdaSessionClient.ts index c4a28fc16b..e9745bd8f6 100644 --- a/src/ida/IdaSessionClient.ts +++ b/src/ida/IdaSessionClient.ts @@ -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> | undefined; + #closeFailures: WeakRef[] = []; #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); } } } diff --git a/src/ida/IdaWorkspace.ts b/src/ida/IdaWorkspace.ts index 9dd0106774..0e0c696dae 100644 --- a/src/ida/IdaWorkspace.ts +++ b/src/ida/IdaWorkspace.ts @@ -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 => { @@ -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 { + copyInput(signal?: AbortSignal): Promise { + 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 { 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 { + await this.#files.close(); await this.runtime.close(); } } diff --git a/tests/boundary/providers/jebFixtureCompletion.test.ts b/tests/boundary/providers/jebFixtureCompletion.test.ts new file mode 100644 index 0000000000..5bedb0fa00 --- /dev/null +++ b/tests/boundary/providers/jebFixtureCompletion.test.ts @@ -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>(); + 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 => + new Promise((resolve) => emitter.once("close", () => resolve())); +const bounded = async (operation: Promise): Promise => { + let timer: ReturnType | undefined; + try { + return await Promise.race([ + operation, + new Promise((_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(); + const release = createDeferred(); + const finished = createDeferred(); + 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(); + 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) => + operation.then( + () => ({ ok: true as const }), + (cause: unknown) => ({ ok: false as const, cause }), + ); + let closing: ReturnType | 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(); + } +}); diff --git a/tests/fixtures/jebMcp.ts b/tests/fixtures/jebMcp.ts index d136a8f705..da43ab6c1a 100644 --- a/tests/fixtures/jebMcp.ts +++ b/tests/fixtures/jebMcp.ts @@ -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, catalogReply?: JsonValue | (() => JsonValue | Promise), ) => { 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((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((resolve, reject) => - server.close((cause) => - cause === undefined ? resolve() : reject(cause), - ), - ); - }, + close, }; }; diff --git a/tests/process-global/idaWorkspaceCleanup.test.ts b/tests/process-global/idaWorkspaceCleanup.test.ts new file mode 100644 index 0000000000..ce6f9c5ee4 --- /dev/null +++ b/tests/process-global/idaWorkspaceCleanup.test.ts @@ -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; + 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(), + handles: new Map(), + denied: new Map(), + install: undefined as ((record: ObservedHandle) => void) | undefined, +})); +vi.mock("node:fs/promises", async (importOriginal) => { + const original = await importOriginal(); + return { + ...original, + open: async (...args: Parameters) => { + 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(); +const clients = new Set(); +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): Promise => + 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(); + const release = createDeferred(); + 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 | 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]); + } +});