fix(process): preserve completed observations and cleanup authority

Retain physical stream errors and late caller cancellation independently. Expose an exact failed Darwin preparation root for cleanup retry without closing the shared reader.

Validation: 24 owned-command tests, 6 cancellation controls, 2 targeted Darwin cleanup retry controls, and check:fast. Controlled macOS processes only; no new real JVM or Ghidra verification.

Refs #4561, #4638

src/process/DarwinProcessRunTokenReader.ts

@@ -111,11 +111,16 @@ const STDERR_EXCERPT_LIMIT = 512;
 
 /** The native Darwin process-ownership reader could not be prepared. */
 export class DarwinProcessOwnershipInspectionError extends Error {
+  readonly #cleanupOwner: { close(): Promise<void> } | undefined;
+  get cleanupOwner(): { close(): Promise<void> } | undefined {
+    return this.#cleanupOwner;
+  }
   constructor(
     message: string,
     options?: ErrorOptions & {
       readonly compilerOutcome?: DarwinCompilerOutcome;
       readonly capturedOutput?: AnalysisCapturedOutput;
+      readonly cleanupOwner?: { close(): Promise<void> } | undefined;
       readonly cleanupFailure?: AnalysisError;
       readonly cancellation?: { readonly reason: unknown };
     },
@@ -124,6 +129,7 @@ export class DarwinProcessOwnershipInspectionError extends Error {
       compilerOutcome,
       capturedOutput,
       cleanupFailure,
+      cleanupOwner,
       cancellation,
       ...errorOptions
     } = options ?? {};
@@ -131,6 +137,7 @@ export class DarwinProcessOwnershipInspectionError extends Error {
     this.name = "DarwinProcessOwnershipInspectionError";
     this.compilerOutcome = compilerOutcome;
     this.capturedOutput = capturedOutput;
+    this.#cleanupOwner = cleanupOwner;
     this.cleanupFailure = cleanupFailure;
     this.cancellation = cancellation;
   }
@@ -143,6 +150,7 @@ export class DarwinProcessOwnershipInspectionError extends Error {
   /** Preserve the physical preparation failure while following the exact cleanup owner. */
   withCleanupFailure(
     cleanup: AnalysisError,
+    owner?: { close(): Promise<void> },
   ): DarwinProcessOwnershipInspectionError {
     const combined =
       this.cleanupFailure === undefined || this.cleanupFailure === cleanup
@@ -160,6 +168,7 @@ export class DarwinProcessOwnershipInspectionError extends Error {
         ? {}
         : { capturedOutput: this.capturedOutput }),
       cleanupFailure: combined,
+      cleanupOwner: owner ?? this.cleanupOwner,
       ...(this.cancellation === undefined
         ? {}
         : { cancellation: this.cancellation }),
@@ -504,7 +513,9 @@ export const createDarwinProcessRunTokenReader = (
                   "Darwin process ownership helper preparation failed",
                   { cause },
                 );
-          throw primary.withCleanupFailure(cleanup);
+          throw primary.withCleanupFailure(cleanup, {
+            close: () => cleanupRoot(operationRoot),
+          });
         }
       }
       throw cause;
@@ -578,6 +589,7 @@ export const createDarwinProcessRunTokenReader = (
               cause: waited.reason,
               cancellation: { reason: waited.reason },
               cleanupFailure: finalized.error.cleanupFailure,
+              cleanupOwner: finalized.error.cleanupOwner,
             },
           );
         }

src/process/OwnedCommand.ts

