fix(artifacts): retain archive streams and source receipts

src/artifacts/GodotPckArtifactReader.test.ts

@@ -9,6 +9,7 @@ import {
 import { mkdir, rename, writeFile } from "node:fs/promises";
 import { join } from "node:path";
 import { text } from "node:stream/consumers";
+import { once } from "node:events";
 
 import { describe, expect, it } from "vitest";
 
@@ -206,6 +207,27 @@ describe("GodotPckArtifactReader", () => {
 });
 
 describe("GodotPckArtifactReader member streams", () => {
+  it.each([8, 128 * 1024])(
+    "retires an unread %i-byte member when its reader closes",
+    async (size) => {
+      await withReader(
+        buildGodotPck(4, [{ path: "member.bin", data: Buffer.alloc(size) }]),
+        async (reader) => {
+          const entries = await list(reader);
+          const stream = await reader.open(entries[0]!);
+          const produced = size === 8 ? once(stream, "finish") : undefined;
+          await once(stream, "readable");
+          if (produced !== undefined) await produced;
+          expect(stream.destroyed).toBe(false);
+          await reader.close();
+          expect(stream.destroyed).toBe(true);
+          expect(stream.closed).toBe(true);
+          await reader.close();
+        },
+      );
+    },
+  );
+
   it("types cancellation during a member read instead of crashing", async () => {
     const data = "x".repeat(4 * 1024 * 1024);
     await withReader(

src/artifacts/GodotPckArtifactReader.ts

@@ -125,12 +125,14 @@ export class GodotPckArtifactReader implements ArtifactReader {
       // retirement before a caller has attached its own error listener.
       output.on("error", () => undefined);
       this.#streams.add(output);
+      // Writable completion can leave an unread member buffered on the readable
+      // side. The returned stream stays owned until that side closes too.
+      output.once("close", () => this.#streams.delete(output));
       void this.#scope
         .run(
           () =>
             new Promise<void>((resolve) => {
               pipeline(source, output, () => {
-                this.#streams.delete(output);
                 resolve();
               });
             }),

src/artifacts/MachOSliceArtifactReader.test.ts

@@ -1,4 +1,4 @@
-import { EventEmitter } from "node:events";
+import { EventEmitter, once } from "node:events";
 import { PassThrough } from "node:stream";
 import { truncate, writeFile } from "node:fs/promises";
 import { join } from "node:path";
@@ -13,6 +13,45 @@ import { openRegularFile } from "../filesystem/RegularFile.js";
 import { ProviderProcessSupervisor } from "../process/ProviderProcess.js";
 import type { NativeCommandRunner } from "../native/CommandRunner.js";
 import { NativeCommandFailure } from "../native/CommandRunner.js";
+import { createDeferred } from "../../tests/fixtures/binarySession.js";
+import { OwnedFileHandle } from "../filesystem/OwnedFileHandle.js";
+import type { ArtifactEntry } from "./ArtifactReader.js";
+
+const sliceOwnershipFixture = async () => {
+  const root = await createTestTempDirectory("rea-slice-ownership-");
+  const path = join(root, "fat");
+  const bytes = Buffer.concat([
+    Buffer.from(machoImage({ cpu: CPU.x86_64 })),
+    Buffer.alloc(256 * 1024),
+  ]);
+  await writeFile(path, fatImage([{ cpu: CPU.x86_64, bytes }]));
+  const runner: NativeCommandRunner = {
+    run: async () =>
+      ok({
+        tool: "lipo",
+        executable: "/usr/bin/lipo",
+        executableSha256: "1".repeat(64),
+        toolVersion: null,
+        versionReason: "fixture",
+        arguments: ["-detailed_info", path],
+        stdout: `architecture x86_64\n cputype 16777223\n cpusubtype 3\n offset 4096\n size ${String(bytes.length)}\n align 2^12 (4096)\n`,
+        stderr: "",
+        stdoutBytes: 1,
+        stderrBytes: 0,
+        exitCode: 0,
+        signal: null,
+      }),
+  };
+  return { path, runner };
+};
+
+const sliceEntries = async (
+  reader: MachOSliceArtifactReader,
+): Promise<ArtifactEntry[]> => {
+  const entries: ArtifactEntry[] = [];
+  for await (const entry of reader.entries()) entries.push(entry);
+  return entries;
+};
 
 const cleanupOwner = (runId: string): ProviderProcessSupervisor => {
   const child = Object.assign(new EventEmitter(), {
@@ -525,3 +564,166 @@ describe("universal slices outside the structural index", () => {
     },
   );
 });
+
+describe("Mach-O slice stream ownership", () => {
+  it.each(["unread", "abandoned", "cancelled"] as const)(
+    "retires an %s opened slice before closing its carrier",
+    async (mode) => {
+      const fixture = await sliceOwnershipFixture();
+      const reader = new MachOSliceArtifactReader(
+        fixture.path,
+        {},
+        fixture.runner,
+      );
+      try {
+        const [entry] = await sliceEntries(reader);
+        const caller = new AbortController();
+        const stream = await reader.open(entry!, caller.signal);
+        await once(stream, "readable");
+        if (mode === "abandoned") {
+          for await (const _chunk of stream) break;
+        } else if (mode === "cancelled") {
+          const failure = once(stream, "error");
+          const reason = { caller: "slice owner cancellation" };
+          caller.abort(reason);
+          await expect(failure).resolves.toMatchObject([
+            { reason: "cancelled", cause: reason },
+          ]);
+        }
+        await Promise.all([reader.close(), reader.close()]);
+        expect(stream.destroyed).toBe(true);
+        expect(stream.closed).toBe(true);
+        await reader.close();
+      } finally {
+        await reader.close();
+      }
+    },
+  );
+
+  it("joins an admitted descriptor read before reader close completes", async () => {
+    const fixture = await sliceOwnershipFixture();
+    const source = await openRegularFile(fixture.path, { symlinks: "reject" });
+    const reader = new MachOSliceArtifactReader(
+      fixture.path,
+      {},
+      fixture.runner,
+      source,
+    );
+    const reached = createDeferred<void>();
+    const release = createDeferred<void>();
+    const nativeRead = source.handle.read.bind(source.handle);
+    let reads: { mockRestore(): void } | undefined;
+    try {
+      const [entry] = await sliceEntries(reader);
+      reads = vi.spyOn(source.handle, "read").mockImplementation((async (
+        ...args: unknown[]
+      ) => {
+        reached.resolve();
+        await release.promise;
+        return Reflect.apply(nativeRead, source.handle, args);
+      }) as typeof source.handle.read);
+      const stream = await reader.open(entry!);
+      stream.read(0);
+      await reached.promise;
+      let closed = false;
+      const closing = reader.close().then(() => {
+        closed = true;
+      });
+      await Promise.resolve();
+      expect(closed).toBe(false);
+      release.resolve();
+      await closing;
+      expect(stream.closed).toBe(true);
+    } finally {
+      release.resolve();
+      await reader.close();
+      reads?.mockRestore();
+      await source.handle.close();
+    }
+  });
+
+  it("keeps descriptor closure retryable after retiring its slice", async () => {
+    const fixture = await sliceOwnershipFixture();
+    const reader = new MachOSliceArtifactReader(
+      fixture.path,
+      {},
+      fixture.runner,
+    );
+    const [entry] = await sliceEntries(reader);
+    const stream = await reader.open(entry!);
+    const failure = new Error("Mach-O descriptor close denied");
+    const close = vi
+      .spyOn(OwnedFileHandle.prototype, "close")
+      .mockRejectedValueOnce(failure);
+    try {
+      await expect(reader.close()).rejects.toBe(failure);
+      expect(stream.closed).toBe(true);
+      await reader.close();
+      expect(close).toHaveBeenCalledTimes(2);
+    } finally {
+      close.mockRestore();
+      await reader.close();
+    }
+  });
+
+  it.runIf(process.platform === "darwin")(
+    "retires a slice discovered by the installed lipo",
+    async () => {
+      const fixture = await sliceOwnershipFixture();
+      const reader = new MachOSliceArtifactReader(fixture.path, process.env);
+      try {
+        const [entry] = await sliceEntries(reader);
+        const stream = await reader.open(entry!);
+        await once(stream, "readable");
+        await reader.close();
+        expect(stream.closed).toBe(true);
+        expect(reader.provenance()).toMatchObject([
+          { tool: "lipo", exit_code: 0 },
+        ]);
+      } finally {
+        await reader.close();
+      }
+    },
+  );
+});
+
+it("joins Mach-O structural enumeration reads admitted before close", async () => {
+  const fixture = await sliceOwnershipFixture();
+  const source = await openRegularFile(fixture.path, { symlinks: "reject" });
+  const reader = new MachOSliceArtifactReader(
+    fixture.path,
+    {},
+    fixture.runner,
+    source,
+  );
+  const reached = createDeferred<void>();
+  const release = createDeferred<void>();
+  const nativeRead = source.handle.read.bind(source.handle);
+  const reads = vi.spyOn(source.handle, "read").mockImplementation((async (
+    ...args: unknown[]
+  ) => {
+    reached.resolve();
+    await release.promise;
+    return Reflect.apply(nativeRead, source.handle, args);
+  }) as typeof source.handle.read);
+  const traversal = reader.entries()[Symbol.asyncIterator]();
+  const next = traversal.next();
+  try {
+    await reached.promise;
+    let closed = false;
+    const closing = reader.close().then(() => {
+      closed = true;
+    });
+    await Promise.resolve();
+    expect(closed).toBe(false);
+    release.resolve();
+    await expect(next).rejects.toMatchObject({ reason: "unavailable" });
+    await closing;
+  } finally {
+    release.resolve();
+    await next.catch(() => undefined);
+    await reader.close();
+    reads.mockRestore();
+    await source.handle.close();
+  }
+});

src/artifacts/MachOSliceArtifactReader.ts

@@ -36,7 +36,12 @@ export class MachOSliceArtifactReader implements ArtifactReader {
   #command: ArtifactCommand | undefined;
   readonly #source: ArtifactFileSource;
   readonly #entries = new Map<string, ArtifactEntry>();
-  readonly #pendingCommands = new Set<Promise<unknown>>();
+  readonly #enumerations = new Set<AsyncGenerator<ArtifactEntry, void>>();
+  readonly #pendingOpens = new Set<Promise<Readable>>();
+  readonly #streams = new Map<
+    Readable,
+    { readonly source: Readable; readonly completion: Promise<unknown> }
+  >();
   #closed = false;
   #closePromise: Promise<void> | undefined;
 
@@ -52,6 +57,18 @@ export class MachOSliceArtifactReader implements ArtifactReader {
   }
 
   async *entries(signal?: AbortSignal): AsyncIterable<ArtifactEntry> {
+    const enumeration = this.#readEntries(signal);
+    this.#enumerations.add(enumeration);
+    try {
+      yield* enumeration;
+    } finally {
+      this.#enumerations.delete(enumeration);
+    }
+  }
+
+  async *#readEntries(
+    signal?: AbortSignal,
+  ): AsyncGenerator<ArtifactEntry, void> {
     this.#entries.clear();
     const source = await this.#ensureSource();
     await this.#source.readSelectedPath();
@@ -65,13 +82,7 @@ export class MachOSliceArtifactReader implements ArtifactReader {
       ["-detailed_info", this.path],
       signal === undefined ? {} : { signal },
     );
-    this.#pendingCommands.add(command);
-    let captured: Awaited<typeof command>;
-    try {
-      captured = await command;
-    } finally {
-      this.#pendingCommands.delete(command);
-    }
+    const captured = await command;
     if (!captured.ok)
       throw new ArtifactReaderFailure(
         captured.error.reason === "cancelled" ? "cancelled" : "unavailable",
@@ -123,6 +134,11 @@ export class MachOSliceArtifactReader implements ArtifactReader {
       signal,
     );
     await this.#source.readSelectedPath();
+    if (this.#closed)
+      throw new ArtifactReaderFailure(
+        "unavailable",
+        "Universal Mach-O reader is closed",
+      );
     if (structural.status === "malformed")
       throw new ArtifactReaderFailure(
         "integrity",
@@ -272,7 +288,15 @@ export class MachOSliceArtifactReader implements ArtifactReader {
     }
   }
 
-  async open(entry: ArtifactEntry, signal?: AbortSignal): Promise<Readable> {
+  open(entry: ArtifactEntry, signal?: AbortSignal): Promise<Readable> {
+    const operation = this.#open(entry, signal);
+    this.#pendingOpens.add(operation);
+    const release = () => this.#pendingOpens.delete(operation);
+    void operation.then(release, release);
+    return operation;
+  }
+
+  async #open(entry: ArtifactEntry, signal?: AbortSignal): Promise<Readable> {
     if (signal?.aborted === true)
       return Promise.reject(
         new ArtifactReaderFailure("cancelled", "Mach-O slice read cancelled", {
@@ -301,6 +325,11 @@ export class MachOSliceArtifactReader implements ArtifactReader {
       );
     const source = await this.#ensureSource();
     await this.#source.readSelectedPath();
+    if (this.#closed)
+      throw new ArtifactReaderFailure(
+        "unavailable",
+        "Universal Mach-O reader is closed",
+      );
     if (
       !Number.isSafeInteger(offset + size) ||
       offset + size > source.initial.size
@@ -369,6 +398,15 @@ export class MachOSliceArtifactReader implements ArtifactReader {
         }
       })(),
     );
+    output.on("error", () => undefined);
+    const completion = Promise.all(
+      [sourceStream, output].map(
+        (stream) =>
+          new Promise<void>((resolve) => stream.once("close", resolve)),
+      ),
+    );
+    this.#streams.set(output, { source: sourceStream, completion });
+    void completion.then(() => this.#streams.delete(output));
     const onSourceError = (cause: unknown): void => {
       const completedCancellation =
         cause instanceof ArtifactReaderFailure &&
@@ -428,7 +466,25 @@ export class MachOSliceArtifactReader implements ArtifactReader {
   }
 
   async #releaseResources(): Promise<void> {
-    await Promise.allSettled(this.#pendingCommands);
+    for (const [output, { source }] of this.#streams) {
+      output.destroy(
+        new ArtifactReaderFailure(
+          "unavailable",
+          "Universal Mach-O reader closed",
+        ),
+      );
+      source.destroy();
+    }
+    const outcomes = await Promise.allSettled([
+      // Failed opens remain their caller's primary result, not a cleanup error.
+      Promise.allSettled(this.#pendingOpens),
+      ...Array.from(this.#enumerations, (enumeration) => enumeration.return()),
+      ...Array.from(this.#streams.values(), ({ completion }) => completion),
+    ]);
+    this.#enumerations.clear();
+    const failures = outcomes.flatMap((outcome) =>
+      outcome.status === "rejected" ? [outcome.reason as unknown] : [],
+    );
     let cleanup: ArtifactReaderFailure["cleanup"];
     try {
       const failures = (await this.runner.retryCleanup?.()) ?? [];
@@ -449,6 +505,16 @@ export class MachOSliceArtifactReader implements ArtifactReader {
         "native-command-process-cleanup",
       );
     }
+    for (const cause of failures)
+      cleanup = {
+        reason: [
+          cleanup?.reason,
+          ArtifactReaderFailure.cleanupObservation(cause, this.path).reason,
+        ]
+          .filter((reason) => reason !== undefined)
+          .join("; "),
+        resources: [...new Set([...(cleanup?.resources ?? []), this.path])],
+      };
     try {
       await this.#source.close();
     } catch (cause: unknown) {
@@ -458,7 +524,7 @@ export class MachOSliceArtifactReader implements ArtifactReader {
     if (cleanup !== undefined)
       throw new ArtifactReaderFailure(
         "unavailable",
-        "Native command process cleanup remains incomplete",
+        "Universal Mach-O reader cleanup remains incomplete",
         { cleanup },
       );
   }

src/artifacts/inventory/classify.ts

@@ -22,11 +22,11 @@ import {
   zipRootPackageFormat,
 } from "../../domain/zipPackageFormat.js";
 import { ArtifactReaderFailure } from "../ArtifactReader.js";
-import type { ArtifactResourceOwner } from "../ArtifactResourceScope.js";
 import type { HashResult } from "../ArtifactHash.js";
 import {
   hashStableRootArtifactHandle,
   openRootArtifact,
+  releaseRootArtifact,
 } from "./hashStableRootArtifact.js";
 import type { ArtifactInventoryOptions } from "./types.js";
 import { readMachoHeaderPrefixFacts } from "../../domain/apple/machoContainer.js";
@@ -78,11 +78,6 @@ export const classifyAndHashRootForInventory = async (
     options.signal,
   );
   const { handle, initial } = source;
-  const owner: ArtifactResourceOwner = {
-    kind: "file-handle",
-    handle: source.owner,
-    resource: `root artifact descriptor for ${path}`,
-  };
   let outcome:
     | {
         readonly kind: "completed";
@@ -133,9 +128,15 @@ export const classifyAndHashRootForInventory = async (
     outcome = { kind: "failed", cause };
   }
   if (outcome.kind === "failed" || outcome.value.rootSource === undefined) {
-    const cleanupAttempt = await options.resourceScope.release(owner);
-    if (cleanupAttempt.kind === "failed")
-      throw options.resourceScope.cleanupFailure(outcome, [cleanupAttempt]);
+    await releaseRootArtifact(
+      path,
+      source.owner,
+      options.resourceScope,
+      outcome.kind === "completed"
+        ? { kind: "completed", digest: outcome.value.digest }
+        : outcome,
+      options.signal,
+    );
   }
   if (outcome.kind === "failed") throw outcome.cause;
   return outcome.value;

src/artifacts/inventory/hashStableRootArtifact.ts

@@ -19,6 +19,47 @@ import {
   type HashResult,
 } from "../ArtifactHash.js";
 import type { ArtifactResourceScope } from "../ArtifactResourceScope.js";
+import { fileSnapshotPartialObservationSchema } from "../../domain/fileSnapshotObservation.js";
+
+/** Finalize the admitted root while preserving its completed identity and exact owner. */
+export const releaseRootArtifact = async (
+  path: string,
+  owner: OwnedFileHandle,
+  resourceScope: ArtifactResourceScope,
+  outcome:
+    | { readonly kind: "completed"; readonly digest: HashResult | null }
+    | { readonly kind: "failed"; readonly cause: unknown },
+  signal?: AbortSignal,
+): Promise<void> => {
+  const cleanup = await resourceScope.release({
+    kind: "file-handle",
+    resource: `root artifact descriptor for ${path}`,
+    handle: owner,
+  });
+  const observation =
+    outcome.kind === "completed" && outcome.digest !== null
+      ? fileSnapshotPartialObservationSchema.parse({
+          kind: "file-snapshot-hash",
+          path,
+          sha256: outcome.digest.sha256,
+          bytes: outcome.digest.bytes,
+        })
+      : undefined;
+  const finalOutcome =
+    outcome.kind === "completed" && signal?.aborted
+      ? {
+          kind: "failed" as const,
+          cause: new ArtifactReaderFailure(
+            "cancelled",
+            "Root artifact inventory cancelled during descriptor finalization",
+            { cause: signal.reason, partialObservation: observation },
+          ),
+        }
+      : outcome;
+  if (cleanup.kind === "failed")
+    throw resourceScope.cleanupFailure(finalOutcome, [cleanup], observation);
+  if (finalOutcome.kind === "failed") throw finalOutcome.cause;
+};
 
 /** Admit a root descriptor without losing ownership when validation cleanup fails. */
 export const openRootArtifact = async (
@@ -87,14 +128,15 @@ export const hashStableRootArtifact = async (
   } catch (cause: unknown) {
     outcome = { kind: "failed", cause };
   }
-  const resource = `root artifact descriptor for ${path}`;
-  const cleanup = await resourceScope.release({
-    kind: "file-handle",
-    resource,
-    handle: owner,
-  });
-  if (cleanup.kind === "failed")
-    throw resourceScope.cleanupFailure(outcome, [cleanup]);
+  await releaseRootArtifact(
+    path,
+    owner,
+    resourceScope,
+    outcome.kind === "completed"
+      ? { kind: "completed", digest: outcome.value }
+      : outcome,
+    signal,
+  );
   if (outcome.kind === "failed") throw outcome.cause;
   return outcome.value;
 };

tests/boundary/filesystem/artifactInventoryDescriptorCleanup.test.ts

@@ -9,6 +9,7 @@ import { ArtifactProvider } from "../../../src/artifacts/ArtifactProvider.js";
 import { classifyAndHashRoot } from "../../fixtures/artifactInventory.js";
 import { asarHeaderBytes } from "../../fixtures/asarHeader.js";
 import { createTestTempDirectory } from "../../fixtures/temporaryDirectory.js";
+import { projectAnalysisError } from "../../../src/domain/analysisErrorProjection.js";
 
 const injected = vi.hoisted(() => ({
   opened: undefined as
@@ -28,6 +29,98 @@ vi.mock("node:fs/promises", async (importOriginal) => {
   };
 });
 
+it.each(
+  [1, 2].flatMap((selectedOpen) =>
+    [
+      "failed close",
+      "cancelled failed close",
+      "cancelled closed",
+      "failed read",
+    ].map((outcome) => ({ selectedOpen, outcome })),
+  ),
+)(
+  "retains completed root identity at open $selectedOpen through $outcome",
+  async ({ selectedOpen, outcome }) => {
+    const root = await realpath(
+      await createTestTempDirectory("rea-root-completed-cleanup-"),
+    );
+    const path = join(root, "payload.bin");
+    const bytes = Buffer.from("completed selected root bytes");
+    const sha256 = createHash("sha256").update(bytes).digest("hex");
+    await writeFile(path, bytes);
+    const client = new ArtifactProvider({}).createClient({
+      path,
+      sourcePath: path,
+      sha256,
+      kind: "artifact",
+      format: "javascript",
+    });
+    const controller = new AbortController();
+    const reason = { source: "root descriptor finalization" };
+    const physical = Object.assign(new Error("physical root read failed"), {
+      code: "EIO",
+      errno: -5,
+      syscall: "read",
+    });
+    let opened = 0;
+    let descriptor: FileHandle | undefined;
+    let blocked = outcome !== "cancelled closed";
+    injected.opened = (candidate, file) => {
+      if (String(candidate) !== path || ++opened !== selectedOpen) return;
+      descriptor = file;
+      if (outcome === "failed read")
+        vi.spyOn(file, "read").mockRejectedValueOnce(physical);
+      const close = file.close.bind(file);
+      vi.spyOn(file, "close").mockImplementation(async () => {
+        if (outcome !== "failed close") controller.abort(reason);
+        if (blocked) throw new Error("root descriptor close denied");
+        await close();
+      });
+    };
+    onTestFinished(async () => {
+      injected.opened = undefined;
+      blocked = false;
+      await client.close();
+      vi.restoreAllMocks();
+    });
+    const result = await client.execute(
+      "inventory_artifact",
+      {},
+      { signal: controller.signal },
+    );
+    if (result.ok) throw new Error("Expected root finalization failure");
+    expect(result.error.partialObservation).toEqual(
+      outcome === "failed read"
+        ? undefined
+        : { kind: "file-snapshot-hash", path, sha256, bytes: bytes.length },
+    );
+    expect(
+      projectAnalysisError(result.error).details?.partial_observation,
+    ).toEqual(result.error.partialObservation);
+    if (outcome.startsWith("cancelled")) {
+      expect(result.error).toMatchObject({
+        reason: "cancelled",
+        cause: reason,
+      });
+    } else if (outcome === "failed read") {
+      expect(result.error).toMatchObject({
+        reason: "io",
+        detail: expect.stringContaining(physical.message),
+        cause: { cause: { cause: physical } },
+      });
+    }
+    expect(result.error.cleanupIncomplete).toBe(blocked);
+    if (blocked) {
+      expect((await client.close()).ok).toBe(false);
+      expect(descriptor!.fd).toBeGreaterThanOrEqual(0);
+      blocked = false;
+    }
+    expect((await client.close()).ok).toBe(true);
+    expect(descriptor!.fd).toBe(-1);
+    expect(result.error.cleanupIncomplete).toBe(false);
+  },
+);
+
 it.each([false, true])(
   "preserves root admission failure and close authority (invalidated=%s)",
   async (invalidated) => {