fix(recovery): retain file outcomes and cleanup ownership

Use shared owned-file and directory primitives throughout Wakaru discovery,
snapshots, report parsing and publication. Preserve primary read failures and
completed fingerprints when final close fails or receives cancellation.

Retain exact cleanup capabilities before retiring workspaces, block new work
while cleanup is unresolved, and bind error status to the aggregate owner so
partial cleanup and later retry keep resource attribution current.

src/javascript/recovery/RecoveryFiles.ownership.test.ts

@@ -0,0 +1,167 @@
+import type { Dir } from "node:fs";
+import { ArtifactReaderFailure } from "../../artifacts/ArtifactReader.js";
+import { createHash } from "node:crypto";
+import { writeFile, type FileHandle } from "node:fs/promises";
+import { dirname, join } from "node:path";
+import { afterEach, expect, it, onTestFinished, vi } from "vitest";
+import { ArtifactResourceScope } from "../../artifacts/ArtifactResourceScope.js";
+import { ArtifactFileCleanupFailure } from "../../artifacts/ArtifactFileOperation.js";
+import { createTestTempDirectory } from "../../../tests/fixtures/temporaryDirectory.js";
+import { inventoryRecoveryFiles, readRecoveryFile } from "./RecoveryFiles.js";
+
+const closeFault = vi.hoisted(() => ({
+  path: undefined as string | undefined,
+  denied: false,
+  handles: [] as FileHandle[],
+  failure: new Error("controlled recovery descriptor close failure"),
+}));
+
+const directoryFault = vi.hoisted(() => ({
+  path: undefined as string | undefined,
+  readFailure: undefined as Error | undefined,
+  cancel: undefined as AbortController | undefined,
+  closeAttempts: 0,
+}));
+
+vi.mock("node:fs/promises", async (importOriginal) => {
+  const actual = await importOriginal<typeof import("node:fs/promises")>();
+  return {
+    ...actual,
+    opendir: async (...args: Parameters<typeof actual.opendir>) => {
+      const handle: Dir = await actual.opendir(...args);
+      if (args[0] === directoryFault.path) {
+        const close = handle.close.bind(handle);
+        onTestFinished(async () => {
+          await close();
+        });
+        const read = handle.read.bind(handle);
+        handle.read = async () => {
+          if (directoryFault.readFailure !== undefined)
+            throw directoryFault.readFailure;
+          return read();
+        };
+        handle.close = async () => {
+          directoryFault.closeAttempts++;
+          directoryFault.cancel?.abort();
+          throw new Error("controlled recovery directory close failure");
+        };
+      }
+      return handle;
+    },
+    open: async (...args: Parameters<typeof actual.open>) => {
+      const handle = await actual.open(...args);
+      if (args[0] === closeFault.path) {
+        const close = handle.close.bind(handle);
+        closeFault.handles.push(handle);
+        onTestFinished(async () => {
+          if (handle.fd >= 0) await close();
+        });
+        handle.close = async () => {
+          if (closeFault.denied) throw closeFault.failure;
+          await close();
+        };
+      }
+      return handle;
+    },
+  };
+});
+
+afterEach(() => {
+  directoryFault.path = undefined;
+  directoryFault.readFailure = undefined;
+  directoryFault.cancel = undefined;
+  directoryFault.closeAttempts = 0;
+  closeFault.path = undefined;
+  closeFault.denied = false;
+  closeFault.handles = [];
+});
+
+const selectedFile = async () => {
+  const root = await createTestTempDirectory("rea-recovery-file-owner-");
+  const path = join(root, "bundle.js");
+  const bytes = Buffer.from("module.exports = 42;\n");
+  await writeFile(path, bytes);
+  closeFault.path = path;
+  return { path, bytes };
+};
+
+it.each(["completed", "size-limit"] as const)(
+  "retains the exact descriptor and %s read outcome through a failed close",
+  async (mode) => {
+    const { path, bytes } = await selectedFile();
+    const resources = new ArtifactResourceScope();
+    closeFault.denied = true;
+    const failure = await readRecoveryFile(
+      path,
+      mode === "completed" ? 1024 : 0,
+      undefined,
+      resources,
+    ).catch((cause: unknown) => cause);
+    expect(failure).toBeInstanceOf(ArtifactFileCleanupFailure);
+    if (!(failure instanceof ArtifactFileCleanupFailure)) return;
+    const physical = failure.fileCleanupFailure;
+    expect(physical.owner.handle).toBe(closeFault.handles[0]);
+    expect(physical.cleanupCause).toBe(closeFault.failure);
+    if (mode === "completed")
+      expect(physical.outcome).toEqual({
+        kind: "completed",
+        value: {
+          sha256: createHash("sha256").update(bytes).digest("hex"),
+          bytes: bytes.length,
+        },
+      });
+    else
+      expect(physical.outcome).toMatchObject({
+        kind: "failed",
+        cause: { maximum: 0 },
+      });
+    expect(resources.hasPending).toBe(true);
+    await expect(resources.retryCleanup()).rejects.toBeDefined();
+    expect(closeFault.handles[0]?.fd).toBeGreaterThanOrEqual(0);
+    closeFault.denied = false;
+    await resources.retryCleanup();
+    expect(resources.hasPending).toBe(false);
+    expect(closeFault.handles[0]?.fd).toBe(-1);
+  },
+);
+
+it.each([
+  [false, false],
+  [true, false],
+  [false, true],
+  [true, true],
+] as const)(
+  "retains an uncertain directory close and its primary read outcome (read denied: %s, cancel on close: %s)",
+  async (readDenied, cancelOnClose) => {
+    const { path } = await selectedFile();
+    const root = dirname(path);
+    directoryFault.path = root;
+    const readFailure = new Error("controlled directory read failure");
+    if (readDenied) directoryFault.readFailure = readFailure;
+    const resources = new ArtifactResourceScope();
+    const controller = new AbortController();
+    if (cancelOnClose) directoryFault.cancel = controller;
+    const failure = await inventoryRecoveryFiles(
+      root,
+      controller.signal,
+      resources,
+    ).catch((cause: unknown) => cause);
+    expect(failure).toBeInstanceOf(ArtifactReaderFailure);
+    if (!(failure instanceof ArtifactReaderFailure)) return;
+    expect(failure.cleanup?.resources).toContain(root);
+    if (readDenied) {
+      expect(failure.cause).toBeInstanceOf(ArtifactReaderFailure);
+      expect((failure.cause as ArtifactReaderFailure).cause).toBe(readFailure);
+    }
+    if (cancelOnClose && !readDenied) {
+      expect(failure.reason).toBe("cancelled");
+      expect(failure.cancellation?.reason).toBe(controller.signal.reason);
+    }
+    expect(resources.hasPending).toBe(true);
+    expect(directoryFault.closeAttempts).toBe(1);
+    await expect(resources.retryCleanup()).rejects.toBeDefined();
+    // Node has no public directory validity signal, so an unconfirmed close cannot be retried as if it were open.
+    expect(directoryFault.closeAttempts).toBe(1);
+    expect(resources.hasPending).toBe(true);
+  },
+);