@@ -31,6 +31,7 @@ export class OwnedCommandFailure extends Error {
   get cancellation(): { readonly reason: unknown } | undefined {
     return this.#cancellation;
   }
+  readonly processError: Error | undefined;
   readonly cleanupResult: ProcessCleanupFailure | undefined;
   readonly collectionFailure: boolean;
   readonly preparationFailure:
@@ -44,6 +45,7 @@ export class OwnedCommandFailure extends Error {
     readonly cleanupFailure: string | null = null,
     options?: ErrorOptions & {
       readonly cancellation?: { readonly reason: unknown } | undefined;
+      readonly processError?: Error | undefined;
       readonly cleanupResult?: ProcessCleanupFailure;
       readonly collectionFailure?: boolean;
       readonly preparationFailure?: DarwinProcessOwnershipInspectionError;
@@ -56,6 +58,7 @@ export class OwnedCommandFailure extends Error {
       options?.cancellation === undefined
         ? undefined
         : { reason: options.cancellation.reason };
+    this.processError = options?.processError;
     this.cleanupResult = options?.cleanupResult;
     this.collectionFailure = options?.collectionFailure ?? false;
     this.preparationFailure = options?.preparationFailure;
@@ -79,6 +82,8 @@ interface OwnedCommandLimits {
   readonly diagnosticBytes: number;
   /** Accept an ordinary nonzero exit after output closes without other process errors. */
   readonly acceptNonZeroExit?: boolean;
+  /** Interpret complete physical output synchronously before awaited final cleanup. */
+  readonly observeCompleted?: (snapshot: ProviderProcessSnapshot) => void;
   /**
    * Receives stdout instead of the retained snapshot, for output parsed as it
    * arrives. The command completes only after the sink finishes. The sink
@@ -184,6 +189,7 @@ const collectOwnedCommand = async (
   }
   let exceeded = false;
   let processFailure: string | undefined;
+  let processError: Error | undefined;
   const supervisor = new ProviderProcessSupervisor(
     {
       ...launched,
@@ -198,7 +204,10 @@ const collectOwnedCommand = async (
       onDiagnostic: (event) => {
         if (event.type === "output" && event.truncated === true)
           exceeded = true;
-        if (event.type === "error") processFailure = event.message;
+        if (event.type === "error") {
+          processFailure ??= event.message;
+          processError ??= event.cause;
+        }
       },
     },
   );
@@ -235,6 +244,7 @@ const collectOwnedCommand = async (
         );
       if (processFailure !== undefined)
         throw new OwnedCommandFailure("process", processFailure, null, null, {
+          processError,
           collectionFailure: true,
         });
       await waitForAbortableDelay(25, signal);
@@ -279,8 +289,9 @@ const collectOwnedCommand = async (
           "Command failed; inspect its exit status and diagnostics.",
         snapshot,
         null,
-        { collectionFailure: processFailure !== undefined },
+        { processError, collectionFailure: processFailure !== undefined },
       );
+    limits.observeCompleted?.(snapshot);
   } catch (cause: unknown) {
     failure =
       cause instanceof OwnedCommandFailure
@@ -297,6 +308,9 @@ const collectOwnedCommand = async (
   abandoned.abort();
   const stopped = await supervisor.stop();
   const snapshot = supervisor.snapshot();
+  const cancellation =
+    failure?.cancellation ??
+    (callerSignal?.aborted ? { reason: callerSignal.reason } : undefined);
   if (stopped.status === "incomplete" && retainCleanup !== undefined)
     retainCleanup(supervisor, stopped);
   else if (stopped.status === "incomplete")
@@ -306,9 +320,10 @@ const collectOwnedCommand = async (
       snapshot,
       stopped.reason,
       {
+        cancellation,
         ...(failure === undefined
           ? {}
-          : { cause: failure, cancellation: failure.cancellation }),
+          : { cause: failure, processError: failure.processError }),
         ...(stopped.cleanupResult === undefined
           ? {}
           : { cleanupResult: stopped.cleanupResult }),
@@ -327,7 +342,8 @@ const collectOwnedCommand = async (
       failure.cleanupFailure,
       {
         cause: failure,
-        cancellation: failure.cancellation,
+        cancellation,
+        processError: failure.processError,
         ...(failure.cleanupResult === undefined
           ? {}
           : { cleanupResult: failure.cleanupResult }),

src/process/ProviderProcess.ts

@@ -163,7 +163,11 @@ export type ProviderProcessDiagnostic =
       readonly signal: NodeJS.Signals | null;
       readonly snapshot: ProviderProcessSnapshot;
     }
-  | { readonly type: "error"; readonly message: string };
+  | {
+      readonly type: "error";
+      readonly message: string;
+      readonly cause?: Error;
+    };
 
 /** Options for process lifecycle diagnostics. */
 export interface ProviderProcessSupervisorOptions {
@@ -329,7 +333,11 @@ export class ProviderProcessSupervisor {
     this.#outputClose.resolve(undefined);
   };
   readonly #onError = (cause: Error): void => {
-    this.#options.onDiagnostic?.({ type: "error", message: cause.message });
+    this.#options.onDiagnostic?.({
+      type: "error",
+      message: cause.message,
+      cause,
+    });
   };
   #exitObservation: ProviderProcessExitObservation | undefined;
   #stopPromise: Promise<ProviderProcessStopResult> | undefined;
@@ -656,6 +664,7 @@ export class ProviderProcessSupervisor {
     this.#options.onDiagnostic?.({
       type: "error",
       message: `${stream} stream failed: ${cause.message}`,
+      cause,
     });
   }
 

tests/boundary/process/darwinProcessRunTokenReader.test.ts

@@ -829,9 +829,11 @@ it.skipIf(process.platform === "win32").each(["first", "second"] as const)(
   },
 );
 
