diff --git a/docs/cli.md b/docs/cli.md index f9062dc9ff..27f9feb215 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -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 diff --git a/src/application/EvidenceBundleCommands.ts b/src/application/EvidenceBundleCommands.ts index a737613bf1..0da3db244f 100644 --- a/src/application/EvidenceBundleCommands.ts +++ b/src/application/EvidenceBundleCommands.ts @@ -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> => { + 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. */ diff --git a/src/cli/analysisEvidenceDelivery.ts b/src/cli/analysisEvidenceDelivery.ts index 2d27e9a4fb..9ee9a8bb39 100644 --- a/src/cli/analysisEvidenceDelivery.ts +++ b/src/cli/analysisEvidenceDelivery.ts @@ -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; }; diff --git a/tests/boundary/cli/commandEvidenceDelivery.test.ts b/tests/boundary/cli/commandEvidenceDelivery.test.ts index eecf5f10f7..1f04135019 100644 --- a/tests/boundary/cli/commandEvidenceDelivery.test.ts +++ b/tests/boundary/cli/commandEvidenceDelivery.test.ts @@ -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 + ) { + 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"]); + }, +); diff --git a/tests/boundary/cli/evidenceOutputPublicationCli.test.ts b/tests/boundary/cli/evidenceOutputPublicationCli.test.ts new file mode 100644 index 0000000000..b448b08d64 --- /dev/null +++ b/tests/boundary/cli/evidenceOutputPublicationCli.test.ts @@ -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 => { + 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 => { + 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, +) => { + 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); + }, +); diff --git a/tests/boundary/cli/managedMemberSummaryCli.test.ts b/tests/boundary/cli/managedMemberSummaryCli.test.ts index dcacd6f4a4..57632379f5 100644 --- a/tests/boundary/cli/managedMemberSummaryCli.test.ts +++ b/tests/boundary/cli/managedMemberSummaryCli.test.ts @@ -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, diff --git a/tests/fixtures/evidencePublicationPreload.mjs b/tests/fixtures/evidencePublicationPreload.mjs new file mode 100644 index 0000000000..b18fe14223 --- /dev/null +++ b/tests/fixtures/evidencePublicationPreload.mjs @@ -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(); diff --git a/tests/fixtures/managedPublicationSignalPreload.mjs b/tests/fixtures/managedPublicationSignalPreload.mjs deleted file mode 100644 index 3f9314e7e1..0000000000 --- a/tests/fixtures/managedPublicationSignalPreload.mjs +++ /dev/null @@ -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();