fix(evidence): own standalone command finalization

This commit is contained in:
morluto committed 2026-10-27 12:24:59 +08:00
1 parent 2e7a206b20
commit c4242cb85e
4 files changed
+458 -119

No files matched your search

+117 -29
View File
@@ -34,24 +34,33 @@ import {
readComparisonEvidenceBundles,
writeEvidenceBundle,
} from "./EvidenceBundleFiles.js";
import { inspectEvidenceBundle } from "./investigation/InspectEvidenceBundle.js";
import type { EvidenceRecordFilters } from "../domain/evidenceRecordSummary.js";
import { inspectAnalysisViewValidated } from "./analysisView/AnalysisViewService.js";
/** Validate a standalone bundle and report its complete record/revision counts. */
export const importEvidenceBundleCommand = async (
path: string,
signal?: AbortSignal,
): Promise<Result<JsonValue, AnalysisError>> => {
const loaded = await readEvidenceBundle(
path,
undefined,
undefined,
return withBundleReadResources(
async (resources) => {
const loaded = await readEvidenceBundle(
path,
signal,
resources,
"import_evidence_bundle",
);
if (!loaded.ok) return loaded;
return ok({
imported: loaded.value.records.length,
unknowns_added: loaded.value.unknowns.length,
total: loaded.value.records.length,
});
},
"import_evidence_bundle",
signal,
);
if (!loaded.ok) return loaded;
return ok({
imported: loaded.value.records.length,
unknowns_added: loaded.value.unknowns.length,
total: loaded.value.records.length,
});
};
/** Validate a source bundle and atomically export canonical bytes. */
@@ -59,19 +68,37 @@ export const exportEvidenceBundleCommand = async (
sourcePath: string,
outputPath: string,
overwrite: boolean,
signal?: AbortSignal,
): Promise<Result<JsonValue, AnalysisError>> => {
// The sealed snapshot lets the writer reuse this authentication.
const loaded = await readImmutableEvidenceBundle(
sourcePath,
undefined,
undefined,
const loaded = await withBundleReadResources(
(resources) =>
readImmutableEvidenceBundle(
sourcePath,
signal,
resources,
"export_evidence_bundle",
),
"export_evidence_bundle",
signal,
);
if (!loaded.ok) return loaded;
return projectWrite(
const written = await persistEvidenceBundleCommand(
loaded.value,
await persistEvidenceBundleCommand(loaded.value, outputPath, overwrite),
outputPath,
overwrite,
signal,
"export_evidence_bundle",
);
await dispatchQueuedInterrupts();
if (written.ok && signal?.aborted)
return err(
new AnalysisCancelledError("export_evidence_bundle", {
cause: signal.reason,
diagnostics: { evidence_bundle: { ...written.value } },
}),
);
return projectWrite(loaded.value, written);
};
/** Publish complete parent Evidence before delivering it or its linked summary. */
@@ -224,23 +251,32 @@ export const compareEvidenceBundlesCommand = async (input: {
readonly leftPath: string;
readonly rightPath: string;
readonly recordPairs?: readonly BundleRecordPair[];
readonly signal?: AbortSignal;
}): Promise<Result<JsonValue, AnalysisError>> => {
// Sealed snapshots let comparison and its bundle digests reuse this
// authentication instead of re-validating every record twice more.
const loaded = await readComparisonEvidenceBundles(
input.leftPath,
input.rightPath,
return withBundleReadResources(
async (resources) => {
const loaded = await readComparisonEvidenceBundles(
input.leftPath,
input.rightPath,
input.signal,
resources,
);
if (!loaded.ok) return loaded;
const { left, right } = loaded.value;
try {
return ok(
jsonValueSchema.parse(compareBundles(left, right, input.recordPairs)),
);
} catch (cause: unknown) {
if (cause instanceof AnalysisInputError) return err(cause);
throw cause;
}
},
"compare_bundles",
input.signal,
);
if (!loaded.ok) return loaded;
const { left, right } = loaded.value;
try {
return ok(
jsonValueSchema.parse(compareBundles(left, right, input.recordPairs)),
);
} catch (cause: unknown) {
if (cause instanceof AnalysisInputError) return err(cause);
throw cause;
}
};
const projectWrite = (
@@ -254,3 +290,55 @@ const projectWrite = (
records: bundle.records.length,
})
: err(written.error);
/** Standalone bundle discovery shares the same read owner as import and comparison. */
export const inspectEvidenceBundleCommand = async (
input: {
readonly path: string;
readonly detail: "complete" | "summary";
readonly filters?: EvidenceRecordFilters | undefined;
},
signal?: AbortSignal,
) => {
return withBundleReadResources(
async (resources) => {
const loaded = await readEvidenceBundle(
input.path,
signal,
resources,
"get_evidence_bundle",
);
return loaded.ok ? inspectEvidenceBundle(loaded.value, input) : loaded;
},
"get_evidence_bundle",
signal,
);
};
/** Final read cleanup retries the exact leases already bound to a failed read. */
const withBundleReadResources = async <Value extends JsonValue>(
read: (
resources: ArtifactResourceScope,
) => Promise<Result<Value, AnalysisError>>,
operation: string,
signal?: AbortSignal,
): Promise<Result<Value, AnalysisError>> => {
const resources = new ArtifactResourceScope();
let result: Result<Value, AnalysisError>;
try {
result = await read(resources);
} finally {
// Read errors own their retained leases and observe resolution or remaining
// failure in place. A retry must not replace the original read failure.
await resources.close().catch(() => undefined);
await dispatchQueuedInterrupts();
}
return result.ok && signal?.aborted
? err(
new AnalysisCancelledError(operation, {
cause: signal.reason,
diagnostics: { completed_result: result.value },
}),
)
: result;
};
+34 -40
View File
@@ -3,8 +3,7 @@ import { z } from "incur";
import { inspectAnalysisView } from "../application/analysisView/AnalysisViewService.js";
import { ArtifactResourceScope } from "../artifacts/ArtifactResourceScope.js";
import { ANALYSIS_VIEW_PROVIDER } from "../domain/investigationProviders.js";
import { readEvidenceBundle } from "../application/EvidenceBundleFiles.js";
import { inspectEvidenceBundle } from "../application/investigation/InspectEvidenceBundle.js";
import { inspectEvidenceBundleCommand } from "../application/EvidenceBundleCommands.js";
import {
getEvidenceBundleInputSchema,
importEvidenceBundleInputSchema,
@@ -36,46 +35,41 @@ export const registerAnalysisViewCommands = (
),
}),
run: ({ args }) =>
logCliCommand(logger, CLI_COMMANDS.inspectEvidenceBundle, async () => {
const input = await parseCliJsonInput(
args.inputJson,
CLI_COMMANDS.inspectEvidenceBundle,
);
if (!input.ok) return input.error;
const schema = getEvidenceBundleInputSchema.safeExtend({
path: importEvidenceBundleInputSchema.shape.path,
});
const parsed = schema.safeParse(input.value);
if (!parsed.success)
return {
error: "Evidence bundle inspection failed",
...projectAnalysisError(
analysisInputErrorFromIssues(
CLI_COMMANDS.inspectEvidenceBundle,
parsed.error.issues,
input.value,
),
),
};
const loaded = await readEvidenceBundle(
parsed.data.path,
undefined,
undefined,
"get_evidence_bundle",
);
if (!loaded.ok)
return {
error: "Evidence bundle inspection failed",
...projectAnalysisError(loaded.error),
};
const result = inspectEvidenceBundle(loaded.value, parsed.data);
return result.ok
? result.value
: {
withCommandCancellation((signal) =>
logCliCommand(logger, CLI_COMMANDS.inspectEvidenceBundle, async () => {
const input = await parseCliJsonInput(
args.inputJson,
CLI_COMMANDS.inspectEvidenceBundle,
signal,
);
if (!input.ok) return input.error;
const schema = getEvidenceBundleInputSchema.safeExtend({
path: importEvidenceBundleInputSchema.shape.path,
});
const parsed = schema.safeParse(input.value);
if (!parsed.success)
return {
error: "Evidence bundle inspection failed",
...projectAnalysisError(result.error),
...projectAnalysisError(
analysisInputErrorFromIssues(
CLI_COMMANDS.inspectEvidenceBundle,
parsed.error.issues,
input.value,
),
),
};
}),
const result = await inspectEvidenceBundleCommand(
parsed.data,
signal,
);
return result.ok
? result.value
: {
error: "Evidence bundle inspection failed",
...projectAnalysisError(result.error),
};
}),
),
});
cli.command(CLI_COMMANDS.inspectAnalysisView, {
description:
+59 -50
View File
@@ -1,3 +1,4 @@
import { withCommandCancellation } from "./cli/commandCancellation.js";
import { safeParseJson } from "./domain/safeJson.js";
import { AnalysisInputError } from "./domain/analysisErrorCore.js";
import { analysisInputErrorFromIssues } from "./domain/inputIssueProjection.js";
@@ -27,10 +28,12 @@ export const registerEvidenceCommands = (
path: z.string().describe("Evidence bundle JSON path"),
}),
run: ({ args }) =>
logCliCommand(logger, "evidence-import", async () => {
const imported = await importEvidenceBundleCommand(args.path);
return imported.ok ? imported.value : cliError(imported.error);
}),
withCommandCancellation((signal) =>
logCliCommand(logger, "evidence-import", async () => {
const imported = await importEvidenceBundleCommand(args.path, signal);
return imported.ok ? imported.value : cliError(imported.error);
}),
),
});
cli.command(CLI_COMMANDS.evidenceExport, {
description: "Validate and atomically export canonical Evidence JSON",
@@ -42,14 +45,17 @@ export const registerEvidenceCommands = (
overwrite: z.boolean().default(false).describe("Replace output file"),
}),
run: ({ args, options }) =>
logCliCommand(logger, "evidence-export", async () => {
const exported = await exportEvidenceBundleCommand(
args.source,
args.output,
options.overwrite,
);
return exported.ok ? exported.value : cliError(exported.error);
}),
withCommandCancellation((signal) =>
logCliCommand(logger, "evidence-export", async () => {
const exported = await exportEvidenceBundleCommand(
args.source,
args.output,
options.overwrite,
signal,
);
return exported.ok ? exported.value : cliError(exported.error);
}),
),
});
cli.command(CLI_COMMANDS.compare, {
description: "Compare two canonical Evidence bundles",
@@ -66,44 +72,47 @@ export const registerEvidenceCommands = (
),
}),
run: ({ args, options }) =>
logCliCommand(logger, "compare", async () => {
const decoded =
options.record_pairs === undefined
? { ok: true as const, value: [] }
: safeParseJson(options.record_pairs);
if (!decoded.ok)
return cliError(
new AnalysisInputError(
"compare_bundles",
{ cause: decoded.cause },
[
{
path: ["record_pairs"],
reason: "invalid_format",
expected: "JSON array",
},
],
),
);
const input = { record_pairs: decoded.value };
const selected = bundleComparisonInputSchema
.pick({ record_pairs: true })
.safeParse(input);
if (!selected.success)
return cliError(
analysisInputErrorFromIssues(
"compare_bundles",
selected.error.issues,
input,
),
);
const compared = await compareEvidenceBundlesCommand({
leftPath: args.left,
rightPath: args.right,
recordPairs: selected.data.record_pairs,
});
return compared.ok ? compared.value : cliError(compared.error);
}),
withCommandCancellation((signal) =>
logCliCommand(logger, "compare", async () => {
const decoded =
options.record_pairs === undefined
? { ok: true as const, value: [] }
: safeParseJson(options.record_pairs);
if (!decoded.ok)
return cliError(
new AnalysisInputError(
"compare_bundles",
{ cause: decoded.cause },
[
{
path: ["record_pairs"],
reason: "invalid_format",
expected: "JSON array",
},
],
),
);
const input = { record_pairs: decoded.value };
const selected = bundleComparisonInputSchema
.pick({ record_pairs: true })
.safeParse(input);
if (!selected.success)
return cliError(
analysisInputErrorFromIssues(
"compare_bundles",
selected.error.issues,
input,
),
);
const compared = await compareEvidenceBundlesCommand({
leftPath: args.left,
rightPath: args.right,
recordPairs: selected.data.record_pairs,
signal,
});
return compared.ok ? compared.value : cliError(compared.error);
}),
),
});
};
@@ -13,8 +13,11 @@ import {
writeEvidenceBundle,
} from "../../src/application/EvidenceBundleFiles.js";
import { ArtifactResourceScope } from "../../src/artifacts/ArtifactResourceScope.js";
import { OwnedStagedOutput } from "../../src/filesystem/OwnedStagedOutput.js";
import { OwnedFileHandle } from "../../src/filesystem/OwnedFileHandle.js";
import { RetainedRegularFileCleanupFailure } from "../../src/application/RegularFileRead.js";
import * as bundleComparison from "../../src/domain/bundleComparison.js";
import * as bundleSummary from "../../src/domain/evidenceRecordSummary.js";
import { bundleComparisonResultSchema } from "../../src/domain/bundleComparison.js";
import { processEvidence } from "../../src/application/ReconstructionObligationLedger.fixture.js";
import { processCaptureSchema } from "../../src/domain/process/processCapture.js";
@@ -315,3 +318,248 @@ it.each([
).rejects.toBe(failure);
},
);
const standaloneBundleArguments = (
command: string,
path: string,
output: string,
) =>
command === "inspect-evidence-bundle"
? [command, JSON.stringify({ path })]
: command === "compare"
? [command, path, path]
: command === "evidence-export"
? [command, path, output]
: [command, path];
it.each([
"evidence-import",
"evidence-export",
"compare",
"inspect-evidence-bundle",
])(
"retries retained read cleanup through the standalone %s owner",
async (command) => {
const directory = await createTestTempDirectory(
"rea-standalone-bundle-cleanup-",
);
const { path } = await validBundleFile(directory);
const output = join(directory, "output.json");
const original = OwnedFileHandle.prototype.close;
const attempts = new Map<OwnedFileHandle, number>();
const close = vi
.spyOn(OwnedFileHandle.prototype, "close")
.mockImplementation(async function (this: OwnedFileHandle) {
const attempt = (attempts.get(this) ?? 0) + 1;
attempts.set(this, attempt);
if (attempt < 3) throw new Error("retryable descriptor close failure");
return original.call(this);
});
onTestFinished(async () => {
close.mockRestore();
for (const owner of attempts.keys()) await owner.close();
});
let stdout = "";
const previousExitCode = process.exitCode;
try {
await createCli({}).serve(
[...standaloneBundleArguments(command, path, output), "--json"],
{
stdout: (part) => {
stdout += part;
},
exit: () => {},
},
);
expect(attempts.size).toBe(command === "compare" ? 2 : 1);
expect([...attempts.values()]).toEqual(
command === "compare" ? [3, 3] : [3],
);
expect([...attempts.keys()].every((owner) => owner.handle.fd < 0)).toBe(
true,
);
const result = JSON.parse(stdout);
// A read which failed to complete its own cleanup remains a failed read;
// final cleanup resolves resource ownership rather than inventing data.
expect(result).toMatchObject({ code: "execution_failure" });
expect(result.details?.cleanup_incomplete).not.toBe(true);
expect(await readdir(directory)).toEqual(["bundle.json"]);
} finally {
process.exitCode = previousExitCode;
}
},
);
it.each([
"evidence-import",
"evidence-export",
"compare",
"inspect-evidence-bundle",
])("handles SIGINT during the standalone %s input close", async (command) => {
const directory = await createTestTempDirectory(
"rea-standalone-bundle-cancel-",
);
const { path } = await validBundleFile(directory);
const output = join(directory, "output.json");
const original = OwnedFileHandle.prototype.close;
const owners = new Set<OwnedFileHandle>();
const close = vi
.spyOn(OwnedFileHandle.prototype, "close")
.mockImplementation(async function (this: OwnedFileHandle) {
owners.add(this);
await original.call(this);
process.emit("SIGINT");
});
onTestFinished(() => close.mockRestore());
const before = process.listenerCount("SIGINT");
const previousExitCode = process.exitCode;
let stdout = "";
try {
process.exitCode = undefined;
await createCli({}).serve(
[...standaloneBundleArguments(command, path, output), "--json"],
{
stdout: (part) => {
stdout += part;
},
exit: () => {},
},
);
expect(JSON.parse(stdout)).toMatchObject({ code: "cancelled" });
expect(process.exitCode).toBe(130);
expect(process.listenerCount("SIGINT")).toBe(before);
expect([...owners].every((owner) => owner.handle.fd < 0)).toBe(true);
expect(await readdir(directory)).toEqual(["bundle.json"]);
} finally {
process.exitCode = previousExitCode;
}
});
it.each(["compare", "inspect-evidence-bundle"])(
"retains completed %s output when SIGINT is queued during projection",
async (command) => {
const directory = await createTestTempDirectory(
"rea-standalone-bundle-projection-",
);
const { path, bundle } = await validBundleFile(directory);
const queueInterrupt = () => {
setImmediate(() => process.emit("SIGINT"));
};
const compare = bundleComparison.compareBundles;
const summarize = bundleSummary.summarizeEvidenceBundle;
const hook =
command === "compare"
? vi
.spyOn(bundleComparison, "compareBundles")
.mockImplementation((...args) => {
const result = compare(...args);
queueInterrupt();
return result;
})
: vi
.spyOn(bundleSummary, "summarizeEvidenceBundle")
.mockImplementation((...args) => {
const result = summarize(...args);
queueInterrupt();
return result;
});
onTestFinished(() => hook.mockRestore());
const previousExitCode = process.exitCode;
const listeners = process.listenerCount("SIGINT");
let stdout = "";
try {
process.exitCode = undefined;
await createCli({}).serve(
[
...(command === "compare"
? [command, path, path]
: [command, JSON.stringify({ path, detail: "summary" })]),
"--json",
],
{
stdout: (part) => {
stdout += part;
},
exit: () => {},
},
);
const expected =
command === "compare" ? compare(bundle, bundle) : summarize(bundle);
expect(JSON.parse(stdout)).toMatchObject({
code: "cancelled",
details: { diagnostics: { completed_result: expected } },
});
expect(hook).toHaveBeenCalledOnce();
expect(process.exitCode).toBe(130);
expect(process.listenerCount("SIGINT")).toBe(listeners);
} finally {
process.exitCode = previousExitCode;
}
},
);
it.each(["before-commit", "after-commit", "cleanup-retry"])(
"preserves standalone export publication facts on SIGINT at %s",
async (phase) => {
const directory = await createTestTempDirectory(
"rea-standalone-export-publication-",
);
const { path, bundle } = await validBundleFile(directory);
const output = join(directory, "output.json");
const method =
phase === "before-commit" ? "assertReadyForPublication" : "close";
const original = OwnedStagedOutput.prototype[method];
let attempts = 0;
const hook = vi
.spyOn(OwnedStagedOutput.prototype, method)
.mockImplementation(async function (
this: OwnedStagedOutput,
...args: Parameters<OwnedStagedOutput[typeof method]>
) {
attempts += 1;
if (phase === "cleanup-retry" && attempts === 1)
throw new Error("retry staged cleanup");
await Reflect.apply(original, this, args);
if (phase === "before-commit") process.emit("SIGINT");
else setImmediate(() => process.emit("SIGINT"));
});
onTestFinished(() => hook.mockRestore());
const previousExitCode = process.exitCode;
const listeners = process.listenerCount("SIGINT");
let stdout = "";
try {
process.exitCode = undefined;
await createCli({}).serve(["evidence-export", path, output, "--json"], {
stdout: (part) => {
stdout += part;
},
exit: () => {},
});
const result = JSON.parse(stdout);
expect(result).toMatchObject({ code: "cancelled" });
expect(process.exitCode).toBe(130);
expect(process.listenerCount("SIGINT")).toBe(listeners);
if (phase === "before-commit") {
expect(await readdir(directory)).toEqual(["bundle.json"]);
expect(result).not.toHaveProperty(
"details.diagnostics.evidence_bundle",
);
} else {
const bytes = await stat(output);
expect(result.details.diagnostics.evidence_bundle).toEqual({
path: output,
bytes: bytes.size,
});
const loaded = await readEvidenceBundle(output);
expect(loaded).toEqual({ ok: true, value: bundle });
expect(await readdir(directory)).toEqual([
"bundle.json",
"output.json",
]);
}
expect(attempts).toBe(phase === "cleanup-retry" ? 2 : 1);
} finally {
process.exitCode = previousExitCode;
}
},
);