fix(process): retain capture resources and bound filesystem snapshots

Validation: integrated changes passed formatting, typecheck, lint, resolved module-boundary verification and git diff --check. Tests and real-provider workflows were not run.
This commit is contained in:
morluto committed 2026-10-11 00:03:23 +08:00
1 parent a84dea5500
commit d438675677
5 files changed
+333 -117

No files matched your search

+4
View File
@@ -187,6 +187,10 @@ Every capture requires `truncation_details`, with separate accounting for:
input, so raw omissions also limit rendered coverage.
- `filesystem_before` and `filesystem_after`: file-count/depth limits,
enumeration failures, whole-file hash budget and bytes successfully hashed.
Directory names are read incrementally within the remaining entry capacity.
When a directory exceeds that capacity, the retained subset follows filesystem
enumeration order and is then sorted; it is explicitly incomplete and cannot
establish that an omitted path is absent.
Each retained regular file without a digest has an aliased path, size,
remaining budget and reason: `file_bytes_budget`,
`file_changed_or_short_read`, or `file_unavailable`. Its `system_code` is
+20 -26
View File
@@ -1,4 +1,3 @@
import { readFile } from "node:fs/promises";
import { z } from "zod";
import { AnalysisError } from "../../domain/analysisErrorBase.js";
@@ -17,9 +16,10 @@ import { compareProcessCaptures } from "../../domain/process/processComparison.j
import { parseProcessCapture } from "../../domain/process/processCaptureParsing.js";
import { createProgressReporter } from "../ProgressReporter.js";
import { captureProcessScenario } from "../../process/capture/ProcessHarness.js";
import { ProcessCaptureResourceScope } from "../../process/capture/ProcessCaptureLifecycle.js";
import { PROCESS_PROVIDER } from "../../domain/process/processEvidenceProvider.js";
import { createProcessCaptureEvidence } from "./ProcessEvidence.js";
import { JSON_BYTE_ORDER_MARK_MESSAGE } from "../Utf8JsonInput.js";
import { readJsonFile } from "../JsonFiles.js";
/** Safe process-command failure returned to the CLI adapter. */
export interface ProcessCliErrorOutput {
@@ -63,6 +63,7 @@ export const captureProcessScenarioFile = async (
},
{ minimumIntervalMs: 100 },
);
const resourceScope = new ProcessCaptureResourceScope();
const captured = await captureProcessScenario(
parsed.data,
signal,
@@ -71,7 +72,14 @@ export const captureProcessScenarioFile = async (
undefined,
undefined,
progress,
resourceScope,
);
try {
await resourceScope.close();
} catch (cause: unknown) {
if (!captured.ok) return cliAnalysisError(captured.error);
throw cause;
}
if (!captured.ok) return cliAnalysisError(captured.error);
return createProcessCaptureEvidence(parsed.data, captured.value);
} catch (cause: unknown) {
@@ -178,30 +186,16 @@ const invalidCaptureEvidence = (cause: unknown): ProcessCliFailure =>
);
const readJson = async (path: string): Promise<unknown> => {
let bytes;
try {
bytes = await readFile(path);
} catch (cause: unknown) {
throw new ProcessCliFailure(
"invalid_input",
`Process input file could not be read: ${path} (${describeValidationFailure(cause)}). Check the reported filesystem constraint and retry.`,
);
}
try {
const text = new TextDecoder("utf-8", {
fatal: true,
ignoreBOM: true,
}).decode(bytes);
if (text.startsWith("\uFEFF"))
throw new Error(JSON_BYTE_ORDER_MARK_MESSAGE);
const parsed: unknown = JSON.parse(text);
return parsed;
} catch (cause: unknown) {
throw new ProcessCliFailure(
"invalid_input",
`Process input file is not valid UTF-8 JSON: ${path} (${describeValidationFailure(cause)}). Repair the file, then try again.`,
);
}
const loaded = await readJsonFile(path);
if (loaded.ok) return loaded.value;
if (loaded.error._tag !== "EvidenceFileError") throw loaded.error;
const detail = describeValidationFailure(loaded.error.cause);
throw new ProcessCliFailure(
"invalid_input",
loaded.error.reason === "invalid-json"
? `Process input file is not valid UTF-8 JSON: ${path} (${detail}). Repair the file, then try again.`
: `Process input file could not be read: ${path} (${detail}). Check the reported filesystem constraint and retry.`,
);
};
class ProcessCliFailure extends Error {
+30 -4
View File
@@ -1,6 +1,6 @@
import { createHash } from "node:crypto";
import { constants as fsConstants } from "node:fs";
import { lstat, open, readdir, readlink } from "node:fs/promises";
import { lstat, open, opendir, readlink } from "node:fs/promises";
import { dirname, join, relative, resolve } from "node:path";
import type { FileHandle } from "node:fs/promises";
import type { ProcessScenario } from "../../domain/process/processScenario.js";
@@ -153,6 +153,7 @@ export const snapshotRoots = async (
const entries: FileState[] = [];
const completeRoots: string[] = [];
let remainingBytes = scenario.limits.file_bytes;
let pendingNames = 0;
let truncated = false;
const enumerationReasons = new Set<
FilesystemCoverage["enumeration_reasons"][number]
@@ -247,15 +248,40 @@ export const snapshotRoots = async (
symlink_target: null,
});
if (type !== "directory") return true;
const children = await readdir(path);
const children: string[] = [];
let complete = true;
const remainingEntries =
scenario.limits.files - entries.length - pendingNames;
const directory = await opendir(path);
for await (const child of directory) {
signal?.throwIfAborted();
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;
}
const afterRead = await lstat(path);
if (!hasSameIdentity(stats, afterRead, "directory")) {
pendingNames -= children.length;
truncated = true;
enumerationReasons.add("identity_changed");
return false;
}
let complete = true;
for (const child of children.sort()) {
children.sort().reverse();
while (children.length > 0) {
const child = children.pop();
if (child === undefined) break;
pendingNames -= 1;
if (!(await visit(root, rootAlias, join(path, child), depth + 1)))
complete = false;
}
+195 -56
View File
@@ -806,7 +806,7 @@ const releaseCapturedProcessGroup = async (
};
};
export const releaseProcessResources = async (options: {
export interface ProcessCaptureResourceCleanupOptions {
readonly timers: ReadonlySet<ProcessTimer>;
readonly terminal: Pick<IPty, "pid"> | undefined;
readonly renderer: TerminalRenderer | undefined;
@@ -815,63 +815,197 @@ export const releaseProcessResources = async (options: {
readonly captureBaseline?: ProcessOwnershipBaseline;
readonly sampledProcessGroupIds?: readonly number[];
readonly host?: ProcessCaptureCleanupHost;
}): Promise<ProcessCaptureCleanupReport> => {
const host = options.host ?? processCaptureCleanupHost;
for (const timer of options.timers) timer.cancel();
let terminalRenderer: ProcessCaptureCleanupReport["terminal_renderer"] = {
state: options.renderer === undefined ? "not_required" : "cleaned",
reason: null,
};
try {
await options.renderer?.dispose();
} catch (cause: unknown) {
terminalRenderer = {
state: "failed",
reason:
cause instanceof Error
? cause.message || cause.name || "renderer dispose failed"
: "renderer dispose failed",
}
/** One capture's exact resource owner, with completion tracked per resource. */
class ProcessCaptureResourceOwner {
readonly #host: ProcessCaptureCleanupHost;
#terminalRenderer:
| ProcessCaptureCleanupReport["terminal_renderer"]
| undefined;
#ownedProcessGroup:
| ProcessCaptureCleanupReport["owned_process_group"]
| undefined;
#temporaryRoot: ProcessCaptureCleanupReport["temporary_root"] | undefined;
constructor(private readonly options: ProcessCaptureResourceCleanupOptions) {
this.#host = options.host ?? processCaptureCleanupHost;
if (options.renderer === undefined)
this.#terminalRenderer = { state: "not_required", reason: null };
if (options.terminal === undefined)
this.#ownedProcessGroup = { state: "not_required", reason: null };
}
async close(): Promise<ProcessCaptureCleanupReport> {
for (const timer of this.options.timers) timer.cancel();
await this.#releaseRenderer();
await this.#releaseOwnedProcessGroup();
await this.#releaseTemporaryRoot();
return {
owned_process_group: this.#ownedProcessGroup ?? {
state: "unverified",
reason: "owned process cleanup was not attempted",
},
terminal_renderer: this.#terminalRenderer ?? {
state: "unverified",
reason: "terminal renderer cleanup was not attempted",
},
temporary_root: this.#temporaryRoot ?? {
state: "unverified",
reason: "temporary root cleanup was not attempted",
},
};
}
const ownedProcessGroup =
options.terminal === undefined
? { state: "not_required" as const, reason: null }
: await releaseCapturedProcessGroup(
{
runId: options.runId,
leaderPid: options.terminal.pid,
processGroupId: options.terminal.pid,
sweepTokenOwnedProcesses: true,
...(options.sampledProcessGroupIds === undefined
? {}
: { sampledProcessGroupIds: options.sampledProcessGroupIds }),
...(options.captureBaseline === undefined
? {}
: { captureBaseline: options.captureBaseline }),
},
host,
async #releaseRenderer(): Promise<void> {
if (
this.options.renderer === undefined ||
this.#terminalRenderer?.state === "cleaned" ||
this.#terminalRenderer?.state === "not_required"
)
return;
try {
await this.options.renderer.dispose();
this.#terminalRenderer = { state: "cleaned", reason: null };
} catch (cause: unknown) {
this.#terminalRenderer = {
state: "failed",
reason: cleanupFailureMessage(cause, "renderer dispose failed"),
};
}
}
async #releaseOwnedProcessGroup(): Promise<void> {
if (
this.options.terminal === undefined ||
this.#ownedProcessGroup?.state === "cleaned" ||
this.#ownedProcessGroup?.state === "not_required"
)
return;
this.#ownedProcessGroup = await releaseCapturedProcessGroup(
{
runId: this.options.runId,
leaderPid: this.options.terminal.pid,
processGroupId: this.options.terminal.pid,
sweepTokenOwnedProcesses: true,
...(this.options.sampledProcessGroupIds === undefined
? {}
: { sampledProcessGroupIds: this.options.sampledProcessGroupIds }),
...(this.options.captureBaseline === undefined
? {}
: { captureBaseline: this.options.captureBaseline }),
},
this.#host,
);
}
async #releaseTemporaryRoot(): Promise<void> {
if (this.#temporaryRoot?.state === "cleaned") return;
const owned = this.#ownedProcessGroup;
if (owned?.state === "failed" || owned?.state === "unverified") {
this.#temporaryRoot = {
state: "unverified",
reason: `temporary root retained because owned process cleanup is ${owned.state}: ${owned.reason ?? "ownership could not be verified"}`,
};
return;
}
try {
await this.#host.removeTemporaryRoot(this.options.temporaryRoot);
this.#temporaryRoot = { state: "cleaned", reason: null };
} catch (cause: unknown) {
this.#temporaryRoot = {
state: "failed",
reason: cleanupFailureMessage(cause, "temporary root removal failed"),
};
}
}
}
const cleanupFailureMessage = (cause: unknown, fallback: string): string =>
cause instanceof Error ? cause.message || cause.name || fallback : fallback;
/** Release one capture's resources, retaining the exact owner for scoped retry. */
export const releaseProcessResources = async (
options: ProcessCaptureResourceCleanupOptions,
): Promise<ProcessCaptureCleanupReport> =>
new ProcessCaptureResourceOwner(options).close();
/** Run-local owner collection for resources whose first cleanup did not finish. */
export class ProcessCaptureResourceScope {
readonly #pending = new Set<ProcessCaptureResourceOwner>();
readonly #active = new Set<Promise<unknown>>();
#draining: Promise<void> | undefined;
#closed = false;
/** Admit one capture before it starts acquiring resources. */
async run<T>(operation: () => Promise<T>): Promise<T> {
while (true) {
while (this.#draining !== undefined) await this.#draining;
if (this.#closed)
throw new ProcessCaptureError(
"process capture lifecycle is closing; start a new server session before capturing again",
);
let temporaryRoot: ProcessCaptureCleanupReport["temporary_root"] = {
state: "cleaned",
reason: null,
};
try {
await host.removeTemporaryRoot(options.temporaryRoot);
} catch (cause: unknown) {
temporaryRoot = {
state: "failed",
reason:
cause instanceof Error
? cause.message || cause.name || "temporary root removal failed"
: "temporary root removal failed",
};
if (this.#pending.size === 0) break;
await this.retryCleanup();
}
const running = Promise.resolve().then(operation);
this.#active.add(running);
try {
return await running;
} finally {
this.#active.delete(running);
}
}
return {
owned_process_group: ownedProcessGroup,
terminal_renderer: terminalRenderer,
temporary_root: temporaryRoot,
};
};
async release(
options: ProcessCaptureResourceCleanupOptions,
): Promise<ProcessCaptureCleanupReport> {
const owner = new ProcessCaptureResourceOwner(options);
const report = await owner.close();
if (cleanupReportFailure(report) !== undefined) {
this.#pending.add(owner);
}
return report;
}
/** Drain active captures, then retry their retained resources without closing admission. */
retryCleanup(): Promise<void> {
if (this.#draining !== undefined) return this.#draining;
this.#draining = this.#drainAndRetry().finally(() => {
this.#draining = undefined;
});
return this.#draining;
}
/** Stop admitting new captures, drain active runs, and retry all retained owners. */
close(): Promise<void> {
this.#closed = true;
return this.retryCleanup();
}
async #drainAndRetry(): Promise<void> {
while (this.#active.size > 0) await Promise.allSettled(this.#active);
const failures: string[] = [];
for (const owner of this.#pending) {
try {
const report = await owner.close();
if (cleanupReportFailure(report) === undefined)
this.#pending.delete(owner);
else
failures.push(cleanupReportFailure(report) ?? "cleanup incomplete");
} catch (cause: unknown) {
failures.push(
cause instanceof Error ? cause.message || cause.name : String(cause),
);
}
}
if (failures.length > 0)
throw new AggregateError(
failures,
`Process capture cleanup remains incomplete: ${failures.join("; ")}`,
);
}
}
export const cleanupReportFailure = (
report: ProcessCaptureCleanupReport,
@@ -1093,6 +1227,7 @@ export const prepareProcessCapture = async (
signal: AbortSignal | undefined,
captureSnapshot: typeof snapshotRoots = snapshotRoots,
host: ProcessPreparationHost = systemProcessPreparationHost,
resourceScope?: ProcessCaptureResourceScope,
): Promise<{
readonly temporaryRoot: string;
readonly runId: string;
@@ -1112,14 +1247,18 @@ export const prepareProcessCapture = async (
assertNotCancelled(signal);
return { temporaryRoot, runId, ownershipBaseline, before };
} catch (cause: unknown) {
const cleanup = await releaseProcessResources({
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);
resolveProcessResult(
undefined,
normalizeCaptureFailure(cause, signal),
+84 -31
View File
@@ -43,6 +43,8 @@ import {
type ProcessCaptureCleanupHost,
type ProcessCaptureObservationBuffer,
type PendingProcessCapture,
type ProcessCaptureResourceCleanupOptions,
ProcessCaptureResourceScope,
} from "./ProcessCaptureLifecycle.js";
import {
assertNotCancelled,
@@ -86,6 +88,7 @@ const cleanupFailedStartup = async (options: {
readonly cleanupHost?: ProcessCaptureCleanupHost;
readonly observations?: ProcessCaptureObservationBuffer;
readonly temporaryRoot: string;
readonly resourceScope?: ProcessCaptureResourceScope;
}): Promise<never> => {
const sampledProcessGroupIds =
options.terminal !== undefined &&
@@ -108,7 +111,7 @@ const cleanupFailedStartup = async (options: {
};
}
}
const cleanupFailure = await releaseProcessResources({
const cleanupOptions: ProcessCaptureResourceCleanupOptions = {
timers: options.timers,
terminal: options.terminal,
renderer: options.renderer,
@@ -117,12 +120,17 @@ const cleanupFailedStartup = async (options: {
...(sampledProcessGroupIds === undefined ? {} : { sampledProcessGroupIds }),
temporaryRoot: options.temporaryRoot,
...(options.cleanupHost === undefined ? {} : { host: options.cleanupHost }),
});
if (cleanupReportFailure(cleanupFailure) !== undefined) {
};
const cleanup =
options.resourceScope === undefined
? await releaseProcessResources(cleanupOptions)
: await options.resourceScope.release(cleanupOptions);
const cleanupFailure = cleanupReportFailure(cleanup);
if (cleanupFailure !== undefined) {
resolveProcessResult(
undefined,
options.cause,
cleanupFailure,
cleanup,
options.observations,
{
scenario: options.scenario,
@@ -158,6 +166,7 @@ interface StartCaptureRuntimeOptions {
readonly runId: string;
readonly ownershipBaseline: ProcessOwnershipBaseline;
readonly cleanupHost?: ProcessCaptureCleanupHost;
readonly resourceScope?: ProcessCaptureResourceScope;
readonly observationBuffer: ProcessCaptureObservationBuffer;
readonly onSpawn: (pid: number) => void;
readonly frames: TerminalFrame[];
@@ -251,6 +260,9 @@ const startCaptureRuntime = async (
: { cleanupHost: options.cleanupHost }),
observations: options.observationBuffer,
temporaryRoot: options.temporaryRoot,
...(options.resourceScope === undefined
? {}
: { resourceScope: options.resourceScope }),
});
}
};
@@ -271,6 +283,7 @@ const finishProcessRun = async (options: {
readonly rootPid?: number;
readonly progress: ProcessCaptureProgressTracker;
readonly progressCounts: () => ProcessCaptureProgressCounts;
readonly resourceScope?: ProcessCaptureResourceScope;
}): Promise<ProcessCapture> => {
options.progress.closeLive();
await options.stopSampler();
@@ -281,7 +294,7 @@ const finishProcessRun = async (options: {
options.runtime.terminal.pid,
options.samples,
).filter((groupId) => groupId !== options.runtime?.terminal.pid);
const cleanup = await releaseProcessResources({
const cleanupOptions: ProcessCaptureResourceCleanupOptions = {
timers: options.timers,
terminal: options.runtime?.terminal,
renderer: options.runtime?.renderer,
@@ -290,7 +303,11 @@ const finishProcessRun = async (options: {
...(sampledProcessGroupIds === undefined ? {} : { sampledProcessGroupIds }),
temporaryRoot: options.temporaryRoot,
...(options.cleanupHost === undefined ? {} : { host: options.cleanupHost }),
});
};
const cleanup =
options.resourceScope === undefined
? await releaseProcessResources(cleanupOptions)
: await options.resourceScope.release(cleanupOptions);
const observedExit =
options.observations?.exit.state === "available"
? options.observations.exit.value.reason
@@ -514,6 +531,7 @@ const runProcessScenario = async (
captureSnapshot: typeof snapshotRoots = snapshotRoots,
cleanupHost?: ProcessCaptureCleanupHost,
progress?: ProcessCaptureProgress,
resourceScope?: ProcessCaptureResourceScope,
): Promise<ProcessCapture> => {
const progressTracker = createProcessCaptureProgressTracker(progress);
const frames: TerminalFrame[] = [];
@@ -539,7 +557,13 @@ const runProcessScenario = async (
phase: "prepare",
counts: progressCounts(),
});
prepared = await prepareProcessCapture(scenario, signal, captureSnapshot);
prepared = await prepareProcessCapture(
scenario,
signal,
captureSnapshot,
undefined,
resourceScope,
);
} catch (cause: unknown) {
progressTracker.closeLive();
await progressTracker.report({
@@ -577,6 +601,7 @@ const runProcessScenario = async (
runId,
ownershipBaseline,
...(cleanupHost === undefined ? {} : { cleanupHost }),
...(resourceScope === undefined ? {} : { resourceScope }),
observationBuffer: observations,
onSpawn: (pid) => {
actualRootPid = pid;
@@ -671,6 +696,7 @@ const runProcessScenario = async (
...(actualRootPid === undefined ? {} : { rootPid: actualRootPid }),
progress: progressTracker,
progressCounts,
...(resourceScope === undefined ? {} : { resourceScope }),
});
};
@@ -683,6 +709,7 @@ export const captureProcessScenario = async (
captureSnapshot?: typeof snapshotRoots,
cleanupHost?: ProcessCaptureCleanupHost,
progress?: ProcessCaptureProgress,
resourceScope?: ProcessCaptureResourceScope,
): Promise<Result<ProcessCapture, ProcessCaptureError | AnalysisError>> => {
const ownershipReason = processCaptureOwnershipUnavailableReason(platform);
if (ownershipReason !== undefined)
@@ -694,14 +721,17 @@ export const captureProcessScenario = async (
{ userMessage: ownershipReason },
),
);
try {
assertNotCancelled(signal);
const resolvedScenario = await resolveProcessScenarioRuntimePaths(
scenario,
environment,
);
return ok(
await runProcessScenario(
const runResourceScope = resourceScope ?? new ProcessCaptureResourceScope();
const execute = async (): Promise<
Result<ProcessCapture, ProcessCaptureError | AnalysisError>
> => {
try {
assertNotCancelled(signal);
const resolvedScenario = await resolveProcessScenarioRuntimePaths(
scenario,
environment,
);
const capture = await runProcessScenario(
resolvedScenario,
signal,
environment,
@@ -709,21 +739,44 @@ export const captureProcessScenario = async (
captureSnapshot ?? snapshotRoots,
cleanupHost,
progress,
),
);
} catch (cause: unknown) {
const failure = normalizeCaptureFailure(cause, signal);
const executionFailureReason =
describeProcessCaptureExecutionFailure(cause);
return err(
failure instanceof ProcessCaptureError
? failure
: new ProcessCaptureError("process capture failed", {
cause,
...(executionFailureReason === undefined
? {}
: { executionFailure: executionFailureReason }),
}),
);
runResourceScope,
);
return ok(capture);
} catch (cause: unknown) {
const failure = normalizeCaptureFailure(cause, signal);
const executionFailureReason =
describeProcessCaptureExecutionFailure(cause);
return err(
failure instanceof ProcessCaptureError
? failure
: new ProcessCaptureError("process capture failed", {
cause,
...(executionFailureReason === undefined
? {}
: { executionFailure: executionFailureReason }),
}),
);
}
};
if (resourceScope !== undefined) {
try {
return await resourceScope.run(execute);
} catch (cause: unknown) {
const failure = normalizeCaptureFailure(cause, signal);
const executionFailure = describeProcessCaptureExecutionFailure(cause);
return err(
failure instanceof ProcessCaptureError
? failure
: new ProcessCaptureError("process capture lifecycle failed", {
cause,
reason: "cleanup_incomplete",
cleanupResources: ["process_capture_lifecycle"],
...(executionFailure === undefined ? {} : { executionFailure }),
}),
);
}
}
const capture = await runResourceScope.run(execute);
await runResourceScope.close().catch(() => undefined);
return capture;
};