mirror of
https://github.com/morluto/rea
synced 2026-10-28 23:11:11 -04:00
fix(reverse): reconcile cleanup ownership and mutation diagnostics (#2799)
* feat(process): add owned debugger shutdown lifecycle * feat(reverse): add local inspection commands * feat(gdb): add persistent debugger integration * feat(cutter): add Cutter bridge integration * test(reverse): enforce complete integration coverage * fix(process): preserve debugger shutdown ownership * fix(reverse): retain evidence across interrupted commands * fix(gdb): honor captured provider environment * test(process): cover debugger shutdown retries * fix(reverse): retain evidence for interrupted providers * test(process): cover debugger shutdown retries * fix(reverse): retain evidence for interrupted providers * fix(server): snapshot provider environment at creation * fix(reverse): close review gaps with complete coverage * fix(reverse): address inspection review findings * fix(gdb): bound sessions and retain cleanup ownership * fix(ci): require reverse-tools quality lane * fix(process): unref inherited debugger output pipes * fix(reverse): address remaining PR 1960 findings * fix(gdb): address remaining PR 1961 findings * fix(reverse): account for unrestricted provider execution * fix(process): retain debugger ownership until safe shutdown * fix(process): preserve replacement debugger sessions * fix(reverse): preserve admission and output provenance * fix(process): retain exited debugger until output settles * fix(mcp): bound provider admission and debugger cleanup * fix(reverse): preserve failure provenance and classify E2BIG * fix(gdb): preserve startup cleanup and detach safety * fix(process): retain debugger ownership after detach failure * fix(reverse): preserve mutation and failure provenance * fix(gdb): address debugger review findings * fix(reverse): rebind sessions after external mutation * fix(process): cap debugger output after exit * fix(gdb): protect one-shot detach verification * feat(debugger): add reverse tool coverage gate * test(debugger): cover integrated failure paths * fix(gdb): bound and preserve session observations * fix(gdb): address review regressions * style: format merged integration files * fix(reverse): preserve session and output evidence * refactor(gdb): share repeated test and partial result setup * fix(gdb): invalidate TTY proof for Python aliases * fix(reverse): retain failure output byte counts * test(process): synchronize late debugger output * fix(gdb): invalidate TTY proof for wrapped commands * fix(gdb): recognize wit command abbreviation * test(reverse): wait for cancelled descendant readiness * test(reverse): bound descendant cancellation cleanup * style(reverse): format integration tests and adapters * style: format reverse engineering integration files * test(reverse): remove duplicate environment harness * test(gdb): share session startup assertions * fix(gdb): invalidate output proof for command abbreviations * test(reverse): share composition edge case scenarios * fix(ci): recognize verification and mutation tool entries * test(rizin): cover successful fake inferior detach * test(cli): align full output failure envelope * test(cli): isolate reverse command lifecycle service * test(process): share debugger output fixture setup * fix(binary): preserve high-bit Windows signature bytes * style(binary): format Windows signature parser * test(mcp): preserve call coverage in unchanged comparison * fix(ci): include reverse tools Vitest configs in lint cache * fix(mcp): preserve target content on delivery errors * ci: align duplication ceiling with expanded suite * test(reverse): allow signal admission mutants * fix(reverse): bound retained cleanup failures to process owners * fix(test): replace drifting mutation ranges with owned selectors * fix(reverse): admit persistent commands and retain cleanup ownership * fix(cutter): recover timed-out commands from exact receipts * fix(cutter): reject unsupported bridge request kinds * fix(ci): repair client format and shared verification fixtures * fix(native): preserve debugger cleanup ownership and bridge authentication * test(native): retain diagnostics without perfect coverage requirements * chore(licenses): refresh locked dependency inventory and notices * fix(test): exclude transient command locks from mutation sandboxes * fix(gdb): admit only platforms with safe detach verification * fix(cutter): retain bounded completion receipts until acknowledgement * fix(cutter): acknowledge commands rejected before UI dispatch * test(reverse): fail cancellation checks before fixture teardown * test(gdb): clean up the retained broken-stdin fixture owner * docs(test): clarify focused provider verification scope * fix(rizin): retain close cleanup failures * fix(gdb): wait for inferior stop and bound MI buffering * fix(cutter): retain failed receipt acknowledgements * fix(rizin): return retained session identity on rebind failure * fix(gdb): preserve startup cancellation and lossy status * fix(rizin): retain launch provenance and assemble frames once * fix(cutter): recover receipts across callers and bound discovery * docs: describe debugger evidence and bridge recovery * fix(process): keep cleanup incomplete while writers remain Stopping a debugger no longer reports success when inherited output streams stay open, so BinarySession cannot reopen a binary a child is still changing. Concurrent Cutter cleanup retries share one receipt acknowledgement instead of treating a consumed receipt as failure. * fix(debuggers): recover cleanup with captured process identities * fix(analysis): retire sessions after unrestricted commands * fix(cli): preserve completed evidence after cleanup recovery * fix(cutter): bound frame collection and coalesce cleanup * fix(windows): resolve selected provider paths absolutely * fix(gdb): report decoding loss in failed startup evidence * fix(rizin): scope decoding loss to collected command frames * test(mcp): cover snapshot retirement in reverse tools CI --------- Co-authored-by: Daniel Gonçalves Araujo <47118053+dandgabr@users.noreply.github.com> Co-authored-by: REA Maintainers <noreply@github.com> Co-authored-by: OpenAI Codex <codex@users.noreply.github.com> Co-authored-by: Cursor Agent <cursoragent@cursor.com>
This commit is contained in:
@@ -1,13 +1,49 @@
|
||||
import { access, readFile } from "node:fs/promises";
|
||||
import { access, readFile, rm } from "node:fs/promises";
|
||||
import { join } from "node:path";
|
||||
|
||||
import { describe, expect, it } from "vitest";
|
||||
|
||||
import { createTestTempDirectory } from "../../tests/fixtures/temporaryDirectory.js";
|
||||
|
||||
import { execFileBytes, execFileOutput } from "./ExecFileOutput.js";
|
||||
import {
|
||||
execFileBytes,
|
||||
execFileOutput,
|
||||
execFileOutputFailure,
|
||||
} from "./ExecFileOutput.js";
|
||||
|
||||
const waitForFile = async (path: string): Promise<boolean> => {
|
||||
for (let attempt = 0; attempt < 500; attempt += 1) {
|
||||
if (
|
||||
await access(path).then(
|
||||
() => true,
|
||||
() => false,
|
||||
)
|
||||
)
|
||||
return true;
|
||||
await new Promise((resolve) => setTimeout(resolve, 10));
|
||||
}
|
||||
return access(path).then(
|
||||
() => true,
|
||||
() => false,
|
||||
);
|
||||
};
|
||||
|
||||
describe("execFileOutput", () => {
|
||||
it("ignores non-Error and incomplete failure metadata", () => {
|
||||
expect(execFileOutputFailure("failure")).toBeUndefined();
|
||||
expect(execFileOutputFailure(new Error("failure"))).toBeUndefined();
|
||||
});
|
||||
|
||||
it("recognizes the execFile output limit code as truncated output", () => {
|
||||
const failure = Object.assign(new Error("output limit"), {
|
||||
stdout: "partial stdout",
|
||||
stderr: "partial stderr",
|
||||
code: "ERR_CHILD_PROCESS_STDIO_MAXBUFFER",
|
||||
});
|
||||
|
||||
expect(execFileOutputFailure(failure)?.outputTruncated).toBe(true);
|
||||
});
|
||||
|
||||
it("preserves captured bytes for callers that own the output format", async () => {
|
||||
const result = await execFileBytes(process.execPath, [
|
||||
"-e",
|
||||
@@ -18,6 +54,20 @@ describe("execFileOutput", () => {
|
||||
expect(result.stderr).toEqual(Buffer.from([0xfe]));
|
||||
});
|
||||
|
||||
it("reports raw byte counts when UTF-8 decoding changes the captured text", async () => {
|
||||
const result = await execFileOutput(process.execPath, [
|
||||
"-e",
|
||||
"process.stdout.write(Buffer.from([0x61, 0xff])); process.stderr.write(Buffer.from([0xe2, 0x82, 0xac]));",
|
||||
]);
|
||||
|
||||
expect(result).toMatchObject({
|
||||
stdout: "a�",
|
||||
stderr: "€",
|
||||
stdoutBytes: 2,
|
||||
stderrBytes: 3,
|
||||
});
|
||||
});
|
||||
|
||||
it("captures output larger than Node's default execFile limit", async () => {
|
||||
const marker = "tail-marker";
|
||||
const result = await execFileOutput(process.execPath, [
|
||||
@@ -48,6 +98,14 @@ describe("execFileOutput", () => {
|
||||
).rejects.toMatchObject({ code: 7, stdout: "out", stderr: "err" });
|
||||
});
|
||||
|
||||
it("uses the built-in execFile timeout when no stop signal is requested", async () => {
|
||||
await expect(
|
||||
execFileOutput(process.execPath, ["-e", "setInterval(() => {}, 1000)"], {
|
||||
timeout: 100,
|
||||
}),
|
||||
).rejects.toMatchObject({ killed: true });
|
||||
});
|
||||
|
||||
it.skipIf(process.platform === "win32")(
|
||||
"stops an aborted process with the caller's signal and settles after it exits",
|
||||
async () => {
|
||||
@@ -75,25 +133,48 @@ describe("execFileOutput", () => {
|
||||
),
|
||||
}),
|
||||
);
|
||||
for (let attempt = 0; attempt < 500; attempt += 1) {
|
||||
if (
|
||||
await access(ready).then(
|
||||
() => true,
|
||||
() => false,
|
||||
)
|
||||
)
|
||||
break;
|
||||
await new Promise((resolve) => setTimeout(resolve, 10));
|
||||
}
|
||||
|
||||
const readyObserved = await waitForFile(ready);
|
||||
controller.abort();
|
||||
const outcome = await execution;
|
||||
|
||||
expect(readyObserved).toBe(true);
|
||||
expect(outcome?.cause).toMatchObject({ name: "AbortError" });
|
||||
expect(outcome?.stoppedBeforeSettling).toBe("SIGINT");
|
||||
},
|
||||
);
|
||||
|
||||
it.skipIf(process.platform === "win32")(
|
||||
"keeps the first stop reason when timeout arrives during abort cleanup",
|
||||
async () => {
|
||||
const directory = await createTestTempDirectory("rea-exec-double-stop-");
|
||||
const ready = join(directory, "ready");
|
||||
const script = [
|
||||
'const fs = require("node:fs");',
|
||||
'for (const name of ["SIGINT", "SIGTERM"]) process.on(name, () => {});',
|
||||
`fs.writeFileSync(${JSON.stringify(ready)}, "");`,
|
||||
"setInterval(() => {}, 1000);",
|
||||
].join("\n");
|
||||
const controller = new AbortController();
|
||||
const execution = execFileOutput(process.execPath, ["-e", script], {
|
||||
signal: controller.signal,
|
||||
timeout: 10_000,
|
||||
stopSignal: "SIGINT",
|
||||
});
|
||||
const startedAt = Date.now();
|
||||
expect(await waitForFile(ready)).toBe(true);
|
||||
const remainingBeforeDeadline = Math.max(
|
||||
0,
|
||||
9_000 - (Date.now() - startedAt),
|
||||
);
|
||||
await new Promise((resolve) =>
|
||||
setTimeout(resolve, remainingBeforeDeadline),
|
||||
);
|
||||
controller.abort();
|
||||
|
||||
await expect(execution).rejects.toMatchObject({ name: "AbortError" });
|
||||
},
|
||||
);
|
||||
|
||||
it("rejects an already aborted stoppable command without starting it", async () => {
|
||||
const controller = new AbortController();
|
||||
controller.abort();
|
||||
@@ -149,6 +230,33 @@ describe.skipIf(process.platform === "win32")(
|
||||
},
|
||||
);
|
||||
|
||||
it("preserves a child failure's missing signal when reporting timeout", async () => {
|
||||
const directory = await createTestTempDirectory(
|
||||
"rea-exec-missing-signal-",
|
||||
);
|
||||
const ready = join(directory, "ready");
|
||||
const script = [
|
||||
'const fs = require("node:fs");',
|
||||
'for (const name of ["SIGINT", "SIGTERM"])',
|
||||
" process.on(name, () => process.exit(1));",
|
||||
`fs.writeFileSync(${JSON.stringify(ready)}, "");`,
|
||||
"setInterval(() => {}, 1000);",
|
||||
].join("\n");
|
||||
try {
|
||||
const execution = execFileOutput(process.execPath, ["-e", script], {
|
||||
timeout: 10_000,
|
||||
stopSignal: "SIGINT",
|
||||
});
|
||||
expect(await waitForFile(ready)).toBe(true);
|
||||
await expect(execution).rejects.toMatchObject({
|
||||
code: "ETIMEDOUT",
|
||||
signal: null,
|
||||
});
|
||||
} finally {
|
||||
await rm(directory, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
it("kills a timed-out process that ignores the stop signal", async () => {
|
||||
const script = [
|
||||
'for (const name of ["SIGINT", "SIGTERM"]) process.on(name, () => {});',
|
||||
@@ -168,3 +276,40 @@ describe.skipIf(process.platform === "win32")(
|
||||
});
|
||||
},
|
||||
);
|
||||
|
||||
describe("execFileOutput failure metadata", () => {
|
||||
it("preserves raw byte counts on a failed command with invalid UTF-8", async () => {
|
||||
const failure: unknown = await execFileOutput(process.execPath, [
|
||||
"-e",
|
||||
"process.stdout.write(Buffer.from([0x61, 0xff])); process.stderr.write(Buffer.from([0xe2, 0x82, 0xac])); process.exitCode = 7;",
|
||||
]).then(
|
||||
() => undefined,
|
||||
(cause: unknown) => cause,
|
||||
);
|
||||
|
||||
expect(execFileOutputFailure(failure)).toMatchObject({
|
||||
stdout: "a�",
|
||||
stderr: "€",
|
||||
stdoutBytes: 2,
|
||||
stderrBytes: 3,
|
||||
code: 7,
|
||||
});
|
||||
});
|
||||
it("preserves explicit truncation metadata from an owned process failure", () => {
|
||||
const failure = Object.assign(new Error("owned output was truncated"), {
|
||||
stdout: "partial stdout",
|
||||
stderr: "partial stderr",
|
||||
code: 7,
|
||||
outputTruncated: true,
|
||||
});
|
||||
|
||||
expect(execFileOutputFailure(failure)).toEqual({
|
||||
stdout: "partial stdout",
|
||||
stderr: "partial stderr",
|
||||
code: 7,
|
||||
signal: null,
|
||||
killed: false,
|
||||
outputTruncated: true,
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
@@ -76,6 +76,9 @@ export interface ExecFileOutputFailure {
|
||||
readonly signal: string | null;
|
||||
readonly killed: boolean;
|
||||
readonly outputTruncated: boolean;
|
||||
readonly stdoutBytes?: number;
|
||||
readonly stderrBytes?: number;
|
||||
readonly collectionFailure?: boolean;
|
||||
}
|
||||
|
||||
/** Read captured output metadata from a failed execFileOutput invocation. */
|
||||
@@ -88,15 +91,31 @@ export const execFileOutputFailure = (
|
||||
const code = Reflect.get(cause, "code");
|
||||
const signal = Reflect.get(cause, "signal");
|
||||
const killed = Reflect.get(cause, "killed");
|
||||
const outputTruncated = Reflect.get(cause, "outputTruncated");
|
||||
const stdoutBytes = Reflect.get(cause, "stdoutBytes");
|
||||
const stderrBytes = Reflect.get(cause, "stderrBytes");
|
||||
const collectionFailure = Reflect.get(cause, "collectionFailure");
|
||||
if (typeof stdout !== "string" || typeof stderr !== "string")
|
||||
return undefined;
|
||||
return {
|
||||
stdout,
|
||||
stderr,
|
||||
...(typeof stdoutBytes === "number" &&
|
||||
Number.isSafeInteger(stdoutBytes) &&
|
||||
stdoutBytes >= 0
|
||||
? { stdoutBytes }
|
||||
: {}),
|
||||
...(typeof stderrBytes === "number" &&
|
||||
Number.isSafeInteger(stderrBytes) &&
|
||||
stderrBytes >= 0
|
||||
? { stderrBytes }
|
||||
: {}),
|
||||
code: typeof code === "string" || typeof code === "number" ? code : null,
|
||||
signal: typeof signal === "string" ? signal : null,
|
||||
killed: killed === true,
|
||||
outputTruncated: code === "ERR_CHILD_PROCESS_STDIO_MAXBUFFER",
|
||||
outputTruncated:
|
||||
outputTruncated === true || code === "ERR_CHILD_PROCESS_STDIO_MAXBUFFER",
|
||||
...(collectionFailure === true ? { collectionFailure: true } : {}),
|
||||
};
|
||||
};
|
||||
|
||||
@@ -150,6 +169,8 @@ export const execFileBytes = (
|
||||
if (failure !== null) {
|
||||
Reflect.set(failure, "stdout", stdout.toString("utf8"));
|
||||
Reflect.set(failure, "stderr", stderr.toString("utf8"));
|
||||
Reflect.set(failure, "stdoutBytes", stdout.byteLength);
|
||||
Reflect.set(failure, "stderrBytes", stderr.byteLength);
|
||||
reject(failure);
|
||||
return;
|
||||
}
|
||||
@@ -179,8 +200,15 @@ export const execFileOutput = (
|
||||
command: string,
|
||||
arguments_: readonly string[],
|
||||
options: ExecFileOutputOptions = {},
|
||||
): Promise<{ readonly stdout: string; readonly stderr: string }> =>
|
||||
): Promise<{
|
||||
readonly stdout: string;
|
||||
readonly stderr: string;
|
||||
readonly stdoutBytes: number;
|
||||
readonly stderrBytes: number;
|
||||
}> =>
|
||||
execFileBytes(command, arguments_, options).then(({ stdout, stderr }) => ({
|
||||
stdout: stdout.toString("utf8"),
|
||||
stderr: stderr.toString("utf8"),
|
||||
stdoutBytes: stdout.byteLength,
|
||||
stderrBytes: stderr.byteLength,
|
||||
}));
|
||||
|
||||
@@ -20,18 +20,23 @@ import {
|
||||
/** A short-lived command failed its lifecycle or complete-output contract. */
|
||||
export class OwnedCommandFailure extends Error {
|
||||
readonly cleanupResult: ProcessCleanupFailure | undefined;
|
||||
readonly collectionFailure: boolean;
|
||||
|
||||
constructor(
|
||||
readonly reason: "cancelled" | "timeout" | "output-limit" | "process",
|
||||
message: string,
|
||||
readonly snapshot: ProviderProcessSnapshot | null = null,
|
||||
readonly cleanupFailure: string | null = null,
|
||||
options?: ErrorOptions & { readonly cleanupResult?: ProcessCleanupFailure },
|
||||
options?: ErrorOptions & {
|
||||
readonly cleanupResult?: ProcessCleanupFailure;
|
||||
readonly collectionFailure?: boolean;
|
||||
},
|
||||
readonly resources: readonly string[] = [],
|
||||
readonly cleanupOwner?: ProviderProcessSupervisor,
|
||||
) {
|
||||
super(message, options);
|
||||
this.cleanupResult = options?.cleanupResult;
|
||||
this.collectionFailure = options?.collectionFailure ?? false;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -56,7 +61,7 @@ export const runOwnedCommand = async (
|
||||
readonly acceptNonZeroExit?: boolean;
|
||||
},
|
||||
options: {
|
||||
readonly signal?: AbortSignal;
|
||||
readonly signal?: AbortSignal | undefined;
|
||||
/** Synchronously accept unresolved cleanup ownership; execution still returns its original outcome. Must not throw. */
|
||||
readonly retainCleanup?: (owner: ProviderProcessSupervisor) => void;
|
||||
readonly launcher?: (
|
||||
@@ -156,7 +161,9 @@ const collectOwnedCommand = async (
|
||||
"Command diagnostic output exceeded its complete-output budget.",
|
||||
);
|
||||
if (processFailure !== undefined)
|
||||
throw new OwnedCommandFailure("process", processFailure);
|
||||
throw new OwnedCommandFailure("process", processFailure, null, null, {
|
||||
collectionFailure: true,
|
||||
});
|
||||
await waitForAbortableDelay(25, signal);
|
||||
}
|
||||
const snapshot = supervisor.snapshot();
|
||||
@@ -191,6 +198,8 @@ const collectOwnedCommand = async (
|
||||
processFailure ??
|
||||
"Command failed; inspect its exit status and diagnostics.",
|
||||
snapshot,
|
||||
null,
|
||||
{ collectionFailure: processFailure !== undefined },
|
||||
);
|
||||
} catch (cause: unknown) {
|
||||
failure =
|
||||
@@ -219,6 +228,9 @@ const collectOwnedCommand = async (
|
||||
...(stopped.cleanupResult === undefined
|
||||
? {}
|
||||
: { cleanupResult: stopped.cleanupResult }),
|
||||
...(failure?.collectionFailure === true
|
||||
? { collectionFailure: true }
|
||||
: {}),
|
||||
},
|
||||
ownedCommandResources(launched.ownership, spawn.cwd),
|
||||
supervisor,
|
||||
@@ -234,6 +246,7 @@ const collectOwnedCommand = async (
|
||||
...(failure.cleanupResult === undefined
|
||||
? {}
|
||||
: { cleanupResult: failure.cleanupResult }),
|
||||
...(failure.collectionFailure ? { collectionFailure: true } : {}),
|
||||
},
|
||||
failure.resources,
|
||||
);
|
||||
|
||||
@@ -6,10 +6,14 @@ import { WindowsOwnedProcess } from "../windows/WindowsOwnedProcess.js";
|
||||
|
||||
import type {
|
||||
OwnedProcessGroup,
|
||||
ProcessOwnershipBaseline,
|
||||
ProcessCleanupFailure,
|
||||
ProcessCleanupResult,
|
||||
} from "./ProcessOwnership.js";
|
||||
import { prepareProcessOwnershipInspection } from "./ProcessOwnershipObservation.js";
|
||||
import {
|
||||
prepareProcessOwnershipInspection,
|
||||
systemProcessOwnershipHost,
|
||||
} from "./ProcessOwnershipObservation.js";
|
||||
|
||||
const DEFAULT_TERMINATION_GRACE_MS = 250;
|
||||
const DEFAULT_KILL_GRACE_MS = 1_000;
|
||||
@@ -36,6 +40,8 @@ export interface OwnedProviderProcessSpawnOptions {
|
||||
* Windows Job Object. Other launchers cannot enforce it and refuse it.
|
||||
*/
|
||||
readonly memoryLimitBytes?: number;
|
||||
/** Sweep exact run-token descendants during POSIX cleanup. */
|
||||
readonly sweepTokenOwnedProcesses?: boolean;
|
||||
/** Cancels native ownership preparation before provider process creation. */
|
||||
readonly signal?: AbortSignal;
|
||||
}
|
||||
@@ -95,6 +101,8 @@ export interface ProviderProcessSnapshot {
|
||||
readonly diagnosticTruncated?: boolean;
|
||||
readonly stdout: {
|
||||
readonly text: string;
|
||||
/** True when retained bytes cannot be represented exactly as UTF-8 text. */
|
||||
readonly invalidUtf8?: true;
|
||||
/** Bytes retained in text, limited by maxDiagnosticBytes when configured. */
|
||||
readonly bytes: number;
|
||||
/** All bytes drained from this stream since construction or resetOutput. */
|
||||
@@ -102,6 +110,8 @@ export interface ProviderProcessSnapshot {
|
||||
};
|
||||
readonly stderr: {
|
||||
readonly text: string;
|
||||
/** True when retained bytes cannot be represented exactly as UTF-8 text. */
|
||||
readonly invalidUtf8?: true;
|
||||
/** Bytes retained in text, limited by maxDiagnosticBytes when configured. */
|
||||
readonly bytes: number;
|
||||
readonly observedBytes: number;
|
||||
@@ -174,6 +184,16 @@ export const spawnOwnedProviderProcess = async (
|
||||
await prepareProcessOwnershipInspection(options.signal);
|
||||
options.signal?.throwIfAborted();
|
||||
const platform = options.platform ?? process.platform;
|
||||
const sweepTokenOwnedProcesses =
|
||||
options.sweepTokenOwnedProcesses === true && platform !== "win32";
|
||||
let captureBaseline: ProcessOwnershipBaseline | undefined;
|
||||
if (sweepTokenOwnedProcesses) {
|
||||
const capture = systemProcessOwnershipHost.captureBaseline;
|
||||
if (capture === undefined)
|
||||
throw new Error("Process ownership baseline is unavailable on this host");
|
||||
captureBaseline = await capture(options.signal);
|
||||
options.signal?.throwIfAborted();
|
||||
}
|
||||
const hostEnvironment = options.hostEnvironment ?? process.env;
|
||||
const environment = {
|
||||
...hostEnvironment,
|
||||
@@ -226,6 +246,12 @@ export const spawnOwnedProviderProcess = async (
|
||||
leaderPid: pid,
|
||||
processGroupId: pid,
|
||||
expectedParentPid: process.pid,
|
||||
...(sweepTokenOwnedProcesses
|
||||
? {
|
||||
sweepTokenOwnedProcesses: true,
|
||||
captureBaseline: captureBaseline!,
|
||||
}
|
||||
: {}),
|
||||
...(options.expectedCommand === null
|
||||
? {}
|
||||
: { expectedCommand: options.expectedCommand ?? options.command }),
|
||||
@@ -282,12 +308,14 @@ export class ProviderProcessSupervisor {
|
||||
#exitObservation: ProviderProcessExitObservation | undefined;
|
||||
#stopPromise: Promise<ProviderProcessStopResult> | undefined;
|
||||
#disposed = false;
|
||||
#maxDiagnosticBytes: number | undefined;
|
||||
|
||||
constructor(
|
||||
readonly launch: ProviderProcessLaunch,
|
||||
options: ProviderProcessSupervisorOptions = {},
|
||||
) {
|
||||
this.#options = options;
|
||||
this.#maxDiagnosticBytes = options.maxDiagnosticBytes;
|
||||
if (options.captureStdout !== false)
|
||||
this.#attach(launch.process.stdout, "stdout");
|
||||
this.#attach(launch.process.stderr, "stderr");
|
||||
@@ -340,7 +368,7 @@ export class ProviderProcessSupervisor {
|
||||
);
|
||||
}
|
||||
|
||||
/** Wait until the producer has closed its output streams; exit alone does not prove complete output. */
|
||||
/** Wait until output streams close; WindowsOwnedProcess closes only after its whole Job Object settles. */
|
||||
async waitForOutputClose(timeoutMs: number): Promise<boolean> {
|
||||
return waitForProcessEvent(
|
||||
this.#outputClosed,
|
||||
@@ -349,6 +377,35 @@ export class ProviderProcessSupervisor {
|
||||
);
|
||||
}
|
||||
|
||||
/** Bound additional retained bytes while still draining all later output. */
|
||||
limitAdditionalOutputBytes(maxBytes: number): void {
|
||||
if (!Number.isSafeInteger(maxBytes) || maxBytes < 0)
|
||||
throw new RangeError(
|
||||
"Additional output limit must be a non-negative safe integer.",
|
||||
);
|
||||
const retained = this.#stdout.bytes + this.#stderr.bytes;
|
||||
const requestedLimit = Math.min(
|
||||
Number.MAX_SAFE_INTEGER,
|
||||
retained + maxBytes,
|
||||
);
|
||||
this.#maxDiagnosticBytes =
|
||||
this.#maxDiagnosticBytes === undefined
|
||||
? requestedLimit
|
||||
: Math.min(this.#maxDiagnosticBytes, requestedLimit);
|
||||
}
|
||||
|
||||
/** Unreference inherited output pipes without closing streams shared with descendants. */
|
||||
unrefOutputStreams(): void {
|
||||
for (const stream of [
|
||||
this.launch.process.stdout,
|
||||
this.launch.process.stderr,
|
||||
]) {
|
||||
const unref = (stream as (Readable & { unref?: () => void }) | null)
|
||||
?.unref;
|
||||
unref?.call(stream);
|
||||
}
|
||||
}
|
||||
|
||||
/** Share each stop attempt; incomplete cleanup retains ownership and permits retry. */
|
||||
stop(
|
||||
options: ProviderProcessStopOptions = {},
|
||||
@@ -509,13 +566,11 @@ export class ProviderProcessSupervisor {
|
||||
if (stream === "stdout") this.#stdoutObservedBytes += bytes.byteLength;
|
||||
else this.#stderrObservedBytes += bytes.byteLength;
|
||||
const remaining =
|
||||
this.#options.maxDiagnosticBytes === undefined
|
||||
this.#maxDiagnosticBytes === undefined
|
||||
? bytes.byteLength
|
||||
: Math.max(
|
||||
0,
|
||||
this.#options.maxDiagnosticBytes -
|
||||
this.#stdout.bytes -
|
||||
this.#stderr.bytes,
|
||||
this.#maxDiagnosticBytes - this.#stdout.bytes - this.#stderr.bytes,
|
||||
);
|
||||
const retained = Math.min(bytes.byteLength, remaining);
|
||||
if (retained < bytes.byteLength) this.#diagnosticTruncated = true;
|
||||
@@ -564,11 +619,19 @@ class ProcessOutputCapture {
|
||||
}
|
||||
|
||||
snapshot(observedBytes: number): ProviderProcessSnapshot["stdout"] {
|
||||
const text = Buffer.concat(this.#chunks, this.#bytes).toString("utf8");
|
||||
const buffer = Buffer.concat(this.#chunks, this.#bytes);
|
||||
const text = buffer.toString("utf8");
|
||||
let invalidUtf8 = false;
|
||||
try {
|
||||
new TextDecoder("utf-8", { fatal: true }).decode(buffer);
|
||||
} catch {
|
||||
invalidUtf8 = true;
|
||||
}
|
||||
return {
|
||||
text,
|
||||
bytes: this.#bytes,
|
||||
observedBytes,
|
||||
...(invalidUtf8 ? { invalidUtf8: true } : {}),
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,311 @@
|
||||
import { readFile } from "node:fs/promises";
|
||||
|
||||
import { parseLinuxProcessStartTime } from "./ProcessOwnershipObservation.js";
|
||||
|
||||
/** Result of terminating one debugger process owned by this application. */
|
||||
export type OwnedDebuggerStopResult =
|
||||
| { readonly status: "stopped" | "already_exited" }
|
||||
| { readonly status: "incomplete"; readonly reason: string };
|
||||
|
||||
const DETACH_TIMEOUT_MS = 1_000;
|
||||
const MAX_LATE_DEBUGGER_OUTPUT_BYTES = 64 * 1024;
|
||||
export type OwnedDebuggerDetachResult =
|
||||
| { readonly status: "no-inferior" }
|
||||
| { readonly status: "detached"; readonly outputSink: "durable" };
|
||||
export interface OwnedInferiorIdentity {
|
||||
readonly pid: string;
|
||||
readonly startTime: string;
|
||||
}
|
||||
|
||||
export type ProcessStartTimeReader = (pid: string) => Promise<string | null>;
|
||||
|
||||
const readLinuxProcessStartTime: ProcessStartTimeReader = async (pid) => {
|
||||
if (process.platform !== "linux")
|
||||
throw new Error(
|
||||
"Inferior process identity verification requires Linux /proc.",
|
||||
);
|
||||
let stat: string;
|
||||
try {
|
||||
stat = await readFile(`/proc/${pid}/stat`, "utf8");
|
||||
} catch (cause: unknown) {
|
||||
if (
|
||||
typeof cause === "object" &&
|
||||
cause !== null &&
|
||||
"code" in cause &&
|
||||
cause.code === "ENOENT"
|
||||
)
|
||||
return null;
|
||||
throw cause;
|
||||
}
|
||||
const startTime = parseLinuxProcessStartTime(stat);
|
||||
if (startTime === undefined)
|
||||
throw new Error(`Inferior PID ${pid} has an unreadable process identity.`);
|
||||
return startTime;
|
||||
};
|
||||
|
||||
/** Confirm only that every previously identified inferior process has ended. */
|
||||
export const verifyOwnedInferiorsEnded = async (
|
||||
identities: Iterable<OwnedInferiorIdentity>,
|
||||
readStartTime: ProcessStartTimeReader = readLinuxProcessStartTime,
|
||||
): Promise<boolean> => {
|
||||
let foundIdentity = false;
|
||||
for (const identity of identities) {
|
||||
foundIdentity = true;
|
||||
const observed = await readStartTime(identity.pid);
|
||||
if (observed === identity.startTime) return false;
|
||||
}
|
||||
return foundIdentity;
|
||||
};
|
||||
|
||||
const detachAttempts = new WeakMap<
|
||||
object,
|
||||
Promise<OwnedDebuggerDetachResult>
|
||||
>();
|
||||
const detachOutcomes = new WeakMap<
|
||||
object,
|
||||
"confirmed" | "unconfirmed" | "inferior-ended"
|
||||
>();
|
||||
|
||||
const isVerifiedDetachResult = (
|
||||
result: unknown,
|
||||
): result is OwnedDebuggerDetachResult =>
|
||||
typeof result === "object" &&
|
||||
result !== null &&
|
||||
(("status" in result && result.status === "no-inferior") ||
|
||||
("status" in result &&
|
||||
result.status === "detached" &&
|
||||
"outputSink" in result &&
|
||||
result.outputSink === "durable"));
|
||||
|
||||
const waitForInferiorDetach = async (
|
||||
session: object,
|
||||
detach: () => Promise<OwnedDebuggerDetachResult>,
|
||||
): Promise<OwnedDebuggerDetachResult | undefined> => {
|
||||
let attempt = detachAttempts.get(session);
|
||||
if (attempt === undefined) {
|
||||
// Providers may use `this` to access their command transport.
|
||||
const pending = Promise.resolve().then(() => detach.call(session));
|
||||
const tracked = pending.then(
|
||||
(result) => {
|
||||
detachAttempts.delete(session);
|
||||
detachOutcomes.set(
|
||||
session,
|
||||
isVerifiedDetachResult(result) ? "confirmed" : "unconfirmed",
|
||||
);
|
||||
return result;
|
||||
},
|
||||
(cause: unknown) => {
|
||||
detachAttempts.delete(session);
|
||||
detachOutcomes.set(session, "unconfirmed");
|
||||
throw cause;
|
||||
},
|
||||
);
|
||||
attempt = tracked;
|
||||
detachAttempts.set(session, tracked);
|
||||
}
|
||||
let timer!: NodeJS.Timeout;
|
||||
try {
|
||||
return await Promise.race([
|
||||
attempt,
|
||||
new Promise<undefined>((resolve) => {
|
||||
timer = setTimeout(() => resolve(undefined), DETACH_TIMEOUT_MS);
|
||||
}),
|
||||
]);
|
||||
} finally {
|
||||
clearTimeout(timer);
|
||||
}
|
||||
};
|
||||
|
||||
interface OwnedDebuggerProcess {
|
||||
readonly id: string;
|
||||
readonly exited: boolean;
|
||||
readonly inferiorStateUnknown?: () => boolean;
|
||||
readonly verifyInferiorsEndedAfterExit?: () => Promise<boolean>;
|
||||
readonly launched: {
|
||||
readonly process: { kill(signal: NodeJS.Signals): boolean };
|
||||
};
|
||||
/** Must return only after the provider verifies detachment or no inferior. */
|
||||
/** Confirms no inferior, or detaches only after verifying a durable output sink. */
|
||||
readonly detachInferior?: () => Promise<OwnedDebuggerDetachResult>;
|
||||
readonly supervisor: {
|
||||
waitForExit(timeoutMs: number): Promise<boolean>;
|
||||
/** WindowsOwnedProcess closes output only after the whole Job Object settles. */
|
||||
waitForOutputClose(timeoutMs: number): Promise<boolean>;
|
||||
unrefOutputStreams(): void;
|
||||
/** Cap bytes retained after debugger shutdown while continuing to drain. */
|
||||
limitAdditionalOutputBytes(maxBytes: number): void;
|
||||
dispose(): void;
|
||||
};
|
||||
}
|
||||
|
||||
const finishExitedDebuggerSession = async (
|
||||
session: OwnedDebuggerProcess,
|
||||
debuggerName: string,
|
||||
platform: NodeJS.Platform,
|
||||
status: "stopped" | "already_exited",
|
||||
): Promise<OwnedDebuggerStopResult> => {
|
||||
try {
|
||||
session.supervisor.limitAdditionalOutputBytes(
|
||||
MAX_LATE_DEBUGGER_OUTPUT_BYTES,
|
||||
);
|
||||
if (!(await session.supervisor.waitForOutputClose(1_000))) {
|
||||
return {
|
||||
status: "incomplete",
|
||||
reason:
|
||||
platform === "win32"
|
||||
? `Owned ${debuggerName} process leader exited, but its Windows Job Object still has active descendants or has not reported settlement.`
|
||||
: `Owned ${debuggerName} process leader exited, but inherited output streams are still open or have not reported settlement.`,
|
||||
};
|
||||
}
|
||||
} catch (cause: unknown) {
|
||||
return {
|
||||
status: "incomplete",
|
||||
reason:
|
||||
platform === "win32"
|
||||
? `Owned ${debuggerName} Windows Job Object settlement could not be verified: ${cause instanceof Error ? cause.message : String(cause)}`
|
||||
: `Owned ${debuggerName} inherited output stream settlement could not be verified: ${cause instanceof Error ? cause.message : String(cause)}`,
|
||||
};
|
||||
}
|
||||
if (platform !== "win32") session.supervisor.unrefOutputStreams();
|
||||
session.supervisor.dispose();
|
||||
return { status };
|
||||
};
|
||||
|
||||
/** Stop only the directly owned debugger process, never its process group or target. */
|
||||
export const stopOwnedDebuggerProcess = async (
|
||||
session: OwnedDebuggerProcess,
|
||||
debuggerName: string,
|
||||
options: { readonly platform?: NodeJS.Platform } = {},
|
||||
): Promise<OwnedDebuggerStopResult> => {
|
||||
const platform = options.platform ?? process.platform;
|
||||
if (session.exited) {
|
||||
if (detachAttempts.has(session))
|
||||
return {
|
||||
status: "incomplete",
|
||||
reason: `Owned ${debuggerName} inferior detach is still pending after the process exited.`,
|
||||
};
|
||||
const detachOutcome = detachOutcomes.get(session);
|
||||
if (
|
||||
detachOutcome === "unconfirmed" ||
|
||||
(detachOutcome === undefined && session.inferiorStateUnknown?.() === true)
|
||||
) {
|
||||
try {
|
||||
if ((await session.verifyInferiorsEndedAfterExit?.()) !== true)
|
||||
return {
|
||||
status: "incomplete",
|
||||
reason: `Owned ${debuggerName} inferior disposition was not verified before the debugger exited, and no owner proof confirms the identified inferior processes have ended.`,
|
||||
};
|
||||
detachOutcomes.set(session, "inferior-ended");
|
||||
} catch (cause: unknown) {
|
||||
return {
|
||||
status: "incomplete",
|
||||
reason: `Owned ${debuggerName} inferior process identity could not be verified after debugger exit: ${cause instanceof Error ? cause.message : String(cause)}`,
|
||||
};
|
||||
}
|
||||
}
|
||||
return finishExitedDebuggerSession(
|
||||
session,
|
||||
debuggerName,
|
||||
platform,
|
||||
"already_exited",
|
||||
);
|
||||
}
|
||||
if (platform === "win32")
|
||||
return {
|
||||
status: "incomplete",
|
||||
reason: `Owned ${debuggerName} process cannot be stopped safely on Windows because job termination can also terminate its inferior processes.`,
|
||||
};
|
||||
try {
|
||||
if (session.detachInferior === undefined) {
|
||||
detachOutcomes.set(session, "unconfirmed");
|
||||
return {
|
||||
status: "incomplete",
|
||||
reason: `Owned ${debuggerName} inferior detach cannot be verified because the provider has no detach capability.`,
|
||||
};
|
||||
}
|
||||
const detached = await waitForInferiorDetach(
|
||||
session,
|
||||
session.detachInferior,
|
||||
);
|
||||
if (detached === undefined)
|
||||
return {
|
||||
status: "incomplete",
|
||||
reason: `Owned ${debuggerName} inferior detach was not verified within ${DETACH_TIMEOUT_MS} ms.`,
|
||||
};
|
||||
if (!isVerifiedDetachResult(detached)) {
|
||||
detachOutcomes.set(session, "unconfirmed");
|
||||
return {
|
||||
status: "incomplete",
|
||||
reason: `Owned ${debuggerName} provider did not confirm no inferior or detachment with a durable output sink.`,
|
||||
};
|
||||
}
|
||||
detachOutcomes.set(session, "confirmed");
|
||||
} catch (cause: unknown) {
|
||||
return {
|
||||
status: "incomplete",
|
||||
reason: `Owned ${debuggerName} inferior detach could not be verified: ${cause instanceof Error ? cause.message : String(cause)}`,
|
||||
};
|
||||
}
|
||||
session.launched.process.kill("SIGTERM");
|
||||
if (!(await session.supervisor.waitForExit(300))) {
|
||||
session.launched.process.kill("SIGKILL");
|
||||
const exitedAfterKill = await session.supervisor.waitForExit(1_000);
|
||||
if (exitedAfterKill)
|
||||
return finishExitedDebuggerSession(
|
||||
session,
|
||||
debuggerName,
|
||||
platform,
|
||||
"stopped",
|
||||
);
|
||||
return {
|
||||
status: "incomplete",
|
||||
reason: `Owned ${debuggerName} process did not exit after direct termination.`,
|
||||
};
|
||||
}
|
||||
return finishExitedDebuggerSession(
|
||||
session,
|
||||
debuggerName,
|
||||
platform,
|
||||
"stopped",
|
||||
);
|
||||
};
|
||||
|
||||
/**
|
||||
* Stop every owned session and retain failed sessions for possible recovery.
|
||||
* Callers must block replacement admission and drain in-flight starts until
|
||||
* this settles; incomplete cleanup leaves the old owner in the map for retry.
|
||||
*/
|
||||
export const closeOwnedDebuggerSessions = async <
|
||||
Session extends { id: string },
|
||||
>(
|
||||
sessions: Map<string, Session>,
|
||||
stop: (session: Session) => Promise<OwnedDebuggerStopResult>,
|
||||
debuggerName: string,
|
||||
): Promise<void> => {
|
||||
const results = await Promise.allSettled(
|
||||
[...sessions.entries()].map(async ([key, session]) => ({
|
||||
key,
|
||||
session,
|
||||
result: await stop(session),
|
||||
})),
|
||||
);
|
||||
const failures: Error[] = [];
|
||||
for (const settled of results) {
|
||||
if (settled.status === "rejected") {
|
||||
failures.push(
|
||||
settled.reason instanceof Error
|
||||
? settled.reason
|
||||
: new Error(String(settled.reason)),
|
||||
);
|
||||
continue;
|
||||
}
|
||||
const { key, session, result } = settled.value;
|
||||
if (result.status === "incomplete") failures.push(new Error(result.reason));
|
||||
else if (sessions.get(key) === session) sessions.delete(key);
|
||||
}
|
||||
if (failures.length > 0)
|
||||
throw new AggregateError(
|
||||
failures,
|
||||
`One or more owned ${debuggerName} processes did not stop cleanly`,
|
||||
);
|
||||
};
|
||||
Reference in New Issue
Block a user