fix(cli): keep committed evidence bundle receipts after cancellation (#6221)

Summary delivery with --evidence_output wrote the bundle, then let the
command's close/cancellation check replace the result. A SIGINT/SIGTERM
during staging cleanup after commit returned typed cancellation without
the receipt, and a persistent staging cleanup failure after commit was
rewrapped as a provider cleanup error that dropped published_path.

Every --evidence_output command now selects its view first and publishes
through publishEvidenceDelivery, the owner inspect-managed-members already
used. Post-commit cancellation keeps {path, bytes} under
details.diagnostics.evidence_bundle; incomplete cleanup keeps
details.published_path and bytes; a cleanup retry that succeeds returns
the normal receipt; pre-commit cancellation creates nothing and names
the command's operation.

Fixes #6095

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Kaoru0822andClaude Opus 5.5 authored and GitHub committed 2026-10-21 20:32:26 -07:00
1 parent e5d2805b42
commit 4a4bbab3c3
8 files changed
+453 -84

No files matched your search

+8
View File
@@ -551,6 +551,14 @@ command, objdump, Rizin and GDB, and the agent environment server
signal-derived status. This describes the current source routing; it is not a
claim that every command or provider runtime has been independently exercised.
Commands with `--evidence_output` select their delivered view, then publish the
complete Evidence bundle atomically. Cancellation before publication leaves no
new file. A publication that has committed can win a later cancellation; the
`cancelled` result then keeps the written path and byte count under
`details.diagnostics.evidence_bundle`. If staging cleanup stays incomplete
after commit, the `cleanup_incomplete` error reports the same file as
`details.published_path` and `details.bytes`.
When the reader of piped output closes early, as `rea ... | head` does, the
CLI stops writing, prints no error and exits `141` (`128 + 13`, the status of a
SIGPIPE-terminated writer). The operation itself may have completed; only its
+40 -6
View File
@@ -98,26 +98,60 @@ export const deliverAnalysisEvidence = async (
)
: ok(evidence);
if (!delivered.ok) return delivered;
const written = await persistEvidenceBundleCommand(
const published = await publishEvidenceDelivery({
bundle,
complete: evidence,
delivered: delivered.value,
outputPath,
operation: evidence.operation,
signal,
});
return published.ok ? ok(published.value.delivered) : published;
};
/** Selected Evidence delivered after its complete bundle was committed. */
export interface PublishedEvidenceDelivery {
readonly delivered: Evidence;
readonly evidenceBundle: PublishedEvidenceFile;
}
/**
* Publish a complete bundle, then deliver Evidence already projected from it.
* Cancellation observed before commit leaves no output and retains the
* complete Evidence. A committed publication wins a later cancellation race;
* its receipt stays in the typed cancellation as diagnostics.evidence_bundle,
* and a cleanup failure after commit keeps its own published receipt.
*/
export const publishEvidenceDelivery = async (input: {
readonly bundle: EvidenceBundle;
readonly complete: Evidence;
readonly delivered: Evidence;
readonly outputPath: string;
readonly operation: string;
readonly signal?: AbortSignal | undefined;
}): Promise<Result<PublishedEvidenceDelivery, AnalysisError>> => {
const { signal, operation } = input;
const written = await persistEvidenceBundleCommand(
input.bundle,
input.outputPath,
false,
signal,
evidence.operation,
operation,
);
// Deliver interrupts queued during final publication cleanup before the
// command's signal owner accepts completion and removes its listeners.
await dispatchQueuedInterrupts();
if (!written.ok) return err(written.error.retainPartialObservation(evidence));
if (!written.ok)
return err(written.error.retainPartialObservation(input.complete));
if (signal?.aborted)
return err(
new AnalysisCancelledError(evidence.operation, {
new AnalysisCancelledError(operation, {
cause: signal.reason,
partialObservation: delivered.value,
partialObservation: input.delivered,
diagnostics: { evidence_bundle: { ...written.value } },
}),
);
return delivered;
return ok({ delivered: input.delivered, evidenceBundle: written.value });
};
/** Resolve the caller's exact retained parent from an authenticated portable bundle. */
+35 -40
View File
@@ -4,12 +4,11 @@ import { parseEvidence, type Evidence } from "../domain/evidence.js";
import type { JsonValue } from "../domain/jsonValue.js";
import { projectAnalysisError } from "../domain/analysisErrorProjection.js";
import type { AnalysisError } from "../domain/analysisErrorBase.js";
import type { Result } from "../domain/result.js";
import { ok, type Result } from "../domain/result.js";
import { AnalysisInputError } from "../domain/analysisErrorCore.js";
import { createEvidenceBundle } from "../domain/evidenceBundle.js";
import { writeEvidenceBundle } from "../application/EvidenceBundleFiles.js";
import { publishEvidenceDelivery } from "../application/EvidenceBundleCommands.js";
import { inspectAnalysisView } from "../application/analysisView/AnalysisViewService.js";
import { ArtifactResourceScope } from "../artifacts/ArtifactResourceScope.js";
import { absoluteCallerPath } from "../filesystem/callerPath.js";
import { finishAnalysisCommand } from "./finishAnalysisCommand.js";
@@ -106,7 +105,7 @@ export const publishCliAnalysisEvidence = async (
value: JsonValue,
input: CliEvidenceDelivery,
) => {
const { providerId, operation, signal, detail, evidenceOutput } = input;
const { operation, signal, detail, evidenceOutput } = input;
if (!isEvidence(value)) return value;
const complete = parseEvidence(value);
const inputError = cliEvidenceDeliveryInputError(input);
@@ -115,43 +114,39 @@ export const publishCliAnalysisEvidence = async (
return projectAnalysisError(inputError);
}
if (evidenceOutput === undefined) return complete;
const resources = new ArtifactResourceScope();
let receipt: { readonly path: string; readonly bytes: number } | undefined;
const delivered = await finishAnalysisCommand(
async () => {
const written = await writeEvidenceBundle(
createEvidenceBundle([
...new Map(
[...(input.sourceEvidence ?? []), complete].map((record) => [
record.evidence_id,
record,
]),
).values(),
// Project before publication so a view failure cannot commit an
// unreported bundle; publication then owns every post-commit outcome.
const delivered =
detail === "summary"
? inspectAnalysisView({
source: { kind: "inline", evidence: complete },
view: input.view ?? { kind: "summary" },
})
: ok(complete);
if (!delivered.ok)
return projectAnalysisError(
delivered.error.retainPartialObservation(complete),
);
const published = await publishEvidenceDelivery({
bundle: createEvidenceBundle([
...new Map(
[...(input.sourceEvidence ?? []), complete].map((record) => [
record.evidence_id,
record,
]),
absoluteCallerPath(evidenceOutput),
false,
resources,
signal,
);
if (!written.ok) {
if (written.error.partialObservation === undefined)
written.error.retainPartialObservation(complete);
return written;
}
receipt = written.value;
return detail === "summary"
? inspectAnalysisView({
source: { kind: "inline", evidence: complete },
view: input.view ?? { kind: "summary" },
})
: { ok: true as const, value: complete };
},
{ close: () => resources.close() },
providerId,
).values(),
]),
complete,
delivered: delivered.value,
outputPath: absoluteCallerPath(evidenceOutput),
operation,
signal,
);
return isEvidence(delivered) && detail === "summary"
? { summary: delivered, evidence_bundle: receipt }
: delivered;
});
if (!published.ok) return projectAnalysisError(published.error);
return detail === "summary"
? {
summary: published.value.delivered,
evidence_bundle: published.value.evidenceBundle,
}
: published.value.delivered;
};
@@ -1,6 +1,6 @@
import { Cli, z } from "incur";
import pino from "pino";
import { readFile, writeFile } from "node:fs/promises";
import { readFile, readdir, writeFile } from "node:fs/promises";
import { join } from "node:path";
import { expect, it, onTestFinished, vi } from "vitest";
import { registerReverseEngineeringCommands } from "../../../src/cli/reverseEngineeringCommands.js";
@@ -11,7 +11,9 @@ import {
parseEvidence,
evidenceSchema,
} from "../../../src/domain/evidence.js";
import { parseEvidenceBundle } from "../../../src/domain/evidenceBundle.js";
import { ok } from "../../../src/domain/result.js";
import { OwnedStagedOutput } from "../../../src/filesystem/OwnedStagedOutput.js";
import { GdbSessionManager } from "../../../src/gdb/GdbSessionManager.js";
import { TestReverseEngineeringService } from "../../support/reverseEngineeringService.js";
import { createTestTempDirectory } from "../../fixtures/temporaryDirectory.js";
@@ -207,3 +209,73 @@ it.each(kinds)(
} else expect(f.closeService).toHaveBeenCalledOnce();
},
);
const interruptAt = (method: "assertReadyForPublication" | "close"): void => {
const original = OwnedStagedOutput.prototype[method];
const seam = vi
.spyOn(OwnedStagedOutput.prototype, method)
.mockImplementationOnce(async function (
this: OwnedStagedOutput,
...arguments_: Parameters<OwnedStagedOutput[typeof method]>
) {
await Reflect.apply(original, this, arguments_);
process.emit("SIGINT");
});
onTestFinished(() => {
seam.mockRestore();
process.exitCode = undefined;
});
};
it.each(kinds)(
"keeps the committed %s receipt when SIGINT arrives during final staging cleanup",
async (kind) => {
const f = await fixture(kind);
const destination = join(f.directory, "complete.json");
interruptAt("close");
const output = await f.invoke({
detail: "summary",
evidence_output: destination,
});
expect(process.exitCode).toBe(130);
const serialized = await readFile(destination, "utf8");
const parent = parseEvidenceBundle(JSON.parse(serialized)).records[0]!;
expect(output).toMatchObject({
code: "cancelled",
details: {
diagnostics: {
evidence_bundle: {
path: destination,
bytes: Buffer.byteLength(serialized),
},
},
partial_observation: {
evidence_links: [parent.evidence_id],
normalized_result: { parent_evidence_id: parent.evidence_id },
},
},
});
expect(await readdir(f.directory)).not.toContainEqual(
expect.stringMatching(/^\.rea-write-/u),
);
},
);
it.each(kinds)(
"publishes no %s bundle or receipt when SIGINT arrives before commit",
async (kind) => {
const f = await fixture(kind);
const destination = join(f.directory, "complete.json");
interruptAt("assertReadyForPublication");
const output = await f.invoke({
detail: "summary",
evidence_output: destination,
});
expect(process.exitCode).toBe(130);
expect(output).toMatchObject({ code: "cancelled" });
expect(output).not.toHaveProperty("details.diagnostics");
expect(JSON.stringify(output)).not.toContain(destination);
expect(output).toHaveProperty("details.partial_observation.raw_result");
expect(await readdir(f.directory)).toEqual(["input.bin"]);
},
);
@@ -0,0 +1,213 @@
import { pathToFileURL } from "node:url";
import {
mkdir,
readFile,
readdir,
realpath,
writeFile,
} from "node:fs/promises";
import { join, resolve } from "node:path";
import { expect } from "vitest";
import { analysisErrorProjectionSchema } from "../../../src/contracts/errorSchemas.js";
import { parseEvidence } from "../../../src/domain/evidence.js";
import { parseEvidenceBundle } from "../../../src/domain/evidenceBundle.js";
import { createTestTempDirectory } from "../../fixtures/temporaryDirectory.js";
import { cliTest, type TestCli } from "../../support/cli/cliFixture.js";
// Two command families share --evidence_output publication; the same real
// staged write is driven across its commit point through the compiled CLI.
const families = [
{
command: "inspect-model-artifact",
operation: "inspect_model_artifact",
target: async (root: string): Promise<string> => {
const header = Buffer.from(
JSON.stringify({
weight: { dtype: "F32", shape: [1], data_offsets: [0, 4] },
}),
);
const length = Buffer.alloc(8);
length.writeBigUInt64LE(BigInt(header.length));
const path = join(root, "fixture.safetensors");
await writeFile(path, Buffer.concat([length, header, Buffer.alloc(4)]));
return path;
},
},
{
command: "inspect-desktop-application",
operation: "inspect_desktop_application",
target: async (root: string): Promise<string> => {
const path = join(root, "installation");
await mkdir(join(path, "resources"), { recursive: true });
await writeFile(join(path, "resources/app.asar"), "path convention");
return path;
},
},
] as const;
type Family = (typeof families)[number];
const publish = async (
cli: TestCli,
family: Family,
detail: "summary" | "complete",
publication: Record<string, unknown>,
) => {
const root = await createTestTempDirectory("rea-evidence-output-");
const target = await family.target(root);
const destination = join(root, "parent.json");
const marker = join(root, "signal.json");
const reply = await cli.run({
cwd: root,
arguments: [
family.command,
target,
"--detail",
detail,
"--evidence_output",
destination,
"--json",
],
environment: {
HOME: root,
REA_LOG_LEVEL: "silent",
REA_TEST_EVIDENCE_PUBLICATION: JSON.stringify({
prefix: join(await realpath(root), ".rea-write-"),
marker,
...publication,
}),
NODE_OPTIONS: `--max-old-space-size=2048 --import=${pathToFileURL(resolve("tests/fixtures/evidencePublicationPreload.mjs")).href}`,
},
});
const staging = (await readdir(root)).filter((name) =>
name.startsWith(".rea-write-"),
);
return { reply, destination, marker, staging };
};
const cliError = (json: unknown) => analysisErrorProjectionSchema.parse(json);
const committedParent = async (destination: string) => {
const bytes = await readFile(destination);
const records = parseEvidenceBundle(JSON.parse(bytes.toString())).records;
expect(records).toHaveLength(1);
return { bytes: bytes.length, parent: records[0]! };
};
const signalDelivery =
process.platform === "win32" ? ("event" as const) : ("os" as const);
cliTest.for(
families.flatMap((family) => [
{ family, detail: "summary" as const, signal: "SIGINT" as const },
{ family, detail: "summary" as const, signal: "SIGTERM" as const },
{ family, detail: "complete" as const, signal: "SIGINT" as const },
]),
)(
"$family.command keeps the committed $detail receipt when $signal arrives during final cleanup",
async ({ family, detail, signal }, { cli }) => {
const { reply, destination, marker, staging } = await publish(
cli,
family,
detail,
{ phase: "cleanup", signal, delivery: signalDelivery },
);
expect(reply.exitCode).toBe(signal === "SIGINT" ? 130 : 143);
const error = cliError(reply.json);
expect(error.code).toBe("cancelled");
expect(error.details?.operation).toBe(family.operation);
const { bytes, parent } = await committedParent(destination);
expect(error.details?.diagnostics).toEqual({
evidence_bundle: { path: destination, bytes },
});
const partial = parseEvidence(error.details?.partial_observation);
if (detail === "complete") expect(partial).toEqual(parent);
else {
expect(partial.evidence_links).toEqual([parent.evidence_id]);
expect(partial.normalized_result).toMatchObject({
kind: "summary",
parent_evidence_id: parent.evidence_id,
});
}
const proof = JSON.parse(await readFile(marker, "utf8"));
expect(proof).toMatchObject({ phase: "cleanup", signal });
expect(proof.listeners).toBeGreaterThan(0);
expect(staging).toEqual([]);
},
);
cliTest.for(families)(
"$command creates no output or receipt when SIGINT arrives before commit",
async (family, { cli }) => {
const { reply, destination, marker, staging } = await publish(
cli,
family,
"summary",
{ phase: "staging", signal: "SIGINT", delivery: signalDelivery },
);
expect(reply.exitCode).toBe(130);
const error = cliError(reply.json);
expect(error.code).toBe("cancelled");
expect(error.details?.operation).toBe(family.operation);
expect(error.details).not.toHaveProperty("diagnostics");
expect(JSON.stringify(reply.json)).not.toContain(destination);
await expect(readFile(destination)).rejects.toMatchObject({
code: "ENOENT",
});
// The complete Evidence remains available when nothing was published.
expect(parseEvidence(error.details?.partial_observation).operation).toBe(
family.operation,
);
const proof = JSON.parse(await readFile(marker, "utf8"));
expect(proof).toMatchObject({ phase: "staging", signal: "SIGINT" });
expect(proof.listeners).toBeGreaterThan(0);
expect(staging).toEqual([]);
},
);
cliTest.for(families)(
"$command delivers the receipt once a failed staging cleanup is confirmed by retry",
async (family, { cli }) => {
const { reply, destination, staging } = await publish(
cli,
family,
"summary",
{ cleanup_failures: 1 },
);
expect(reply.exitCode).toBe(0);
const { bytes, parent } = await committedParent(destination);
expect(reply.json).toMatchObject({
evidence_bundle: { path: destination, bytes },
summary: {
normalized_result: { parent_evidence_id: parent.evidence_id },
},
});
expect(staging).toEqual([]);
},
);
cliTest.for(families)(
"$command reports the committed receipt when staging cleanup stays incomplete",
async (family, { cli }) => {
const { reply, destination, staging } = await publish(
cli,
family,
"summary",
{ cleanup_failures: 1_000 },
);
expect(reply.exitCode).not.toBe(0);
const error = cliError(reply.json);
expect(error.code).toBe("cleanup_incomplete");
const { bytes, parent } = await committedParent(destination);
expect(error.details).toMatchObject({
published: true,
published_path: destination,
bytes,
cleanup: "incomplete",
});
expect(parseEvidence(error.details?.partial_observation)).toEqual(parent);
expect(staging).toHaveLength(1);
},
);
@@ -386,9 +386,7 @@ cliTest.for(
const target = join(root, "fixture.dll");
const destination = join(root, "parent.json");
const marker = join(root, "signal.json");
const preload = resolve(
"tests/fixtures/managedPublicationSignalPreload.mjs",
);
const preload = resolve("tests/fixtures/evidencePublicationPreload.mjs");
const prefix = join(await realpath(root), ".rea-write-");
await writeFile(target, memberFixture());
const reply = await cli.run({
@@ -407,7 +405,7 @@ cliTest.for(
environment: {
HOME: root,
REA_LOG_LEVEL: "silent",
REA_TEST_MANAGED_PUBLICATION: JSON.stringify({
REA_TEST_EVIDENCE_PUBLICATION: JSON.stringify({
prefix,
marker,
signal,
+82
View File
@@ -0,0 +1,82 @@
import assert from "node:assert/strict";
import { createRequire, syncBuiltinESMExports } from "node:module";
import { writeFileSync } from "node:fs";
// Drives one real staged Evidence publication at an exact lifecycle point.
// "cleanup" queues the signal after the committed output's staging directory
// is removed; "staging" delivers it once the staging directory exists, before
// any byte can be committed. cleanup_failures rejects that many removals of
// the staging directory first, without removing it.
const input = JSON.parse(process.env.REA_TEST_EVIDENCE_PUBLICATION ?? "null");
assert.ok(
input && typeof input.prefix === "string" && typeof input.marker === "string",
);
const phase = input.phase ?? "cleanup";
assert.ok(["staging", "cleanup"].includes(phase));
if (input.signal !== undefined) {
assert.ok(["SIGINT", "SIGTERM"].includes(input.signal));
assert.ok(["event", "os"].includes(input.delivery));
}
let cleanupFailures = input.cleanup_failures ?? 0;
assert.ok(Number.isSafeInteger(cleanupFailures) && cleanupFailures >= 0);
const require = createRequire(import.meta.url);
const fs = require("node:fs/promises");
const owned = (path) => String(path).startsWith(input.prefix);
let signalled = input.signal === undefined;
const deliver = (observe) => {
writeFileSync(
input.marker,
JSON.stringify({
phase,
signal: input.signal,
delivery: input.delivery,
listeners: process.listenerCount(input.signal),
}),
);
// Registered after the marker so it never stands in for the CLI handler.
// Signal watchers do not keep the event loop alive; the timer does, and
// reports a signal that was never observed. It never selects the phase.
const observed = observe
? new Promise((resolve, reject) => {
const timeout = setTimeout(
() => reject(new Error(`${input.signal} was not observed`)),
5_000,
);
process.once(input.signal, () => {
clearTimeout(timeout);
resolve();
});
})
: undefined;
if (input.delivery === "os") process.kill(process.pid, input.signal);
else process.emit(input.signal);
return observed;
};
const remove = fs.rmdir;
fs.rmdir = async (...arguments_) => {
if (cleanupFailures > 0 && owned(arguments_[0])) {
cleanupFailures -= 1;
throw Object.assign(new Error("injected staging cleanup failure"), {
code: "EBUSY",
});
}
const result = await remove(...arguments_);
if (phase === "cleanup" && !signalled && owned(arguments_[0])) {
signalled = true;
setImmediate(() => deliver(false));
}
return result;
};
const stage = fs.mkdtemp;
fs.mkdtemp = async (...arguments_) => {
const directory = await stage(...arguments_);
if (phase === "staging" && !signalled && owned(directory)) {
signalled = true;
await deliver(true);
}
return directory;
};
syncBuiltinESMExports();
-33
View File
@@ -1,33 +0,0 @@
import assert from "node:assert/strict";
import { createRequire, syncBuiltinESMExports } from "node:module";
import { writeFileSync } from "node:fs";
const input = JSON.parse(process.env.REA_TEST_MANAGED_PUBLICATION ?? "null");
assert.ok(
input && typeof input.prefix === "string" && typeof input.marker === "string",
);
assert.ok(["SIGINT", "SIGTERM"].includes(input.signal));
assert.ok(["event", "os"].includes(input.delivery));
const require = createRequire(import.meta.url);
const fs = require("node:fs/promises");
const remove = fs.rmdir;
let queued = false;
fs.rmdir = async (...arguments_) => {
const result = await remove(...arguments_);
if (!queued && String(arguments_[0]).startsWith(input.prefix)) {
queued = true;
setImmediate(() => {
writeFileSync(
input.marker,
JSON.stringify({
signal: input.signal,
delivery: input.delivery,
listeners: process.listenerCount(input.signal),
}),
);
if (input.delivery === "os") process.kill(process.pid, input.signal);
else process.emit(input.signal);
});
}
return result;
};
syncBuiltinESMExports();