fix(process): retain snapshot handles and partial read outcomes

This commit is contained in:
morluto
2026-10-19 11:37:41 +08:00
parent 974487f00f
commit 9af0b6113c
15 changed files with 2217 additions and 214 deletions
+9
View File
@@ -457,6 +457,15 @@ owner confirms release, the live error status clears `cleanup_incomplete`.
The original failure, observations and `cleanup_report` retain the facts from
the earlier attempt, including any cleanup failure recorded then.
Snapshot file handles, directory handles, and the selected executable reader
follow the same ownership rule. A failed close preserves completed observations
and any primary read failure; it cannot turn a completed digest into a successful
capture with unresolved cleanup. Interrupted snapshots retain examined entries
with incomplete coverage in `partial_observation`, alongside the failure.
File-handle retry uses the exact retained owner. After a directory close rejects,
Node does not expose enough handle validity to authorize another native close;
REA retains that unresolved owner and blocks further captures in the session.
On macOS, changing a Node process's `process.title` can make its run token
unreadable. Node documents that [setting the title overwrites argv memory](https://nodejs.org/download/release/v24.18.0/docs/api/process.html#processtitle).
With the pinned toolchain, a live child retained a readable start identity
+12 -1
View File
@@ -1,6 +1,7 @@
import { z } from "zod";
import { AnalysisError } from "../../domain/analysisErrorBase.js";
import { analysisErrorWithCleanupFailure } from "../../domain/analysisErrorCleanup.js";
import {
AnalysisCancelledError,
AnalysisInputError,
@@ -20,6 +21,7 @@ import { parseProcessCapture } from "../../domain/process/processCaptureParsing.
import { createProgressReporter } from "../ProgressReporter.js";
import { captureProcessScenario } from "../../process/capture/ProcessHarness.js";
import { ProcessCaptureResourceScope } from "../../process/capture/ProcessCaptureLifecycle.js";
import { ProcessCaptureError } from "../../process/capture/ProcessCaptureError.js";
import {
PROCESS_PROVIDER,
isProcessEvidenceProvider,
@@ -84,7 +86,16 @@ export const captureProcessScenarioFile = async (
try {
await resourceScope.close();
} catch (cause: unknown) {
if (!captured.ok) return cliAnalysisError(captured.error);
if (!captured.ok && cause instanceof AnalysisError)
return cliAnalysisError(
captured.error instanceof ProcessCaptureError
? captured.error.withFinalization({ cleanupFailure: cause })
: analysisErrorWithCleanupFailure(
captured.error,
cause,
"capture_process_scenario",
).bindCleanupDependencies([captured.error, cause]),
);
throw cause;
}
if (!captured.ok) return cliAnalysisError(captured.error);
+6 -5
View File
@@ -107,7 +107,7 @@ export abstract class AnalysisError extends Error {
}
/** Later typed failures retained without changing the primary error cause. */
get secondaryFailures(): readonly AnalysisError[] {
return this.#secondaryFailures;
return [...this.#secondaryFailures];
}
readonly cleanupReport: ProcessCaptureCleanupReport | undefined = undefined;
@@ -141,13 +141,14 @@ export abstract class AnalysisError extends Error {
/** Retain a later lifecycle failure without replacing the original typed error. */
retainSecondaryFailure(
failure: AnalysisError,
diagnostics: Readonly<Record<string, JsonValue>>,
diagnostics?: Readonly<Record<string, JsonValue>>,
): this {
// Secondary resources remain owned by their cleanup lifecycle; preserve
// the primary error's live cleanup status and original output.
Object.assign(this, {
diagnostics: structuredClone({ ...this.diagnostics, ...diagnostics }),
});
if (diagnostics !== undefined)
Object.assign(this, {
diagnostics: structuredClone({ ...this.diagnostics, ...diagnostics }),
});
this.#secondaryFailures.push(failure);
return this;
}
+10 -1
View File
@@ -52,7 +52,13 @@ export const projectAnalysisError = (
): AnalysisErrorProjection => {
const composed = error;
error = resolvedAnalysisError(error);
const diagnostics = error.diagnostics;
let diagnosticSource = error;
while (
diagnosticSource.diagnostics === undefined &&
diagnosticSource.cleanupPrimary !== undefined
)
diagnosticSource = diagnosticSource.cleanupPrimary;
const diagnostics = diagnosticSource.diagnostics;
const capturedOutput = composed.capturedOutput ?? error.capturedOutput;
const partialObservation =
composed.partialObservation ?? error.partialObservation;
@@ -62,6 +68,9 @@ export const projectAnalysisError = (
const primaryDetails = errorDetails(error);
const details = {
...primaryDetails,
...(composed === error || composed.cleanupReport === undefined
? {}
: { cleanup_report: jsonValueSchema.parse(composed.cleanupReport) }),
...(diagnostics === undefined ? {} : { diagnostics }),
...(composed === error || composed.diagnostics === undefined
? {}
+13 -1
View File
@@ -16,6 +16,7 @@ import { join } from "node:path";
import { expect, it } from "vitest";
import { hashFile } from "./FilesystemSnapshot.js";
import { ProcessCaptureReadFailure } from "./ProcessCaptureRead.js";
const countOpenDescriptorsFor = async (path: string): Promise<number> => {
const directory = process.platform === "linux" ? "/proc/self/fd" : "/dev/fd";
@@ -170,7 +171,18 @@ it.skipIf(process.platform === "win32")(
});
controller.abort(reason);
await expect(hashing).rejects.toBe(reason);
let failure: unknown;
try {
await hashing;
} catch (cause: unknown) {
failure = cause;
}
expect(failure).toBeInstanceOf(ProcessCaptureReadFailure);
expect(failure).toMatchObject({
cause: reason,
userCategory: "cancelled",
outcome: { kind: "failed" },
});
expect(await countOpenDescriptorsFor(path)).toBe(0);
} finally {
await rm(directory, { recursive: true, force: true });
+334 -129
View File
@@ -11,6 +11,21 @@ import type {
FileState,
ProcessFilesystemSnapshot,
} from "../../domain/process/processCapture.js";
import {
AnalysisCleanupLease,
AnalysisCleanupLeaseGroup,
} from "../../artifacts/AnalysisCleanupLease.js";
import {
normalizeCaptureFailure,
describeProcessCaptureExecutionFailure,
ProcessCaptureError,
} from "./ProcessCaptureError.js";
import {
ProcessCaptureReadFailure,
withProcessCaptureDirectory,
withProcessCaptureFile,
type ProcessCaptureCleanupRetainer,
} from "./ProcessCaptureRead.js";
const hasSameIdentity = (
before: Stats,
@@ -69,7 +84,6 @@ const readEntryObservation = async <T>(
signal?.throwIfAborted();
return ok(value);
} catch (cause: unknown) {
signal?.throwIfAborted();
const code = filesystemSystemCode(cause);
// readlink reports EINVAL if the observed symlink was replaced by a file.
if (
@@ -77,6 +91,7 @@ const readEntryObservation = async <T>(
(!unavailableSystemCodes.has(code) && code !== "EINVAL")
)
throw cause;
signal?.throwIfAborted();
return err({ reason: "entry_unavailable", system_code: code });
}
};
@@ -106,9 +121,9 @@ const lstatIfPresent = async (
signal?.throwIfAborted();
return stats;
} catch (cause: unknown) {
signal?.throwIfAborted();
const code = filesystemSystemCode(cause);
if (code === "ENOENT" || (previouslyObservedRoot && code === "ENOTDIR")) {
signal?.throwIfAborted();
return undefined;
}
throw cause;
@@ -124,12 +139,91 @@ export type FileHashOutcome =
readonly system_code: string | null;
};
/** Snapshot facts retained alongside close failures from its owned reads. */
export class ProcessCaptureSnapshotFailure extends ProcessCaptureError {
readonly snapshot: ProcessFilesystemSnapshot | undefined;
readonly cleanupLeases: readonly AnalysisCleanupLease[];
readonly readFailures: readonly ProcessCaptureReadFailure<unknown>[];
constructor(options: {
readonly cause: unknown;
readonly snapshot?: ProcessFilesystemSnapshot;
readonly cleanupLeases?: readonly AnalysisCleanupLease[];
readonly readFailures?: readonly ProcessCaptureReadFailure<unknown>[];
}) {
const cleanupLeases = [...new Set(options.cleanupLeases ?? [])];
const readFailures = [...new Set(options.readFailures ?? [])];
const readFailure =
options.cause instanceof ProcessCaptureReadFailure
? options.cause
: undefined;
const primary =
readFailure !== undefined && Object.hasOwn(readFailure, "cause")
? readFailure.cause
: options.cause instanceof ProcessCaptureError &&
options.cause.userCategory === "cancelled" &&
Object.hasOwn(options.cause, "cause")
? options.cause.cause
: options.cause;
const cancellation =
readFailure?.userCategory === "cancelled" ||
(options.cause instanceof ProcessCaptureError &&
options.cause.userCategory === "cancelled");
const closeDetails = readFailures
.filter((failure) => failure.cleanupLeases.length > 0)
.map((failure) => {
const reason = failure.cleanup?.reason ?? "descriptor close failed";
return `${failure.path}: ${reason}`;
});
const primaryMessage =
describeProcessCaptureExecutionFailure(options.cause) ??
"snapshot read rejected without a value";
const message =
closeDetails.length === 0
? `Process filesystem snapshot failed: ${primaryMessage}`
: `Process filesystem snapshot failed: ${primaryMessage}; ${closeDetails.join("; ")}`;
const cleanupReason =
closeDetails.join("; ") ||
cleanupLeases
.map((lease) => lease.cleanupResources ?? [])
.flat()
.join(", ") ||
"snapshot handle cleanup failed";
super(message, {
cause: primary,
...(cancellation
? { userCategory: "cancelled" as const, reason: "cancelled" as const }
: {}),
...(cleanupLeases.length === 0
? {}
: {
cleanup: {
reason: cleanupReason,
resources: [
...new Set(
cleanupLeases.flatMap(
(lease) => lease.cleanupResources ?? [],
),
),
],
},
}),
});
this.snapshot = options.snapshot;
this.cleanupLeases = cleanupLeases;
this.readFailures = readFailures;
if (cleanupLeases.length > 0)
new AnalysisCleanupLeaseGroup(this, cleanupLeases);
}
}
/** Hash a file only when its opened descriptor still matches the captured path state. */
export const hashFile = async (
path: string,
expected: Stats,
maxBytes: number,
signal?: AbortSignal,
retainer?: ProcessCaptureCleanupRetainer,
): Promise<FileHashOutcome> => {
signal?.throwIfAborted();
const noFollowAvailable =
@@ -143,84 +237,90 @@ export const hashFile = async (
(fsConstants.O_NONBLOCK ?? 0),
);
} catch (cause: unknown) {
signal?.throwIfAborted();
const systemCode = fileUnavailableSystemCode(cause);
if (systemCode === undefined) throw cause;
signal?.throwIfAborted();
return {
state: "omitted",
reason: "file_unavailable",
system_code: systemCode,
};
}
try {
signal?.throwIfAborted();
// On hosts without O_NOFOLLOW, compare the leaf's identity with the
// captured regular file. Ancestor aliases do not change file identity.
if (!noFollowAvailable) {
const named = await lstat(path);
signal?.throwIfAborted();
if (!hasSameFileState(expected, named))
return await withProcessCaptureFile(
path,
handle,
signal,
retainer,
async (ownedHandle) => {
try {
signal?.throwIfAborted();
// On hosts without O_NOFOLLOW, compare the leaf's identity with the
// captured regular file. Ancestor aliases do not change file identity.
if (!noFollowAvailable) {
const named = await lstat(path);
signal?.throwIfAborted();
if (!hasSameFileState(expected, named))
return {
state: "omitted",
reason: "file_changed_or_short_read",
system_code: null,
};
}
const stats = await ownedHandle.stat();
signal?.throwIfAborted();
if (!hasSameFileState(expected, stats) || stats.size > maxBytes)
return {
state: "omitted",
reason: "file_changed_or_short_read",
system_code: null,
};
const hash = createHash("sha256");
const buffer = Buffer.allocUnsafe(Math.min(64 * 1024, maxBytes));
let position = 0;
while (position < stats.size) {
signal?.throwIfAborted();
const { bytesRead } = await ownedHandle.read(
buffer,
0,
Math.min(buffer.length, stats.size - position),
position,
);
if (bytesRead === 0)
return {
state: "omitted",
reason: "file_changed_or_short_read",
system_code: null,
};
hash.update(buffer.subarray(0, bytesRead));
position += bytesRead;
}
const after = await ownedHandle.stat();
signal?.throwIfAborted();
let unchanged = hasSameFileState(stats, after);
if (unchanged && !noFollowAvailable) {
const namedAfter = await lstat(path);
signal?.throwIfAborted();
unchanged = hasSameFileState(after, namedAfter);
}
if (!unchanged)
return {
state: "omitted",
reason: "file_changed_or_short_read",
system_code: null,
};
return { state: "hashed", sha256: hash.digest("hex") };
} catch (cause: unknown) {
const systemCode = fileUnavailableSystemCode(cause);
if (systemCode === undefined) throw cause;
signal?.throwIfAborted();
return {
state: "omitted",
reason: "file_changed_or_short_read",
system_code: null,
reason: "file_unavailable",
system_code: systemCode,
};
}
const stats = await handle.stat();
signal?.throwIfAborted();
if (!hasSameFileState(expected, stats) || stats.size > maxBytes)
return {
state: "omitted",
reason: "file_changed_or_short_read",
system_code: null,
};
const hash = createHash("sha256");
const buffer = Buffer.allocUnsafe(Math.min(64 * 1024, maxBytes));
let position = 0;
while (position < stats.size) {
signal?.throwIfAborted();
const { bytesRead } = await handle.read(
buffer,
0,
Math.min(buffer.length, stats.size - position),
position,
);
if (bytesRead === 0)
return {
state: "omitted",
reason: "file_changed_or_short_read",
system_code: null,
};
hash.update(buffer.subarray(0, bytesRead));
position += bytesRead;
}
const after = await handle.stat();
signal?.throwIfAborted();
let unchanged = hasSameFileState(stats, after);
if (unchanged && !noFollowAvailable) {
const namedAfter = await lstat(path);
signal?.throwIfAborted();
unchanged = hasSameFileState(after, namedAfter);
}
if (!unchanged)
return {
state: "omitted",
reason: "file_changed_or_short_read",
system_code: null,
};
return { state: "hashed", sha256: hash.digest("hex") };
} catch (cause: unknown) {
signal?.throwIfAborted();
const systemCode = fileUnavailableSystemCode(cause);
if (systemCode === undefined) throw cause;
return {
state: "omitted",
reason: "file_unavailable",
system_code: systemCode,
};
} finally {
await handle.close();
}
}
},
);
};
/** Capture bounded, root-aliased filesystem state without following symlinks. */
@@ -228,6 +328,7 @@ export const snapshotRoots = async (
scenario: ProcessScenario,
signal?: AbortSignal,
previous?: ProcessFilesystemSnapshot,
retainer?: ProcessCaptureCleanupRetainer,
): Promise<ProcessFilesystemSnapshot> => {
// A prior root entry establishes same-run admission. Preserve initial
// obstruction errors; a later non-directory ancestor makes it absent.
@@ -242,12 +343,52 @@ export const snapshotRoots = async (
>();
const enumerationOmissions: FilesystemCoverage["enumeration_omissions"] = [];
const hashOmissions: FilesystemCoverage["hash_omissions"] = [];
const closeFailures: ProcessCaptureReadFailure<unknown>[] = [];
const omitEntry = (path: string, omission: EntryOmission): false => {
truncated = true;
enumerationReasons.add(omission.reason);
enumerationOmissions.push({ path, ...omission });
return false;
};
const recordFile = (
rootAlias: string,
relativePath: string,
stats: Stats,
outcome:
| FileHashOutcome
| {
readonly state: "omitted";
readonly reason: "file_bytes_budget";
readonly system_code: null;
},
): void => {
let sha256: string | null = null;
let omission: FilesystemCoverage["hash_omissions"][number] | undefined;
if (outcome.state === "hashed") {
sha256 = outcome.sha256;
remainingBytes -= stats.size;
} else {
omission = {
path: `${rootAlias}:${relativePath}`,
size_bytes: stats.size,
remaining_budget_bytes: remainingBytes,
reason: outcome.reason,
system_code: outcome.system_code,
};
}
if (omission !== undefined) {
truncated = true;
hashOmissions.push(omission);
}
entries.push({
path: `${rootAlias}:${relativePath}`,
type: "file",
mode: stats.mode,
size: stats.size,
sha256,
symlink_target: null,
});
};
const visit = async (
root: string,
rootAlias: string,
@@ -288,43 +429,45 @@ export const snapshotRoots = async (
return true;
}
if (stats.isFile()) {
let sha256: string | null = null;
let omission: FilesystemCoverage["hash_omissions"][number] | undefined;
if (remainingBytes < stats.size) {
omission = {
path: `${rootAlias}:${relativePath}`,
size_bytes: stats.size,
remaining_budget_bytes: remainingBytes,
recordFile(rootAlias, relativePath, stats, {
state: "omitted",
reason: "file_bytes_budget",
system_code: null,
};
});
} else {
const outcome = await hashFile(path, stats, remainingBytes, signal);
if (outcome.state === "hashed") {
sha256 = outcome.sha256;
remainingBytes -= stats.size;
} else {
omission = {
path: `${rootAlias}:${relativePath}`,
size_bytes: stats.size,
remaining_budget_bytes: remainingBytes,
reason: outcome.reason,
system_code: outcome.system_code,
};
let outcome: FileHashOutcome;
try {
outcome = await hashFile(
path,
stats,
remainingBytes,
signal,
retainer,
);
} catch (cause: unknown) {
if (
cause instanceof ProcessCaptureReadFailure &&
cause.outcome.kind === "completed" &&
cause.userCategory === "cancelled"
) {
recordFile(rootAlias, relativePath, stats, cause.outcome.value);
if (depth === 0) completeRoots.push(rootAlias);
throw cause;
}
if (
cause instanceof ProcessCaptureReadFailure &&
cause.outcome.kind === "completed" &&
cause.cleanupLeases.length > 0
) {
closeFailures.push(cause);
outcome = cause.outcome.value;
} else {
throw cause;
}
}
recordFile(rootAlias, relativePath, stats, outcome);
}
if (omission !== undefined) {
truncated = true;
hashOmissions.push(omission);
}
entries.push({
path: `${rootAlias}:${relativePath}`,
type: "file",
mode: stats.mode,
size: stats.size,
sha256,
symlink_target: null,
});
return true;
}
const type = stats.isDirectory() ? "directory" : "other";
@@ -345,9 +488,9 @@ export const snapshotRoots = async (
try {
directory = await opendir(path);
} catch (cause: unknown) {
signal?.throwIfAborted();
const systemCode = fileUnavailableSystemCode(cause);
if (systemCode === undefined) throw cause;
signal?.throwIfAborted();
truncated = true;
complete = false;
enumerationReasons.add("directory_unavailable");
@@ -358,22 +501,47 @@ export const snapshotRoots = async (
});
return false;
}
for await (const child of directory) {
signal?.throwIfAborted();
try {
await withProcessCaptureDirectory(
path,
directory,
signal,
retainer,
async (ownedDirectory) => {
while (true) {
signal?.throwIfAborted();
const child = await ownedDirectory.read();
if (child === null) break;
if (
children.length >= remainingEntries ||
depth >= scenario.limits.filesystem_depth
) {
truncated = true;
complete = false;
if (children.length >= remainingEntries)
enumerationReasons.add("files_limit");
if (depth >= scenario.limits.filesystem_depth)
enumerationReasons.add("depth_limit");
break;
}
children.push(child.name);
pendingNames += 1;
}
},
);
} catch (cause: unknown) {
if (
children.length >= remainingEntries ||
depth >= scenario.limits.filesystem_depth
cause instanceof ProcessCaptureReadFailure &&
cause.outcome.kind === "completed" &&
cause.cleanupLeases.length > 0
) {
truncated = true;
complete = false;
if (children.length >= remainingEntries)
enumerationReasons.add("files_limit");
if (depth >= scenario.limits.filesystem_depth)
enumerationReasons.add("depth_limit");
break;
if (cause.userCategory === "cancelled") {
throw cause;
}
closeFailures.push(cause);
} else {
throw cause;
}
children.push(child.name);
pendingNames += 1;
}
const afterRead = await readEntryObservation(() => lstat(path), signal);
if (
@@ -398,20 +566,7 @@ export const snapshotRoots = async (
}
return complete;
};
for (const [index, root] of scenario.filesystem_observation_paths.entries()) {
const alias = `root_${String(index)}`;
const admitted = await lstatIfPresent(
root,
previousPaths.has(`${alias}:.`),
signal,
);
if (admitted === undefined) {
completeRoots.push(alias);
continue;
}
if (await visit(root, alias, root, 0, admitted)) completeRoots.push(alias);
}
return {
const buildSnapshot = (): ProcessFilesystemSnapshot => ({
files: entries,
truncated,
completeRoots,
@@ -426,5 +581,55 @@ export const snapshotRoots = async (
hashed_bytes: scenario.limits.file_bytes - remainingBytes,
hash_omissions: hashOmissions,
},
};
});
try {
signal?.throwIfAborted();
for (const [
index,
root,
] of scenario.filesystem_observation_paths.entries()) {
const alias = `root_${String(index)}`;
const admitted = await lstatIfPresent(
root,
previousPaths.has(`${alias}:.`),
signal,
);
if (admitted === undefined) {
completeRoots.push(alias);
continue;
}
if (await visit(root, alias, root, 0, admitted))
completeRoots.push(alias);
}
const snapshot = buildSnapshot();
if (closeFailures.length > 0) {
const causes = closeFailures.map(({ cleanupCause }) => cleanupCause);
throw new ProcessCaptureSnapshotFailure({
cause:
causes.length === 1
? (causes[0] as unknown)
: new AggregateError(causes, "Snapshot handle closes failed"),
snapshot,
cleanupLeases: closeFailures.flatMap(
({ cleanupLeases }) => cleanupLeases,
),
readFailures: closeFailures,
});
}
return snapshot;
} catch (cause: unknown) {
if (cause instanceof ProcessCaptureSnapshotFailure) throw cause;
truncated ||=
completeRoots.length !== scenario.filesystem_observation_paths.length;
const readFailures = [
...closeFailures,
...(cause instanceof ProcessCaptureReadFailure ? [cause] : []),
];
throw new ProcessCaptureSnapshotFailure({
cause: normalizeCaptureFailure(cause, signal),
snapshot: buildSnapshot(),
cleanupLeases: readFailures.flatMap(({ cleanupLeases }) => cleanupLeases),
readFailures,
});
}
};
@@ -1,4 +1,6 @@
import { expect, it } from "vitest";
import { AnalysisCleanupLease } from "../../artifacts/AnalysisCleanupLease.js";
import { ProcessCaptureReadFailure } from "./ProcessCaptureRead.js";
import { ProviderCleanupError } from "../../domain/providerCleanupError.js";
import { projectAnalysisError } from "../../domain/analysisErrorProjection.js";
import { DarwinProcessOwnershipInspectionError } from "../DarwinProcessRunTokenReader.js";
@@ -171,3 +173,72 @@ it("does not infer captured stdout from partial compiler diagnostics", () => {
capturedOutput: undefined,
});
});
it("retains the typed completed read through finalization and confirmed cleanup", async () => {
const value = { sha256: "completed digest", identity: { dev: 1n, ino: 2n } };
const lease = new AnalysisCleanupLease(
async () => undefined,
["selected-file"],
);
const original = new ProcessCaptureReadFailure({
path: "selected-file",
outcome: { kind: "completed", value },
cleanupCause: new Error("close denied after completed read"),
cleanupLease: lease,
signal: undefined,
});
const finalized = original.withFinalization({ cleanupReport });
expect(finalized.cleanupPrimary).toBe(original);
expect(original.outcome).toEqual({ kind: "completed", value });
await lease.release();
expect(finalized.cleanupIncomplete).toBe(false);
expect(projectAnalysisError(finalized)).toMatchObject({
code: "process_capture_failed",
details: { cleanup_report: cleanupReport },
});
expect(finalized.cleanupPrimary).toBe(original);
expect(original.outcome).toEqual({ kind: "completed", value });
});
it("retains an undefined physical read rejection as an explicit typed outcome", async () => {
const lease = new AnalysisCleanupLease(
async () => undefined,
["selected-file"],
);
const original = new ProcessCaptureReadFailure({
path: "selected-file",
outcome: { kind: "failed", cause: undefined },
cleanupCause: new Error("close failure following undefined read rejection"),
cleanupLease: lease,
signal: undefined,
});
const finalized = original.withFinalization({ cleanupReport });
expect(finalized.cleanupPrimary).toBe(original);
expect(original.outcome).toEqual({ kind: "failed", cause: undefined });
expect(Object.hasOwn(original.outcome, "cause")).toBe(true);
expect(finalized.cleanupIncomplete).toBe(true);
expect(projectAnalysisError(finalized)).toMatchObject({
code: "cleanup_incomplete",
details: { cleanup_report: cleanupReport },
});
await lease.release();
expect(finalized.cleanupIncomplete).toBe(false);
expect(finalized.cleanupPrimary).toBe(original);
expect(original.outcome).toEqual({ kind: "failed", cause: undefined });
expect(projectAnalysisError(finalized)).toMatchObject({
details: { cleanup_report: cleanupReport },
});
});
it("inherits primary diagnostics without duplicating finalization history", async () => {
const primary = new ProcessCaptureError("read failed", {
diagnostics: { read_stage: "metadata" },
});
const finalized = primary.withFinalization({ cleanupReport });
const projected = projectAnalysisError(finalized);
expect(projected.details).toMatchObject({
diagnostics: { read_stage: "metadata" },
cleanup_report: cleanupReport,
});
expect(projected.details).not.toHaveProperty("cleanup_history");
});
+4 -4
View File
@@ -61,7 +61,8 @@ export class ProcessCaptureError extends AnalysisError {
const partialObservation =
this.partialObservation ?? options.partialObservation;
const cleanupReport = this.cleanupReport ?? options.cleanupReport;
return new ProcessCaptureError(this.message, {
const finalized = new ProcessCaptureError(this.message, {
cleanupPrimary: this,
...(Object.hasOwn(this, "cause") ? { cause: this.cause } : {}),
...(this.userMessage === undefined
? {}
@@ -70,9 +71,6 @@ export class ProcessCaptureError extends AnalysisError {
? {}
: { userCategory: this.userCategory }),
reason: this.reason,
...(this.diagnostics === undefined
? {}
: { diagnostics: this.diagnostics }),
...(this.capturedOutput === undefined
? {}
: { capturedOutput: this.capturedOutput }),
@@ -88,6 +86,8 @@ export class ProcessCaptureError extends AnalysisError {
this,
...(secondary === undefined ? [] : [secondary]),
]);
if (secondary !== undefined) finalized.retainSecondaryFailure(secondary);
return finalized;
}
}
+124 -67
View File
@@ -19,7 +19,15 @@ import {
} from "../../domain/process/processScenario.js";
import { parseProcessCapture } from "../../domain/process/processCaptureParsing.js";
import { PRODUCT_IDENTITY } from "../../identity.js";
import { snapshotRoots } from "./FilesystemSnapshot.js";
import {
snapshotRoots,
ProcessCaptureSnapshotFailure,
} from "./FilesystemSnapshot.js";
import {
withProcessCaptureFile,
ProcessCaptureReadFailure,
type ProcessCaptureCleanupRetainer,
} from "./ProcessCaptureRead.js";
import { classifyFilesystemEffects } from "./ProcessFilesystemEffects.js";
import {
cleanupOwnedProcessGroup,
@@ -186,7 +194,7 @@ export const createProcessCaptureObservationBuffer = (options: {
readonly interactions: readonly InteractionEvent[];
readonly samples: readonly ProcessSample[];
readonly eventJournal: readonly ProcessCaptureEventJournalEntry[];
readonly before: ProcessFilesystemSnapshot;
readonly before: ProcessFilesystemSnapshot | undefined;
/** Seed a finalization observation only for a scenario with an interval. */
readonly finalizationEnabled?: boolean;
}): ProcessCaptureObservationBuffer => ({
@@ -221,10 +229,13 @@ export const createProcessCaptureObservationBuffer = (options: {
},
process_samples: { state: "available", value: options.samples },
filesystem_snapshots: {
before: {
state: "available",
value: options.before,
},
before:
options.before === undefined
? {
state: "unavailable",
reason: "Initial filesystem snapshot did not complete.",
}
: { state: "available", value: options.before },
after: {
state: "unavailable",
reason:
@@ -517,63 +528,83 @@ export const observeSelectedExecutable = async (
path: string,
signal?: AbortSignal,
fileSystem: SelectedExecutableFileSystem = selectedExecutableFileSystem,
cleanupRetainer?: ProcessCaptureCleanupRetainer,
): Promise<SelectedExecutableObservation> => {
assertNotCancelled(signal);
let handle: Awaited<ReturnType<typeof open>> | undefined;
try {
handle = await fileSystem.open(
const handle = await fileSystem.open(
path,
constants.O_RDONLY | constants.O_NONBLOCK,
);
const openedStats = await handle.stat({ bigint: true });
if (!openedStats.isFile())
return {
sha256: null,
identity: null,
reason: "Selected executable is not a regular file.",
};
const openedIdentity = identityFromStats(openedStats);
const selectedPathIdentity = identityFromStats(await fileSystem.stat(path));
if (!sameExecutableIdentity(openedIdentity, selectedPathIdentity))
return {
sha256: null,
identity: null,
reason: "Selected executable path changed while it was being opened.",
};
// IO above can settle after cancellation. Do not construct a stream with an
// already-aborted signal: Node can emit a second, unhandled AbortError.
assertNotCancelled(signal);
const hash = createHash("sha256");
for await (const chunk of handle.createReadStream({
autoClose: false,
return await withProcessCaptureFile(
path,
handle,
signal,
})) {
assertNotCancelled(signal);
hash.update(chunk);
}
assertNotCancelled(signal);
const finalIdentity = identityFromStats(
await handle.stat({ bigint: true }),
cleanupRetainer,
async (handle): Promise<SelectedExecutableObservation> => {
const openedStats = await handle.stat({ bigint: true });
if (!openedStats.isFile())
return {
sha256: null,
identity: null,
reason: "Selected executable is not a regular file.",
};
const openedIdentity = identityFromStats(openedStats);
const selectedPathIdentity = identityFromStats(
await fileSystem.stat(path),
);
if (!sameExecutableIdentity(openedIdentity, selectedPathIdentity))
return {
sha256: null,
identity: null,
reason:
"Selected executable path changed while it was being opened.",
};
// IO above can settle after cancellation. Do not construct a stream with an
// already-aborted signal: Node can emit a second, unhandled AbortError.
assertNotCancelled(signal);
const hash = createHash("sha256");
for await (const chunk of handle.createReadStream({
autoClose: false,
signal,
})) {
assertNotCancelled(signal);
hash.update(chunk);
}
assertNotCancelled(signal);
const finalIdentity = identityFromStats(
await handle.stat({ bigint: true }),
);
if (!sameExecutableIdentity(openedIdentity, finalIdentity))
return {
sha256: null,
identity: null,
reason:
"Selected executable changed while its contents were sampled.",
};
return {
sha256: hash.digest("hex"),
identity: openedIdentity,
reason: null,
};
},
);
if (!sameExecutableIdentity(openedIdentity, finalIdentity))
return {
sha256: null,
identity: null,
reason: "Selected executable changed while its contents were sampled.",
};
return {
sha256: hash.digest("hex"),
identity: openedIdentity,
reason: null,
};
} catch (cause: unknown) {
assertNotCancelled(signal);
const reason = filesystemFailure(cause);
if (
cause instanceof ProcessCaptureReadFailure &&
!cause.cleanupIncomplete &&
cause.userCategory !== "cancelled" &&
cause.outcome.kind === "failed"
) {
const reason = filesystemFailure(cause.outcome.cause);
if (reason !== undefined) return { sha256: null, identity: null, reason };
}
const failure = normalizeCaptureFailure(cause, signal);
if (failure instanceof ProcessCaptureError) throw failure;
const reason = filesystemFailure(failure);
if (reason !== undefined) return { sha256: null, identity: null, reason };
throw cause;
} finally {
await handle?.close();
throw failure;
}
};
@@ -1282,6 +1313,11 @@ export class ProcessCaptureResourceScope {
}
}
/** Retain an exact child owner acquired by an already admitted capture. */
retain(lease: AnalysisCleanupLease): void {
if (!lease.released) this.#pending.add(lease);
}
async release(
options: ProcessCaptureResourceCleanupOptions,
): Promise<ProcessCaptureCleanupOutcome> {
@@ -1603,28 +1639,49 @@ export const prepareProcessCapture = async (
assertNotCancelled(signal);
await host.prepareOwnershipInspector?.(signal);
assertNotCancelled(signal);
const before = await captureSnapshot(scenario, signal);
let before: ProcessFilesystemSnapshot | undefined;
let temporaryRoot: string | undefined;
const runId = randomUUID();
const temporaryRoot = await host.createTemporaryRoot();
try {
before = await captureSnapshot(scenario, signal, undefined, resourceScope);
temporaryRoot = await host.createTemporaryRoot();
assertNotCancelled(signal);
const ownershipBaseline =
(await host.captureOwnershipBaseline?.(signal)) ?? [];
assertNotCancelled(signal);
return { temporaryRoot, runId, ownershipBaseline, before };
} catch (cause: unknown) {
const cleanupOptions: ProcessCaptureResourceCleanupOptions = {
timers: new Set(),
terminal: undefined,
renderer: undefined,
runId,
temporaryRoot,
host: { ...processCaptureCleanupHost, removeTemporaryRoot: host.cleanup },
};
const cleanup =
resourceScope === undefined
? await releaseProcessResources(cleanupOptions)
: await resourceScope.release(cleanupOptions);
if (cause instanceof ProcessCaptureSnapshotFailure) {
before = cause.snapshot;
for (const lease of cause.cleanupLeases) resourceScope?.retain(lease);
}
const cleanupOptions =
temporaryRoot === undefined
? undefined
: ({
timers: new Set<ProcessTimer>(),
terminal: undefined,
renderer: undefined,
runId,
temporaryRoot,
host: {
...processCaptureCleanupHost,
removeTemporaryRoot: host.cleanup,
},
} satisfies ProcessCaptureResourceCleanupOptions);
const cleanup: ProcessCaptureCleanupOutcome =
cleanupOptions === undefined
? {
report: {
terminal_renderer: { state: "not_required", reason: null },
owned_process_group: { state: "not_required", reason: null },
temporary_root: { state: "not_required", reason: null },
},
cleanupLease: undefined,
}
: resourceScope === undefined
? await releaseProcessResources(cleanupOptions)
: await resourceScope.release(cleanupOptions);
resolveProcessResult(
undefined,
normalizeCaptureFailure(cause, signal),
+241
View File
@@ -0,0 +1,241 @@
import type { Dir } from "node:fs";
import type { FileHandle } from "node:fs/promises";
import {
AnalysisCleanupLease,
AnalysisCleanupLeaseGroup,
} from "../../artifacts/AnalysisCleanupLease.js";
import {
ProcessCaptureError,
describeProcessCaptureExecutionFailure,
} from "./ProcessCaptureError.js";
import { OwnedDirectoryHandle } from "../../filesystem/OwnedDirectoryHandle.js";
import { OwnedFileHandle } from "../../filesystem/OwnedFileHandle.js";
import {
RegularFileCancelled,
RegularFileCleanupFailure,
withOwnedFileHandle,
type RegularFileOperationOutcome,
} from "../../filesystem/OwnedRegularFile.js";
import { RegularFileOperationCancelled } from "../../filesystem/RegularFile.js";
/** The capture's existing scope retains exact leases without admitting work. */
export interface ProcessCaptureCleanupRetainer {
retain(lease: AnalysisCleanupLease): void;
}
type Cancellation = { readonly reason: unknown };
type CloseRetention = RegularFileCleanupFailure["retention"];
/** One physical read outcome plus any exact descriptor owners still unresolved. */
export class ProcessCaptureReadFailure<
Value = unknown,
> extends ProcessCaptureError {
readonly path: string;
readonly outcome: RegularFileOperationOutcome<Value>;
readonly cleanupLeases: readonly AnalysisCleanupLease[];
readonly cleanupCause: unknown;
readonly retention: CloseRetention;
readonly cancellation: Cancellation | undefined;
constructor(options: {
readonly path: string;
readonly outcome: RegularFileOperationOutcome<Value>;
readonly cleanupCause?: unknown;
readonly cleanupLease?: AnalysisCleanupLease;
readonly retention?: CloseRetention;
readonly completionCancellation?: Cancellation | undefined;
readonly signal: AbortSignal | undefined;
}) {
const cleanupLeases =
options.cleanupLease === undefined ? [] : [options.cleanupLease];
const cancellation =
options.outcome.kind === "failed"
? operationCancellation(options.outcome.cause, options.signal)
: options.completionCancellation;
const primary =
cancellation !== undefined
? cancellation.reason
: options.outcome.kind === "failed"
? options.outcome.cause
: options.cleanupCause;
const cleanupCause = options.cleanupCause;
const cleanupReason =
describeProcessCaptureExecutionFailure(cleanupCause) ??
"descriptor close rejected without a value";
const reasons = [
...(options.outcome.kind === "failed"
? [
describeProcessCaptureExecutionFailure(options.outcome.cause) ??
"read rejected without a value",
]
: []),
...(options.cleanupLease === undefined
? []
: [`descriptor close failed: ${cleanupReason}`]),
];
super(
`Process capture read failed for ${options.path}${reasons.length === 0 ? "" : `: ${reasons.join("; ")}`}`,
{
cause: primary,
...(cancellation === undefined
? {}
: {
userCategory: "cancelled" as const,
reason: "cancelled" as const,
}),
...(options.cleanupLease === undefined
? {}
: {
cleanup: {
reason: cleanupReason,
resources: options.cleanupLease?.cleanupResources ?? [
options.path,
],
},
}),
},
);
this.path = options.path;
this.outcome = options.outcome;
this.cleanupLeases = cleanupLeases;
this.cleanupCause = cleanupCause;
this.retention = options.retention ?? { kind: "not_requested" };
this.cancellation = cancellation;
if (cleanupLeases.length > 0)
new AnalysisCleanupLeaseGroup(this, cleanupLeases);
}
}
const operationCancellation = (
cause: unknown,
signal: AbortSignal | undefined,
): Cancellation | undefined => {
if (cause instanceof RegularFileOperationCancelled)
return { reason: cause.reason };
if (
cause instanceof ProcessCaptureError &&
cause.userCategory === "cancelled"
)
return {
reason: Object.hasOwn(cause, "cause") ? cause.cause : cause,
};
if (
signal?.aborted === true &&
(cause === signal.reason ||
(cause instanceof Error && cause.name === "AbortError"))
)
return { reason: signal.reason };
return undefined;
};
/** Keep an admitted file descriptor, its operation result, and close owner together. */
export const withProcessCaptureFile = async <Value>(
path: string,
handle: FileHandle,
signal: AbortSignal | undefined,
retainer: ProcessCaptureCleanupRetainer | undefined,
operation: (file: FileHandle) => Promise<Value>,
): Promise<Value> => {
const owner = new OwnedFileHandle(handle);
const cleanupLease = AnalysisCleanupLease.forCloseOwner(owner, path);
try {
return await withOwnedFileHandle(
path,
owner,
{
...(signal === undefined ? {} : { signal }),
...(retainer === undefined
? {}
: { retainFailedClose: () => retainer.retain(cleanupLease) }),
},
operation,
);
} catch (cause: unknown) {
if (cause instanceof RegularFileCleanupFailure)
throw new ProcessCaptureReadFailure({
path,
outcome: cause.outcome as RegularFileOperationOutcome<Value>,
cleanupCause: cause.cleanupCause,
cleanupLease,
retention: cause.retention,
...(cause.completionCancellation === undefined
? {}
: { completionCancellation: cause.completionCancellation }),
signal,
});
if (cause instanceof RegularFileCancelled)
throw new ProcessCaptureReadFailure({
path,
outcome: cause.outcome,
completionCancellation: { reason: cause.reason },
signal,
});
throw new ProcessCaptureReadFailure({
path,
outcome: { kind: "failed", cause },
signal,
});
}
};
/** Read and close an explicit directory without losing completed entries on close failure. */
export const withProcessCaptureDirectory = async <Value>(
path: string,
directory: Dir,
signal: AbortSignal | undefined,
retainer: ProcessCaptureCleanupRetainer | undefined,
operation: (directory: Dir) => Promise<Value>,
): Promise<Value> => {
const owner = new OwnedDirectoryHandle(directory);
const cleanupLease = AnalysisCleanupLease.forCloseOwner(owner, path);
let outcome: RegularFileOperationOutcome<Value>;
try {
if (signal?.aborted === true)
throw new RegularFileOperationCancelled(signal.reason);
outcome = { kind: "completed", value: await operation(directory) };
} catch (cause: unknown) {
outcome = { kind: "failed", cause };
}
try {
await owner.close();
} catch (cleanupCause: unknown) {
let retention: CloseRetention = { kind: "not_requested" };
if (retainer !== undefined) {
try {
retainer.retain(cleanupLease);
retention = { kind: "callback_returned" };
} catch (cause: unknown) {
retention = { kind: "callback_threw", cause };
}
}
throw new ProcessCaptureReadFailure({
path,
outcome,
cleanupCause,
cleanupLease,
retention,
...(outcome.kind === "completed" && signal?.aborted === true
? { completionCancellation: { reason: signal.reason } }
: {}),
signal,
});
}
if (outcome.kind === "failed") {
throw new ProcessCaptureReadFailure({
path,
outcome,
signal,
});
}
if (signal?.aborted === true)
throw new ProcessCaptureReadFailure({
path,
outcome,
completionCancellation: { reason: signal.reason },
signal,
});
return outcome.value;
};
+37 -6
View File
@@ -5,6 +5,7 @@ import { parseProcessCapture } from "../../domain/process/processCaptureParsing.
import { err, ok, type Result } from "../../domain/result.js";
import { AnalysisCapabilityUnavailableError } from "../../domain/analysisErrorCore.js";
import { AnalysisError } from "../../domain/analysisErrorBase.js";
import { analysisErrorWithCleanupFailure } from "../../domain/analysisErrorCleanup.js";
import {
describeProcessCaptureExecutionFailure,
ProcessCaptureError,
@@ -20,7 +21,10 @@ import type {
TerminalFrame,
} from "../../domain/process/processCapture.js";
import { startProcessSampler } from "./ProcessSampling.js";
import { snapshotRoots } from "./FilesystemSnapshot.js";
import {
snapshotRoots,
ProcessCaptureSnapshotFailure,
} from "./FilesystemSnapshot.js";
import { TerminalRenderer } from "./TerminalRenderer.js";
import type { ProcessTimer } from "./ProcessTimer.js";
import {
@@ -271,6 +275,8 @@ const startCaptureRuntime = async (
const selectedExecutable = await observeSelectedExecutable(
scenario.executable,
options.signal,
undefined,
options.resourceScope,
);
const startedAt = new Date();
const started = monotonicProcessCaptureClock();
@@ -478,6 +484,7 @@ const completeCapture = async (options: {
>;
readonly signal?: AbortSignal;
readonly captureSnapshot: typeof snapshotRoots;
readonly resourceScope?: ProcessCaptureResourceScope;
readonly eventJournal: readonly ProcessCaptureEventJournalEntry[];
readonly observationBuffer: ProcessCaptureObservationBuffer;
readonly recordEvent: RecordProcessCaptureEvent;
@@ -521,14 +528,22 @@ const completeCapture = async (options: {
scenario,
options.signal,
options.before,
options.resourceScope,
);
} catch (cause: unknown) {
if (cause instanceof ProcessCaptureSnapshotFailure)
for (const lease of cause.cleanupLeases)
options.resourceScope?.retain(lease);
options.observationBuffer.filesystem_snapshots = {
...options.observationBuffer.filesystem_snapshots,
after: {
state: "unavailable",
reason: unavailableReason("Final filesystem snapshot", cause),
},
after:
cause instanceof ProcessCaptureSnapshotFailure &&
cause.snapshot !== undefined
? { state: "available", value: cause.snapshot }
: {
state: "unavailable",
reason: unavailableReason("Final filesystem snapshot", cause),
},
};
throw cause;
}
@@ -794,6 +809,7 @@ const runProcessScenario = async (
observationBuffer: observations,
recordEvent,
captureSnapshot,
...(resourceScope === undefined ? {} : { resourceScope }),
...(signal === undefined ? {} : { signal }),
...(progress === undefined ? {} : { onLiveProgress: reportSettling }),
});
@@ -926,6 +942,21 @@ export const captureProcessScenario = async (
}
}
const capture = await runResourceScope.run(execute);
await runResourceScope.close().catch(() => undefined);
try {
await runResourceScope.close();
} catch (cause: unknown) {
if (!(cause instanceof AnalysisError)) throw cause;
return err(
capture.ok
? cause
: capture.error instanceof ProcessCaptureError
? capture.error.withFinalization({ cleanupFailure: cause })
: analysisErrorWithCleanupFailure(
capture.error,
cause,
"capture_process_scenario",
).bindCleanupDependencies([capture.error, cause]),
);
}
return capture;
};
@@ -0,0 +1,177 @@
import { mkdir, mkdtemp, open, rm, writeFile } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join, resolve } from "node:path";
import { Client, InMemoryTransport } from "@modelcontextprotocol/client";
import { expect, it, vi } from "vitest";
const injected = vi.hoisted(() => ({
path: "",
failClose: false,
opens: 0,
handle: undefined as Awaited<ReturnType<typeof open>> | undefined,
nativeClose: undefined as (() => Promise<void>) | undefined,
}));
vi.mock("node:fs/promises", async (importOriginal) => {
const original = await importOriginal<typeof import("node:fs/promises")>();
return {
...original,
open: async (...args: Parameters<typeof original.open>) => {
const handle = await original.open(...args);
if (resolve(String(args[0])) !== injected.path) return handle;
injected.opens += 1;
injected.handle = handle;
const nativeClose = handle.close.bind(handle);
injected.nativeClose = nativeClose;
handle.close = async () => {
if (injected.failClose)
throw new Error("controlled file close failure");
await nativeClose();
};
return handle;
},
};
});
import { createTestBinarySession } from "../../fixtures/binarySession.js";
import { parseMcpToolError } from "../../fixtures/mcpToolError.js";
import type { AnalysisClient } from "../../../src/application/AnalysisProvider.js";
import { observed } from "../../fixtures/analysisExecution.js";
import { createServer } from "../../../src/server/createServer.js";
import { silentLogger } from "../../../src/logger.js";
import { ToolResultDelivery } from "../../../src/server/toolResult.js";
import { ok } from "../../../src/domain/result.js";
const connectSession = async () => {
const session = createTestBinarySession(
() =>
({
execute: () => Promise.resolve(observed(null)),
close: () => Promise.resolve(ok(null)),
}) satisfies AnalysisClient,
);
const server = createServer(
{ kind: "session", session },
{ logger: silentLogger },
);
const client = new Client({ name: "process-snapshot-cleanup", version: "1" });
const [clientTransport, serverTransport] =
InMemoryTransport.createLinkedPair();
await server.connect(serverTransport);
await client.connect(clientTransport);
return { client, server, session };
};
it("retains and retries the exact snapshot file handle through public MCP capture", async () => {
const root = await mkdtemp(join(tmpdir(), "rea-process-snapshot-cleanup-"));
const observedRoot = join(root, "observed");
const filePath = join(observedRoot, "input.txt");
await mkdir(observedRoot);
await writeFile(filePath, "snapshot bytes");
injected.path = resolve(filePath);
injected.failClose = true;
injected.opens = 0;
const resources = await connectSession();
const delivery = vi.spyOn(ToolResultDelivery.prototype, "toCallToolResult");
const capture = () =>
resources.client.callTool({
name: "capture_process_scenario",
arguments: {
executable: process.execPath,
arguments: ["-e", "process.stdout.write('captured')"],
working_directory: root,
filesystem_observation_paths: [observedRoot],
timeout_ms: 10_000,
finalization_ms: 250,
},
});
try {
const failed = parseMcpToolError(await capture());
expect(injected.opens).toBe(1);
expect(injected.handle?.fd).toBeGreaterThanOrEqual(0);
expect(failed.error).toMatchObject({
code: "cleanup_incomplete",
details: {
resources: [injected.path],
partial_observation: {
observations: {
filesystem_snapshots: {
before: {
state: "available",
value: {
files: expect.arrayContaining([
expect.objectContaining({ path: "root_0:input.txt" }),
]),
},
},
},
},
},
},
});
const firstDelivered = delivery.mock.calls.at(-1)?.[0];
expect(firstDelivered?.ok).toBe(false);
if (firstDelivered === undefined || firstDelivered.ok)
throw new Error("Expected a delivered snapshot failure");
expect(firstDelivered.error.cleanupResources).toContain(injected.path);
const retryBlocked = parseMcpToolError(await capture());
expect(retryBlocked.error.code).toBe("cleanup_incomplete");
expect(retryBlocked.error.details).toMatchObject({
resources: [injected.path],
});
expect(injected.opens).toBe(1);
expect(injected.handle?.fd).toBeGreaterThanOrEqual(0);
injected.failClose = false;
const exactFd = injected.handle?.fd;
const closed = await resources.client.callTool({
name: "close_binary",
arguments: {},
});
expect(closed.isError, JSON.stringify(closed.content)).not.toBe(true);
expect(injected.handle?.fd).toBe(-1);
expect(exactFd).toBeGreaterThanOrEqual(0);
if (injected.handle === undefined)
throw new Error("Expected the retained snapshot file descriptor");
await expect(injected.handle.stat()).rejects.toMatchObject({
code: "EBADF",
});
expect(firstDelivered.error.cleanupIncomplete).toBe(false);
expect(firstDelivered.error.cleanupResources).toEqual([]);
const recovered = await capture();
expect(recovered.isError, JSON.stringify(recovered.content)).not.toBe(true);
expect(injected.opens).toBe(3);
expect(recovered.structuredContent).toMatchObject({
normalized_result: {
files_before: expect.arrayContaining([
expect.objectContaining({ path: "root_0:input.txt" }),
]),
files_after: expect.arrayContaining([
expect.objectContaining({ path: "root_0:input.txt" }),
]),
},
});
} finally {
injected.failClose = false;
if (injected.handle?.fd !== undefined && injected.handle.fd >= 0) {
const cleanup = await resources.client.callTool({
name: "close_binary",
arguments: {},
});
expect(cleanup.isError, JSON.stringify(cleanup.content)).not.toBe(true);
}
delivery.mockRestore();
await resources.client.close();
await resources.server.close();
const closed = await resources.session.close();
expect(closed.ok).toBe(true);
await rm(root, { recursive: true, force: true });
injected.path = "";
injected.handle = undefined;
injected.nativeClose = undefined;
}
});
@@ -0,0 +1,197 @@
import { createHash } from "node:crypto";
import { access, mkdir, readFile, rm, writeFile } from "node:fs/promises";
import { execFile } from "node:child_process";
import { join, resolve } from "node:path";
import { pathToFileURL } from "node:url";
import { promisify } from "node:util";
import { expect } from "vitest";
import { z } from "zod";
import { analysisCliErrorEnvelopeSchema } from "../../../src/contracts/errorSchemas.js";
import { createTestTempDirectory } from "../../fixtures/temporaryDirectory.js";
import { itWithCaptureCapability } from "./processCaptureCapability.js";
const execFileAsync = promisify(execFile);
const CLI_TIMEOUT_MS = 30_000;
const snapshotClosePreload = `
import fs from "node:fs";
import { resolve } from "node:path";
import { syncBuiltinESMExports } from "node:module";
const cliEntry = process.env.REA_TEST_CLI_ENTRY;
if (cliEntry !== undefined && resolve(process.argv[1] ?? "") === cliEntry) {
const targetPath = process.env.REA_TEST_SNAPSHOT_FILE;
const markerPath = process.env.REA_TEST_SNAPSHOT_CLOSE_MARKER;
const originalOpen = fs.promises.open;
let closeAttempts = 0;
let released = false;
let descriptorClosed = false;
let closeFailureMessage;
const record = () => {
if (markerPath !== undefined) {
fs.writeFileSync(markerPath, JSON.stringify({
closeAttempts,
released,
descriptorClosed,
closeFailureMessage: closeFailureMessage ?? null,
}));
}
};
fs.promises.open = async function (path, ...args) {
const handle = await originalOpen.call(this, path, ...args);
if (targetPath === undefined || resolve(String(path)) !== targetPath)
return handle;
const nativeClose = handle.close.bind(handle);
const nativeStat = handle.stat.bind(handle);
handle.close = async function () {
closeAttempts++;
if (closeAttempts === 1) {
closeFailureMessage = "controlled final snapshot close failure";
record();
throw new Error(closeFailureMessage);
}
await nativeClose();
released = true;
try {
await nativeStat();
} catch (cause) {
descriptorClosed = cause?.code === "EBADF";
}
record();
};
return handle;
};
syncBuiltinESMExports();
}
`;
const snapshotCloseMarkerSchema = z
.object({
closeAttempts: z.number().int().nonnegative(),
released: z.boolean(),
descriptorClosed: z.boolean(),
closeFailureMessage: z.string().nullable(),
})
.strict();
itWithCaptureCapability(
"retains the completed final snapshot and retries its exact file descriptor in the compiled CLI",
async () => {
const fixtureRoot = await createTestTempDirectory(
"rea-process-snapshot-close-cli-",
);
const workingRoot = join(fixtureRoot, "working");
const scenarioPath = join(fixtureRoot, "scenario.json");
const preloadPath = join(fixtureRoot, "snapshot-close-preload.mjs");
const markerPath = join(fixtureRoot, "snapshot-close-marker.json");
const observedPath = join(workingRoot, "final-observation.txt");
const observedBytes = "completed final filesystem observation";
await mkdir(workingRoot);
await writeFile(
scenarioPath,
JSON.stringify({
executable: process.execPath,
arguments: [
"-e",
`require("node:fs").writeFileSync(${JSON.stringify(observedPath)}, ${JSON.stringify(observedBytes)}); process.stdout.write("capture completed\\n");`,
],
working_directory: workingRoot,
filesystem_observation_paths: [workingRoot],
}),
);
await writeFile(preloadPath, snapshotClosePreload);
try {
const cliEntry = resolve("scripts/rea.mjs");
const result = await execFileAsync(
process.execPath,
[cliEntry, "capture-process", scenarioPath, "--json"],
{
cwd: fixtureRoot,
env: {
...process.env,
NODE_OPTIONS: `--import=${pathToFileURL(preloadPath).href}`,
REA_TEST_CLI_ENTRY: cliEntry,
REA_TEST_SNAPSHOT_FILE: observedPath,
REA_TEST_SNAPSHOT_CLOSE_MARKER: markerPath,
},
timeout: CLI_TIMEOUT_MS,
},
).then(
({ stdout, stderr }) => ({ exitCode: 0, stdout, stderr }),
(cause: unknown) => {
if (!(cause instanceof Error)) throw cause;
const failure = cause as Error & {
readonly code?: number | string;
readonly stdout?: string;
readonly stderr?: string;
};
return {
exitCode: failure.code,
stdout: failure.stdout ?? "",
stderr: failure.stderr ?? "",
};
},
);
const marker = snapshotCloseMarkerSchema.parse(
JSON.parse(await readFile(markerPath, "utf8")),
);
expect(result.exitCode, result.stderr || result.stdout).toBe(1);
expect(
marker,
JSON.stringify({
exitCode: result.exitCode,
stdout: result.stdout,
stderr: result.stderr,
}),
).toEqual({
closeAttempts: 2,
released: true,
descriptorClosed: true,
closeFailureMessage: "controlled final snapshot close failure",
});
await expect(access(observedPath)).resolves.toBeUndefined();
const output = analysisCliErrorEnvelopeSchema.parse(
JSON.parse(result.stdout),
);
const expectedHash = createHash("sha256")
.update(observedBytes)
.digest("hex");
expect(output).toMatchObject({
code: "process_capture_failed",
details: {
partial_observation: {
observations: {
exit: {
state: "available",
value: { code: 0, reason: "exited" },
},
filesystem_snapshots: {
after: {
state: "available",
value: {
files: expect.arrayContaining([
expect.objectContaining({
path: "root_0:final-observation.txt",
sha256: expectedHash,
}),
]),
},
},
},
},
},
execution_failure: expect.stringContaining(
"controlled final snapshot close failure",
),
},
});
} finally {
await rm(fixtureRoot, { recursive: true, force: true });
}
},
CLI_TIMEOUT_MS,
);
@@ -0,0 +1,754 @@
import { createHash } from "node:crypto";
import { Dir, type Dirent } from "node:fs";
import { mkdir, mkdtemp, open, rm, stat, writeFile } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { afterEach, expect, it, vi } from "vitest";
const injected = vi.hoisted(() => ({
filePath: "",
fileCloseFailure: undefined as Error | undefined,
fileCloseRejectionEnabled: false,
fileCloseRejection: undefined as unknown,
fileStatFailure: undefined as Error | undefined,
openedPaths: [] as string[],
abortOnClose: undefined as (() => void) | undefined,
abortOnStat: undefined as (() => void) | undefined,
directoryPath: "",
directoryCloseFailure: undefined as Error | undefined,
fileHandle: undefined as Awaited<ReturnType<typeof open>> | undefined,
restoreFileHandle: undefined as (() => void) | undefined,
nativeFileClose: undefined as (() => Promise<void>) | undefined,
nativeFileStat: undefined as
| (() => Promise<Awaited<ReturnType<typeof stat>>>)
| undefined,
directory: undefined as Dir | undefined,
nativeDirectoryClose: undefined as
| ((callback: (error?: NodeJS.ErrnoException | null) => void) => void)
| undefined,
}));
vi.mock("node:fs/promises", async (importOriginal) => {
const original = await importOriginal<typeof import("node:fs/promises")>();
return {
...original,
open: async (...args: Parameters<typeof original.open>) => {
const handle = await original.open(...args);
const path = String(args[0]);
injected.openedPaths.push(path);
const statFailure = injected.fileStatFailure;
if (path !== injected.filePath) return handle;
const nativeClose = handle.close.bind(handle);
const nativeStat = handle.stat.bind(handle);
if (path === injected.filePath) {
injected.fileHandle = handle;
injected.nativeFileClose = nativeClose;
injected.nativeFileStat = nativeStat;
injected.restoreFileHandle = () => {
handle.close = nativeClose;
handle.stat = nativeStat as typeof handle.stat;
};
}
handle.close = async () => {
injected.abortOnClose?.();
if (path === injected.filePath && injected.fileCloseRejectionEnabled)
throw injected.fileCloseRejection;
const currentCloseFailure = injected.fileCloseFailure;
if (currentCloseFailure !== undefined) throw currentCloseFailure;
await nativeClose();
};
if (statFailure !== undefined)
handle.stat = async () => {
injected.abortOnStat?.();
throw statFailure;
};
return handle;
},
opendir: async (...args: Parameters<typeof original.opendir>) => {
const directory = await original.opendir(...args);
if (args[0] === injected.directoryPath) injected.directory = directory;
return directory;
},
};
});
import { AnalysisCleanupLease } from "../../../src/artifacts/AnalysisCleanupLease.js";
import { parseProcessScenario } from "../../../src/domain/process/processScenario.js";
import {
hashFile,
snapshotRoots,
ProcessCaptureSnapshotFailure,
} from "../../../src/process/capture/FilesystemSnapshot.js";
import { ProcessCaptureReadFailure } from "../../../src/process/capture/ProcessCaptureRead.js";
import { ProcessCaptureResourceScope } from "../../../src/process/capture/ProcessCaptureLifecycle.js";
const roots: string[] = [];
const trackedRoot = async (): Promise<string> => {
const root = await mkdtemp(join(tmpdir(), "rea-snapshot-handle-owner-"));
roots.push(root);
return root;
};
type DirectoryReadCallback = (
error: NodeJS.ErrnoException | null,
value: Dirent | null,
) => void;
type DirectoryCloseCallback = (error?: NodeJS.ErrnoException | null) => void;
const scenarioFor = (root: string) =>
parseProcessScenario({
executable: process.execPath,
working_directory: root,
filesystem_observation_paths: [root],
});
const expectSnapshotFailure = (cause: unknown) => {
expect(cause).toBeInstanceOf(Error);
expect(cause).toMatchObject({
cleanupLeases: expect.any(Array),
});
const failure = cause as Error & {
cleanupIncomplete: boolean;
snapshot?: {
files: readonly { path: string; type?: string; sha256?: string | null }[];
truncated?: boolean;
completeRoots?: readonly string[];
coverage?: { enumeration_truncated?: boolean };
};
cleanupLeases: readonly AnalysisCleanupLease[];
};
expect(failure.cleanupLeases.length).toBeGreaterThan(0);
expect(failure.cleanupLeases[0]).toBeInstanceOf(AnalysisCleanupLease);
return failure;
};
afterEach(async () => {
if (injected.fileHandle !== undefined) {
injected.restoreFileHandle?.();
if (injected.fileHandle.fd >= 0 && injected.nativeFileClose !== undefined) {
try {
await injected.nativeFileClose();
} catch (cause: unknown) {
if (
injected.fileHandle.fd >= 0 ||
!(cause instanceof Error && "code" in cause && cause.code === "EBADF")
)
throw cause;
}
}
}
injected.fileHandle = undefined;
injected.restoreFileHandle = undefined;
injected.nativeFileClose = undefined;
injected.nativeFileStat = undefined;
injected.filePath = "";
injected.fileCloseFailure = undefined;
injected.fileCloseRejectionEnabled = false;
injected.fileCloseRejection = undefined;
injected.fileStatFailure = undefined;
injected.openedPaths = [];
injected.abortOnClose = undefined;
injected.abortOnStat = undefined;
injected.directoryPath = "";
injected.directoryCloseFailure = undefined;
injected.directory = undefined;
const closeDirectory = injected.nativeDirectoryClose;
injected.nativeDirectoryClose = undefined;
if (closeDirectory !== undefined) {
await new Promise<void>((resolve, reject) => {
try {
closeDirectory((error) => {
if (
error !== undefined &&
error !== null &&
error.code !== "ERR_DIR_CLOSED"
) {
reject(error);
} else {
resolve();
}
});
} catch (cause: unknown) {
reject(cause);
}
});
}
await Promise.all(
roots.splice(0).map((root) => rm(root, { recursive: true, force: true })),
);
});
it("retains a completed file hash and its exact descriptor owner when close fails", async () => {
const root = await trackedRoot();
const path = join(root, "input");
const bytes = "retained file bytes";
await writeFile(path, bytes);
injected.filePath = path;
injected.fileCloseFailure = new Error("injected file close denial");
let cause: unknown;
try {
await snapshotRoots(scenarioFor(path));
} catch (error: unknown) {
cause = error;
}
expect(cause).not.toBe(injected.fileCloseFailure);
const failure = expectSnapshotFailure(cause);
expect(failure.cause).toBe(injected.fileCloseFailure);
expect(failure.snapshot?.files).toEqual([
expect.objectContaining({
path: "root_0:.",
type: "file",
sha256: createHash("sha256").update(bytes).digest("hex"),
}),
]);
const [lease] = failure.cleanupLeases;
expect(lease).toBeDefined();
if (lease === undefined) throw new Error("file cleanup lease missing");
await expect(injected.nativeFileStat?.()).resolves.toMatchObject({
isFile: expect.any(Function),
});
injected.fileCloseFailure = undefined;
await lease.release();
expect(lease.released).toBe(true);
await expect(injected.nativeFileStat?.()).rejects.toMatchObject({
code: "EBADF",
});
});
it("exposes a completed direct hash outcome while retaining its failed descriptor close", async () => {
const root = await trackedRoot();
const path = join(root, "input");
const bytes = "direct hash outcome bytes";
await writeFile(path, bytes);
injected.filePath = path;
injected.fileCloseFailure = new Error("injected direct hash close denial");
const scope = new ProcessCaptureResourceScope();
let cause: unknown;
try {
await hashFile(path, await stat(path), 1024, undefined, scope);
} catch (error: unknown) {
cause = error;
}
expect(cause).toBeInstanceOf(ProcessCaptureReadFailure);
const failure = cause as ProcessCaptureReadFailure<{
readonly state: "hashed";
readonly sha256: string;
}>;
expect(failure.outcome).toEqual({
kind: "completed",
value: {
state: "hashed",
sha256: createHash("sha256").update(bytes).digest("hex"),
},
});
expect(failure.cleanupCause).toBe(injected.fileCloseFailure);
expect(failure.cleanupLeases).toHaveLength(1);
await expect(scope.retryCleanup()).rejects.toMatchObject({
cleanupIncomplete: true,
});
injected.fileCloseFailure = undefined;
await scope.retryCleanup();
expect(failure.cleanupIncomplete).toBe(false);
await expect(injected.nativeFileStat?.()).rejects.toMatchObject({
code: "EBADF",
});
await scope.close();
});
it("preserves a primary file stat failure when descriptor close also fails", async () => {
const root = await trackedRoot();
const path = join(root, "input");
await writeFile(path, "retained file bytes");
const primary = new Error("injected file stat failure");
injected.filePath = path;
injected.fileStatFailure = primary;
injected.fileCloseFailure = new Error("injected file close denial");
let cause: unknown;
try {
await snapshotRoots(scenarioFor(path));
} catch (error: unknown) {
cause = error;
}
expect(cause).not.toBe(injected.fileCloseFailure);
const failure = expectSnapshotFailure(cause);
expect(failure.cause).toBe(primary);
expect(failure.cleanupLeases).toHaveLength(1);
expect(failure.snapshot).toMatchObject({
files: [],
truncated: true,
completeRoots: [],
coverage: { enumeration_truncated: true },
});
const [lease] = failure.cleanupLeases;
expect(lease).toBeDefined();
if (lease === undefined) throw new Error("file cleanup lease missing");
injected.fileCloseFailure = undefined;
await lease.release();
expect(lease.released).toBe(true);
await expect(injected.nativeFileStat?.()).rejects.toMatchObject({
code: "EBADF",
});
});
it("keeps the physical file error primary when cancellation arrives during failed close", async () => {
const root = await trackedRoot();
const path = join(root, "input");
await writeFile(path, "primary file and close cancellation");
const primary = new Error("injected physical file stat failure");
const reason = new Error("cancel during final file close");
injected.filePath = path;
injected.fileStatFailure = primary;
injected.fileCloseFailure = new Error("injected file close denial");
const controller = new AbortController();
injected.abortOnClose = () => controller.abort(reason);
const scope = new ProcessCaptureResourceScope();
let cause: unknown;
try {
await hashFile(path, await stat(path), 1024, controller.signal, scope);
} catch (error: unknown) {
cause = error;
}
expect(cause).toBeInstanceOf(ProcessCaptureReadFailure);
const failure = cause as ProcessCaptureReadFailure<unknown>;
expect(failure.cause).toBe(primary);
expect(failure.outcome).toMatchObject({ kind: "failed", cause: primary });
expect(failure.cleanupCause).toBe(injected.fileCloseFailure);
expect(failure.userCategory).toBeUndefined();
expect(failure.cleanupIncomplete).toBe(true);
expect(failure.cleanupLeases).toHaveLength(1);
await expect(scope.retryCleanup()).rejects.toMatchObject({
cleanupIncomplete: true,
});
expect(failure.cause).toBe(primary);
expect(failure.cleanupIncomplete).toBe(true);
injected.fileCloseFailure = undefined;
injected.abortOnClose = undefined;
await scope.retryCleanup();
expect(failure.cause).toBe(primary);
expect(failure.cleanupIncomplete).toBe(false);
await scope.close();
});
it("keeps a completed file hash in the partial snapshot when cancellation arrives during close", async () => {
const root = await trackedRoot();
const first = join(root, "a-first");
const second = join(root, "b-not-visited");
const firstBytes = "completed before final-close cancellation";
await writeFile(first, firstBytes);
await writeFile(second, "not visited after cancellation");
injected.filePath = first;
injected.fileCloseFailure = new Error(
"injected close failure after cancellation",
);
const controller = new AbortController();
const reason = new Error("cancel during snapshot file close");
injected.abortOnClose = () => controller.abort(reason);
const scope = new ProcessCaptureResourceScope();
let cause: unknown;
try {
await scope.run(() =>
snapshotRoots(scenarioFor(root), controller.signal, undefined, scope),
);
} catch (error: unknown) {
cause = error;
}
expect(cause).toBeInstanceOf(ProcessCaptureSnapshotFailure);
const failure = cause as ProcessCaptureSnapshotFailure;
expect(failure.cause).toBe(reason);
expect(failure.userCategory).toBe("cancelled");
expect(failure.snapshot).toMatchObject({
truncated: true,
completeRoots: [],
coverage: { enumeration_truncated: true },
});
expect(failure.snapshot?.files).toContainEqual(
expect.objectContaining({
path: "root_0:a-first",
sha256: createHash("sha256").update(firstBytes).digest("hex"),
}),
);
expect(failure.snapshot?.files).not.toContainEqual(
expect.objectContaining({ path: "root_0:b-not-visited" }),
);
expect(failure.cleanupLeases).toHaveLength(1);
expect(failure.cleanupIncomplete).toBe(true);
injected.fileCloseFailure = undefined;
injected.abortOnClose = undefined;
await scope.retryCleanup();
expect(failure.cleanupIncomplete).toBe(false);
await scope.close();
});
it("keeps a completed file hash when cancellation arrives during confirmed close", async () => {
const root = await trackedRoot();
const first = join(root, "a-first");
const second = join(root, "b-not-visited");
const firstBytes = "completed before confirmed-close cancellation";
await writeFile(first, firstBytes);
await writeFile(second, "not visited after cancellation");
injected.filePath = first;
const controller = new AbortController();
const reason = new Error("cancel during confirmed snapshot file close");
injected.abortOnClose = () => controller.abort(reason);
const scope = new ProcessCaptureResourceScope();
let cause: unknown;
try {
await scope.run(() =>
snapshotRoots(scenarioFor(root), controller.signal, undefined, scope),
);
} catch (error: unknown) {
cause = error;
}
expect(cause).toBeInstanceOf(ProcessCaptureSnapshotFailure);
const failure = cause as ProcessCaptureSnapshotFailure;
expect(failure.cause).toBe(reason);
expect(failure.userCategory).toBe("cancelled");
expect(failure.snapshot).toMatchObject({
truncated: true,
completeRoots: [],
coverage: { enumeration_truncated: true },
});
expect(failure.snapshot?.files).toContainEqual(
expect.objectContaining({
path: "root_0:a-first",
sha256: createHash("sha256").update(firstBytes).digest("hex"),
}),
);
expect(failure.cleanupLeases).toEqual([]);
expect(failure.cleanupIncomplete).toBe(false);
await scope.close();
});
it.each([
{ label: "undefined", rejection: undefined },
{ label: "string", rejection: "injected non-Error close rejection" },
])(
"keeps cleanup incomplete for a $label descriptor close rejection",
async ({ rejection }) => {
const root = await trackedRoot();
const path = join(root, "input");
await writeFile(path, "non-Error close rejection bytes");
injected.filePath = path;
injected.fileCloseRejectionEnabled = true;
injected.fileCloseRejection = rejection;
const scope = new ProcessCaptureResourceScope();
let cause: unknown;
try {
await snapshotRoots(scenarioFor(path), undefined, undefined, scope);
} catch (error: unknown) {
cause = error;
}
expect(cause).toBeInstanceOf(ProcessCaptureSnapshotFailure);
const failure = cause as ProcessCaptureSnapshotFailure;
expect(failure.readFailures[0]?.outcome.kind).toBe("completed");
expect(failure.readFailures[0]?.cleanupCause).toBe(rejection);
if (typeof rejection === "string")
expect(failure.message).toContain(rejection);
expect(failure.cleanupIncomplete).toBe(true);
expect(failure.cleanupLeases).toHaveLength(1);
await expect(scope.retryCleanup()).rejects.toMatchObject({
cleanupIncomplete: true,
});
expect(failure.cleanupIncomplete).toBe(true);
injected.fileCloseRejectionEnabled = false;
await scope.retryCleanup();
expect(failure.cleanupIncomplete).toBe(false);
await scope.close();
},
);
it("retains the full directory snapshot and exact directory owner when close fails", async () => {
const root = await trackedRoot();
const child = join(root, "observed");
await mkdir(child);
await writeFile(join(child, "payload"), "complete observation");
injected.directoryPath = child;
injected.directoryCloseFailure = new Error("injected directory close denial");
const originalClose = Dir.prototype.close;
let ownedDirectory: Dir | undefined;
let nativeClose: ((callback: DirectoryCloseCallback) => void) | undefined;
function closeWithFailure(this: Dir): Promise<void>;
function closeWithFailure(this: Dir, callback: DirectoryCloseCallback): void;
function closeWithFailure(
this: Dir,
callback?: DirectoryCloseCallback,
): void | Promise<void> {
const boundClose = originalClose.bind(this);
if (this !== injected.directory) {
if (callback !== undefined) return boundClose(callback);
return new Promise<void>((resolve, reject) =>
boundClose((error) =>
error === undefined || error === null ? resolve() : reject(error),
),
);
}
ownedDirectory = this;
nativeClose ??= (done: DirectoryCloseCallback) =>
originalClose.call(this, done);
if (callback !== undefined) {
queueMicrotask(() => callback(injected.directoryCloseFailure));
return;
}
return Promise.reject(injected.directoryCloseFailure);
}
Dir.prototype.close = closeWithFailure;
const scope = new ProcessCaptureResourceScope();
let cause: unknown;
try {
await scope.run(() =>
snapshotRoots(scenarioFor(root), undefined, undefined, scope),
);
} catch (error: unknown) {
cause = error;
} finally {
Dir.prototype.close = originalClose;
injected.directory = ownedDirectory;
injected.nativeDirectoryClose = nativeClose;
}
expect(cause).not.toBe(injected.directoryCloseFailure);
const failure = expectSnapshotFailure(cause);
expect(failure.cause).toBe(injected.directoryCloseFailure);
expect(failure.snapshot?.files).toEqual(
expect.arrayContaining([
expect.objectContaining({ path: "root_0:observed" }),
expect.objectContaining({ path: "root_0:observed/payload" }),
]),
);
expect(failure.cleanupLeases).toHaveLength(1);
await expect(
scope.run(async () => "unexpected replacement capture"),
).rejects.toMatchObject({ cleanupIncomplete: true });
await expect(scope.close()).rejects.toMatchObject({
cleanupIncomplete: true,
});
expect(failure.cleanupIncomplete).toBe(true);
expect(failure.cleanupLeases[0]?.released).toBe(false);
expect(ownedDirectory).toBeDefined();
});
it("preserves a primary directory read failure when final close also fails", async () => {
const root = await trackedRoot();
const selected = join(root, "selected");
await mkdir(selected);
await writeFile(
join(selected, "payload"),
"never visited after read failure",
);
injected.directoryPath = selected;
injected.directoryCloseFailure = new Error("injected directory close denial");
const primary = new Error("injected directory read failure");
const originalRead = Dir.prototype.read;
const originalClose = Dir.prototype.close;
let ownedDirectory: Dir | undefined;
let nativeClose: ((callback: DirectoryCloseCallback) => void) | undefined;
function failRead(this: Dir): Promise<Dirent | null>;
function failRead(this: Dir, callback: DirectoryReadCallback): void;
function failRead(
this: Dir,
callback?: DirectoryReadCallback,
): void | Promise<Dirent | null> {
if (this !== injected.directory) {
if (callback !== undefined) return originalRead.call(this, callback);
return new Promise<Dirent | null>((resolve, reject) =>
originalRead.call(this, (error, value) =>
error === null ? resolve(value) : reject(error),
),
);
}
if (callback !== undefined) {
queueMicrotask(() => callback(primary as NodeJS.ErrnoException, null));
return;
}
return Promise.reject(primary);
}
function closeWithFailure(this: Dir): Promise<void>;
function closeWithFailure(this: Dir, callback: DirectoryCloseCallback): void;
function closeWithFailure(
this: Dir,
callback?: DirectoryCloseCallback,
): void | Promise<void> {
const boundClose = originalClose.bind(this);
if (this !== injected.directory) {
if (callback !== undefined) return boundClose(callback);
return new Promise<void>((resolve, reject) =>
boundClose((error) =>
error === undefined || error === null ? resolve() : reject(error),
),
);
}
ownedDirectory = this;
nativeClose ??= (done: DirectoryCloseCallback) =>
originalClose.call(this, done);
if (callback !== undefined) {
queueMicrotask(() => callback(injected.directoryCloseFailure));
return;
}
return Promise.reject(injected.directoryCloseFailure);
}
Dir.prototype.read = failRead;
Dir.prototype.close = closeWithFailure;
let cause: unknown;
try {
await snapshotRoots(scenarioFor(selected));
} catch (error: unknown) {
cause = error;
} finally {
Dir.prototype.read = originalRead;
Dir.prototype.close = originalClose;
injected.directory = ownedDirectory;
injected.nativeDirectoryClose = nativeClose;
}
expect(cause).not.toBe(injected.directoryCloseFailure);
const failure = expectSnapshotFailure(cause);
expect(failure.cause).toBe(primary);
expect(failure.snapshot).toMatchObject({
files: [expect.objectContaining({ path: "root_0:.", type: "directory" })],
truncated: true,
completeRoots: [],
coverage: { enumeration_truncated: true },
});
expect(failure.cleanupLeases).toHaveLength(1);
});
it("keeps a later file hash, blocks scope admission on the first file owner, then resumes after retry", async () => {
const root = await trackedRoot();
const first = join(root, "a-first");
const second = join(root, "b-second");
const firstBytes = "first completed hash";
const secondBytes = "second completed hash";
await writeFile(first, firstBytes);
await writeFile(second, secondBytes);
injected.filePath = first;
injected.fileCloseFailure = new Error("injected first file close denial");
const scope = new ProcessCaptureResourceScope();
let cause: unknown;
try {
await scope.run(() =>
snapshotRoots(scenarioFor(root), undefined, undefined, scope),
);
} catch (error: unknown) {
cause = error;
}
expect(cause).not.toBe(injected.fileCloseFailure);
const failure = expectSnapshotFailure(cause);
expect(failure.snapshot?.files).toEqual(
expect.arrayContaining([
expect.objectContaining({
path: "root_0:a-first",
sha256: createHash("sha256").update(firstBytes).digest("hex"),
}),
expect.objectContaining({
path: "root_0:b-second",
sha256: createHash("sha256").update(secondBytes).digest("hex"),
}),
]),
);
expect(failure.cleanupLeases).toHaveLength(1);
expect(injected.openedPaths).toEqual([first, second]);
await expect(
scope.run(() =>
snapshotRoots(scenarioFor(root), undefined, undefined, scope),
),
).rejects.toMatchObject({ cleanupIncomplete: true });
expect(injected.openedPaths).toEqual([first, second]);
injected.fileCloseFailure = undefined;
await scope.retryCleanup();
expect(failure.cleanupIncomplete).toBe(false);
await expect(
scope.run(() =>
snapshotRoots(scenarioFor(root), undefined, undefined, scope),
),
).resolves.toMatchObject({ files: expect.any(Array) });
expect(injected.openedPaths).toEqual([first, second, first, second]);
await scope.close();
});
it("preserves an IO failure that arrives with cancellation before descriptor cleanup", async () => {
const root = await trackedRoot();
const path = join(root, "input");
await writeFile(path, "failed metadata with late cancellation");
const primary = new Error("physical metadata failure");
const reason = new Error("cancelled while metadata rejected");
const controller = new AbortController();
injected.filePath = path;
injected.fileStatFailure = primary;
injected.abortOnStat = () => controller.abort(reason);
let failure: unknown;
try {
await hashFile(path, await stat(path), 1024, controller.signal);
} catch (cause: unknown) {
failure = cause;
}
expect(failure).toBeInstanceOf(ProcessCaptureReadFailure);
expect(failure).toMatchObject({
cause: primary,
outcome: { kind: "failed", cause: primary },
cleanupIncomplete: false,
});
expect((failure as ProcessCaptureReadFailure).userCategory).toBeUndefined();
expect(injected.fileHandle?.fd).toBe(-1);
});
it.each([false, true])(
"preserves complete root-file coverage after close-time cancellation (close denied: %s)",
async (denied) => {
const root = await trackedRoot();
const path = join(root, "input");
const bytes = "all selected snapshot facts completed";
await writeFile(path, bytes);
injected.filePath = path;
injected.fileCloseFailure = denied
? new Error("close denied after full root read")
: undefined;
const controller = new AbortController();
const reason = new Error("cancelled during final root-file close");
injected.abortOnClose = () => controller.abort(reason);
const scope = new ProcessCaptureResourceScope();
let failure: unknown;
try {
await scope.run(() =>
snapshotRoots(scenarioFor(path), controller.signal, undefined, scope),
);
} catch (cause: unknown) {
failure = cause;
}
expect(failure).toBeInstanceOf(ProcessCaptureSnapshotFailure);
expect(failure).toMatchObject({
cause: reason,
userCategory: "cancelled",
cleanupIncomplete: denied,
snapshot: {
truncated: false,
completeRoots: ["root_0"],
coverage: { enumeration_truncated: false },
files: [
expect.objectContaining({
sha256: createHash("sha256").update(bytes).digest("hex"),
}),
],
},
});
injected.abortOnClose = undefined;
injected.fileCloseFailure = undefined;
await scope.close();
expect(injected.fileHandle?.fd).toBe(-1);
},
);
@@ -0,0 +1,228 @@
import { createHash } from "node:crypto";
import { mkdtemp, open, rm, stat, writeFile } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { afterEach, expect, it, vi } from "vitest";
const injected = vi.hoisted(() => ({
path: "",
closeFailure: undefined as Error | undefined,
primaryFailure: undefined as Error | undefined,
abortOnClose: undefined as (() => void) | undefined,
opened: [] as string[],
closed: [] as string[],
handles: [] as {
isOpen: () => boolean;
close: () => Promise<void>;
restore: () => void;
}[],
}));
vi.mock("node:fs/promises", async (importOriginal) => {
const original = await importOriginal<typeof import("node:fs/promises")>();
return {
...original,
open: async (...args: Parameters<typeof original.open>) => {
const handle = await original.open(...args);
const path = String(args[0]);
injected.opened.push(path);
if (path !== injected.path) return handle;
const nativeClose = handle.close.bind(handle);
injected.handles.push({
isOpen: () => handle.fd >= 0,
close: nativeClose,
restore: () => {
handle.close = nativeClose;
},
});
handle.close = async () => {
injected.closed.push(path);
injected.abortOnClose?.();
if (injected.closeFailure !== undefined) throw injected.closeFailure;
await nativeClose();
};
return handle;
},
};
});
import { ProcessCaptureResourceScope } from "../../../src/process/capture/ProcessCaptureLifecycle.js";
import { ProcessCaptureReadFailure } from "../../../src/process/capture/ProcessCaptureRead.js";
import { observeSelectedExecutable } from "../../../src/process/capture/ProcessCaptureLifecycle.js";
const roots: string[] = [];
const createExecutable = async (content: string): Promise<string> => {
const root = await mkdtemp(join(tmpdir(), "rea-selected-executable-owner-"));
roots.push(root);
const path = join(root, "selected-fixture");
await writeFile(path, content);
return path;
};
const caughtFailure = async (
operation: () => Promise<unknown>,
): Promise<ProcessCaptureReadFailure<unknown>> => {
let cause: unknown;
try {
await operation();
} catch (error: unknown) {
cause = error;
}
expect(cause).toBeInstanceOf(ProcessCaptureReadFailure);
return cause as ProcessCaptureReadFailure<unknown>;
};
afterEach(async () => {
for (const handle of injected.handles.splice(0)) {
handle.restore();
if (!handle.isOpen()) continue;
try {
await handle.close();
} catch (cause: unknown) {
if (
handle.isOpen() ||
!(cause instanceof Error && "code" in cause && cause.code === "EBADF")
)
throw cause;
}
}
injected.path = "";
injected.closeFailure = undefined;
injected.primaryFailure = undefined;
injected.abortOnClose = undefined;
injected.opened = [];
injected.closed = [];
await Promise.all(
roots.splice(0).map((root) => rm(root, { recursive: true, force: true })),
);
});
it("retains the completed executable observation and descriptor owner after close failure", async () => {
const bytes = "small selected executable fixture";
const path = await createExecutable(bytes);
injected.path = path;
injected.closeFailure = new Error(
"injected selected executable close denial",
);
const scope = new ProcessCaptureResourceScope();
const failure = await caughtFailure(() =>
scope.run(() =>
observeSelectedExecutable(path, undefined, undefined, scope),
),
);
expect(failure.outcome.kind).toBe("completed");
if (failure.outcome.kind === "completed")
expect(failure.outcome.value).toMatchObject({
sha256: createHash("sha256").update(bytes).digest("hex"),
identity: expect.any(Object),
reason: null,
});
expect(failure.cleanupCause).toBe(injected.closeFailure);
expect(failure.cleanupLeases).toHaveLength(1);
expect(injected.opened).toEqual([path]);
await expect(
scope.run(() =>
observeSelectedExecutable(path, undefined, undefined, scope),
),
).rejects.toMatchObject({ cleanupIncomplete: true });
expect(injected.opened).toEqual([path]);
injected.closeFailure = undefined;
await scope.retryCleanup();
expect(failure.cleanupIncomplete).toBe(false);
await expect(
scope.run(() =>
observeSelectedExecutable(path, undefined, undefined, scope),
),
).resolves.toMatchObject({
sha256: createHash("sha256").update(bytes).digest("hex"),
});
expect(injected.opened).toEqual([path, path]);
expect(injected.handles).toHaveLength(2);
expect(injected.handles.every((handle) => !handle.isOpen())).toBe(true);
await scope.close();
});
it("preserves selected executable IO failure when descriptor close also fails", async () => {
const path = await createExecutable(
"selected executable with primary IO failure",
);
const primary = new Error("injected selected executable metadata failure");
injected.path = path;
injected.closeFailure = new Error(
"injected selected executable close denial",
);
const scope = new ProcessCaptureResourceScope();
const fileSystem = {
open,
stat: async () => {
throw primary;
},
};
const failure = await caughtFailure(() =>
observeSelectedExecutable(path, undefined, fileSystem, scope),
);
expect(failure.outcome).toMatchObject({ kind: "failed", cause: primary });
expect(failure.cause).toBe(primary);
expect(failure.cleanupCause).toBe(injected.closeFailure);
expect(failure.cleanupLeases).toHaveLength(1);
injected.closeFailure = undefined;
await scope.retryCleanup();
expect(failure.cleanupIncomplete).toBe(false);
await scope.close();
});
it("keeps the exact cancellation reason when cancellation arrives during descriptor close", async () => {
const path = await createExecutable(
"selected executable cancelled during close",
);
const controller = new AbortController();
const reason = new Error("cancel during selected executable close");
const closeFailure = new Error("injected close denial after cancellation");
injected.path = path;
injected.closeFailure = closeFailure;
injected.abortOnClose = () => controller.abort(reason);
const scope = new ProcessCaptureResourceScope();
const failure = await caughtFailure(() =>
observeSelectedExecutable(path, controller.signal, undefined, scope),
);
expect(failure.outcome.kind).toBe("completed");
expect(failure.cause).toBe(reason);
expect(failure.cancellation).toEqual({ reason });
expect(failure.cleanupCause).toBe(closeFailure);
expect(failure.userCategory).toBe("cancelled");
expect(failure.cleanupLeases).toHaveLength(1);
injected.abortOnClose = undefined;
injected.closeFailure = undefined;
await scope.retryCleanup();
expect(failure.cleanupIncomplete).toBe(false);
await scope.close();
});
it("returns a completed executable observation when descriptor close succeeds without cancellation", async () => {
const bytes = "successful selected executable fixture";
const path = await createExecutable(bytes);
injected.path = path;
const scope = new ProcessCaptureResourceScope();
await expect(
scope.run(() =>
observeSelectedExecutable(path, undefined, undefined, scope),
),
).resolves.toMatchObject({
sha256: createHash("sha256").update(bytes).digest("hex"),
identity: expect.any(Object),
reason: null,
});
expect(injected.opened).toEqual([path]);
expect(injected.handles).toHaveLength(1);
expect(injected.handles[0]?.isOpen()).toBe(false);
await scope.close();
});