-it.skipIf(process.platform === "win32" || process.getuid?.() === 0)(
-  "retains a failed compilation cleanup for a later close retry",
-  async () => {
+it
+  .skipIf(process.platform === "win32" || process.getuid?.() === 0)
+  .each(["reader-close", "exact-root"] as const)(
+  "retains a failed compilation cleanup for a later %s retry",
+  async (retry) => {
     const directory = await mkdtemp(
       join(tmpdir(), "rea-process-token-cleanup-retry-test-"),
     );
@@ -854,10 +856,31 @@ it.skipIf(process.platform === "win32" || process.getuid?.() === 0)(
       );
       await chmod(state.root, 0o000);
       controller.abort();
-      expect((await preparation).state).toBe("rejected");
-      await expect(reader.close()).rejects.toThrow();
-      await chmod(state.root, 0o700);
-      await reader.close();
+      const outcome = await preparation;
+      expect(outcome.state).toBe("rejected");
+      if (retry === "reader-close") {
+        await expect(reader.close()).rejects.toThrow();
+        await chmod(state.root, 0o700);
+        await reader.close();
+      } else {
+        if (
+          outcome.state !== "rejected" ||
+          !(outcome.cause instanceof DarwinProcessOwnershipInspectionError)
+        )
+          throw new Error("Expected owned compilation cleanup failure");
+        const failure = outcome.cause;
+        expect(failure.cleanupOwner).toBeDefined();
+        expect(failure.cleanupFailure?.cleanupIncomplete).toBe(true);
+        await expect(failure.cleanupOwner?.close()).rejects.toThrow();
+        await chmod(state.root, 0o700);
+        await failure.cleanupOwner?.close();
+        expect(failure.cleanupFailure?.cleanupIncomplete).toBe(false);
+        // Releasing the failed root does not close the shared reader itself.
+        const stopped = new AbortController();
+        const reason = { request: "probe reader remains open" };
+        stopped.abort(reason);
+        await expect(reader.prepare(stopped.signal)).rejects.toBe(reason);
+      }
       await expect(access(state.root)).rejects.toThrow();
     } finally {
       if (!controller.signal.aborted) controller.abort();

tests/boundary/process/ownedCommand.test.ts

@@ -103,6 +103,99 @@ it("collects a real owned command and verifies its exit before returning diagnos
   });
 });
 
+it.each([0, 2])(
+  "observes complete accepted output once before final cleanup (exit %s)",
+  async (exitCode) => {
+    const stages: string[] = [];
+    const snapshot = await runOwnedCommand(
+      command(
+        `process.stdout.write("complete"); process.exitCode = ${exitCode}`,
+      ),
+      {
+        timeoutMs: 2000,
+        diagnosticBytes: 1024,
+        acceptNonZeroExit: true,
+        observeCompleted: (observed) => {
+          stages.push("observed");
+          expect(observed.exitCode).toBe(exitCode);
+          expect(observed.stdout.text).toBe("complete");
+        },
+      },
+      {
+        launcher: async (input) => {
+          const acquired = await spawnOwnedProviderProcess(input);
+          return {
+            ...acquired,
+            cleanup: async () => {
+              stages.push("cleanup");
+              return cleanupOwnedProcessGroup(acquired.ownership);
+            },
+          };
+        },
+      },
+    );
+    expect(snapshot.exitCode).toBe(exitCode);
+    expect(stages).toEqual(["observed", "cleanup"]);
+  },
+);
+
+it("does not interpret output rejected by its complete-output budget", async () => {
+  const observeCompleted = vi.fn();
+  await expect(
+    runOwnedCommand(command('process.stdout.write("x".repeat(2048))'), {
+      timeoutMs: 2000,
+      diagnosticBytes: 1024,
+      observeCompleted,
+    }),
+  ).rejects.toMatchObject({ reason: "output-limit" });
+  expect(observeCompleted).not.toHaveBeenCalled();
+});
+
+it("retains late cancellation separately from a physical failure with refused cleanup", async () => {
+  const controller = new AbortController();
+  const callerReason = { request: "cancel during refused finalization" };
+  let denied = true;
+  let failure: OwnedCommandFailure | undefined;
+  try {
+    await runOwnedCommand(
+      command("process.exit(2)"),
+      { timeoutMs: 2000, diagnosticBytes: 1024 },
+      {
+        signal: controller.signal,
+        launcher: async (input) => {
+          const acquired = await spawnOwnedProviderProcess(input);
+          return {
+            ...acquired,
+            cleanup: async () => {
+              if (denied) {
+                controller.abort(callerReason);
+                return { cleaned: false, reason: "controlled cleanup refusal" };
+              }
+              return cleanupOwnedProcessGroup(acquired.ownership);
+            },
+          };
+        },
+      },
+    ).catch((cause: unknown) => {
+      if (!(cause instanceof OwnedCommandFailure)) throw cause;
+      failure = cause;
+    });
+    expect(failure).toMatchObject({
+      reason: "process",
+      snapshot: { exitCode: 2 },
+      cleanupFailure: "controlled cleanup refusal",
+    });
+    expect(failure?.cancellation?.reason).toBe(callerReason);
+    denied = false;
+    expect(await failure?.cleanupOwner?.stop()).toEqual({
+      status: "verified-cleanup",
+    });
+  } finally {
+    denied = false;
+    await failure?.cleanupOwner?.stop();
+  }
+});
+
 it("streams stdout beyond the diagnostic budget into a sink and returns after it finishes", async () => {
   let received = 0;
   let finished = false;
@@ -175,6 +268,9 @@ it("reports a failed stdout sink as a collection failure", async () => {
 it("fails an accepted nonzero command when an output stream errors and cleans it up", async () => {
   const stdout = new PassThrough();
   const stderr = new PassThrough();
+  const readFailure = Object.assign(new Error("fixture read failure"), {
+    code: "EIO",
+  });
   class FakeProcess extends EventEmitter {
     readonly stdout = stdout;
     readonly stderr = stderr;
@@ -199,7 +295,7 @@ it("fails an accepted nonzero command when an output stream errors and cleans it
       launcher: async () => {
         setImmediate(() => {
           stdout.write("partial output");
-          stdout.destroy(new Error("fixture read failure"));
+          stdout.destroy(readFailure);
           stderr.end();
           child.exitCode = 7;
           child.emit("exit", 7, null);
@@ -223,6 +319,7 @@ it("fails an accepted nonzero command when an output stream errors and cleans it
     message: "stdout stream failed: fixture read failure",
     snapshot: { stdout: { text: "partial output" }, exitCode: 7 },
   });
+  await expect(failure).rejects.toHaveProperty("processError", readFailure);
   expect(cleanup).toHaveBeenCalledOnce();
 });
 
@@ -286,6 +383,9 @@ it.skipIf(process.platform === "win32")(
       hostEnvironment: process.env,
     });
     let descendantPid: number | undefined;
+    let descendantOwner:
+      | Awaited<ReturnType<typeof spawnOwnedProviderProcess>>
+      | undefined;
     try {
       const result = await runOwnedCommand(
         {
@@ -302,6 +402,12 @@ it.skipIf(process.platform === "win32")(
           sweepTokenOwnedProcesses: true,
         },
         { timeoutMs: 5_000, diagnosticBytes: 1024 },
+        {
+          launcher: async (input) => {
+            descendantOwner = await spawnOwnedProviderProcess(input);
+            return descendantOwner;
+          },
+        },
       );
       descendantPid = Number(result.stdout.text);
       expect(Number.isSafeInteger(descendantPid)).toBe(true);
@@ -316,13 +422,13 @@ it.skipIf(process.platform === "win32")(
       expect(remainsLive).toBe(false);
       expect(() => process.kill(unrelated.process.pid!, 0)).not.toThrow();
     } finally {
-      if (descendantPid !== undefined) {
-        try {
-          process.kill(descendantPid, "SIGKILL");
-        } catch {
-          // The run-token sweep may already have removed it.
-        }
-      }
+      if (descendantOwner !== undefined)
+        expect(
+          (
+            await (descendantOwner.cleanup?.() ??
+              cleanupOwnedProcessGroup(descendantOwner.ownership))
+          ).cleaned,
+        ).toBe(true);
       await cleanupOwnedProcessGroup(unrelated.ownership);
     }
   },