src/javascript/recovery/RecoveryFiles.test.ts

@@ -7,6 +7,7 @@ import {
   AnalysisInputError,
   AnalysisOutputError,
 } from "../../domain/analysisErrorCore.js";
+import { ArtifactFileCleanupFailure } from "../../artifacts/ArtifactFileOperation.js";
 import { OwnedFileHandle } from "../../filesystem/OwnedFileHandle.js";
 import {
   openRegularFile,
@@ -92,12 +93,24 @@ it("keeps a failed admission close owned when cancellation is also requested", a
     new Error("close failed"),
     owner,
   );
-  vi.mocked(openRegularFile).mockRejectedValueOnce(failure);
   const controller = new AbortController();
-  controller.abort();
-  await expect(
-    readRecoveryFile("/selected.js", 1024, controller.signal),
-  ).rejects.toBe(failure);
+  vi.mocked(openRegularFile).mockImplementationOnce(async () => {
+    controller.abort();
+    throw failure;
+  });
+  const observed = await readRecoveryFile(
+    "/selected.js",
+    1024,
+    controller.signal,
+  ).catch((cause: unknown) => cause);
+  expect(observed).toBeInstanceOf(ArtifactFileCleanupFailure);
+  if (!(observed instanceof ArtifactFileCleanupFailure)) return;
+  expect(observed.fileCleanupFailure.owner).toBe(owner);
+  expect(observed.fileCleanupFailure.outcome).toEqual({
+    kind: "failed",
+    cause: failure.cause,
+  });
+  expect(observed.fileCleanupFailure.cleanupCause).toBe(failure.cleanupCause);
 });
 
 it("reports one file selected under two letter cases as an input error", async (context) => {

src/javascript/recovery/RecoveryFiles.ts

@@ -1,4 +1,12 @@
-import { regularFileCancellation } from "../../filesystem/OwnedRegularFile.js";
+import {
+  artifactFileCancellationCause,
+  ArtifactFileCancelled,
+  ArtifactFileCleanupFailure,
+  withArtifactRegularFile,
+} from "../../artifacts/ArtifactFileOperation.js";
+import { ArtifactReaderFailure } from "../../artifacts/ArtifactReader.js";
+import { ArtifactResourceScope } from "../../artifacts/ArtifactResourceScope.js";
+import { OwnedDirectoryHandle } from "../../filesystem/OwnedDirectoryHandle.js";
 import { createHash } from "node:crypto";
 import { lstat, mkdir, opendir, writeFile } from "node:fs/promises";
 import { dirname, isAbsolute, join, relative, resolve, sep } from "node:path";
@@ -13,8 +21,7 @@ import {
 } from "../../domain/analysisErrorCore.js";
 import {
   NonRegularFileReadError,
-  openRegularFile,
-  RegularFileAdmissionFailure,
+  sameRegularFileState,
 } from "../../filesystem/RegularFile.js";
 import { RECOVERY_LIMITS, wakaruFixedLimitRecovery } from "./WakaruRelease.js";
 
@@ -48,19 +55,56 @@ const scanRecoveryFile = async (
   maximum: number,
   consume: (chunk: Buffer) => void,
   signal?: AbortSignal,
+  resources?: ArtifactResourceScope,
 ) => {
-  // A FIFO without a writer must be rejected before it can block the open.
-  const handle = await openRegularFile(path, {
-    symlinks: "reject",
-    signal,
-  }).catch((cause: unknown) => {
-    // A descriptor whose close failed stays owned by its error, even when
-    // cancellation was also requested.
-    if (cause instanceof RegularFileAdmissionFailure) throw cause;
-    const cancellation = regularFileCancellation(cause);
-    if (cancellation !== undefined)
+  try {
+    return await withArtifactRegularFile(
+      path,
+      resources ?? new ArtifactResourceScope(),
+      signal,
+      async (handle) => {
+        const hash = createHash("sha256");
+        const before = await handle.stat();
+        if (before.size > maximum)
+          throw new RecoveryFileSizeError(
+            `Expected a regular file no larger than ${String(maximum)} bytes: ${path}`,
+            maximum,
+          );
+        let bytes = 0;
+        for await (const raw of handle.createReadStream({ autoClose: false })) {
+          checkRecoveryCancellation(signal);
+          const chunk = Buffer.isBuffer(raw) ? raw : Buffer.from(raw);
+          bytes += chunk.length;
+          if (bytes > maximum)
+            throw new RecoveryFileSizeError(
+              `File exceeds ${String(maximum)} bytes: ${path}`,
+              maximum,
+            );
+          hash.update(chunk);
+          consume(chunk);
+        }
+        const after = await handle.stat();
+        const current = await lstat(path);
+        if (
+          before.size !== bytes ||
+          [after, current].some((state) => !sameRegularFileState(before, state))
+        )
+          throw new AnalysisOutputError(
+            OPERATION,
+            `File identity changed while reading: ${path}`,
+          );
+        return { sha256: hash.digest("hex"), bytes };
+      },
+    );
+  } catch (cause: unknown) {
+    if (
+      cause instanceof ArtifactFileCleanupFailure ||
+      cause instanceof ArtifactFileCancelled
+    )
+      throw cause;
+    if (cause instanceof ArtifactReaderFailure && cause.reason === "cancelled")
       throw new AnalysisCancelledError(OPERATION, {
-        cause: cancellation.reason,
+        cause: artifactFileCancellationCause(cause),
       });
     if (cause instanceof NonRegularFileReadError)
       throw new AnalysisOutputError(
@@ -68,45 +112,6 @@ const scanRecoveryFile = async (
         `Expected a regular file no larger than ${String(maximum)} bytes: ${path}`,
       );
     throw cause;
-  });
-  const hash = createHash("sha256");
-  try {
-    const before = await handle.stat();
-    if (before.size > maximum)
-      throw new RecoveryFileSizeError(
-        `Expected a regular file no larger than ${String(maximum)} bytes: ${path}`,
-        maximum,
-      );
-    let bytes = 0;
-    for await (const raw of handle.createReadStream({ autoClose: false })) {
-      checkRecoveryCancellation(signal);
-      const chunk = Buffer.isBuffer(raw) ? raw : Buffer.from(raw);
-      bytes += chunk.length;
-      if (bytes > maximum)
-        throw new RecoveryFileSizeError(
-          `File exceeds ${String(maximum)} bytes: ${path}`,
-          maximum,
-        );
-      hash.update(chunk);
-      consume(chunk);
-    }
-    const after = await handle.stat();
-    const current = await lstat(path);
-    if (
-      before.dev !== current.dev ||
-      before.ino !== current.ino ||
-      before.size !== bytes ||
-      before.mtimeMs !== after.mtimeMs ||
-      before.ctimeMs !== after.ctimeMs ||
-      before.size !== after.size
-    )
-      throw new AnalysisOutputError(
-        OPERATION,
-        `File identity changed while reading: ${path}`,
-      );
-    return { sha256: hash.digest("hex"), bytes };
-  } finally {
-    await handle.close();
   }
 };
 
@@ -115,13 +120,15 @@ export const fingerprintRecoveryFile = async (
   path: string,
   maximum: number,
   signal?: AbortSignal,
-) => scanRecoveryFile(path, maximum, () => {}, signal);
+  resources?: ArtifactResourceScope,
+) => scanRecoveryFile(path, maximum, () => {}, signal, resources);
 
 /** Read only a regular file through one handle with a bounded, stable identity. */
 export const readRecoveryFile = async (
   path: string,
   maximum: number,
   signal?: AbortSignal,
+  resources?: ArtifactResourceScope,
 ): Promise<Buffer> => {
   const chunks: Buffer[] = [];
   const result = await scanRecoveryFile(
@@ -129,6 +136,7 @@ export const readRecoveryFile = async (
     maximum,
     (chunk) => chunks.push(chunk),
     signal,
+    resources,
   );
   return Buffer.concat(chunks, result.bytes);
 };
@@ -148,6 +156,7 @@ const snapshotRecoveryInput = async (
   relativePath: string,
   maximum: number,
   signal?: AbortSignal,
+  resources?: ArtifactResourceScope,
 ): Promise<RecoveryInputSnapshot> => {
   if (!isAbsolute(path))
     throw recoveryInputFailure(
@@ -156,9 +165,14 @@ const snapshotRecoveryInput = async (
     );
   let bytes: Buffer;
   try {
-    bytes = await readRecoveryFile(path, maximum, signal);
+    bytes = await readRecoveryFile(path, maximum, signal, resources);
   } catch (cause: unknown) {
-    if (cause instanceof AnalysisCancelledError) throw cause;
+    if (
+      cause instanceof AnalysisCancelledError ||
+      cause instanceof ArtifactFileCleanupFailure ||
+      cause instanceof ArtifactFileCancelled
+    )
+      throw cause;
     // The operation-wide input allowance is a deterministic byte budget, not
     // a malformed argument; the same selection fails again unchanged.
     if (cause instanceof RecoveryFileSizeError)
@@ -227,6 +241,7 @@ export const snapshotRecoveryInputs = async (
   paths: readonly [string, ...string[]],
   directory: string,
   signal?: AbortSignal,
+  resources?: ArtifactResourceScope,
 ): Promise<readonly [RecoveryInputSnapshot, ...RecoveryInputSnapshot[]]> => {
   let commonDirectory = dirname(resolve(paths[0]));
   for (const path of paths) {
@@ -277,6 +292,7 @@ export const snapshotRecoveryInputs = async (
       relativePath,
       remaining,
       signal,
+      resources,
     );
     snapshots.push(snapshot);
     remaining -= snapshot.bytes;
@@ -292,6 +308,7 @@ export const snapshotRecoveryInputs = async (
 export const inventoryRecoveryFiles = async (
   root: string,
   signal?: AbortSignal,
+  resources: ArtifactResourceScope = new ArtifactResourceScope(),
 ): Promise<
   readonly { relativePath: string; path: string; bytes: number }[]
 > => {
@@ -306,9 +323,15 @@ export const inventoryRecoveryFiles = async (
         `Unsafe staging directory: ${directory}`,
       );
     const handle = await opendir(directory);
+    const owner = new OwnedDirectoryHandle(handle);
     const children: string[] = [];
+    let outcome:
+      | { readonly kind: "completed" }
+      | { readonly kind: "failed"; readonly cause: unknown };
     try {
-      for await (const entry of handle) {
+      for (;;) {
+        const entry = await handle.read();
+        if (entry === null) break;
         checkRecoveryCancellation(signal);
         entries++;
         if (entries > RECOVERY_LIMITS.outputEntries)
@@ -321,9 +344,32 @@ export const inventoryRecoveryFiles = async (
           );
         children.push(entry.name);
       }
-    } finally {
-      await handle.close().catch(() => undefined);
+      outcome = { kind: "completed" };
+    } catch (cause: unknown) {
+      outcome = { kind: "failed", cause };
     }
+    const cleanup = await resources.release({
+      kind: "directory-handle",
+      handle: owner,
+      resource: directory,
+    });
+    if (cleanup.kind === "failed") {
+      const failure = resources.cleanupFailure(outcome, [cleanup]);
+      if (outcome.kind === "completed" && signal?.aborted === true)
+        throw new ArtifactReaderFailure(
+          "cancelled",
+          "Recovery inventory cancelled during directory close",
+          {
+            cause: failure,
+            cancellation: { reason: signal.reason },
+            cleanup: failure.cleanup,
+            cleanupLeases: failure.cleanupLeases,
+          },
+        );
+      throw failure;
+    }
+    if (outcome.kind === "failed") throw outcome.cause;
+    checkRecoveryCancellation(signal);
     for (const name of children) {
       checkRecoveryCancellation(signal);
       const relativePath = prefix === "" ? name : `${prefix}/${name}`;
@@ -369,7 +415,8 @@ export const recoveryFailureMessage = (cause: unknown): string =>
 
 /** Cancellation remains distinct from malformed source or provider output. */
 export const checkRecoveryCancellation = (signal?: AbortSignal): void => {
-  if (signal?.aborted === true) throw new AnalysisCancelledError(OPERATION);
+  if (signal?.aborted === true)
+    throw new AnalysisCancelledError(OPERATION, { cause: signal.reason });
 };
 
 /** Keep both tool invocations within one operation's execution budget. */

src/javascript/recovery/WakaruCommand.ts

@@ -7,6 +7,8 @@ import {
   AnalysisCapabilityUnavailableError,
   AnalysisOutputError,
 } from "../../domain/analysisErrorCore.js";
+import { ArtifactReaderFailure } from "../../artifacts/ArtifactReader.js";
+import type { ArtifactResourceScope } from "../../artifacts/ArtifactResourceScope.js";
 import { ProviderAdapterError } from "../../domain/providerAdapterError.js";
 import { ProviderCleanupError } from "../../domain/providerCleanupError.js";
 import { prlimitArgumentVector } from "../../process/LinuxResourceLimits.js";
@@ -68,6 +70,8 @@ export class WakaruCleanupFailure extends ProviderCleanupError {
 /** Resolve and fingerprint explicitly configured tools; acquire nothing at startup. */
 export const resolveWakaruCommand = async (
   environment: Readonly<Record<string, string | undefined>>,
+  resources?: ArtifactResourceScope,
+  signal?: AbortSignal,
 ) => {
   const configured = environment.REA_WAKARU_COMMAND;
   if (
@@ -87,6 +91,8 @@ export const resolveWakaruCommand = async (
     const fingerprint = await fingerprintRecoveryFile(
       command,
       RECOVERY_LIMITS.outputBytes,
+      signal,
+      resources,
     );
     const limiter =
       environment.REA_JAVASCRIPT_PRLIMIT_COMMAND ?? "/usr/bin/prlimit";
@@ -97,6 +103,11 @@ export const resolveWakaruCommand = async (
       throw new TypeError(`Resource limiter is not a regular file: ${limiter}`);
     return { command, limiter, ...fingerprint };
   } catch (cause: unknown) {
+    if (
+      cause instanceof ArtifactReaderFailure &&
+      (cause.cleanup !== undefined || cause.reason === "cancelled")
+    )
+      throw cause;
     const reason = `Configured Wakaru or util-linux prlimit is unavailable: ${recoveryFailureMessage(cause)}`;
     throw new AnalysisCapabilityUnavailableError("wakaru", OPERATION, reason, {
       cause,
@@ -115,6 +126,7 @@ type WakaruCommandContext = {
   signal?: AbortSignal;
   launcher?: WakaruLauncher;
   staging?: string;
+  resources?: ArtifactResourceScope;
 };
 
 /** Run one bounded command and verify cleanup before consuming its output. */
@@ -183,7 +195,11 @@ export const runWakaruCommand = async (context: WakaruCommandContext) => {
           ),
         );
       if (context.staging !== undefined)
-        await inventoryRecoveryFiles(context.staging, context.signal);
+        await inventoryRecoveryFiles(
+          context.staging,
+          context.signal,
+          context.resources,
+        );
     }
     checkRecoveryDeadline(context.deadline, context.signal);
     if (oversized)

src/javascript/recovery/WakaruExecution.ts

@@ -4,6 +4,7 @@ import {
   type ExecutionOptions,
 } from "../../application/AnalysisProvider.js";
 import type { SafeOutputTree } from "../../artifacts/SafeOutputTree.js";
+import type { ArtifactResourceScope } from "../../artifacts/ArtifactResourceScope.js";
 import {
   AnalysisCapabilityUnavailableError,
   AnalysisOutputError,
@@ -43,12 +44,21 @@ type RecoveryWorkspace = {
   launcher: WakaruLauncher | undefined;
   options: ExecutionOptions | undefined;
   deadline: number;
+  resources: ArtifactResourceScope;
 };
 
 /** Consume the pinned producer representation in a private workspace. */
 const executeWakaruRecovery = async (context: RecoveryWorkspace) => {
-  const { root, input, sources, engine, environment, launcher, options } =
-    context;
+  const {
+    root,
+    input,
+    sources,
+    engine,
+    environment,
+    launcher,
+    options,
+    resources,
+  } = context;
   const staging = join(root, "modules");
   const commandContext = {
     ...engine,
@@ -57,6 +67,7 @@ const executeWakaruRecovery = async (context: RecoveryWorkspace) => {
     environment: environment,
     ...(options?.signal === undefined ? {} : { signal: options.signal }),
     ...(launcher === undefined ? {} : { launcher: launcher }),
+    resources,
   };
   const version = await runWakaruCommand({
     ...commandContext,
@@ -90,7 +101,11 @@ const executeWakaruRecovery = async (context: RecoveryWorkspace) => {
       staging,
     ],
   });
-  const files = await inventoryRecoveryFiles(staging, options?.signal);
+  const files = await inventoryRecoveryFiles(
+    staging,
+    options?.signal,
+    resources,
+  );
   const snapshots = new Map<string, Buffer>();
   for (const source of sources)
     snapshots.set(
@@ -99,6 +114,7 @@ const executeWakaruRecovery = async (context: RecoveryWorkspace) => {
         source.snapshot_path,
         source.bytes,
         options?.signal,
+        resources,
       ),
     );
   const parsed = await parseWakaruReports({
@@ -106,13 +122,15 @@ const executeWakaruRecovery = async (context: RecoveryWorkspace) => {
     provenancePath: join(staging, "provenance.json"),
     snapshots,
     files,
+    ...(options?.signal === undefined ? {} : { signal: options.signal }),
+    resources,
   });
   return { version, admitted, run, parsed };
 };
 
 /** Publish validated recovery and retain exact source and engine bindings. */
 export const prepareWakaruExecution = async (context: RecoveryWorkspace) => {
-  const { root, tree, input, sources, engine, options } = context;
+  const { root, tree, input, sources, engine, options, resources } = context;
   const [source] = sources;
   const staging = join(root, "modules");
   const { version, admitted, run, parsed } =
@@ -127,12 +145,14 @@ export const prepareWakaruExecution = async (context: RecoveryWorkspace) => {
     parsed,
     execution: run,
     ...(options?.signal === undefined ? {} : { signal: options.signal }),
+    resources,
   });
   for (const selected of sources) {
     const current = await fingerprintRecoveryFile(
       selected.path,
       RECOVERY_LIMITS.inputBytes,
       options?.signal,
+      resources,
     );
     if (current.sha256 !== selected.sha256 || current.bytes !== selected.bytes)
       throw new AnalysisOutputError(
@@ -146,6 +166,7 @@ export const prepareWakaruExecution = async (context: RecoveryWorkspace) => {
         engine.command,
         RECOVERY_LIMITS.outputBytes,
         options?.signal,
+        resources,
       )
     ).sha256 !== engine.sha256
   )

src/javascript/recovery/WakaruProvider.ts

@@ -6,6 +6,8 @@ import type {
   AnalysisExecution,
   ExecutionOptions,
 } from "../../application/AnalysisProvider.js";
+import { ArtifactResourceScope } from "../../artifacts/ArtifactResourceScope.js";
+import { AnalysisCleanupLease } from "../../artifacts/AnalysisCleanupLease.js";
 import { SafeOutputTree } from "../../artifacts/SafeOutputTree.js";
 import { SafeOutputTreeCreationFailure } from "../../artifacts/SafeOutputTreeCreationFailure.js";
 import { AnalysisError } from "../../domain/analysisErrorBase.js";
@@ -39,16 +41,22 @@ import { RECOVERY_LIMITS } from "./WakaruRelease.js";
 
 const OPERATION = "recover_javascript_sources";
 
+interface PendingRecoveryCleanup {
+  root: string | undefined;
+  process: ProviderProcessSupervisor | undefined;
+  tree: SafeOutputTree | undefined;
+}
+
 /** Serialized recovery adapter retaining failed process, workspace and publication cleanup. */
 export class WakaruProvider implements JavaScriptRecoveryPort {
+  readonly #resources = new ArtifactResourceScope();
   #tail: Promise<void> = Promise.resolve();
   #closing = false;
   #closePromise: Promise<void> | undefined;
   #pendingCleanup:
     | {
-        root: string | undefined;
-        process: ProviderProcessSupervisor | undefined;
-        tree: SafeOutputTree | undefined;
+        readonly resources: PendingRecoveryCleanup;
+        readonly lease: AnalysisCleanupLease;
       }
     | undefined;
 
@@ -107,7 +115,11 @@ export class WakaruProvider implements JavaScriptRecoveryPort {
     const deadline = performance.now() + RECOVERY_LIMITS.timeoutMs;
     try {
       checkRecoveryCancellation(options?.signal);
-      const engine = await resolveWakaruCommand(this.environment);
+      const engine = await resolveWakaruCommand(
+        this.environment,
+        this.#resources,
+        options?.signal,
+      );
       root = await mkdtemp(join(tmpdir(), "rea-wakaru-"));
       const inputs = join(root, "inputs");
       const staging = join(root, "modules");
@@ -117,6 +129,7 @@ export class WakaruProvider implements JavaScriptRecoveryPort {
         [input.path, ...(input.additional_paths ?? [])],
         inputs,
         options?.signal,
+        this.#resources,
       );
       try {
         tree = await SafeOutputTree.create(input.output_directory);
@@ -131,6 +144,7 @@ export class WakaruProvider implements JavaScriptRecoveryPort {
       execution = await prepareWakaruExecution({
         root,
         tree,
+        resources: this.#resources,
         input,
         sources,
         engine,
@@ -150,7 +164,8 @@ export class WakaruProvider implements JavaScriptRecoveryPort {
     if (failed !== undefined) {
       const failure =
         failed.cause instanceof ArtifactReaderFailure &&
-        failed.cause.reason === "cancelled"
+        failed.cause.reason === "cancelled" &&
+        failed.cause.cleanup === undefined
           ? new AnalysisCancelledError(OPERATION, { cause: failed.cause })
           : failed.cause;
       const residuals: string[] = [];
@@ -174,7 +189,9 @@ export class WakaruProvider implements JavaScriptRecoveryPort {
       }
       if (cleanupOwner !== undefined)
         residuals.push(cleanupOwner.launch.ownership?.runId ?? "wakaru-worker");
-      if (root !== undefined && cleanupOwner === undefined) {
+      const filesPending = this.#resources.hasPending;
+      if (filesPending) residuals.push(...this.#resources.cleanupResources);
+      if (root !== undefined && cleanupOwner === undefined && !filesPending) {
         const failedRoot = root;
         try {
           await rm(failedRoot, { recursive: true, force: true });
@@ -190,47 +207,75 @@ export class WakaruProvider implements JavaScriptRecoveryPort {
       if (
         cleanupOwner !== undefined ||
         failedTree !== undefined ||
-        root !== undefined
+        root !== undefined ||
+        filesPending
       ) {
-        this.#pendingCleanup = {
+        const resources: PendingRecoveryCleanup = {
           root,
           process: cleanupOwner,
           tree: failedTree,
         };
+        const files = this.#resources;
+        const lease = new AnalysisCleanupLease(
+          () => this.#releasePending(resources),
+          {
+            get cleanupResources() {
+              return [
+                ...files.cleanupResources,
+                ...(resources.process === undefined
+                  ? []
+                  : [
+                      resources.process.launch.ownership?.runId ??
+                        "wakaru-worker",
+                    ]),
+                ...(resources.tree === undefined
+                  ? []
+                  : [resources.tree.outputRoot]),
+                ...(resources.root === undefined ? [] : [resources.root]),
+              ];
+            },
+          },
+        );
+        this.#pendingCleanup = { resources, lease };
         if (root !== undefined) residuals.push(root);
         root = undefined;
       }
-      if (residuals.length > 0)
-        return err(
-          new ProviderCleanupError(
-            "wakaru",
-            residuals,
-            {
-              ...(failure instanceof WakaruCleanupFailure
-                ? failure.diagnostics
-                : {}),
-              previous_error:
-                failure instanceof WakaruCleanupFailure
-                  ? (failure.diagnostics?.previous_error ?? null)
-                  : recoveryFailureMessage(failure),
-              cleanup_failures: cleanupFailures,
-            },
-            {
-              operation: OPERATION,
-              cause: failure,
-              ...(tree?.published && execution !== undefined
-                ? {
-                    partialObservation: {
-                      kind: "javascript-recovery" as const,
-                      result: javascriptRecoveryResultSchema.parse(
-                        execution.result,
-                      ),
-                    },
-                  }
-                : {}),
-            },
-          ),
+      if (residuals.length > 0) {
+        const error = new ProviderCleanupError(
+          "wakaru",
+          residuals,
+          {
+            ...(failure instanceof WakaruCleanupFailure
+              ? failure.diagnostics
+              : {}),
+            previous_error:
+              failure instanceof WakaruCleanupFailure
+                ? (failure.diagnostics?.previous_error ?? null)
+                : recoveryFailureMessage(failure),
+            cleanup_failures: cleanupFailures,
+          },
+          {
+            operation: OPERATION,
+            cause: failure,
+            ...(tree?.published && execution !== undefined
+              ? {
+                  partialObservation: {
+                    kind: "javascript-recovery" as const,
+                    result: javascriptRecoveryResultSchema.parse(
+                      execution.result,
+                    ),
+                  },
+                }
+              : {}),
+          },
         );
+        const lease = this.#pendingCleanup?.lease;
+        if (lease !== undefined) {
+          lease.bindCleanupError(error);
+          error.bindCleanupResourceOwner(lease);
+        }
+        return err(error);
+      }
       if (tree?.published === true && execution !== undefined)
         return ok(execution);
       return err(
@@ -266,49 +311,36 @@ export class WakaruProvider implements JavaScriptRecoveryPort {
   async #retire(): Promise<ProviderCleanupError | undefined> {
     const pending = this.#pendingCleanup;
     if (pending === undefined) return undefined;
+    try {
+      await pending.lease.release();
+      this.#pendingCleanup = undefined;
+      return undefined;
+    } catch (cause: unknown) {
+      const error = new ProviderCleanupError(
+        "wakaru",
+        pending.lease.cleanupResources ?? [],
+        { reason: recoveryFailureMessage(cause) },
+        { operation: OPERATION, cause },
+      ).bindCleanupResourceOwner(pending.lease);
+      pending.lease.bindCleanupError(error);
+      return error;
+    }
+  }
+
+  async #releasePending(pending: PendingRecoveryCleanup): Promise<void> {
     if (pending.process !== undefined) {
       const stopped = await pending.process.stop();
-      if (stopped.status === "incomplete")
-        return new ProviderCleanupError(
-          "wakaru",
-          [
-            pending.process.launch.ownership?.runId ?? "wakaru-worker",
-            ...(pending.root === undefined ? [] : [pending.root]),
-          ],
-          { reason: stopped.reason },
-          { operation: OPERATION },
-        );
+      if (stopped.status === "incomplete") throw new Error(stopped.reason);
       pending.process = undefined;
     }
+    await this.#resources.retryCleanup();
     if (pending.tree !== undefined) {
-      const tree = pending.tree;
-      try {
-        await tree.rollback();
-        pending.tree = undefined;
-      } catch (cause: unknown) {
-        return new ProviderCleanupError(
-          "wakaru",
-          [tree.outputRoot],
-          { reason: recoveryFailureMessage(cause) },
-          { operation: OPERATION, cause },
-        );
-      }
+      await pending.tree.rollback();
+      pending.tree = undefined;
     }
     if (pending.root !== undefined) {
-      const root = pending.root;
-      try {
-        await rm(root, { recursive: true, force: true });
-        pending.root = undefined;
-      } catch (cause: unknown) {
-        return new ProviderCleanupError(
-          "wakaru",
-          [root],
-          { reason: recoveryFailureMessage(cause) },
-          { operation: OPERATION, cause },
-        );
-      }
+      await rm(pending.root, { recursive: true, force: true });
+      pending.root = undefined;
     }
-    this.#pendingCleanup = undefined;
-    return undefined;
   }
 }

src/javascript/recovery/WakaruPublication.ts

@@ -1,6 +1,7 @@
 import { join } from "node:path";
 import { Readable } from "node:stream";
 import type { SafeOutputTree } from "../../artifacts/SafeOutputTree.js";
+import type { ArtifactResourceScope } from "../../artifacts/ArtifactResourceScope.js";
 import { AnalysisOutputError } from "../../domain/analysisErrorCore.js";
 import {
   javascriptRecoveryResultSchema,
@@ -37,6 +38,7 @@ type PublicationContext = {
   admitted: Exclude<WakaruVersionAdmission, { status: "unsupported" }>;
   parsed: Awaited<ReturnType<typeof parseWakaruReports>>;
   execution: Awaited<ReturnType<typeof runWakaruCommand>>;
+  resources?: ArtifactResourceScope;
   signal?: AbortSignal;
 };
 type PublishedFile = JavaScriptRecoveryResult["manifest"];
@@ -46,7 +48,7 @@ type RecoveryWriter = (path: string, bytes: Buffer) => Promise<PublishedFile>;
 export const publishWakaruArtifacts = async (
   context: PublicationContext,
 ): Promise<JavaScriptRecoveryResult> => {
-  const { tree, sources, parsed, staging, signal } = context;
+  const { tree, sources, parsed, staging, signal, resources } = context;
   const write = async (path: string, bytes: Buffer) => {
     const sha256 = recoveryDigest(bytes);
     const file = await tree.write(
@@ -68,6 +70,7 @@ export const publishWakaruArtifacts = async (
       source.snapshot_path,
       source.bytes,
       signal,
+      resources,
     );
     if (recoveryDigest(snapshot) !== source.sha256)
       throw new AnalysisOutputError(
@@ -94,6 +97,7 @@ export const publishWakaruArtifacts = async (
       join(staging, "provenance.json"),
       RECOVERY_LIMITS.reportBytes,
       signal,
+      resources,
     ),
   );
   const details = {
@@ -161,13 +165,14 @@ const publishModule = async (
   module: PublicationContext["parsed"]["report"]["modules"][number],
   write: RecoveryWriter,
 ): Promise<JavaScriptRecoveryResult["modules"][number]> => {
-  const { staging, sources, parsed, signal } = context;
+  const { staging, sources, parsed, signal, resources } = context;
   const artifact = await write(
     `modules/${module.filename}`,
     await readRecoveryFile(
       join(staging, module.filename),
       RECOVERY_LIMITS.outputBytes,
       signal,
+      resources,
     ),
   );
   const mapPath = join(staging, `${module.filename}.map`);
@@ -177,6 +182,7 @@ const publishModule = async (
     mapPath,
     RECOVERY_LIMITS.outputBytes,
     signal,
+    resources,
   ).catch((cause: unknown) => {
     if (cause instanceof Error && "code" in cause && cause.code === "ENOENT")
       return null;

src/javascript/recovery/WakaruReport.ts

@@ -1,4 +1,6 @@
 import { z } from "zod";
+import { ArtifactReaderFailure } from "../../artifacts/ArtifactReader.js";
+import type { ArtifactResourceScope } from "../../artifacts/ArtifactResourceScope.js";
 import {
   normalizeArtifactPath,
   ArtifactPathRegistry,
@@ -57,6 +59,8 @@ type ReportContext = {
   provenancePath: string;
   snapshots: ReadonlyMap<string, Buffer>;
   files: readonly { relativePath: string }[];
+  resources?: ArtifactResourceScope;
+  signal?: AbortSignal;
 };
 
 /**
@@ -66,7 +70,7 @@ type ReportContext = {
  * checks below remain exact.
  */
 export const parseWakaruReports = async (context: ReportContext) => {
-  const { execution, provenancePath } = context;
+  const { execution, provenancePath, resources, signal } = context;
 
   try {
     if (execution.exit_code !== 0 && execution.exit_code !== 1)
@@ -76,7 +80,9 @@ export const parseWakaruReports = async (context: ReportContext) => {
     const report = reportSchema.parse(JSON.parse(execution.stdout));
     const provenance = provenanceSchema.parse(
       JSON.parse(
-        (await readWakaruProvenanceReport(provenancePath)).toString("utf8"),
+        (
+          await readWakaruProvenanceReport(provenancePath, signal, resources)
+        ).toString("utf8"),
       ),
     );
     if (report.total !== report.modules.length || report.failed > report.total)
@@ -101,6 +107,11 @@ export const parseWakaruReports = async (context: ReportContext) => {
           : ("complete" as const),
     };
   } catch (cause: unknown) {
+    if (
+      cause instanceof ArtifactReaderFailure &&
+      (cause.cleanup !== undefined || cause.reason === "cancelled")
+    )
+      throw cause;
     throw new ProviderAdapterError("wakaru", OPERATION, {
       cause,
       ...wakaruReportGuidance(cause),
@@ -118,9 +129,18 @@ export const parseWakaruReports = async (context: ReportContext) => {
 const PROVENANCE_REPORT_ALLOWANCE = `Wakaru provenance report exceeds the ${String(RECOVERY_LIMITS.reportBytes)}-byte report allowance`;
 
 /** Report-size refusals name the fixed allowance; other read failures stay unchanged. */
-const readWakaruProvenanceReport = async (path: string): Promise<Buffer> => {
+const readWakaruProvenanceReport = async (
+  path: string,
+  signal: AbortSignal | undefined,
+  resources: ArtifactResourceScope | undefined,
+): Promise<Buffer> => {
   try {
-    return await readRecoveryFile(path, RECOVERY_LIMITS.reportBytes);
+    return await readRecoveryFile(
+      path,
+      RECOVERY_LIMITS.reportBytes,
+      signal,
+      resources,
+    );
   } catch (cause: unknown) {
     if (!isRecoveryFileSizeError(cause, RECOVERY_LIMITS.reportBytes))
       throw cause;