fix(providers): preserve artifact producer cancellation facts

Wire sigrok and tshark finalization to the effective signal. Services and file-failure adapters retain exact opaque reasons and completed Evidence instead of guessing from error names. Preserve typed primary outcomes through failed cleanup and recovery.

Validation: latest per-file coverage is 114 passing tests across nine selected files with 14 explicit Linux-only skips; core foundation passes independently. check:fast, docs and discovery pass. Four finalization controls failed before the signal ownership fix. Bounded completed authored-producer service workflows have identical Evidence with no performance improvement claim. Real sigrok/tshark and native Linux/Windows were not verified.

Closes #4858
This commit is contained in:
morluto committed 2026-10-18 02:38:30 +08:00
1 parent 147ae9718a
commit 8a247fc70f
7 files changed
+341 -18

No files matched your search

@@ -49,7 +49,9 @@ export class PacketCaptureService {
): Promise<Result<Evidence, AnalysisError>> {
const operation = "inspect_packet_capture";
if (options?.signal?.aborted)
return err(new AnalysisCancelledError(operation));
return err(
new AnalysisCancelledError(operation, { cause: options.signal.reason }),
);
const input = inspectPacketCaptureInputSchema.safeParse(raw);
if (!input.success)
return err(
@@ -92,7 +94,9 @@ export class PacketCaptureService {
): Promise<Result<Evidence, AnalysisError>> {
const operation = "inspect_packet_fields";
if (options?.signal?.aborted)
return err(new AnalysisCancelledError(operation));
return err(
new AnalysisCancelledError(operation, { cause: options.signal.reason }),
);
const input = inspectPacketFieldsInputSchema.safeParse(raw);
if (!input.success)
return err(
@@ -139,6 +143,7 @@ export class PacketCaptureService {
return options?.signal?.aborted
? err(
new AnalysisCancelledError(operation, {
cause: options.signal.reason,
partialObservation: evidence,
}),
)
@@ -49,7 +49,9 @@ export class SignalRecordingService {
): Promise<Result<Evidence, AnalysisError>> {
const operation = "decode_signal_recording";
if (options?.signal?.aborted)
return err(new AnalysisCancelledError(operation));
return err(
new AnalysisCancelledError(operation, { cause: options.signal.reason }),
);
const input = decodeSignalRecordingInputSchema.safeParse(raw);
if (!input.success)
return err(
@@ -92,7 +94,9 @@ export class SignalRecordingService {
): Promise<Result<Evidence, AnalysisError>> {
const operation = "inspect_signal_decoders";
if (options?.signal?.aborted)
return err(new AnalysisCancelledError(operation));
return err(
new AnalysisCancelledError(operation, { cause: options.signal.reason }),
);
const input = inspectSignalDecodersInputSchema.safeParse(raw);
if (!input.success)
return err(
@@ -139,6 +143,7 @@ export class SignalRecordingService {
return options?.signal?.aborted
? err(
new AnalysisCancelledError(operation, {
cause: options.signal.reason,
partialObservation: evidence,
}),
)
+13 -6
View File
@@ -1,3 +1,4 @@
import { artifactFileCancellationCause } from "../artifacts/ArtifactFileOperation.js";
import {
type InspectPacketFieldsInput,
packetFieldsCatalogSchema,
@@ -130,7 +131,10 @@ export class TsharkCaptureProvider implements PacketCapturePort {
let root: ArtifactAnalysisRuntime | undefined;
let result: Result<AnalysisExecution, AnalysisError>;
try {
if (options?.signal?.aborted) throw new AnalysisCancelledError(operation);
if (options?.signal?.aborted)
throw new AnalysisCancelledError(operation, {
cause: options.signal.reason,
});
await this.#resources.release();
if (this.platform !== "linux")
throw new Error(
@@ -419,6 +423,7 @@ export class TsharkCaptureProvider implements PacketCapturePort {
limitations: execution.limitations,
}),
operation,
options?.signal,
);
}
#failure(
@@ -447,11 +452,13 @@ export class TsharkCaptureProvider implements PacketCapturePort {
observedOutput,
operation,
);
if (
(cause instanceof Error && cause.name === "AbortError") ||
(cause instanceof ArtifactReaderFailure && cause.reason === "cancelled")
)
return new AnalysisCancelledError(operation, { cause });
if (cause instanceof ArtifactReaderFailure && cause.reason === "cancelled")
return new AnalysisCancelledError(operation, {
cause: artifactFileCancellationCause(cause),
...(observedOutput === undefined
? {}
: { capturedOutput: observedOutput }),
});
if (cause instanceof FieldDiscoveryBudgetError)
return new AnalysisResourceConstraintError(
operation,
+13 -6
View File
@@ -1,3 +1,4 @@
import { artifactFileCancellationCause } from "../artifacts/ArtifactFileOperation.js";
import {
type InspectSignalDecodersInput,
signalDecodersSchema,
@@ -126,7 +127,10 @@ export class SigrokRecordingProvider implements SignalRecordingPort {
let root: ArtifactAnalysisRuntime | undefined;
let result: Result<AnalysisExecution, AnalysisError>;
try {
if (options?.signal?.aborted) throw new AnalysisCancelledError(operation);
if (options?.signal?.aborted)
throw new AnalysisCancelledError(operation, {
cause: options.signal.reason,
});
await this.#resources.release();
if (this.platform !== "linux")
throw new Error(
@@ -383,6 +387,7 @@ export class SigrokRecordingProvider implements SignalRecordingPort {
limitations: execution.limitations,
}),
operation,
options?.signal,
);
}
#failure(
@@ -407,11 +412,13 @@ export class SigrokRecordingProvider implements SignalRecordingPort {
observedOutput,
operation,
);
if (
(cause instanceof Error && cause.name === "AbortError") ||
(cause instanceof ArtifactReaderFailure && cause.reason === "cancelled")
)
return new AnalysisCancelledError(operation, { cause });
if (cause instanceof ArtifactReaderFailure && cause.reason === "cancelled")
return new AnalysisCancelledError(operation, {
cause: artifactFileCancellationCause(cause),
...(observedOutput === undefined
? {}
: { capturedOutput: observedOutput }),
});
if (cause instanceof DecoderDiscoveryBudgetError)
return new AnalysisResourceConstraintError(
operation,
@@ -97,6 +97,8 @@ else { if(readFileSync('capture', 'utf8') !== 'recording bytes') throw Error('wr
return {
path: root.path,
close: async () => {
if (mode === "close-abort")
controller.abort({ request: "cancel during workspace close" });
if (!mayClean) throw new Error("cleanup fixture failure");
await root.close();
},
@@ -248,6 +250,8 @@ it.each([
);
if (result.ok) throw new Error("Expected capture analysis failure");
expect(result.error._tag).toBe(tag);
if (mode === "wait")
expect(result.error.cause).toBe(f.controller.signal.reason);
expect(result.error.cleanupIncomplete).toBe(false);
expect(
(await readdir(f.directory)).filter((name) =>
@@ -327,7 +331,7 @@ it("retains a completed observation when cancellation arrives at the service bou
throw new Error("Unexpected field discovery call");
},
inspect: async () => {
f.controller.abort();
f.controller.abort({ request: "cancel at Evidence boundary" });
return execution;
},
close: async () => {},
@@ -335,6 +339,7 @@ it("retains a completed observation when cancellation arrives at the service bou
const result = await service.inspect(input, { signal: f.controller.signal });
if (result.ok) throw new Error("Expected late cancellation");
expect(result.error._tag).toBe("AnalysisCancelledError");
expect(result.error.cause).toBe(f.controller.signal.reason);
expect(result.error.partialObservation).toMatchObject({
operation: "inspect_packet_capture",
normalized_result: {
@@ -676,3 +681,36 @@ it("preserves discovery resource ownership through CLI shutdown retries", async
await f.service.close();
}
});
it.each(
[false, true].flatMap((failsClose) =>
[false, true].map((discovery) => ({ failsClose, discovery })),
),
)(
"retains completed Evidence and exact cancellation during workspace close: %j",
async ({ failsClose, discovery }) => {
const f = await fixture("close-abort", failsClose);
const result = discovery
? await f.service.inspectFields({}, { signal: f.controller.signal })
: await f.service.inspect(
{ path: f.path, fields: ["ip.addr"] },
{ signal: f.controller.signal },
);
if (result.ok) throw new Error("Expected finalization cancellation");
const primary = failsClose ? result.error.cleanupPrimary : result.error;
expect(primary).toMatchObject({ _tag: "AnalysisCancelledError" });
expect(primary?.cause).toBe(f.controller.signal.reason);
expect(result.error.partialObservation).toMatchObject({
operation: discovery ? "inspect_packet_fields" : "inspect_packet_capture",
});
expect(result.error.cleanupIncomplete).toBe(failsClose);
if (failsClose)
await expect(f.service.close()).rejects.toMatchObject({
cleanupIncomplete: true,
});
f.allowCleanup();
await f.service.close();
expect(result.error.cleanupIncomplete).toBe(false);
expect(primary?.cause).toBe(f.controller.signal.reason);
},
);
@@ -0,0 +1,221 @@
import { stat, writeFile } from "node:fs/promises";
import { join } from "node:path";
import { expect, it, vi } from "vitest";
import { SigrokRecordingProvider } from "../../../src/signals/SigrokRecordingProvider.js";
import { TsharkCaptureProvider } from "../../../src/captures/TsharkCaptureProvider.js";
import { SignalRecordingService } from "../../../src/application/signals/SignalRecordingService.js";
import { PacketCaptureService } from "../../../src/application/captures/PacketCaptureService.js";
import { decodeSignalRecordingInputSchema } from "../../../src/domain/signals/signalRecording.js";
import { inspectSignalDecodersInputSchema } from "../../../src/domain/signals/signalDecoders.js";
import { inspectPacketCaptureInputSchema } from "../../../src/domain/captures/packetCapture.js";
import { inspectPacketFieldsInputSchema } from "../../../src/domain/captures/packetFields.js";
import { OwnedFileHandle } from "../../../src/filesystem/OwnedFileHandle.js";
import { readStableArtifact } from "../../../src/artifacts/readStableArtifact.js";
import { ArtifactReaderFailure } from "../../../src/artifacts/ArtifactReader.js";
import { ProviderCleanupError } from "../../../src/domain/providerCleanupError.js";
import { AnalysisError } from "../../../src/domain/analysisErrorBase.js";
import type { ExecutionOptions } from "../../../src/application/AnalysisProvider.js";
import { createTestTempDirectory } from "../../fixtures/temporaryDirectory.js";
// Simulated Linux admission, real descriptors; no producer, limiter or target runs.
const fixture = async (
kind: "signal" | "capture",
readArtifact = readStableArtifact,
) => {
const root = await createTestTempDirectory("rea-producer-cancellation-");
const command = join(root, "selected-tool");
const path = join(root, "recording");
await writeFile(command, "not an executable producer", { mode: 0o700 });
await writeFile(path, "recording bytes");
const Provider =
kind === "signal" ? SigrokRecordingProvider : TsharkCaptureProvider;
const provider = new Provider(
{
REA_SIGROK_COMMAND: command,
REA_SIGROK_PRLIMIT_COMMAND: command,
REA_TSHARK_COMMAND: command,
REA_TSHARK_PRLIMIT_COMMAND: command,
},
async () => {
throw new Error("Must not launch a producer");
},
async () => {
throw new Error("Must not allocate a workspace");
},
"linux",
readArtifact,
);
const requests =
provider instanceof SigrokRecordingProvider
? [
(options: ExecutionOptions) =>
provider.decode(
decodeSignalRecordingInputSchema.parse({
path,
decoders: "uart",
}),
options,
),
(options: ExecutionOptions) =>
provider.inspectDecoders(
inspectSignalDecodersInputSchema.parse({}),
options,
),
(options: ExecutionOptions) =>
new SignalRecordingService(provider).decode(
{ path, decoders: "uart" },
options,
),
(options: ExecutionOptions) =>
new SignalRecordingService(provider).inspectDecoders({}, options),
]
: [
(options: ExecutionOptions) =>
provider.inspect(
inspectPacketCaptureInputSchema.parse({
path,
fields: ["frame.number"],
}),
options,
),
(options: ExecutionOptions) =>
provider.inspectFields(
inspectPacketFieldsInputSchema.parse({}),
options,
),
(options: ExecutionOptions) =>
new PacketCaptureService(provider).inspect(
{ path, fields: ["frame.number"] },
options,
),
(options: ExecutionOptions) =>
new PacketCaptureService(provider).inspectFields({}, options),
];
return { command, provider, requests };
};
const reasons = [
null,
"opaque caller reason",
new ProviderCleanupError("foreign", ["foreign-resource"], {
reason: "caller-owned cleanup",
}),
];
it.each(["signal", "capture"] as const)(
"preserves opaque reasons at every %s service/provider admission",
async (kind) => {
const f = await fixture(kind);
try {
for (const reason of reasons)
for (const request of f.requests) {
const result = await request({ signal: AbortSignal.abort(reason) });
if (result.ok) throw new Error("Expected cancelled admission");
expect(result.error).toMatchObject({
_tag: "AnalysisCancelledError",
cleanupIncomplete: false,
});
expect(result.error.cause).toBe(reason);
expect(result.error.cleanupResources).toEqual([]);
if (reason instanceof ProviderCleanupError)
expect(reason.cleanupIncomplete).toBe(true);
}
} finally {
await f.provider.close();
}
},
);
it.each(["signal", "capture"] as const)(
"uses artifact cancellation facts for the %s producer",
async (kind) => {
for (const reason of reasons) {
const f = await fixture(kind, async () => {
throw new ArtifactReaderFailure("cancelled", "read interrupted", {
cause: reason,
});
});
try {
const result = await f.requests[0]!({});
if (result.ok) throw new Error("Expected artifact cancellation");
expect(result.error).toMatchObject({ _tag: "AnalysisCancelledError" });
expect(result.error.cause).toBe(reason);
expect(result.error.cleanupResources).toEqual([]);
} finally {
await f.provider.close();
}
}
},
);
it.each(["signal", "capture"] as const)(
"does not infer %s cancellation from an arbitrary error name",
async (kind) => {
const physical = new Error("physical read failed");
physical.name = "AbortError";
const f = await fixture(kind, async () => {
throw physical;
});
try {
const result = await f.requests[0]!({});
if (result.ok) throw new Error("Expected read failure");
expect(result.error).toMatchObject({
_tag: "AnalysisCapabilityUnavailableError",
cause: physical,
});
} finally {
await f.provider.close();
}
},
);
it.each(
(["signal", "capture"] as const).flatMap((kind) =>
[false, true].map((failsClose) => ({ kind, failsClose })),
),
)(
"preserves the selected-tool file outcome through final-close cancellation: %j",
async ({ kind, failsClose }) => {
const f = await fixture(kind);
const identity = await stat(f.command);
const reason = { request: "cancel during selected-tool descriptor close" };
const caller = new AbortController();
const original = OwnedFileHandle.prototype.close;
let owner: OwnedFileHandle | undefined;
let denied = failsClose;
const close = vi
.spyOn(OwnedFileHandle.prototype, "close")
.mockImplementation(async function (this: OwnedFileHandle) {
const observed = await this.handle.stat();
if (observed.dev === identity.dev && observed.ino === identity.ino) {
owner = this;
caller.abort(reason);
if (denied) throw new Error("injected selected-tool close failure");
}
await original.call(this);
});
try {
const result = await f.requests[0]!({ signal: caller.signal });
if (result.ok) throw new Error("Expected final-close cancellation");
const primary = failsClose ? result.error.cleanupPrimary : result.error;
expect(primary).toMatchObject({ _tag: "AnalysisCancelledError" });
expect(primary?.cause).toBe(reason);
expect(result.error.cleanupIncomplete).toBe(failsClose);
if (failsClose) {
expect(owner?.handle.fd).toBeGreaterThanOrEqual(0);
await expect(f.provider.close()).rejects.toMatchObject({
cleanupIncomplete: true,
});
}
denied = false;
await f.provider.close();
expect(owner?.handle.fd).toBe(-1);
expect(result.error.cleanupIncomplete).toBe(false);
expect(primary?.cause).toBe(reason);
} finally {
denied = false;
close.mockRestore();
await f.provider.close();
if (owner !== undefined) await original.call(owner);
}
},
);
@@ -97,6 +97,8 @@ else { if(readFileSync('recording', 'utf8') !== 'recording bytes') throw Error('
return {
path: root.path,
close: async () => {
if (mode === "close-abort")
controller.abort({ request: "cancel during workspace close" });
if (!mayClean) throw new Error("cleanup fixture failure");
await root.close();
},
@@ -236,6 +238,8 @@ it.each([
);
if (result.ok) throw new Error("Expected decoder failure");
expect(result.error._tag).toBe(tag);
if (mode === "wait")
expect(result.error.cause).toBe(f.controller.signal.reason);
expect(result.error.cleanupIncomplete).toBe(false);
expect(
(await readdir(f.directory)).filter((name) =>
@@ -311,7 +315,7 @@ it("retains a completed observation when cancellation arrives at the service bou
throw new Error("Unexpected discovery call");
},
decode: async () => {
f.controller.abort();
f.controller.abort({ request: "cancel at Evidence boundary" });
return execution;
},
close: async () => {},
@@ -319,6 +323,7 @@ it("retains a completed observation when cancellation arrives at the service bou
const result = await service.decode(input, { signal: f.controller.signal });
if (result.ok) throw new Error("Expected late cancellation");
expect(result.error._tag).toBe("AnalysisCancelledError");
expect(result.error.cause).toBe(f.controller.signal.reason);
expect(result.error.partialObservation).toMatchObject({
operation: "decode_signal_recording",
normalized_result: { annotations_text: "10-20 uart-1: 42\n" },
@@ -626,3 +631,38 @@ it("retains completed discovery on cancellation during identity verification", a
},
});
});
it.each(
[false, true].flatMap((failsClose) =>
[false, true].map((discovery) => ({ failsClose, discovery })),
),
)(
"retains completed Evidence and exact cancellation during workspace close: %j",
async ({ failsClose, discovery }) => {
const f = await fixture("close-abort", failsClose);
const result = discovery
? await f.service.inspectDecoders({}, { signal: f.controller.signal })
: await f.service.decode(
{ path: f.path, decoders: "uart" },
{ signal: f.controller.signal },
);
if (result.ok) throw new Error("Expected finalization cancellation");
const primary = failsClose ? result.error.cleanupPrimary : result.error;
expect(primary).toMatchObject({ _tag: "AnalysisCancelledError" });
expect(primary?.cause).toBe(f.controller.signal.reason);
expect(result.error.partialObservation).toMatchObject({
operation: discovery
? "inspect_signal_decoders"
: "decode_signal_recording",
});
expect(result.error.cleanupIncomplete).toBe(failsClose);
if (failsClose)
await expect(f.service.close()).rejects.toMatchObject({
cleanupIncomplete: true,
});
f.allowCleanup();
await f.service.close();
expect(result.error.cleanupIncomplete).toBe(false);
expect(primary?.cause).toBe(f.controller.signal.reason);
},
);