fix(providers): own profile selection through finalization
Join pending profile producers on cancellation and shutdown. Preserve failed results before later signals, retain every automatic-selection cleanup owner, and move profile file ownership into the registry. Related: #4569
This commit is contained in:
1 parent
bf7fad390c
commit
f26973a529
13 files changed
+768
-163
No files matched your search
@@ -132,6 +132,14 @@ providers need not expose identical state machines.
|
||||
while it remains unresolved. Link the originating cleanup error to its exact
|
||||
owner so confirmed release resolves its cleanup marker while preserving the
|
||||
earlier diagnostic. Shared CLI/MCP adapters cannot infer that confirmation.
|
||||
- Provider profile selection owns every asynchronous candidate through its
|
||||
terminal result and finalization. The registry owns profile-file cleanup,
|
||||
joins all automatic-selection children, and closes those file owners before
|
||||
reporting candidate shutdown. Final caller cancellation joins the selection;
|
||||
an adapter that ignores cancellation remains active and owned until it settles.
|
||||
Block new admission during shutdown and while cleanup remains unresolved;
|
||||
reopen after confirmed retry. Preserve failed results before considering a
|
||||
later signal, and retain every child's attributed cleanup in mixed outcomes.
|
||||
- Release child descriptors before removing their containing root. Retain a
|
||||
composite owner when either stage fails so retry preserves that ordering.
|
||||
A Windows command leader's exit does not establish Job Object closure. Keep
|
||||
|
||||
@@ -290,6 +290,12 @@ export interface AnalysisProviderReadinessSnapshot {
|
||||
|
||||
/** Deep provider candidate discoverable before one target-bound route is chosen. */
|
||||
export interface AnalysisProviderCandidate extends AnalysisProvider {
|
||||
/**
|
||||
* Own inspection through final reads and cleanup. Observe the supplied owner
|
||||
* signal cooperatively; resolve only after finalization settles, retaining any
|
||||
* failed cleanup capability. Registry cancellation and shutdown join this
|
||||
* completion rather than abandoning the promise.
|
||||
*/
|
||||
resolveAnalysisProfile(
|
||||
target: BinaryTarget,
|
||||
options?: AnalysisProfileResolutionOptions,
|
||||
|
||||
@@ -7,12 +7,12 @@ import {
|
||||
AnalysisCancelledError,
|
||||
AnalysisUnsupportedTargetError,
|
||||
} from "../../domain/analysisErrorCore.js";
|
||||
import type { AnalysisError } from "../../domain/analysisErrorBase.js";
|
||||
import { AnalysisError } from "../../domain/analysisErrorBase.js";
|
||||
import { jsonObjectSchema, type JsonValue } from "../../domain/jsonValue.js";
|
||||
import { err, ok, type Result } from "../../domain/result.js";
|
||||
import type { AnalysisProviderCandidate } from "../AnalysisProvider.js";
|
||||
import type { AnalysisProviderCandidateStatus } from "./AnalysisProviderRegistry.js";
|
||||
import { ABORTED, waitForAbortable } from "../../process/AbortablePromise.js";
|
||||
import { ProviderAdapterError } from "../../domain/providerAdapterError.js";
|
||||
import type { FileCloseResourceOwner } from "../AnalysisProvider.js";
|
||||
|
||||
/** Profile-resolution truth for one discovered provider candidate. */
|
||||
@@ -34,34 +34,29 @@ export const evaluateAnalysisProviderCandidate = async (
|
||||
Result<AnalysisProviderCandidateEvaluation, AnalysisCancelledError>
|
||||
> => {
|
||||
if (signalIsAborted(signal))
|
||||
return err(new AnalysisCancelledError("open_binary"));
|
||||
return err(
|
||||
new AnalysisCancelledError("open_binary", { cause: signal?.reason }),
|
||||
);
|
||||
let resolved: Awaited<ReturnType<typeof candidate.resolveAnalysisProfile>>;
|
||||
try {
|
||||
const resolution = await waitForAbortable(
|
||||
candidate.resolveAnalysisProfile(
|
||||
target,
|
||||
signal === undefined && fileCloseResources === undefined
|
||||
? undefined
|
||||
: {
|
||||
...(signal === undefined ? {} : { signal }),
|
||||
...(fileCloseResources === undefined
|
||||
? {}
|
||||
: { fileCloseResources }),
|
||||
},
|
||||
),
|
||||
signal,
|
||||
);
|
||||
if (resolution === ABORTED)
|
||||
return err(new AnalysisCancelledError("open_binary"));
|
||||
resolved = resolution;
|
||||
resolved = await candidate.resolveAnalysisProfile(target, {
|
||||
...(signal === undefined ? {} : { signal }),
|
||||
...(fileCloseResources === undefined ? {} : { fileCloseResources }),
|
||||
});
|
||||
} catch (cause: unknown) {
|
||||
void cause;
|
||||
return ok(
|
||||
rejectedProfile(candidate, status, "Provider profile resolution threw"),
|
||||
resolved = err(
|
||||
cause instanceof AnalysisError
|
||||
? cause
|
||||
: new ProviderAdapterError(
|
||||
status.provider.id,
|
||||
"resolve_analysis_profile",
|
||||
{
|
||||
cause,
|
||||
diagnostics: { reason: "Provider profile resolution threw" },
|
||||
},
|
||||
),
|
||||
);
|
||||
}
|
||||
if (signalIsAborted(signal))
|
||||
return err(new AnalysisCancelledError("open_binary"));
|
||||
if (!resolved.ok && resolved.error instanceof AnalysisCancelledError)
|
||||
return err(resolved.error);
|
||||
if (!resolved.ok) {
|
||||
@@ -85,6 +80,10 @@ export const evaluateAnalysisProviderCandidate = async (
|
||||
});
|
||||
return ok({ ...rejection, profileError: error });
|
||||
}
|
||||
if (signalIsAborted(signal))
|
||||
return err(
|
||||
new AnalysisCancelledError("open_binary", { cause: signal?.reason }),
|
||||
);
|
||||
try {
|
||||
if (resolved.value.profile === null)
|
||||
return ok(
|
||||
|
||||
@@ -16,6 +16,7 @@ import { ProviderAdapterError } from "../../domain/providerAdapterError.js";
|
||||
import { ProviderCleanupError } from "../../domain/providerCleanupError.js";
|
||||
import { ProviderSelectionError } from "../../domain/providerSelectionError.js";
|
||||
import { projectAnalysisError } from "../../domain/analysisErrorProjection.js";
|
||||
import { AnalysisCancelledError } from "../../domain/analysisErrorCore.js";
|
||||
import { err, ok } from "../../domain/result.js";
|
||||
|
||||
const DATABASE_TARGET: BinaryTarget = {
|
||||
@@ -395,29 +396,63 @@ describe("analysis provider registry: profile and cleanup failures", () => {
|
||||
expect(alternate.created).toEqual([]);
|
||||
});
|
||||
|
||||
it("cancels a pending profile probe even when the adapter ignores its signal", async () => {
|
||||
it("owns a signal-ignoring profile probe until final completion and joins shutdown", async () => {
|
||||
const alpha = candidate("alpha");
|
||||
let observedSignal: AbortSignal | undefined;
|
||||
let finish: (() => void) | undefined;
|
||||
let started: (() => void) | undefined;
|
||||
const pending = new Promise<void>((resolve) => {
|
||||
started = resolve;
|
||||
});
|
||||
const gate = new Promise<void>((resolve) => {
|
||||
finish = resolve;
|
||||
});
|
||||
const ignoresCancellation: AnalysisProviderCandidate = {
|
||||
...alpha.provider,
|
||||
resolveAnalysisProfile: (_target, options) => {
|
||||
resolveAnalysisProfile: async (target, options) => {
|
||||
observedSignal = options?.signal;
|
||||
return new Promise<never>(() => undefined);
|
||||
started?.();
|
||||
await gate;
|
||||
return alpha.provider.resolveAnalysisProfile(target);
|
||||
},
|
||||
};
|
||||
const registry = new AnalysisProviderRegistry([ignoresCancellation]);
|
||||
const controller = new AbortController();
|
||||
const selecting = new AnalysisProviderRegistry([
|
||||
ignoresCancellation,
|
||||
]).select(DATABASE_TARGET, "alpha", { signal: controller.signal });
|
||||
expect(observedSignal).toBe(controller.signal);
|
||||
|
||||
controller.abort();
|
||||
|
||||
await expect(selecting).resolves.toMatchObject({
|
||||
ok: false,
|
||||
error: { _tag: "AnalysisCancelledError", operation: "open_binary" },
|
||||
const reason = { caller: "profile finalization" };
|
||||
let selected = false;
|
||||
const selecting = registry
|
||||
.select(DATABASE_TARGET, "alpha", { signal: controller.signal })
|
||||
.then((result) => {
|
||||
selected = true;
|
||||
return result;
|
||||
});
|
||||
await pending;
|
||||
controller.abort(reason);
|
||||
let closed = false;
|
||||
const closing = registry.closeOwnedResources().then((result) => {
|
||||
closed = true;
|
||||
return result;
|
||||
});
|
||||
expect(alpha.created).toEqual([]);
|
||||
try {
|
||||
await new Promise<void>((resolve) => setImmediate(resolve));
|
||||
expect(selected).toBe(false);
|
||||
expect(closed).toBe(false);
|
||||
expect(registry.hasPendingOwnedResources()).toBe(true);
|
||||
expect(observedSignal?.aborted).toBe(true);
|
||||
expect(observedSignal?.reason).toBe(reason);
|
||||
finish?.();
|
||||
const result = await selecting;
|
||||
expect(result).toMatchObject({
|
||||
ok: false,
|
||||
error: { _tag: "AnalysisCancelledError", cause: reason },
|
||||
});
|
||||
expect(await closing).toEqual(ok(null));
|
||||
expect(registry.hasPendingOwnedResources()).toBe(false);
|
||||
expect(alpha.created).toEqual([]);
|
||||
} finally {
|
||||
finish?.();
|
||||
await Promise.all([selecting, closing]);
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
@@ -524,3 +559,207 @@ const capability = (provider: ProviderIdentity): CapabilityDescriptor => ({
|
||||
},
|
||||
limitations: [],
|
||||
});
|
||||
|
||||
describe("profile resolution terminal facts", () => {
|
||||
it.each([false, true])(
|
||||
"preserves a physical profile failure across late caller abort (throws: %s)",
|
||||
async (throws) => {
|
||||
const alpha = candidate("alpha");
|
||||
const controller = new AbortController();
|
||||
const primary = new ProviderAdapterError(
|
||||
"alpha",
|
||||
"resolve_analysis_profile",
|
||||
{ cause: new Error("physical profile failure") },
|
||||
);
|
||||
const registry = new AnalysisProviderRegistry([
|
||||
{
|
||||
...alpha.provider,
|
||||
resolveAnalysisProfile: async () => {
|
||||
controller.abort({ caller: "after physical failure" });
|
||||
if (throws) throw primary;
|
||||
return err(primary);
|
||||
},
|
||||
},
|
||||
]);
|
||||
const result = await registry.select(DATABASE_TARGET, "alpha", {
|
||||
signal: controller.signal,
|
||||
});
|
||||
expect(result).toEqual(err(primary));
|
||||
expect(registry.hasPendingOwnedResources()).toBe(false);
|
||||
expect(await registry.closeOwnedResources()).toEqual(ok(null));
|
||||
},
|
||||
);
|
||||
|
||||
it("preserves exact cancellation with cleanup and blocks admission until its owner confirms release", async () => {
|
||||
const alpha = candidate("alpha");
|
||||
const controller = new AbortController();
|
||||
const reason = { caller: "final cleanup" };
|
||||
const primary = new AnalysisCancelledError("resolve_analysis_profile", {
|
||||
cause: reason,
|
||||
cleanup: { reason: "file close rejected", resources: ["profile-file"] },
|
||||
});
|
||||
let attempts = 0;
|
||||
let allowClose = false;
|
||||
const registry = new AnalysisProviderRegistry([
|
||||
{
|
||||
...alpha.provider,
|
||||
resolveAnalysisProfile: async () => {
|
||||
attempts++;
|
||||
controller.abort(reason);
|
||||
return err(primary);
|
||||
},
|
||||
closeOwnedResources: async () => {
|
||||
if (!allowClose) return err(primary);
|
||||
primary.resolveCleanup();
|
||||
return ok(null);
|
||||
},
|
||||
},
|
||||
]);
|
||||
const result = await registry.select(DATABASE_TARGET, "alpha", {
|
||||
signal: controller.signal,
|
||||
});
|
||||
expect(result).toEqual(err(primary));
|
||||
expect(primary.cause).toBe(reason);
|
||||
expect(registry.hasPendingOwnedResources()).toBe(true);
|
||||
expect(await registry.select(DATABASE_TARGET, "alpha")).toEqual(
|
||||
err(primary),
|
||||
);
|
||||
expect(attempts).toBe(1);
|
||||
allowClose = true;
|
||||
expect(await registry.closeOwnedResources()).toEqual(ok(null));
|
||||
expect(primary.cleanupIncomplete).toBe(false);
|
||||
expect(registry.hasPendingOwnedResources()).toBe(false);
|
||||
});
|
||||
});
|
||||
|
||||
describe("automatic profile cleanup ownership", () => {
|
||||
it.each(["alpha", "beta"])(
|
||||
"keeps automatic mixed-child cleanup conjunctive when %s releases first",
|
||||
async (first) => {
|
||||
const alpha = candidate("alpha");
|
||||
const beta = candidate("beta");
|
||||
const controller = new AbortController();
|
||||
const reason = { caller: "mixed profile finalization" };
|
||||
const failures = {
|
||||
alpha: new AnalysisCancelledError("resolve_analysis_profile", {
|
||||
cause: reason,
|
||||
cleanup: { reason: "alpha descriptor", resources: ["alpha-file"] },
|
||||
}),
|
||||
beta: new ProviderCleanupError("beta", ["beta-file"], {
|
||||
reason: "beta descriptor",
|
||||
}),
|
||||
};
|
||||
let entered = 0;
|
||||
let admitChildren: (() => void) | undefined;
|
||||
const admitted = new Promise<void>((resolve) => {
|
||||
admitChildren = resolve;
|
||||
});
|
||||
const released = new Set<string>();
|
||||
const registry = new AnalysisProviderRegistry(
|
||||
[alpha, beta].map((fixture) => {
|
||||
const id = fixture.provider.identity().id as "alpha" | "beta";
|
||||
return {
|
||||
...fixture.provider,
|
||||
resolveAnalysisProfile: async () => {
|
||||
entered++;
|
||||
if (entered === 2) {
|
||||
controller.abort(reason);
|
||||
admitChildren?.();
|
||||
}
|
||||
await admitted;
|
||||
return err(failures[id]);
|
||||
},
|
||||
closeOwnedResources: async () => {
|
||||
if (!released.has(id)) return err(failures[id]);
|
||||
failures[id].resolveCleanup();
|
||||
return ok(null);
|
||||
},
|
||||
};
|
||||
}),
|
||||
);
|
||||
const result = await registry.select(DATABASE_TARGET, "auto", {
|
||||
signal: controller.signal,
|
||||
});
|
||||
if (result.ok) throw new Error("Mixed failed profiles admitted");
|
||||
expect(result.error.cleanupIncomplete).toBe(true);
|
||||
expect(result.error.cleanupResources).toEqual([
|
||||
"alpha-file",
|
||||
"beta-file",
|
||||
]);
|
||||
expect(result.error.diagnostics?.profile_errors).toEqual(
|
||||
expect.arrayContaining([
|
||||
expect.objectContaining({ provider_id: "alpha" }),
|
||||
expect.objectContaining({ provider_id: "beta" }),
|
||||
]),
|
||||
);
|
||||
const close1 = await registry.closeOwnedResources();
|
||||
expect(close1.ok).toBe(false);
|
||||
released.add(first);
|
||||
const close2 = await registry.closeOwnedResources();
|
||||
expect(close2.ok).toBe(false);
|
||||
expect(result.error.cleanupIncomplete).toBe(true);
|
||||
expect(result.error.cleanupResources).toEqual([
|
||||
first === "alpha" ? "beta-file" : "alpha-file",
|
||||
]);
|
||||
expect(close1.ok ? [] : close1.error.cleanupResources).toEqual([
|
||||
first === "alpha" ? "beta-file" : "alpha-file",
|
||||
]);
|
||||
released.add(first === "alpha" ? "beta" : "alpha");
|
||||
expect(await registry.closeOwnedResources()).toEqual(ok(null));
|
||||
expect(result.error.cleanupIncomplete).toBe(false);
|
||||
if (close1.ok || close2.ok)
|
||||
throw new Error("Cleanup unexpectedly succeeded");
|
||||
expect(close1.error.cleanupIncomplete).toBe(false);
|
||||
expect(close2.error.cleanupIncomplete).toBe(false);
|
||||
expect(registry.hasPendingOwnedResources()).toBe(false);
|
||||
},
|
||||
);
|
||||
});
|
||||
|
||||
it("registers before production and gates fresh selections until shutdown has joined finalization", async () => {
|
||||
const alpha = candidate("alpha");
|
||||
let calls = 0;
|
||||
let enter: (() => void) | undefined;
|
||||
let finish: (() => void) | undefined;
|
||||
const entered = new Promise<void>((resolve) => {
|
||||
enter = resolve;
|
||||
});
|
||||
const gate = new Promise<void>((resolve) => {
|
||||
finish = resolve;
|
||||
});
|
||||
const registry = new AnalysisProviderRegistry([
|
||||
{
|
||||
...alpha.provider,
|
||||
resolveAnalysisProfile: async (target) => {
|
||||
calls++;
|
||||
if (calls === 1) {
|
||||
enter?.();
|
||||
await gate;
|
||||
}
|
||||
return alpha.provider.resolveAnalysisProfile(target);
|
||||
},
|
||||
},
|
||||
]);
|
||||
const selecting = registry.select(DATABASE_TARGET, "alpha");
|
||||
await entered;
|
||||
const closing = registry.closeOwnedResources();
|
||||
const next = registry.select(DATABASE_TARGET, "alpha");
|
||||
try {
|
||||
await new Promise<void>((resolve) => setImmediate(resolve));
|
||||
expect(calls).toBe(1);
|
||||
expect(registry.hasPendingOwnedResources()).toBe(true);
|
||||
finish?.();
|
||||
expect(await selecting).toMatchObject({
|
||||
ok: false,
|
||||
error: { _tag: "AnalysisCancelledError" },
|
||||
});
|
||||
expect(await closing).toEqual(ok(null));
|
||||
expect((await next).ok).toBe(true);
|
||||
expect(calls).toBe(2);
|
||||
expect(registry.hasPendingOwnedResources()).toBe(false);
|
||||
} finally {
|
||||
finish?.();
|
||||
await Promise.all([selecting, closing, next]);
|
||||
await registry.closeOwnedResources();
|
||||
}
|
||||
});
|
||||
@@ -1,3 +1,10 @@
|
||||
import { AnalysisCleanupLease } from "../../artifacts/AnalysisCleanupLease.js";
|
||||
import { ArtifactResourceScope } from "../../artifacts/ArtifactResourceScope.js";
|
||||
import { ArtifactReaderFailure } from "../../artifacts/ArtifactReader.js";
|
||||
import { bindArtifactFileCleanup } from "../../artifacts/ArtifactFileOperation.js";
|
||||
import { SharedStartupAttempt } from "../../process/SharedStartupAttempt.js";
|
||||
import { ProviderAdapterError } from "../../domain/providerAdapterError.js";
|
||||
import { projectAnalysisError } from "../../domain/analysisErrorProjection.js";
|
||||
import { ProviderCleanupError } from "../../domain/providerCleanupError.js";
|
||||
import {
|
||||
analysisProviderIdSchema,
|
||||
@@ -67,9 +74,10 @@ interface TargetSelectionContext {
|
||||
readonly target: BinaryTarget;
|
||||
readonly baseline: readonly AnalysisProviderCandidateStatus[];
|
||||
readonly signal: AbortSignal | undefined;
|
||||
readonly fileCloseResources?: FileCloseResourceOwner;
|
||||
}
|
||||
|
||||
type SelectionResult = Result<AnalysisProviderSelection, AnalysisError>;
|
||||
|
||||
/**
|
||||
* Discover and select overlapping deep providers without starting their clients.
|
||||
* Registration order never affects selection or caller-visible output.
|
||||
@@ -80,6 +88,20 @@ export class AnalysisProviderRegistry {
|
||||
readonly #declaredOperations: readonly string[];
|
||||
readonly #defaultSelector: AnalysisProviderSelector;
|
||||
|
||||
readonly #selections = new Set<SharedStartupAttempt<SelectionResult>>();
|
||||
readonly #selectionCleanup = new Set<AnalysisError>();
|
||||
readonly #profileFiles = new ArtifactResourceScope();
|
||||
readonly #profileFileOwner: FileCloseResourceOwner = {
|
||||
retain: (owner, resource) =>
|
||||
this.#profileFiles.retain({
|
||||
kind: "lease",
|
||||
resource,
|
||||
lease: new AnalysisCleanupLease(() => owner.close(), [resource]),
|
||||
}),
|
||||
run: (operation) => this.#profileFiles.run(operation),
|
||||
};
|
||||
#closing: Promise<Result<null, AnalysisError>> | undefined;
|
||||
|
||||
constructor(
|
||||
providers: readonly AnalysisProviderCandidate[],
|
||||
defaultSelector: AnalysisProviderSelector = "auto",
|
||||
@@ -143,29 +165,67 @@ export class AnalysisProviderRegistry {
|
||||
|
||||
/** Report retained candidate resources without retrying cleanup. */
|
||||
hasPendingOwnedResources(): boolean {
|
||||
return this.#providers.some(
|
||||
(provider) => provider.hasPendingOwnedResources?.() === true,
|
||||
return (
|
||||
this.#selections.size > 0 ||
|
||||
this.#profileFiles.hasPending ||
|
||||
[...this.#selectionCleanup].some((error) => error.cleanupIncomplete) ||
|
||||
this.#providers.some(
|
||||
(provider) => provider.hasPendingOwnedResources?.() === true,
|
||||
)
|
||||
);
|
||||
}
|
||||
|
||||
/** Close candidate-owned resources even when no target client was opened. */
|
||||
async closeOwnedResources(): Promise<Result<null, AnalysisError>> {
|
||||
closeOwnedResources(): Promise<Result<null, AnalysisError>> {
|
||||
this.#closing ??= Promise.resolve()
|
||||
.then(() => this.#closeOwnedResources())
|
||||
.finally(() => {
|
||||
this.#closing = undefined;
|
||||
});
|
||||
return this.#closing;
|
||||
}
|
||||
|
||||
async #closeOwnedResources(): Promise<Result<null, AnalysisError>> {
|
||||
const selections = [...this.#selections];
|
||||
for (const selection of selections) selection.close();
|
||||
await Promise.allSettled(selections.map((selection) => selection.promise));
|
||||
const failures: AnalysisError[] = [];
|
||||
try {
|
||||
await this.#profileFiles.retryCleanup();
|
||||
} catch (cause: unknown) {
|
||||
const cleanup = ArtifactReaderFailure.cleanupObservation(
|
||||
cause,
|
||||
"provider profile files",
|
||||
);
|
||||
const failure = new ProviderCleanupError(
|
||||
"rea",
|
||||
cleanup.resources,
|
||||
{ reason: cleanup.reason },
|
||||
{ cause },
|
||||
);
|
||||
bindArtifactFileCleanup(cause, failure);
|
||||
failures.push(failure);
|
||||
}
|
||||
for (const provider of this.#providers) {
|
||||
if (provider.closeOwnedResources === undefined) continue;
|
||||
const result = await provider.closeOwnedResources();
|
||||
if (!result.ok) failures.push(result.error);
|
||||
}
|
||||
if (failures.length === 0) return ok(null);
|
||||
if (failures.length === 1) return err(failures[0]!);
|
||||
for (const failure of this.#selectionCleanup) {
|
||||
if (failure.cleanupIncomplete) failures.push(failure);
|
||||
else this.#selectionCleanup.delete(failure);
|
||||
}
|
||||
const distinct = [...new Set(failures)];
|
||||
if (distinct.length === 0) return ok(null);
|
||||
if (distinct.length === 1) return err(distinct[0]!);
|
||||
return err(
|
||||
new ProviderCleanupError(
|
||||
"rea",
|
||||
failures.flatMap((failure) => failure.cleanupResources),
|
||||
{
|
||||
reason: failures.map((failure) => failure.message).join("; "),
|
||||
},
|
||||
{ cause: new AggregateError(failures) },
|
||||
[...new Set(distinct.flatMap((failure) => failure.cleanupResources))],
|
||||
{ reason: distinct.map((failure) => failure.message).join("; ") },
|
||||
{ cause: new AggregateError(distinct) },
|
||||
).bindCleanupDependencies(
|
||||
distinct.filter((failure) => failure.cleanupIncomplete),
|
||||
),
|
||||
);
|
||||
}
|
||||
@@ -186,13 +246,84 @@ export class AnalysisProviderRegistry {
|
||||
async select(
|
||||
target: BinaryTarget,
|
||||
requestedSelector?: AnalysisProviderSelector,
|
||||
options: {
|
||||
readonly signal?: AbortSignal;
|
||||
readonly fileCloseResources?: FileCloseResourceOwner;
|
||||
} = {},
|
||||
): Promise<Result<AnalysisProviderSelection, AnalysisError>> {
|
||||
options: { readonly signal?: AbortSignal } = {},
|
||||
): Promise<SelectionResult> {
|
||||
if (signalIsAborted(options.signal))
|
||||
return err(new AnalysisCancelledError("open_binary"));
|
||||
return err(
|
||||
new AnalysisCancelledError("open_binary", {
|
||||
cause: options.signal?.reason,
|
||||
}),
|
||||
);
|
||||
if (this.#closing !== undefined) {
|
||||
const closed = await this.#closing;
|
||||
if (!closed.ok) return closed;
|
||||
}
|
||||
for (const failure of this.#selectionCleanup)
|
||||
if (!failure.cleanupIncomplete) this.#selectionCleanup.delete(failure);
|
||||
if (
|
||||
[...this.#selectionCleanup].some((failure) => failure.cleanupIncomplete)
|
||||
) {
|
||||
// A retry may release retained owners; never stop another live selection
|
||||
// merely because a new caller wants admission.
|
||||
if (this.#selections.size > 0)
|
||||
return err(
|
||||
[...this.#selectionCleanup].find(
|
||||
(failure) => failure.cleanupIncomplete,
|
||||
)!,
|
||||
);
|
||||
const closed = await this.closeOwnedResources();
|
||||
if (!closed.ok) return closed;
|
||||
}
|
||||
if (signalIsAborted(options.signal))
|
||||
return err(
|
||||
new AnalysisCancelledError("open_binary", {
|
||||
cause: options.signal?.reason,
|
||||
}),
|
||||
);
|
||||
const selection = new SharedStartupAttempt<SelectionResult>(
|
||||
(signal) =>
|
||||
Promise.resolve().then(() =>
|
||||
this.#selectTarget(target, requestedSelector, signal),
|
||||
),
|
||||
options.signal,
|
||||
);
|
||||
this.#selections.add(selection);
|
||||
void selection.promise.then(
|
||||
(result) => {
|
||||
this.#selections.delete(selection);
|
||||
if (!result.ok && result.error.cleanupIncomplete)
|
||||
this.#selectionCleanup.add(result.error);
|
||||
},
|
||||
() => {
|
||||
this.#selections.delete(selection);
|
||||
},
|
||||
);
|
||||
const waited = await selection.wait(options.signal);
|
||||
if (waited.status !== "cancelled") {
|
||||
if (!waited.value.ok) return waited.value;
|
||||
if (!selection.signal.aborted) return waited.value;
|
||||
return err(
|
||||
new AnalysisCancelledError("open_binary", {
|
||||
cause: selection.signal.reason,
|
||||
}),
|
||||
);
|
||||
}
|
||||
return err(
|
||||
new AnalysisCancelledError("open_binary", {
|
||||
cause: options.signal?.reason,
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
async #selectTarget(
|
||||
target: BinaryTarget,
|
||||
requestedSelector: AnalysisProviderSelector | undefined,
|
||||
signal: AbortSignal,
|
||||
): Promise<SelectionResult> {
|
||||
if (signalIsAborted(signal))
|
||||
return err(
|
||||
new AnalysisCancelledError("open_binary", { cause: signal.reason }),
|
||||
);
|
||||
const parsedRequest =
|
||||
requestedSelector === undefined
|
||||
? undefined
|
||||
@@ -210,8 +341,10 @@ export class AnalysisProviderRegistry {
|
||||
const request = parsedRequest?.data;
|
||||
const selector = request ?? this.#defaultSelector;
|
||||
const baseline = this.candidates(target);
|
||||
if (signalIsAborted(options.signal))
|
||||
return err(new AnalysisCancelledError("open_binary"));
|
||||
if (signalIsAborted(signal))
|
||||
return err(
|
||||
new AnalysisCancelledError("open_binary", { cause: signal.reason }),
|
||||
);
|
||||
if (selector === "auto" && this.canInspectWithoutDeepProvider?.(target))
|
||||
return ok({
|
||||
requestedProviderId: selector,
|
||||
@@ -236,10 +369,7 @@ export class AnalysisProviderRegistry {
|
||||
{
|
||||
target,
|
||||
baseline,
|
||||
signal: options.signal,
|
||||
...(options.fileCloseResources === undefined
|
||||
? {}
|
||||
: { fileCloseResources: options.fileCloseResources }),
|
||||
signal: signal,
|
||||
},
|
||||
selector,
|
||||
request === undefined ? "environment" : "request",
|
||||
@@ -247,10 +377,7 @@ export class AnalysisProviderRegistry {
|
||||
return this.#selectAutomatically({
|
||||
target,
|
||||
baseline,
|
||||
signal: options.signal,
|
||||
...(options.fileCloseResources === undefined
|
||||
? {}
|
||||
: { fileCloseResources: options.fileCloseResources }),
|
||||
signal: signal,
|
||||
});
|
||||
}
|
||||
|
||||
@@ -299,7 +426,7 @@ export class AnalysisProviderRegistry {
|
||||
initial,
|
||||
target,
|
||||
signal,
|
||||
context.fileCloseResources,
|
||||
this.#profileFileOwner,
|
||||
);
|
||||
if (!evaluation.ok) return evaluation;
|
||||
const evaluated = evaluation.value;
|
||||
@@ -346,7 +473,7 @@ export class AnalysisProviderRegistry {
|
||||
status,
|
||||
target,
|
||||
signal,
|
||||
context.fileCloseResources,
|
||||
this.#profileFileOwner,
|
||||
)
|
||||
: ok({
|
||||
candidate: provider,
|
||||
@@ -355,17 +482,21 @@ export class AnalysisProviderRegistry {
|
||||
},
|
||||
),
|
||||
);
|
||||
const cancelled = results.find((result) => !result.ok);
|
||||
if (cancelled !== undefined && !cancelled.ok) return cancelled;
|
||||
const evaluations = results.map((result) => {
|
||||
if (!result.ok)
|
||||
throw new TypeError("Cancelled candidate evaluation disappeared");
|
||||
return result.value;
|
||||
const evaluations = results.flatMap((result) =>
|
||||
result.ok ? [result.value] : [],
|
||||
);
|
||||
const failures = results.flatMap((result, index) => {
|
||||
const error = result.ok ? result.value.profileError : result.error;
|
||||
return error === undefined
|
||||
? []
|
||||
: [{ providerId: this.#providers[index]!.identity().id, error }];
|
||||
});
|
||||
const cleanupFailure = evaluations.find(
|
||||
({ profileError }) => profileError?.cleanupIncomplete === true,
|
||||
)?.profileError;
|
||||
if (cleanupFailure !== undefined) return err(cleanupFailure);
|
||||
if (
|
||||
results.some((result) => !result.ok) ||
|
||||
failures.some(({ error }) => error.cleanupIncomplete) ||
|
||||
signalIsAborted(signal)
|
||||
)
|
||||
return err(profileSelectionFailure(failures, signal));
|
||||
const statuses = evaluations.map(({ status }) => status);
|
||||
const usable = evaluations.filter(
|
||||
(
|
||||
@@ -507,3 +638,64 @@ const cloneTargetSupport = (
|
||||
...support,
|
||||
diagnostics: jsonObjectSchema.parse(support.diagnostics),
|
||||
});
|
||||
|
||||
/** Every terminal child remains attributed; cleanup is conjunctive across owners. */
|
||||
const profileSelectionFailure = (
|
||||
failures: readonly {
|
||||
readonly providerId: string;
|
||||
readonly error: AnalysisError;
|
||||
}[],
|
||||
signal?: AbortSignal,
|
||||
): AnalysisError => {
|
||||
const primary =
|
||||
failures.find(({ error }) => !(error instanceof AnalysisCancelledError)) ??
|
||||
failures[0];
|
||||
if (primary === undefined)
|
||||
return new AnalysisCancelledError("open_binary", { cause: signal?.reason });
|
||||
if (failures.length === 1) return primary.error;
|
||||
const cleanupFailures = failures
|
||||
.filter(({ error }) => error.cleanupIncomplete)
|
||||
.map(({ error }) => error);
|
||||
const options = {
|
||||
diagnostics: {
|
||||
profile_errors: failures.map(({ providerId, error }) => ({
|
||||
provider_id: providerId,
|
||||
error: projectAnalysisError(error),
|
||||
})),
|
||||
},
|
||||
...(primary.error.capturedOutput === undefined
|
||||
? {}
|
||||
: { capturedOutput: primary.error.capturedOutput }),
|
||||
...(primary.error.partialObservation === undefined
|
||||
? {}
|
||||
: { partialObservation: primary.error.partialObservation }),
|
||||
...(cleanupFailures.length === 0
|
||||
? {}
|
||||
: {
|
||||
cleanup: {
|
||||
reason: cleanupFailures.map((error) => error.message).join("; "),
|
||||
resources: [
|
||||
...new Set(
|
||||
cleanupFailures.flatMap((error) => error.cleanupResources),
|
||||
),
|
||||
],
|
||||
},
|
||||
}),
|
||||
};
|
||||
const failure =
|
||||
primary.error instanceof AnalysisCancelledError
|
||||
? new AnalysisCancelledError("open_binary", {
|
||||
...options,
|
||||
cause: primary.error.cause,
|
||||
})
|
||||
: new ProviderAdapterError("rea", "resolve_analysis_profile", {
|
||||
...options,
|
||||
userMessage: projectAnalysisError(primary.error).message,
|
||||
cause: new AggregateError(
|
||||
failures.map(({ error }) => error),
|
||||
"Provider profile resolutions failed",
|
||||
{ cause: primary.error },
|
||||
),
|
||||
});
|
||||
return failure.bindCleanupDependencies(cleanupFailures);
|
||||
};
|
||||
@@ -200,12 +200,15 @@ export class BinarySession
|
||||
current: this.#active,
|
||||
target,
|
||||
options,
|
||||
fileCloseResources: this.fileCloseResourceOwner,
|
||||
stagedSnapshotMatches: (target, profile) =>
|
||||
this.matchesSnapshot(target, profile),
|
||||
}).then((resolved) =>
|
||||
isAborted(options.signal)
|
||||
? err(new AnalysisCancelledError("open_binary"))
|
||||
resolved.ok && isAborted(options.signal)
|
||||
? err(
|
||||
new AnalysisCancelledError("open_binary", {
|
||||
cause: options.signal?.reason,
|
||||
}),
|
||||
)
|
||||
: resolved,
|
||||
);
|
||||
}
|
||||
@@ -225,7 +228,6 @@ export class BinarySession
|
||||
current: this.#active,
|
||||
path,
|
||||
options: admitted,
|
||||
fileCloseResources: this.fileCloseResourceOwner,
|
||||
stagedSnapshotMatches: (target, profile) =>
|
||||
this.matchesSnapshot(target, profile),
|
||||
}),
|
||||
@@ -503,11 +505,9 @@ export class BinarySession
|
||||
if (clearRecords) this.clearSessionRecords();
|
||||
this.#clearRuntimeAvailability();
|
||||
const fileResources = await this.closeWriteResources();
|
||||
const hashResources = await this.closeFailedFileCloses();
|
||||
const candidateResources = await this.#providerRouter.closeOwnedResources();
|
||||
const resourceFailures = [
|
||||
...(fileResources.ok ? [] : [fileResources.error]),
|
||||
...(hashResources.ok ? [] : [hashResources.error]),
|
||||
...(candidateResources.ok ? [] : [candidateResources.error]),
|
||||
];
|
||||
let resourceFailure = resourceFailures[0];
|
||||
|
||||
@@ -15,7 +15,6 @@ import {
|
||||
AnalysisSnapshotMismatchError,
|
||||
} from "../../domain/evidenceErrors.js";
|
||||
import type { AnalysisError } from "../../domain/analysisErrorBase.js";
|
||||
import type { FileCloseResourceOwner } from "../AnalysisProvider.js";
|
||||
import { err, type Result } from "../../domain/result.js";
|
||||
import type { SessionProviderRoute } from "./SessionProviderRouter.js";
|
||||
import { SessionProviderRouter } from "./SessionProviderRouter.js";
|
||||
@@ -61,7 +60,6 @@ interface ResolveSessionOpenInput {
|
||||
target: BinaryTarget,
|
||||
profile: AnalysisProfileCommitment | null,
|
||||
) => boolean;
|
||||
readonly fileCloseResources?: FileCloseResourceOwner;
|
||||
}
|
||||
|
||||
interface ResolveSessionTargetInput {
|
||||
@@ -73,7 +71,6 @@ interface ResolveSessionTargetInput {
|
||||
target: BinaryTarget,
|
||||
profile: AnalysisProfileCommitment | null,
|
||||
) => boolean;
|
||||
readonly fileCloseResources?: FileCloseResourceOwner;
|
||||
}
|
||||
|
||||
/** Parse a target, resolve its provider route, and validate snapshot binding. */
|
||||
@@ -109,12 +106,7 @@ export const resolveSessionTarget = async (
|
||||
const resolvedRoute =
|
||||
sameTarget && options.providerId === undefined && current !== undefined
|
||||
? ({ ok: true, value: current.route } as const)
|
||||
: await router.resolve(
|
||||
target,
|
||||
options.providerId,
|
||||
options.signal,
|
||||
input.fileCloseResources,
|
||||
);
|
||||
: await router.resolve(target, options.providerId, options.signal);
|
||||
if (!resolvedRoute.ok) return resolvedRoute;
|
||||
return validateResolvedSessionTarget({
|
||||
...input,
|
||||
|
||||
@@ -32,8 +32,6 @@ import { AnalysisSnapshotCache } from "./AnalysisSnapshotCache.js";
|
||||
import { InvestigationRecords } from "../investigation/InvestigationRecords.js";
|
||||
import { ArtifactResourceScope } from "../../artifacts/ArtifactResourceScope.js";
|
||||
import { ArtifactReaderFailure } from "../../artifacts/ArtifactReader.js";
|
||||
import { ProviderCleanupError } from "../../domain/providerCleanupError.js";
|
||||
import type { FileCloseResourceOwner } from "../AnalysisProvider.js";
|
||||
import { writeAnalysisSnapshot as writeSnapshotFile } from "./AnalysisSnapshotFiles.js";
|
||||
|
||||
const SNAPSHOT_MUTATION_RECOVERY =
|
||||
@@ -57,21 +55,9 @@ export interface WorkflowSnapshotRecordInput {
|
||||
export abstract class BinarySessionRecords {
|
||||
readonly #records: InvestigationRecords;
|
||||
readonly #writeResources = new ArtifactResourceScope();
|
||||
readonly #fileCloseResources = new ArtifactResourceScope();
|
||||
protected readonly fileCloseResourceOwner: FileCloseResourceOwner = {
|
||||
retain: (owner, resource) =>
|
||||
this.#fileCloseResources.retain({
|
||||
kind: "lease",
|
||||
resource,
|
||||
lease: { release: () => owner.close() },
|
||||
}),
|
||||
run: (operation) => this.#fileCloseResources.run(operation),
|
||||
};
|
||||
|
||||
protected get hasPendingFileResources(): boolean {
|
||||
return (
|
||||
this.#writeResources.hasPending || this.#fileCloseResources.hasPending
|
||||
);
|
||||
return this.#writeResources.hasPending;
|
||||
}
|
||||
readonly #snapshot = new AnalysisSnapshotCache();
|
||||
#snapshotInvalidated = false;
|
||||
@@ -150,27 +136,6 @@ export abstract class BinarySessionRecords {
|
||||
}
|
||||
}
|
||||
|
||||
/** Retry retained descriptors owned by this composed binary session. */
|
||||
async closeFailedFileCloses(): Promise<Result<null, AnalysisError>> {
|
||||
try {
|
||||
await this.#fileCloseResources.retryCleanup();
|
||||
return ok(null);
|
||||
} catch (cause: unknown) {
|
||||
const observation = ArtifactReaderFailure.cleanupObservation(
|
||||
cause,
|
||||
"provider file descriptor",
|
||||
);
|
||||
return err(
|
||||
new ProviderCleanupError(
|
||||
"rea",
|
||||
observation.resources,
|
||||
{ reason: observation.reason },
|
||||
{ cause, operation: "close_binary" },
|
||||
),
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
importEvidenceBundle(
|
||||
bundle: unknown,
|
||||
): Result<number, EvidenceIntegrityError | AnalysisInputError> {
|
||||
|
||||
@@ -12,7 +12,6 @@ import type {
|
||||
CapabilityDescriptor,
|
||||
ProviderIdentity,
|
||||
} from "../AnalysisProvider.js";
|
||||
import type { FileCloseResourceOwner } from "../AnalysisProvider.js";
|
||||
import {
|
||||
AnalysisProviderRegistry,
|
||||
type AnalysisProviderBinding,
|
||||
@@ -86,15 +85,8 @@ export class SessionProviderRouter {
|
||||
target: BinaryTarget,
|
||||
providerId?: AnalysisProviderSelector,
|
||||
signal?: AbortSignal,
|
||||
fileCloseResources?: FileCloseResourceOwner,
|
||||
): Promise<Result<SessionProviderRoute, AnalysisError>> {
|
||||
const resolutionOptions =
|
||||
signal === undefined && fileCloseResources === undefined
|
||||
? undefined
|
||||
: {
|
||||
...(signal === undefined ? {} : { signal }),
|
||||
...(fileCloseResources === undefined ? {} : { fileCloseResources }),
|
||||
};
|
||||
const resolutionOptions = signal === undefined ? undefined : { signal };
|
||||
const selected = await this.registry.select(
|
||||
target,
|
||||
providerId,
|
||||
|
||||
@@ -1,3 +1,5 @@
|
||||
import { AnalysisProviderRegistry } from "../../../src/application/binary/AnalysisProviderRegistry.js";
|
||||
import { composeBinarySession } from "../../../src/application/binary/BinarySessionComposition.js";
|
||||
import { createHash } from "node:crypto";
|
||||
import type { Dir } from "node:fs";
|
||||
import { GhidraClient } from "../../../src/ghidra/GhidraClient.js";
|
||||
@@ -970,3 +972,79 @@ describe.skipIf(process.platform === "win32")(
|
||||
});
|
||||
},
|
||||
);
|
||||
|
||||
describe.skipIf(process.platform === "win32")(
|
||||
"PSP profile selection completion through composed sessions",
|
||||
() => {
|
||||
it.each(["preview", "open"] as const)(
|
||||
"preserves physical failure and live cleanup after late cancellation during %s",
|
||||
async (operation) => {
|
||||
const { directory, root, provider } = await installation();
|
||||
const path = join(directory, "controlled-psp.elf");
|
||||
await writeFile(path, pspElfBytes());
|
||||
const parsed = await parseBinaryTarget(path);
|
||||
if (!parsed.ok) throw parsed.error;
|
||||
const controller = new AbortController();
|
||||
const readFailure = new Error(
|
||||
"controlled session profile read failure",
|
||||
);
|
||||
let allowClose = false;
|
||||
let closes = 0;
|
||||
controls.file = (handle, openedPath) => {
|
||||
if (openedPath !== join(root, "Module.manifest")) return;
|
||||
controls.file = undefined;
|
||||
vi.spyOn(handle, "read").mockRejectedValue(readFailure);
|
||||
const close = handle.close.bind(handle);
|
||||
vi.spyOn(handle, "close").mockImplementation(async () => {
|
||||
closes++;
|
||||
controller.abort({ caller: "profile inspection finalization" });
|
||||
if (!allowClose)
|
||||
throw new Error("controlled session profile close failure");
|
||||
await close();
|
||||
});
|
||||
};
|
||||
const createClient = vi
|
||||
.spyOn(provider, "createClient")
|
||||
.mockImplementation(() => {
|
||||
throw new Error("Failed profile must not start an engine");
|
||||
});
|
||||
const binary = composeBinarySession(
|
||||
new AnalysisProviderRegistry([provider]),
|
||||
);
|
||||
try {
|
||||
const result =
|
||||
operation === "preview"
|
||||
? await binary.previewTarget(parsed.value, {
|
||||
providerId: "ghidra",
|
||||
signal: controller.signal,
|
||||
})
|
||||
: await binary.open(path, {
|
||||
providerId: "ghidra",
|
||||
signal: controller.signal,
|
||||
});
|
||||
expect(result.ok).toBe(false);
|
||||
if (result.ok) throw new Error("Failed profile admitted");
|
||||
expect(result.error._tag).toBe("AnalysisCapabilityUnavailableError");
|
||||
expect(result.error.message).toContain(readFailure.message);
|
||||
expect(result.error.cleanupIncomplete).toBe(true);
|
||||
expect(binary.status()).toMatchObject({ cleanup_pending: true });
|
||||
expect(createClient).not.toHaveBeenCalled();
|
||||
const failedClose = await binary.close();
|
||||
expect(failedClose.ok).toBe(false);
|
||||
expect(closes).toBe(2);
|
||||
expect(result.error.cleanupIncomplete).toBe(true);
|
||||
allowClose = true;
|
||||
expect(await binary.close()).toEqual({ ok: true, value: null });
|
||||
expect(closes).toBe(3);
|
||||
expect(result.error.cleanupIncomplete).toBe(false);
|
||||
if (failedClose.ok) throw new Error("Close unexpectedly succeeded");
|
||||
expect(failedClose.error.cleanupIncomplete).toBe(false);
|
||||
expect(binary.status()).toMatchObject({ cleanup_pending: false });
|
||||
} finally {
|
||||
allowClose = true;
|
||||
await binary.close();
|
||||
}
|
||||
},
|
||||
);
|
||||
},
|
||||
);
|
||||
@@ -18,7 +18,9 @@ import {
|
||||
import { AnalysisProviderRegistry } from "../../../src/application/binary/AnalysisProviderRegistry.js";
|
||||
import { composeBinarySession } from "../../../src/application/binary/BinarySessionComposition.js";
|
||||
import { createAnalysisProfile } from "../../../src/domain/analysisProfile.js";
|
||||
import { ok } from "../../../src/domain/result.js";
|
||||
import { err, ok } from "../../../src/domain/result.js";
|
||||
import { AnalysisCancelledError } from "../../../src/domain/analysisErrorCore.js";
|
||||
import { ProviderCleanupError } from "../../../src/domain/providerCleanupError.js";
|
||||
import { silentLogger } from "../../../src/logger.js";
|
||||
import { createServer } from "../../../src/server/createServer.js";
|
||||
|
||||
@@ -235,3 +237,81 @@ const structured = (result: CallToolResult): Record<string, unknown> =>
|
||||
result.isError === true
|
||||
? parseMcpToolError(result)
|
||||
: z.record(z.string(), z.unknown()).parse(result.structuredContent);
|
||||
|
||||
it("retains attributed mixed profile cleanup through registered MCP calls and partial retry", async () => {
|
||||
directory = await createTestTempDirectory("rea-profile-completion-mcp-");
|
||||
const path = join(directory, "fixture.hop");
|
||||
await writeFile(path, "fixture");
|
||||
const starts: string[] = [];
|
||||
const released = new Set<string>();
|
||||
const failures = {
|
||||
alpha: new AnalysisCancelledError("resolve_analysis_profile", {
|
||||
cause: { caller: "controlled producer cancellation" },
|
||||
cleanup: { reason: "alpha close failed", resources: ["alpha-file"] },
|
||||
}),
|
||||
beta: new ProviderCleanupError("beta", ["beta-file"], {
|
||||
reason: "beta close failed",
|
||||
}),
|
||||
};
|
||||
const providers = (["alpha", "beta"] as const).map((id) => ({
|
||||
...candidate(id, starts),
|
||||
resolveAnalysisProfile: async () => err(failures[id]),
|
||||
closeOwnedResources: async () => {
|
||||
if (!released.has(id)) return err(failures[id]);
|
||||
failures[id].resolveCleanup();
|
||||
return ok(null);
|
||||
},
|
||||
}));
|
||||
const session = composeBinarySession(new AnalysisProviderRegistry(providers));
|
||||
const server = createServer(
|
||||
{ kind: "session", session },
|
||||
{ logger: silentLogger },
|
||||
);
|
||||
const mcp = new Client({ name: "profile-completion", version: "1.0.0" });
|
||||
const [clientTransport, serverTransport] =
|
||||
InMemoryTransport.createLinkedPair();
|
||||
resources.push(mcp, server, { close: () => session.close() });
|
||||
try {
|
||||
await server.connect(serverTransport);
|
||||
await mcp.connect(clientTransport);
|
||||
const opening = parseMcpToolError(
|
||||
await mcp.callTool({
|
||||
name: "open_binary",
|
||||
arguments: { path, provider_id: "auto" },
|
||||
}),
|
||||
);
|
||||
expect(opening.error).toMatchObject({
|
||||
code: "cleanup_incomplete",
|
||||
details: {
|
||||
resources: ["alpha-file", "beta-file"],
|
||||
diagnostics: {
|
||||
profile_errors: expect.arrayContaining([
|
||||
expect.objectContaining({ provider_id: "alpha" }),
|
||||
expect.objectContaining({ provider_id: "beta" }),
|
||||
]),
|
||||
},
|
||||
},
|
||||
});
|
||||
expect(starts).toEqual([]);
|
||||
released.add("alpha");
|
||||
const closing = parseMcpToolError(
|
||||
await mcp.callTool({ name: "close_binary", arguments: {} }),
|
||||
);
|
||||
expect(closing.error).toMatchObject({
|
||||
code: "cleanup_incomplete",
|
||||
details: { resources: ["beta-file"] },
|
||||
});
|
||||
expect(session.status()).toMatchObject({ cleanup_pending: true });
|
||||
released.add("beta");
|
||||
expect(
|
||||
(await mcp.callTool({ name: "close_binary", arguments: {} })).isError,
|
||||
).not.toBe(true);
|
||||
expect(session.status()).toMatchObject({ cleanup_pending: false });
|
||||
expect(failures.alpha.cleanupIncomplete).toBe(false);
|
||||
expect(failures.beta.cleanupIncomplete).toBe(false);
|
||||
} finally {
|
||||
released.add("alpha");
|
||||
released.add("beta");
|
||||
await session.close();
|
||||
}
|
||||
});
|
||||
@@ -129,19 +129,33 @@ it("waits for an aborted pre-session hash before retrying its late close owner",
|
||||
});
|
||||
try {
|
||||
const controller = new AbortController();
|
||||
const previewWork = binary.previewTarget(fixture.target, {
|
||||
signal: controller.signal,
|
||||
});
|
||||
let previewSettled = false;
|
||||
const previewWork = binary
|
||||
.previewTarget(fixture.target, {
|
||||
signal: controller.signal,
|
||||
})
|
||||
.then((result) => {
|
||||
previewSettled = true;
|
||||
return result;
|
||||
});
|
||||
await closeEntered;
|
||||
controller.abort();
|
||||
let closeSettled = false;
|
||||
const closing = binary.close().then((result) => {
|
||||
closeSettled = true;
|
||||
return result;
|
||||
});
|
||||
await new Promise<void>((resolve) => setImmediate(resolve));
|
||||
expect(previewSettled).toBe(false);
|
||||
expect(closeSettled).toBe(false);
|
||||
expect(binary.status()).toMatchObject({ cleanup_pending: true });
|
||||
releaseClose();
|
||||
const preview = await previewWork;
|
||||
expect(preview).toMatchObject({
|
||||
ok: false,
|
||||
error: { _tag: "AnalysisCancelledError" },
|
||||
error: { _tag: "ProviderAdapterError", cleanupIncomplete: true },
|
||||
});
|
||||
|
||||
const closing = binary.close();
|
||||
releaseClose();
|
||||
const firstClose = await closing;
|
||||
expect(firstClose).toMatchObject({
|
||||
ok: false,
|
||||
@@ -152,6 +166,10 @@ it("waits for an aborted pre-session hash before retrying its late close owner",
|
||||
close.mockRestore();
|
||||
expect(await binary.close()).toEqual({ ok: true, value: null });
|
||||
expect(binary.status()).toMatchObject({ cleanup_pending: false });
|
||||
if (preview.ok || firstClose.ok)
|
||||
throw new Error("Cleanup unexpectedly succeeded");
|
||||
expect(preview.error.cleanupIncomplete).toBe(false);
|
||||
expect(firstClose.error.cleanupIncomplete).toBe(false);
|
||||
} finally {
|
||||
releaseClose();
|
||||
close.mockRestore();
|
||||
|
||||
@@ -1327,9 +1327,10 @@ describe("cancellation during profile resolution", () => {
|
||||
const [first] = await createBinarySessionTargets();
|
||||
const provider = createCacheProvider([]);
|
||||
const controller = new AbortController();
|
||||
const failure = new Error("Profile resolution failed before cancellation");
|
||||
provider.resolveAnalysisProfile = () => {
|
||||
controller.abort();
|
||||
return Promise.reject(new Error("Profile resolution cancelled"));
|
||||
controller.abort({ caller: "late profile rejection" });
|
||||
return Promise.reject(failure);
|
||||
};
|
||||
const session = createTestBinarySession(provider);
|
||||
try {
|
||||
@@ -1337,7 +1338,11 @@ describe("cancellation during profile resolution", () => {
|
||||
session.open(first, { signal: controller.signal }),
|
||||
).resolves.toMatchObject({
|
||||
ok: false,
|
||||
error: { _tag: "AnalysisCancelledError", operation: "open_binary" },
|
||||
error: {
|
||||
_tag: "ProviderAdapterError",
|
||||
operation: "resolve_analysis_profile",
|
||||
cause: failure,
|
||||
},
|
||||
});
|
||||
// Allow Node to report any rejection abandoned by the cancellation path.
|
||||
await new Promise<void>((resolve) => setImmediate(resolve));
|
||||
@@ -1347,35 +1352,66 @@ describe("cancellation during profile resolution", () => {
|
||||
}
|
||||
});
|
||||
|
||||
it("cancels profile resolution even when the provider ignores its signal", async () => {
|
||||
it("joins a signal-ignoring profile producer before cancellation and session close settle", async () => {
|
||||
const [first] = await createBinarySessionTargets();
|
||||
const provider = createCacheProvider([]);
|
||||
if (provider.resolveAnalysisProfile === undefined)
|
||||
throw new Error("Expected the fixture profile resolver");
|
||||
const resolveProfile = provider.resolveAnalysisProfile.bind(provider);
|
||||
const admitted = createDeferred<void>();
|
||||
const release = createDeferred<void>();
|
||||
let observedSignal: AbortSignal | undefined;
|
||||
let created = 0;
|
||||
provider.resolveAnalysisProfile = (_target, options) => {
|
||||
provider.resolveAnalysisProfile = async (target, options) => {
|
||||
observedSignal = options?.signal;
|
||||
return new Promise<never>(() => undefined);
|
||||
admitted.resolve();
|
||||
await release.promise;
|
||||
return resolveProfile(target);
|
||||
};
|
||||
provider.createClient = () => {
|
||||
created += 1;
|
||||
created++;
|
||||
return new ControllableAnalysisClient();
|
||||
};
|
||||
const session = createTestBinarySession(provider);
|
||||
const controller = new AbortController();
|
||||
const opening = session.open(first, { signal: controller.signal });
|
||||
while (observedSignal === undefined)
|
||||
await new Promise<void>((resolve) => setImmediate(resolve));
|
||||
expect(observedSignal).toBe(controller.signal);
|
||||
|
||||
controller.abort();
|
||||
|
||||
await expect(opening).resolves.toMatchObject({
|
||||
ok: false,
|
||||
error: { _tag: "AnalysisCancelledError", operation: "open_binary" },
|
||||
const reason = { caller: "profile cancellation" };
|
||||
let openSettled = false;
|
||||
const opening = session
|
||||
.open(first, { signal: controller.signal })
|
||||
.then((result) => {
|
||||
openSettled = true;
|
||||
return result;
|
||||
});
|
||||
await admitted.promise;
|
||||
controller.abort(reason);
|
||||
let closeSettled = false;
|
||||
const closing = session.close().then((result) => {
|
||||
closeSettled = true;
|
||||
return result;
|
||||
});
|
||||
expect(created).toBe(0);
|
||||
try {
|
||||
await new Promise<void>((resolve) => setImmediate(resolve));
|
||||
expect(observedSignal?.reason).toBe(reason);
|
||||
expect(openSettled).toBe(false);
|
||||
expect(closeSettled).toBe(false);
|
||||
expect(session.status()).toMatchObject({ cleanup_pending: true });
|
||||
release.resolve();
|
||||
await expect(opening).resolves.toMatchObject({
|
||||
ok: false,
|
||||
error: { _tag: "AnalysisCancelledError", cause: reason },
|
||||
});
|
||||
expect(await closing).toEqual(resultOk(null));
|
||||
expect(created).toBe(0);
|
||||
expect(session.status()).toMatchObject({ cleanup_pending: false });
|
||||
} finally {
|
||||
release.resolve();
|
||||
await Promise.all([opening, closing]);
|
||||
await session.close();
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
describe("cancellation during client transitions", () => {
|
||||
it("cancels a call while it waits for a transition", async () => {
|
||||
const [first] = await createBinarySessionTargets();
|
||||
const health = createDeferred<ReturnType<typeof ok>>();
|
||||
|
||||
Reference in new issue
Block a user