fix(process): retain lease directory cleanup ownership

src/application/LinuxHopperPublisher.ts

@@ -306,6 +306,7 @@ export class LinuxHopperPublisher {
       );
       const lease = await this.#acquireCacheLease(
         verified.release.main_exe_hash,
+        resources,
         signal,
       );
       await this.#withOwnedResource(
@@ -412,6 +413,7 @@ export class LinuxHopperPublisher {
 
   async #acquireCacheLease(
     digest: string,
+    resources: ArtifactResourceScope,
     signal?: AbortSignal,
   ): Promise<OwnedLease> {
     const deadline = performance.now() + 120_000;
@@ -423,6 +425,7 @@ export class LinuxHopperPublisher {
         cacheDirectory: this.cacheRoot,
         launcherSha256: digest,
         runId,
+        resources,
       });
       if (acquired.acquired) return acquired.lease;
       const remaining = deadline - performance.now();

src/application/decompileDirectory/exportTree.ts

@@ -10,6 +10,11 @@ import {
   type FileHandle,
 } from "node:fs/promises";
 import { basename, dirname, join } from "node:path";
+import {
+  ArtifactFileCleanupFailure,
+  bindArtifactFileCleanup,
+} from "../../artifacts/ArtifactFileOperation.js";
+import { ArtifactResourceScope } from "../../artifacts/ArtifactResourceScope.js";
 
 import {
   DECOMPILE_DIRECTORY_FUNCTIONS_DIRECTORY,
@@ -249,7 +254,7 @@ export class DecompileExport {
     if (!created.ok) return created;
     // The directory exists before its journal, so a resume that starts in
     // between takes the same lease and cannot write alongside this export.
-    const lease = await acquireExportLease(root);
+    const lease = await acquireExportLease(root, scope);
     // Even an empty directory may now have another lease holder preparing its
     // journal. Without the lease, preserve it rather than disrupting that run.
     if (!lease.ok) {
@@ -286,7 +291,7 @@ export class DecompileExport {
   ): Promise<Result<DecompileExport, AnalysisError>> {
     // The lease is held before the journal is read, so no other run appends
     // between this read and this run's own appends, repair or index.
-    const lease = await acquireExportLease(root);
+    const lease = await acquireExportLease(root, scope);
     if (!lease.ok) return lease;
     const resources = new ExportResources(
       root,
@@ -805,6 +810,7 @@ const DECOMPILE_EXPORT_LEASE_DIRECTORY = join(
  */
 const acquireExportLease = async (
   root: string,
+  scope: RetainedResourceScope | undefined,
 ): Promise<
   Result<
     {
@@ -824,6 +830,7 @@ const acquireExportLease = async (
         "The output directory could not be inspected.",
       ),
     );
+  const leaseResources = new ArtifactResourceScope();
   try {
     const acquisition = await acquireOwnedLease({
       identity: {
@@ -834,6 +841,7 @@ const acquireExportLease = async (
       runId: randomUUID(),
       directory: DECOMPILE_EXPORT_LEASE_DIRECTORY,
       failure: exportLeaseFailure,
+      resources: leaseResources,
     });
     return acquisition.acquired
       ? ok({ lease: acquisition.lease, identity: stat })
@@ -851,6 +859,16 @@ const acquireExportLease = async (
           ),
         );
   } catch (cause: unknown) {
+    if (leaseResources.hasPending)
+      scope?.retain(
+        {
+          close: () => leaseResources.close(),
+          get cleanupResources() {
+            return leaseResources.cleanupResources;
+          },
+        },
+        "decompile lease directory",
+      );
     return err(
       cause instanceof AnalysisError
         ? cause
@@ -866,15 +884,30 @@ const acquireExportLease = async (
 const exportLeaseFailure = (
   failure: LeaseFailure,
   cause?: unknown,
-): AnalysisOutputError =>
-  new AnalysisOutputError(
+): AnalysisOutputError => {
+  const primary =
+    cause instanceof ArtifactFileCleanupFailure &&
+    cause.fileCleanupFailure.outcome.kind === "failed"
+      ? cause.fileCleanupFailure.outcome.cause
+      : undefined;
+  const error = new AnalysisOutputError(
     "decompile_all",
     "output directory lease could not be acquired",
     {
       ...(cause === undefined ? {} : { cause }),
-      userMessage: exportLeaseMessage(failure),
+      ...(primary instanceof AnalysisError ? { cleanupPrimary: primary } : {}),
+      ...(cause instanceof ArtifactFileCleanupFailure
+        ? { cleanup: cause.cleanup }
+        : {}),
+      userMessage:
+        primary instanceof AnalysisError
+          ? (primary.userMessage ?? exportLeaseMessage(failure))
+          : exportLeaseMessage(failure),
     },
   );
+  bindArtifactFileCleanup(cause, error);
+  return error;
+};
 
 const exportLeaseMessage = (failure: LeaseFailure): string => {
   switch (failure.kind) {
@@ -886,6 +919,8 @@ const exportLeaseMessage = (failure: LeaseFailure): string => {
       return `REA could not remove its lease recovery directory at ${JSON.stringify(failure.path)}. Check its write access before retrying.`;
     case "unrecoverable":
       return "REA could not take this output directory's lease. Retry after other decompile_all runs finish.";
+    case "directory_close":
+      return "REA could not confirm closure of its output lease directory handle.";
     case "directory_access_denied":
       return `REA could not access its lease directory at ${JSON.stringify(failure.directory)}. Restore owner-only permissions (0700) on a directory you own, then retry.`;
     case "directory_create":

src/hopper/BridgeLauncher.ts

@@ -507,7 +507,7 @@ export class HopperApplicationLauncher implements BridgeLauncher {
         ? await (
             this.dependencies.acquireApplicationLease ??
             acquireLinuxHopperApplicationLease
-          )({ runId })
+          )({ runId, resources: this.#probeResources })
         : await (
             this.dependencies.acquireTargetLease ?? acquireHopperTargetLease
           )({
@@ -515,6 +515,7 @@ export class HopperApplicationLauncher implements BridgeLauncher {
             targetKind: this.options.targetKind,
             loaderArgs: this.options.loaderArgs,
             runId,
+            resources: this.#probeResources,
           });
       if (!acquisition.acquired)
         return err(

src/process/HopperLease.ts

@@ -1,5 +1,11 @@
 import { realpath } from "node:fs/promises";
 import { join } from "node:path";
+
+import {
+  ArtifactFileCleanupFailure,
+  bindArtifactFileCleanup,
+} from "../artifacts/ArtifactFileOperation.js";
+import type { ArtifactResourceScope } from "../artifacts/ArtifactResourceScope.js";
 import { HopperStartError } from "../domain/hopperErrors.js";
 import {
   acquireOwnedLease,
@@ -20,6 +26,7 @@ export const acquireHopperTargetLease = async (input: {
   readonly loaderArgs: readonly string[];
   readonly runId: string;
   readonly directory?: string;
+  readonly resources?: ArtifactResourceScope | undefined;
 }): Promise<LeaseAcquisition> => {
   // Canonicalization is required for lease identity. A lexical fallback could
   // let two aliases reserve independent sockets for the same Hopper document.
@@ -46,6 +53,7 @@ export const acquireHopperTargetLease = async (input: {
 export const acquireLinuxHopperApplicationLease = (input: {
   readonly runId: string;
   readonly directory?: string;
+  readonly resources?: ArtifactResourceScope | undefined;
 }): Promise<LeaseAcquisition> =>
   acquireLease({ ...input, identity: { application: "linux-hopper-demo" } });
 
@@ -54,9 +62,11 @@ export const acquireHopperReleaseCacheLease = async (input: {
   readonly cacheDirectory: string;
   readonly launcherSha256: string;
   readonly runId: string;
+  readonly resources?: ArtifactResourceScope | undefined;
 }): Promise<LeaseAcquisition> =>
   acquireLease({
     runId: input.runId,
+    resources: input.resources,
     identity: {
       cacheDirectory: await realpath(input.cacheDirectory),
       launcherSha256: input.launcherSha256,
@@ -68,22 +78,39 @@ const acquireLease = async (input: {
   readonly identity: Readonly<Record<string, unknown>>;
   readonly runId: string;
   readonly directory?: string;
+  readonly resources?: ArtifactResourceScope | undefined;
 }): Promise<LeaseAcquisition> =>
   acquireOwnedLease({
     identity: input.identity,
     runId: input.runId,
     directory: input.directory ?? HOPPER_TARGET_LEASE_DIRECTORY,
     failure: hopperLeaseFailure,
+    resources: input.resources,
   });
 
 const hopperLeaseFailure = (
   failure: LeaseFailure,
   cause?: unknown,
-): HopperStartError =>
-  new HopperStartError({
+): HopperStartError => {
+  const primary =
+    cause instanceof ArtifactFileCleanupFailure &&
+    cause.fileCleanupFailure.outcome.kind === "failed"
+      ? cause.fileCleanupFailure.outcome.cause
+      : undefined;
+  const error = new HopperStartError({
     ...(cause === undefined ? {} : { cause }),
-    userMessage: hopperLeaseMessage(failure),
+    ...(primary instanceof HopperStartError ? { cleanupPrimary: primary } : {}),
+    ...(cause instanceof ArtifactFileCleanupFailure
+      ? { cleanup: cause.cleanup }
+      : {}),
+    userMessage:
+      primary instanceof HopperStartError
+        ? (primary.userMessage ?? hopperLeaseMessage(failure))
+        : hopperLeaseMessage(failure),
   });
+  bindArtifactFileCleanup(cause, error);
+  return error;
+};
 
 const hopperLeaseMessage = (failure: LeaseFailure): string => {
   switch (failure.kind) {
@@ -95,6 +122,8 @@ const hopperLeaseMessage = (failure: LeaseFailure): string => {
       return "REA's Hopper lease directory permissions could not be secured.";
     case "directory_access_denied":
       return `REA could not access its Hopper lease directory at ${JSON.stringify(failure.directory)}. Check its ownership and parent-directory access. For a directory you own, restore owner-only read, write, and search permissions (0700), then retry.`;
+    case "directory_close":
+      return "REA could not confirm closure of its Hopper lease directory handle.";
     case "directory_inspect":
       return "REA could not inspect its private Hopper lease directory.";
     case "recovery_busy":

src/process/OwnedLease.test.ts

@@ -0,0 +1,168 @@
+import { mkdtemp, rm } from "node:fs/promises";
+import { join } from "node:path";
+import { afterEach, describe, expect, it, onTestFinished, vi } from "vitest";
+
+import { ArtifactResourceScope } from "../artifacts/ArtifactResourceScope.js";
+import { HopperStartError } from "../domain/hopperErrors.js";
+const createTestTempDirectory = async (prefix: string): Promise<string> => {
+  const directory = await mkdtemp(join("/tmp", prefix));
+  onTestFinished(() => rm(directory, { recursive: true, force: true }));
+  return directory;
+};
+
+const injection = vi.hoisted(() => ({
+  directory: undefined as string | undefined,
+  admission: "accepted" as "accepted" | "not_private" | "access_denied",
+  invalidateOnClose: false,
+  closeCalls: 0,
+}));
+
+vi.mock("node:fs/promises", async (importOriginal) => {
+  const actual = await importOriginal<typeof import("node:fs/promises")>();
+  return {
+    ...actual,
+    open: async (...args: Parameters<typeof actual.open>) => {
+      const handle = await actual.open(...args);
+      if (args[0] !== injection.directory) return handle;
+      const close = handle.close.bind(handle);
+      vi.spyOn(handle, "close").mockImplementation(async () => {
+        injection.closeCalls += 1;
+        if (injection.closeCalls === 1) {
+          if (injection.invalidateOnClose) await close();
+          throw new Error("injected directory close failure");
+        }
+        await close();
+      });
+      if (injection.admission !== "accepted") {
+        const stat = handle.stat.bind(handle);
+        vi.spyOn(handle, "stat").mockImplementation(async () => {
+          if (injection.admission === "access_denied")
+            throw Object.assign(new Error("denied"), { code: "EACCES" });
+          const observed = await stat();
+          observed.isDirectory = () => false;
+          return observed;
+        });
+      }
+      return handle;
+    },
+  };
+});
+
+const { acquireLinuxHopperApplicationLease } = await import("./HopperLease.js");
+const { HopperApplicationLauncher } =
+  await import("../hopper/BridgeLauncher.js");
+
+afterEach(() => {
+  injection.directory = undefined;
+  injection.admission = "accepted";
+  injection.invalidateOnClose = false;
+  injection.closeCalls = 0;
+  vi.restoreAllMocks();
+});
+
+describe.skipIf(process.platform === "win32")(
+  "lease directory finalization",
+  () => {
+    it.each(["accepted", "not_private", "access_denied"] as const)(
+      "retains failed directory close after admission failure=%s",
+      async (admission) => {
+        const root = await createTestTempDirectory("rea-lease-close-");
+        injection.directory = join(root, "leases");
+        injection.admission = admission;
+        const resources = new ArtifactResourceScope();
+        const failure = await acquireLinuxHopperApplicationLease({
+          runId: "directory-close-test",
+          directory: injection.directory,
+          resources,
+        }).catch((cause: unknown) => cause);
+        expect(failure).toBeInstanceOf(HopperStartError);
+        if (!(failure instanceof HopperStartError)) return;
+        expect(failure.cleanupIncomplete).toBe(true);
+        expect(resources.cleanupResources).toEqual([injection.directory]);
+        expect(failure.cleanupResources).toEqual([injection.directory]);
+        if (admission !== "accepted") {
+          expect(failure.userMessage).toContain(
+            admission === "not_private"
+              ? "not a private directory"
+              : "could not access",
+          );
+          expect(failure.cleanupPrimary).toBeInstanceOf(HopperStartError);
+        } else
+          expect(failure.userMessage).toContain("could not confirm closure");
+        await resources.retryCleanup();
+        expect(injection.closeCalls).toBe(2);
+        expect(failure.cleanupIncomplete).toBe(false);
+        expect(resources.hasPending).toBe(false);
+        injection.directory = undefined;
+        const retried = await acquireLinuxHopperApplicationLease({
+          runId: "directory-close-retry",
+          directory: join(root, "leases"),
+          resources,
+        });
+        expect(retried.acquired).toBe(true);
+        if (retried.acquired) await retried.lease.release();
+      },
+    );
+
+    it("retries the directory owner through the actual Hopper launcher cleanup", async () => {
+      const root = await createTestTempDirectory("rea-launch-lease-");
+      injection.directory = join(root, "leases");
+      injection.admission = "not_private";
+      const launcher = new HopperApplicationLauncher(
+        {
+          environment: {},
+          launcherPath: join(root, "hopper"),
+          targetPath: join(root, "target"),
+          targetKind: "database",
+          loaderArgs: [],
+          bridgeScriptPath: join(root, "bridge.py"),
+          launchMode: "verified_linux_demo",
+          demoHelperPath: join(root, "helper"),
+        },
+        {
+          platform: "linux",
+          acquireApplicationLease: (input) =>
+            acquireLinuxHopperApplicationLease({
+              ...input,
+              directory: join(root, "leases"),
+            }),
+          retryPublisherCleanup: async () => undefined,
+        },
+      );
+      const launched = await launcher.launch({
+        directory: root,
+        socketPath: join(root, "bridge.sock"),
+        token: "fixture-token",
+        runId: "launcher-lease-close-test",
+      });
+      expect(launched.ok).toBe(false);
+      if (launched.ok) return;
+      expect(launched.error.userMessage).toContain("not a private directory");
+      expect(launched.error.cleanupIncomplete).toBe(true);
+      expect(injection.closeCalls).toBe(1);
+      expect((await launcher.cleanup()).ok).toBe(true);
+      expect(injection.closeCalls).toBe(2);
+      expect(launched.error.cleanupIncomplete).toBe(false);
+      expect((await launcher.cleanup()).ok).toBe(true);
+      expect(injection.closeCalls).toBe(2);
+    });
+
+    it("retains uncertainty when Node invalidates the directory handle before rejecting close", async () => {
+      const root = await createTestTempDirectory("rea-lease-invalidated-");
+      injection.directory = join(root, "leases");
+      injection.invalidateOnClose = true;
+      const resources = new ArtifactResourceScope();
+      const failure = await acquireLinuxHopperApplicationLease({
+        runId: "invalidated-close-test",
+        directory: injection.directory,
+        resources,
+      }).catch((cause: unknown) => cause);
+      expect(failure).toBeInstanceOf(HopperStartError);
+      await expect(resources.retryCleanup()).rejects.toThrow();
+      expect(injection.closeCalls).toBe(1);
+      expect(resources.hasPending).toBe(true);
+      if (failure instanceof HopperStartError)
+        expect(failure.cleanupIncomplete).toBe(true);
+    });
+  },
+);

src/process/OwnedLease.ts

@@ -9,6 +9,12 @@ import {
 } from "node:net";
 import { join } from "node:path";
 
+import {
+  ArtifactFileCleanupFailure,
+  withArtifactFileHandle,
+} from "../artifacts/ArtifactFileOperation.js";
+import type { ArtifactResourceScope } from "../artifacts/ArtifactResourceScope.js";
+
 import { safeParseJson } from "../domain/safeJson.js";
 
 /** The process that holds a lease, as that process reports itself. */
@@ -34,6 +40,7 @@ export type LeaseFailure =
   | { readonly kind: "directory_permissions" }
   | { readonly kind: "directory_access_denied"; readonly directory: string }
   | { readonly kind: "directory_inspect" }
+  | { readonly kind: "directory_close" }
   | { readonly kind: "recovery_busy"; readonly path: string }
   | { readonly kind: "recovery_create"; readonly path: string }
   | { readonly kind: "recovery_remove"; readonly path: string }
@@ -59,6 +66,7 @@ export const acquireOwnedLease = async (input: {
   /** Private per-user directory for POSIX sockets. Unused on Windows. */
   readonly directory: string;
   readonly failure: LeaseFailureFactory;
+  readonly resources?: ArtifactResourceScope | undefined;
 }): Promise<LeaseAcquisition> => {
   const produced = new WeakSet<Error>();
   const fail = (failure: LeaseFailure, cause?: unknown): Error => {
@@ -81,7 +89,7 @@ export const acquireOwnedLease = async (input: {
     }
     throw fail({ kind: "unrecoverable" });
   }
-  await ensureLeaseDirectory(input.directory, fail, produced);
+  await ensureLeaseDirectory(input.directory, fail, produced, input.resources);
   const socketPath = join(input.directory, `${key}.sock`);
   const acquisition = await tryAcquireLease(socketPath, owner, fail, true);
   if (typeof acquisition !== "string") return acquisition;
@@ -173,60 +181,76 @@ const ensureLeaseDirectory = async (
   directory: string,
   fail: (failure: LeaseFailure, cause?: unknown) => Error,
   produced: WeakSet<Error>,
+  resources: ArtifactResourceScope | undefined,
 ): Promise<void> => {
   try {
     await mkdir(directory, { recursive: true, mode: 0o700 });
   } catch (cause: unknown) {
     throw fail({ kind: "directory_create" }, cause);
   }
-  // Use O_DIRECTORY|O_NOFOLLOW to open the directory without following symlinks,
-  // then inspect and secure that descriptor. Later socket operations still use
-  // the path and are not made atomic by this directory validation.
-  let handle;
-  try {
-    handle = await open(
-      directory,
-      fsConstants.O_RDONLY |
-        (fsConstants.O_DIRECTORY ?? 0) |
-        (fsConstants.O_NOFOLLOW ?? 0),
-    );
-    const directoryStat = await handle.stat();
-    if (
-      directoryStat.isSymbolicLink() ||
-      directoryStat.isDirectory() === false ||
-      (process.getuid !== undefined && directoryStat.uid !== process.getuid())
-    )
-      throw fail({ kind: "directory_not_private" });
-    // POSIX permission hardening: chmod on the already-opened descriptor avoids
-    // a second TOCTOU window where the directory could be swapped between check
-    // and permission fix.
-    const posixPermissionsApply = process.getuid !== undefined;
-    if (posixPermissionsApply && (directoryStat.mode & 0o077) !== 0) {
-      await handle.chmod(0o700);
-      // Re-verify permissions on the same opened directory.
-      const afterChmod = await handle.stat();
-      if (
-        afterChmod.isSymbolicLink() ||
-        afterChmod.isDirectory() === false ||
-        (afterChmod.mode & 0o077) !== 0
-      )
-        throw fail({ kind: "directory_permissions" });
-    }
-  } catch (cause: unknown) {
-    if (cause instanceof Error && produced.has(cause)) throw cause;
+  const inspectionFailure = (cause: unknown): Error => {
+    if (cause instanceof Error && produced.has(cause)) return cause;
     const accessDenied =
       process.getuid !== undefined &&
       cause instanceof Error &&
       "code" in cause &&
       cause.code === "EACCES";
-    throw fail(
+    return fail(
       accessDenied
         ? { kind: "directory_access_denied", directory }
         : { kind: "directory_inspect" },
       cause,
     );
-  } finally {
-    await handle?.close();
+  };
+  // Use O_DIRECTORY|O_NOFOLLOW to open the directory without following symlinks,
+  // then inspect and secure that descriptor. Later socket operations still use
+  // the path and are not made atomic by this directory validation.
+  try {
+    const handle = await open(
+      directory,
+      fsConstants.O_RDONLY |
+        (fsConstants.O_DIRECTORY ?? 0) |
+        (fsConstants.O_NOFOLLOW ?? 0),
+    );
+    await withArtifactFileHandle(
+      directory,
+      handle,
+      resources,
+      undefined,
+      async (handle) => {
+        try {
+          const directoryStat = await handle.stat();
+          if (
+            directoryStat.isSymbolicLink() ||
+            directoryStat.isDirectory() === false ||
+            (process.getuid !== undefined &&
+              directoryStat.uid !== process.getuid())
+          )
+            throw fail({ kind: "directory_not_private" });
+          // POSIX permission hardening: chmod on the already-opened descriptor avoids
+          // a second TOCTOU window where the directory could be swapped between check
+          // and permission fix.
+          const posixPermissionsApply = process.getuid !== undefined;
+          if (posixPermissionsApply && (directoryStat.mode & 0o077) !== 0) {
+            await handle.chmod(0o700);
+            // Re-verify permissions on the same opened directory.
+            const afterChmod = await handle.stat();
+            if (
+              afterChmod.isSymbolicLink() ||
+              afterChmod.isDirectory() === false ||
+              (afterChmod.mode & 0o077) !== 0
+            )
+              throw fail({ kind: "directory_permissions" });
+          }
+        } catch (cause: unknown) {
+          throw inspectionFailure(cause);
+        }
+      },
+    );
+  } catch (cause: unknown) {
+    if (cause instanceof ArtifactFileCleanupFailure)
+      throw fail({ kind: "directory_close" }, cause);
+    throw inspectionFailure(cause);
   }
 };