fix(browser): preserve caller cancellation through CDP observations
This commit is contained in:
1 parent
9aca543f46
commit
7f4e7d5577
15 files changed
+678
-48
No files matched your search
@@ -197,3 +197,21 @@ it("retains completed bundle facts as bundle Evidence after cancellation", async
|
||||
normalized_result: bundle,
|
||||
});
|
||||
});
|
||||
|
||||
it("preserves a prior caller reason without admitting browser work", async () => {
|
||||
const reason = new AnalysisOutputError(
|
||||
"inspect_web_page",
|
||||
"caller supplied reason",
|
||||
);
|
||||
const result = await listBrowserTargets(
|
||||
provider(async () => {
|
||||
throw new Error("Cancelled inventory must not start");
|
||||
}),
|
||||
input,
|
||||
{ signal: AbortSignal.abort(reason) },
|
||||
);
|
||||
expect(result.ok).toBe(false);
|
||||
if (result.ok) throw new Error("Expected cancelled inventory");
|
||||
expect(result.error._tag).toBe("AnalysisCancelledError");
|
||||
expect(result.error.cause).toBe(reason);
|
||||
});
|
||||
@@ -38,7 +38,11 @@ export const listBrowserTargets = async (
|
||||
const ready = requireBrowser(browser, "list_browser_targets");
|
||||
if (!ready.ok) return ready;
|
||||
if (options.signal?.aborted)
|
||||
return err(new AnalysisCancelledError("list_browser_targets"));
|
||||
return err(
|
||||
new AnalysisCancelledError("list_browser_targets", {
|
||||
cause: options.signal.reason,
|
||||
}),
|
||||
);
|
||||
const result = await ready.value.listTargets(input, options);
|
||||
return result.ok
|
||||
? browserEvidence(
|
||||
@@ -60,7 +64,11 @@ export const inspectWebPage = async (
|
||||
const ready = requireBrowser(browser, "inspect_web_page");
|
||||
if (!ready.ok) return ready;
|
||||
if (options.signal?.aborted)
|
||||
return err(new AnalysisCancelledError("inspect_web_page"));
|
||||
return err(
|
||||
new AnalysisCancelledError("inspect_web_page", {
|
||||
cause: options.signal.reason,
|
||||
}),
|
||||
);
|
||||
const result = await ready.value.inspectPage(input, options);
|
||||
if (!result.ok)
|
||||
return retainBrowserObservation(
|
||||
@@ -89,7 +97,11 @@ export const analyzeWebBundle = async (
|
||||
const ready = requireBrowser(browser, "analyze_web_bundle");
|
||||
if (!ready.ok) return ready;
|
||||
if (options.signal?.aborted)
|
||||
return err(new AnalysisCancelledError("analyze_web_bundle"));
|
||||
return err(
|
||||
new AnalysisCancelledError("analyze_web_bundle", {
|
||||
cause: options.signal.reason,
|
||||
}),
|
||||
);
|
||||
const analyzed = await ready.value.analyzeBundle(input, options);
|
||||
if (!analyzed.ok)
|
||||
return retainBrowserObservation(
|
||||
@@ -116,7 +128,11 @@ export const observeWebSession = async (
|
||||
const ready = requireBrowser(browser, "observe_web_session");
|
||||
if (!ready.ok) return ready;
|
||||
if (options.signal?.aborted)
|
||||
return err(new AnalysisCancelledError("observe_web_session"));
|
||||
return err(
|
||||
new AnalysisCancelledError("observe_web_session", {
|
||||
cause: options.signal.reason,
|
||||
}),
|
||||
);
|
||||
const observed = await ready.value.observeSession(input, options);
|
||||
return observed.ok
|
||||
? browserEvidence(
|
||||
@@ -138,7 +154,11 @@ export const discoverWebMcpTools = async (
|
||||
const ready = requireBrowser(browser, "discover_webmcp_tools");
|
||||
if (!ready.ok) return ready;
|
||||
if (options.signal?.aborted)
|
||||
return err(new AnalysisCancelledError("discover_webmcp_tools"));
|
||||
return err(
|
||||
new AnalysisCancelledError("discover_webmcp_tools", {
|
||||
cause: options.signal.reason,
|
||||
}),
|
||||
);
|
||||
const discovered = await ready.value.discoverWebMcpTools(input, options);
|
||||
return discovered.ok
|
||||
? browserEvidence(
|
||||
@@ -180,7 +200,11 @@ export const captureWebScreenshot = async (
|
||||
const ready = requireBrowser(browser, "capture_web_screenshot");
|
||||
if (!ready.ok) return ready;
|
||||
if (options.signal?.aborted)
|
||||
return err(new AnalysisCancelledError("capture_web_screenshot"));
|
||||
return err(
|
||||
new AnalysisCancelledError("capture_web_screenshot", {
|
||||
cause: options.signal.reason,
|
||||
}),
|
||||
);
|
||||
const captured = await ready.value.captureScreenshot(input, options);
|
||||
return captured.ok
|
||||
? browserEvidence(
|
||||
|
||||
@@ -29,10 +29,15 @@ it("rejects malformed input and prior cancellation without calling a provider",
|
||||
if (malformed.ok)
|
||||
throw new Error("Invalid endpoint must fail before execution");
|
||||
expect(malformed.error._tag).toBe("AnalysisInputError");
|
||||
const cancelled = await service.inspect({}, { signal: AbortSignal.abort() });
|
||||
const reason = { requested: "stop before provider admission" };
|
||||
const cancelled = await service.inspect(
|
||||
{},
|
||||
{ signal: AbortSignal.abort(reason) },
|
||||
);
|
||||
if (cancelled.ok)
|
||||
throw new Error("Prior cancellation must fail before inspection");
|
||||
expect(cancelled.error._tag).toBe("AnalysisCancelledError");
|
||||
expect(cancelled.error.cause).toBe(reason);
|
||||
});
|
||||
|
||||
it("preserves actual provider failures and rejects changed target/selector identities", async () => {
|
||||
|
||||
@@ -187,7 +187,9 @@ const parseInput = <T>(
|
||||
options: ExecutionOptions | undefined,
|
||||
): Result<T, AnalysisError> => {
|
||||
if (options?.signal?.aborted)
|
||||
return err(new AnalysisCancelledError(operation));
|
||||
return err(
|
||||
new AnalysisCancelledError(operation, { cause: options.signal.reason }),
|
||||
);
|
||||
const parsed = schema.safeParse(rawInput);
|
||||
return parsed.success
|
||||
? ok(parsed.data)
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
import { afterEach, describe, expect, it, vi } from "vitest";
|
||||
|
||||
import { AnalysisCancelledError } from "../domain/analysisErrorCore.js";
|
||||
import { delayWithCancellation } from "./CdpCaptureValues.js";
|
||||
|
||||
describe("cancellable browser observation delays", () => {
|
||||
@@ -24,4 +25,33 @@ describe("cancellable browser observation delays", () => {
|
||||
await waiting;
|
||||
expect(completed).toBe(true);
|
||||
});
|
||||
|
||||
it("preserves exact cancellation reasons before and during a capture wait", async () => {
|
||||
const preAbortedReason = { _tag: "pre-aborted" };
|
||||
const preAborted = new AbortController();
|
||||
preAborted.abort(preAbortedReason);
|
||||
const beforeStart = await delayWithCancellation(
|
||||
1,
|
||||
"observe_web_execution",
|
||||
preAborted.signal,
|
||||
).catch((error: unknown) => error);
|
||||
|
||||
expect(beforeStart).toBeInstanceOf(AnalysisCancelledError);
|
||||
expect(beforeStart).toMatchObject({ cause: preAbortedReason });
|
||||
|
||||
vi.useFakeTimers();
|
||||
const reason = new Error("capture wait cancelled");
|
||||
const active = new AbortController();
|
||||
const waiting = delayWithCancellation(
|
||||
10_000,
|
||||
"observe_web_execution",
|
||||
active.signal,
|
||||
).catch((error: unknown) => error);
|
||||
active.abort(reason);
|
||||
const duringWait = await waiting;
|
||||
|
||||
expect(duringWait).toBeInstanceOf(AnalysisCancelledError);
|
||||
expect(duringWait).toMatchObject({ cause: reason });
|
||||
expect(vi.getTimerCount()).toBe(0);
|
||||
});
|
||||
});
|
||||
@@ -125,7 +125,8 @@ export const delayWithCancellation = async (
|
||||
const maximumTimerDelay = 2_147_483_647;
|
||||
let remaining = durationMs;
|
||||
while (remaining > 0) {
|
||||
if (signal?.aborted === true) throw new AnalysisCancelledError(operation);
|
||||
if (signal?.aborted === true)
|
||||
throw new AnalysisCancelledError(operation, { cause: signal.reason });
|
||||
const delay = Math.min(remaining, maximumTimerDelay);
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
const timer = setTimeout(() => {
|
||||
@@ -135,7 +136,9 @@ export const delayWithCancellation = async (
|
||||
const onAbort = (): void => {
|
||||
clearTimeout(timer);
|
||||
signal?.removeEventListener("abort", onAbort);
|
||||
reject(new AnalysisCancelledError(operation));
|
||||
reject(
|
||||
new AnalysisCancelledError(operation, { cause: signal?.reason }),
|
||||
);
|
||||
};
|
||||
signal?.addEventListener("abort", onAbort, { once: true });
|
||||
if (signal?.aborted === true) onAbort();
|
||||
|
||||
@@ -66,7 +66,8 @@ export class CdpConnection {
|
||||
limits?: { readonly maxPayloadBytes: number },
|
||||
onConnected?: (connection: CdpConnection) => void,
|
||||
): Promise<CdpConnection> {
|
||||
if (signal?.aborted === true) throw new AnalysisCancelledError(operation);
|
||||
if (signal?.aborted === true)
|
||||
throw new AnalysisCancelledError(operation, { cause: signal.reason });
|
||||
const maxPayloadBytes =
|
||||
limits === undefined
|
||||
? WEB_RUNTIME_LIMITS.protocolBytes
|
||||
@@ -152,7 +153,9 @@ export class CdpConnection {
|
||||
if (this.#closed || this.socket.readyState !== WebSocket.OPEN)
|
||||
throw this.#transportError(this.#transportFailure);
|
||||
if (signal?.aborted === true)
|
||||
throw new AnalysisCancelledError(this.operation);
|
||||
throw new AnalysisCancelledError(this.operation, {
|
||||
cause: signal.reason,
|
||||
});
|
||||
const id = this.#nextId;
|
||||
this.#nextId += 1;
|
||||
const message = JSON.stringify({
|
||||
@@ -166,7 +169,11 @@ export class CdpConnection {
|
||||
const pending = this.#pending.get(id);
|
||||
if (pending === undefined) return;
|
||||
this.#complete(id, pending);
|
||||
reject(new AnalysisCancelledError(this.operation));
|
||||
reject(
|
||||
new AnalysisCancelledError(this.operation, {
|
||||
cause: signal?.reason,
|
||||
}),
|
||||
);
|
||||
};
|
||||
const timer = setTimeout(() => {
|
||||
const pending = this.#pending.get(id);
|
||||
@@ -446,11 +453,9 @@ const waitForOpen = async (
|
||||
const onFailure = (cause: unknown): void => {
|
||||
cleanup();
|
||||
reject(
|
||||
signal?.aborted === true
|
||||
? new AnalysisCancelledError(operation)
|
||||
: new BrowserObservationError(operation, "endpoint_unreachable", {
|
||||
cause,
|
||||
}),
|
||||
new BrowserObservationError(operation, "endpoint_unreachable", {
|
||||
cause,
|
||||
}),
|
||||
);
|
||||
};
|
||||
const onClose = (code: number, reason: Buffer): void => {
|
||||
@@ -461,8 +466,9 @@ const waitForOpen = async (
|
||||
);
|
||||
};
|
||||
const onAbort = (): void => {
|
||||
cleanup();
|
||||
terminateSocket(socket);
|
||||
onFailure(new AnalysisCancelledError(operation));
|
||||
reject(new AnalysisCancelledError(operation, { cause: signal?.reason }));
|
||||
};
|
||||
const cleanup = (): void => {
|
||||
clearTimeout(timer);
|
||||
|
||||
@@ -0,0 +1,223 @@
|
||||
import { createServer, type RequestListener, type Server } from "node:http";
|
||||
import { once } from "node:events";
|
||||
|
||||
import { AnalysisCancelledError } from "../domain/analysisErrorCore.js";
|
||||
import { BrowserObservationError } from "../domain/browserObservationError.js";
|
||||
|
||||
import { afterEach, describe, expect, it } from "vitest";
|
||||
|
||||
import {
|
||||
discoverCdpEndpoint,
|
||||
readCdpJson,
|
||||
withCdpDiscoveryDeadline,
|
||||
} from "./CdpEndpoint.js";
|
||||
|
||||
const servers: Server[] = [];
|
||||
|
||||
const listen = async (
|
||||
handler: RequestListener,
|
||||
): Promise<{ readonly server: Server; readonly endpoint: string }> => {
|
||||
const server = createServer(handler);
|
||||
servers.push(server);
|
||||
server.listen(0, "127.0.0.1");
|
||||
await once(server, "listening");
|
||||
const address = server.address();
|
||||
if (address === null || typeof address === "string")
|
||||
throw new Error("Expected a TCP listener address.");
|
||||
return {
|
||||
server,
|
||||
endpoint: `http://127.0.0.1:${String(address.port)}`,
|
||||
};
|
||||
};
|
||||
|
||||
afterEach(async () => {
|
||||
await Promise.all(
|
||||
servers.splice(0).map(async (server) => {
|
||||
server.closeAllConnections();
|
||||
if (server.listening) {
|
||||
server.close();
|
||||
await once(server, "close");
|
||||
}
|
||||
}),
|
||||
);
|
||||
});
|
||||
|
||||
describe("CDP discovery cancellation ownership", () => {
|
||||
it("preserves the exact reason for a pre-aborted discovery request", async () => {
|
||||
let requests = 0;
|
||||
const { endpoint } = await listen((_request, response) => {
|
||||
requests += 1;
|
||||
response.end("{}");
|
||||
});
|
||||
const reason = { _tag: "caller-stop" };
|
||||
const controller = new AbortController();
|
||||
controller.abort(reason);
|
||||
|
||||
const failure = await readCdpJson(
|
||||
new URL("/json/version", endpoint),
|
||||
"inspect_web_page",
|
||||
controller.signal,
|
||||
).catch((error: unknown) => error);
|
||||
|
||||
expect(failure).toBeInstanceOf(AnalysisCancelledError);
|
||||
expect(failure).toMatchObject({ cause: reason });
|
||||
expect(requests).toBe(0);
|
||||
});
|
||||
|
||||
it("cancels an in-flight discovery at the HTTP producer with the original reason", async () => {
|
||||
let requestedPath: string | undefined;
|
||||
let requestSeen!: () => void;
|
||||
const seen = new Promise<void>((resolve) => {
|
||||
requestSeen = resolve;
|
||||
});
|
||||
const { endpoint } = await listen((request) => {
|
||||
requestedPath = request.url;
|
||||
requestSeen();
|
||||
});
|
||||
const reason = new Error("operator stopped discovery");
|
||||
const controller = new AbortController();
|
||||
const discovery = discoverCdpEndpoint(
|
||||
endpoint,
|
||||
"inspect_web_page",
|
||||
controller.signal,
|
||||
).catch((error: unknown) => error);
|
||||
|
||||
await seen;
|
||||
controller.abort(reason);
|
||||
const failure = await discovery;
|
||||
|
||||
expect(requestedPath).toBe("/json/version");
|
||||
expect(failure).toBeInstanceOf(AnalysisCancelledError);
|
||||
expect(failure).toMatchObject({ cause: reason });
|
||||
});
|
||||
|
||||
it("does not replace a primary discovery failure when cancellation arrives later", async () => {
|
||||
const primary = new BrowserObservationError(
|
||||
"inspect_web_page",
|
||||
"protocol_error",
|
||||
);
|
||||
const controller = new AbortController();
|
||||
const failure = await withCdpDiscoveryDeadline(
|
||||
"inspect_web_page",
|
||||
controller.signal,
|
||||
async () => {
|
||||
throw primary;
|
||||
},
|
||||
100,
|
||||
).catch((error: unknown) => error);
|
||||
controller.abort(new Error("late caller cancellation"));
|
||||
|
||||
expect(failure).toBe(primary);
|
||||
});
|
||||
|
||||
it("does not replace a plain physical failure rejected before a late abort", async () => {
|
||||
const primary = new Error("socket failed before cancellation");
|
||||
const controller = new AbortController();
|
||||
const pending = withCdpDiscoveryDeadline(
|
||||
"inspect_web_page",
|
||||
controller.signal,
|
||||
() => Promise.reject(primary),
|
||||
100,
|
||||
).catch((error: unknown) => error);
|
||||
controller.abort(new Error("later caller cancellation"));
|
||||
const failure = await pending;
|
||||
|
||||
expect(failure).toBe(primary);
|
||||
});
|
||||
|
||||
it("classifies its own discovery deadline as timeout", async () => {
|
||||
const failure = await withCdpDiscoveryDeadline(
|
||||
"inspect_web_page",
|
||||
undefined,
|
||||
(signal) =>
|
||||
new Promise((_resolve, reject) => {
|
||||
signal.addEventListener("abort", () => reject(signal.reason), {
|
||||
once: true,
|
||||
});
|
||||
}),
|
||||
1,
|
||||
).catch((error: unknown) => error);
|
||||
|
||||
expect(failure).toBeInstanceOf(BrowserObservationError);
|
||||
expect(failure).toMatchObject({ reason: "timeout" });
|
||||
});
|
||||
|
||||
it("normalizes a typed cancellation caused by its own deadline to timeout", async () => {
|
||||
const failure = await withCdpDiscoveryDeadline(
|
||||
"inspect_web_page",
|
||||
undefined,
|
||||
(signal) =>
|
||||
new Promise((_resolve, reject) => {
|
||||
signal.addEventListener(
|
||||
"abort",
|
||||
() =>
|
||||
reject(
|
||||
new AnalysisCancelledError("inspect_web_page", {
|
||||
cause: signal.reason,
|
||||
}),
|
||||
),
|
||||
{ once: true },
|
||||
);
|
||||
}),
|
||||
1,
|
||||
).catch((error: unknown) => error);
|
||||
|
||||
expect(failure).toBeInstanceOf(BrowserObservationError);
|
||||
expect(failure).toMatchObject({ reason: "timeout" });
|
||||
});
|
||||
|
||||
it("treats the caller's AnalysisError abort reason as cancellation when caller wins first", async () => {
|
||||
const reason = new BrowserObservationError(
|
||||
"inspect_web_page",
|
||||
"protocol_error",
|
||||
);
|
||||
const controller = new AbortController();
|
||||
const failure = await withCdpDiscoveryDeadline(
|
||||
"inspect_web_page",
|
||||
controller.signal,
|
||||
async () => {
|
||||
controller.abort(reason);
|
||||
throw reason;
|
||||
},
|
||||
0,
|
||||
).catch((error: unknown) => error);
|
||||
|
||||
expect(failure).toBeInstanceOf(AnalysisCancelledError);
|
||||
expect(failure).toMatchObject({ cause: reason });
|
||||
});
|
||||
|
||||
it("keeps a deadline rejection when the caller aborts only after timeout", async () => {
|
||||
const controller = new AbortController();
|
||||
const failure = await withCdpDiscoveryDeadline(
|
||||
"inspect_web_page",
|
||||
controller.signal,
|
||||
(signal) =>
|
||||
new Promise((_resolve, reject) => {
|
||||
signal.addEventListener("abort", () => reject(signal.reason), {
|
||||
once: true,
|
||||
});
|
||||
}),
|
||||
0,
|
||||
).catch((error: unknown) => error);
|
||||
controller.abort(new Error("caller stopped after timeout"));
|
||||
|
||||
expect(failure).toBeInstanceOf(BrowserObservationError);
|
||||
expect(failure).toMatchObject({ reason: "timeout" });
|
||||
});
|
||||
});
|
||||
|
||||
it("preserves an independent typed cancellation when the discovery deadline also expires", async () => {
|
||||
const primary = new AnalysisCancelledError("inspect_web_page", {
|
||||
cause: "independent cancellation",
|
||||
});
|
||||
const failure = await withCdpDiscoveryDeadline(
|
||||
"inspect_web_page",
|
||||
undefined,
|
||||
(signal) =>
|
||||
new Promise((_resolve, reject) => {
|
||||
signal.addEventListener("abort", () => reject(primary), { once: true });
|
||||
}),
|
||||
1,
|
||||
).catch((cause: unknown) => cause);
|
||||
expect(failure).toBe(primary);
|
||||
});
|
||||
+43
-13
@@ -157,21 +157,50 @@ export const withCdpDiscoveryDeadline = async <T>(
|
||||
timeoutMs: number = CDP_DISCOVERY_RESPONSE_LIMITS.timeoutMs,
|
||||
): Promise<T> => {
|
||||
const deadline = new AbortController();
|
||||
const timer = setTimeout(() => deadline.abort(), timeoutMs);
|
||||
let abortSource: "caller" | "deadline" | undefined;
|
||||
let abortReason: unknown;
|
||||
const abort = (source: "caller" | "deadline", reason: unknown): void => {
|
||||
if (abortSource !== undefined) return;
|
||||
abortSource = source;
|
||||
abortReason = reason;
|
||||
deadline.abort(reason);
|
||||
};
|
||||
const onCallerAbort = (): void => {
|
||||
if (signal?.aborted === true) abort("caller", signal.reason);
|
||||
};
|
||||
const timer = setTimeout(
|
||||
() => abort("deadline", new Error("CDP discovery deadline exceeded")),
|
||||
timeoutMs,
|
||||
);
|
||||
signal?.addEventListener("abort", onCallerAbort, { once: true });
|
||||
onCallerAbort();
|
||||
try {
|
||||
return await work(
|
||||
signal === undefined
|
||||
? deadline.signal
|
||||
: AbortSignal.any([signal, deadline.signal]),
|
||||
);
|
||||
return await work(deadline.signal);
|
||||
} catch (cause: unknown) {
|
||||
if (deadline.signal.aborted && signal?.aborted !== true)
|
||||
throw new BrowserObservationError(operation, "timeout", {
|
||||
detail: `CDP browser-socket discovery exceeded its ${String(timeoutMs)} ms deadline.`,
|
||||
});
|
||||
if (cause === abortReason) {
|
||||
if (abortSource === "caller")
|
||||
throw new AnalysisCancelledError(operation, { cause: abortReason });
|
||||
if (abortSource === "deadline")
|
||||
throw new BrowserObservationError(operation, "timeout", {
|
||||
detail: `CDP browser-socket discovery exceeded its ${String(timeoutMs)} ms deadline.`,
|
||||
});
|
||||
}
|
||||
// A concrete failure produced before cancellation remains authoritative.
|
||||
if (cause instanceof AnalysisError) {
|
||||
if (
|
||||
abortSource === "deadline" &&
|
||||
cause instanceof AnalysisCancelledError &&
|
||||
cause.cause === abortReason
|
||||
)
|
||||
throw new BrowserObservationError(operation, "timeout", {
|
||||
detail: `CDP browser-socket discovery exceeded its ${String(timeoutMs)} ms deadline.`,
|
||||
});
|
||||
throw cause;
|
||||
}
|
||||
throw cause;
|
||||
} finally {
|
||||
clearTimeout(timer);
|
||||
signal?.removeEventListener("abort", onCallerAbort);
|
||||
}
|
||||
};
|
||||
|
||||
@@ -409,7 +438,7 @@ export const readCdpJson = async (
|
||||
): Promise<unknown> =>
|
||||
await new Promise((resolve, reject) => {
|
||||
if (signal?.aborted === true) {
|
||||
reject(new AnalysisCancelledError(operation));
|
||||
reject(new AnalysisCancelledError(operation, { cause: signal.reason }));
|
||||
return;
|
||||
}
|
||||
let response: IncomingMessage | undefined;
|
||||
@@ -431,7 +460,7 @@ export const readCdpJson = async (
|
||||
};
|
||||
const onAbort = (): void => {
|
||||
if (signal?.aborted === true)
|
||||
finish(new AnalysisCancelledError(operation));
|
||||
finish(new AnalysisCancelledError(operation, { cause: signal.reason }));
|
||||
};
|
||||
request_ = request(
|
||||
url,
|
||||
@@ -523,7 +552,8 @@ const endpointFailure = (
|
||||
signal?: AbortSignal,
|
||||
): AnalysisError => {
|
||||
if (cause instanceof AnalysisError) return cause;
|
||||
if (signal?.aborted === true) return new AnalysisCancelledError(operation);
|
||||
if (signal?.aborted === true)
|
||||
return new AnalysisCancelledError(operation, { cause: signal.reason });
|
||||
return new BrowserObservationError(operation, "endpoint_unreachable", {
|
||||
cause,
|
||||
});
|
||||
|
||||
@@ -370,8 +370,13 @@ const waitForWindow = async (
|
||||
if (settled) return;
|
||||
settled = true;
|
||||
clearTimeout(timer);
|
||||
signal?.removeEventListener("abort", onAbort);
|
||||
removeEnd();
|
||||
reject(new AnalysisCancelledError("observe_web_session"));
|
||||
reject(
|
||||
new AnalysisCancelledError("observe_web_session", {
|
||||
cause: signal?.reason,
|
||||
}),
|
||||
);
|
||||
};
|
||||
const timer = setTimeout(() => finish("window_elapsed"), durationMs);
|
||||
removeEnd = capture.onEnd(finish);
|
||||
|
||||
@@ -53,7 +53,9 @@ export const captureWebExecution = async (
|
||||
const reason = await waiting;
|
||||
session.beginCompletion();
|
||||
if (session.options.signal?.aborted)
|
||||
throw new AnalysisCancelledError(session.operation);
|
||||
throw new AnalysisCancelledError(session.operation, {
|
||||
cause: session.options.signal.reason,
|
||||
});
|
||||
sources.check();
|
||||
requests.check();
|
||||
let raw: unknown;
|
||||
|
||||
@@ -10,6 +10,7 @@ it("settles an already lost transport without arming a timer that cannot resolve
|
||||
window.end("target_terminated");
|
||||
await expect(window.start(30_000)).resolves.toBe("target_terminated");
|
||||
expect(window.active).toBe(false);
|
||||
expect(window.armedAt).toBeUndefined();
|
||||
});
|
||||
|
||||
it("ends the actual armed window at its selected duration and disposes its timer", async () => {
|
||||
@@ -41,11 +42,20 @@ it("settles caller cancellation and removes its local timer", async () => {
|
||||
options: { signal: controller.signal },
|
||||
});
|
||||
try {
|
||||
const assertion = expect(window.start(30_000)).rejects.toMatchObject({
|
||||
const reason = new Error("operator stopped the armed window");
|
||||
const result = window.start(30_000).catch((error: unknown) => error);
|
||||
controller.abort(reason);
|
||||
const failure = await result;
|
||||
expect(failure).toMatchObject({
|
||||
_tag: "AnalysisCancelledError",
|
||||
cause: reason,
|
||||
});
|
||||
controller.abort();
|
||||
await assertion;
|
||||
expect(() => window.check()).toThrow(failure as Error);
|
||||
const endedAt = window.endedAt;
|
||||
window.end("target_terminated");
|
||||
expect(window.reason).toBeUndefined();
|
||||
expect(window.endedAt).toBe(endedAt);
|
||||
await expect(window.start(30_000)).rejects.toBe(failure);
|
||||
expect(window.active).toBe(false);
|
||||
expect(vi.getTimerCount()).toBe(0);
|
||||
} finally {
|
||||
|
||||
@@ -63,18 +63,20 @@ export class CdpExecutionWindow {
|
||||
|
||||
/** The timer starts at arming, before progress delivery can delay the external action. */
|
||||
start(duration: number): Promise<EndReason> {
|
||||
this.armedAt = new Date().toISOString();
|
||||
if (this.#failure !== undefined) return Promise.reject(this.#failure);
|
||||
if (this.reason !== undefined) return Promise.resolve(this.reason);
|
||||
this.armedAt = new Date().toISOString();
|
||||
return new Promise((resolve, reject) => {
|
||||
this.#resolve = resolve;
|
||||
this.#reject = reject;
|
||||
this.#timer = setTimeout(() => this.end("window_elapsed"), duration);
|
||||
const signal = this.session.options.signal;
|
||||
const abort = (): void => {
|
||||
this.endedAt = new Date().toISOString();
|
||||
this.dispose();
|
||||
this.#reject?.(new AnalysisCancelledError(this.session.operation));
|
||||
this.fail(
|
||||
new AnalysisCancelledError(this.session.operation, {
|
||||
cause: signal?.reason,
|
||||
}),
|
||||
);
|
||||
};
|
||||
this.#removeAbort = () => signal?.removeEventListener("abort", abort);
|
||||
signal?.addEventListener("abort", abort, { once: true });
|
||||
|
||||
@@ -0,0 +1,95 @@
|
||||
import { expect, it, vi } from "vitest";
|
||||
import { WebRuntimeService } from "../../../src/application/WebRuntimeService.js";
|
||||
import { CdpWebRuntimeProvider } from "../../../src/browser/execution/CdpWebRuntimeProvider.js";
|
||||
import { AnalysisOutputError } from "../../../src/domain/analysisErrorCore.js";
|
||||
import { startRuntimeBrowser } from "../../fixtures/webRuntime.js";
|
||||
|
||||
it.each([
|
||||
new Error("operator stopped observation"),
|
||||
new AnalysisOutputError("inspect_web_page", "caller supplied reason"),
|
||||
{ requested: "stop the armed observation" },
|
||||
])(
|
||||
"preserves caller reason through the armed service workflow and stops instrumentation (%s)",
|
||||
async (reason) => {
|
||||
const browser = await startRuntimeBrowser();
|
||||
const provider = new CdpWebRuntimeProvider();
|
||||
const service = new WebRuntimeService(provider);
|
||||
const input = {
|
||||
cdp_endpoint: browser.endpoint,
|
||||
target_id: "allowed-page",
|
||||
observation_ms: 5,
|
||||
};
|
||||
try {
|
||||
const before = await service.observe(input);
|
||||
if (!before.ok) throw before.error;
|
||||
const controller = new AbortController();
|
||||
const cancelled = await service.observe(input, {
|
||||
signal: controller.signal,
|
||||
progress: {
|
||||
report: async (update) => {
|
||||
if (update.completed === 1) controller.abort(reason);
|
||||
},
|
||||
},
|
||||
});
|
||||
expect(cancelled.ok).toBe(false);
|
||||
if (cancelled.ok) throw new Error("Expected cancellation at arming");
|
||||
expect(cancelled.error._tag).toBe("AnalysisCancelledError");
|
||||
expect(cancelled.error.cause).toBe(controller.signal.reason);
|
||||
expect(cancelled.error.cause).toBe(reason);
|
||||
expect(cancelled.error.cleanupIncomplete).toBe(false);
|
||||
expect(
|
||||
browser.commands.filter(
|
||||
(command) => command.method === "Profiler.stopPreciseCoverage",
|
||||
),
|
||||
).toHaveLength(2);
|
||||
await vi.waitFor(() => expect(browser.activeConnectionCount()).toBe(0));
|
||||
const after = await service.observe(input);
|
||||
if (!after.ok) throw after.error;
|
||||
await vi.waitFor(() => expect(browser.activeConnectionCount()).toBe(0));
|
||||
} finally {
|
||||
await provider.close();
|
||||
await browser.close();
|
||||
}
|
||||
},
|
||||
);
|
||||
|
||||
it("preserves the caller reason when a navigation observation is cancelled at arming", async () => {
|
||||
const { CdpBrowserProvider } =
|
||||
await import("../../../src/browser/CdpBrowserProvider.js");
|
||||
const { observeWebSession } =
|
||||
await import("../../../src/application/BrowserObservationService.js");
|
||||
const { observeWebSessionInputSchema } =
|
||||
await import("../../../src/domain/browserSession.js");
|
||||
const browser = await startRuntimeBrowser();
|
||||
const provider = new CdpBrowserProvider();
|
||||
const controller = new AbortController();
|
||||
const reason = { requested: "stop navigation observation" };
|
||||
try {
|
||||
const result = await observeWebSession(
|
||||
provider,
|
||||
observeWebSessionInputSchema.parse({
|
||||
cdp_endpoint: browser.endpoint,
|
||||
allowed_origins: [browser.allowedOrigin],
|
||||
target_id: "allowed-page",
|
||||
observation_ms: 30_000,
|
||||
}),
|
||||
{
|
||||
signal: controller.signal,
|
||||
progress: {
|
||||
report: async (update) => {
|
||||
if (update.completed === 1) controller.abort(reason);
|
||||
},
|
||||
},
|
||||
},
|
||||
);
|
||||
expect(result.ok).toBe(false);
|
||||
if (result.ok) throw new Error("Expected cancelled navigation observation");
|
||||
expect(result.error._tag).toBe("AnalysisCancelledError");
|
||||
expect(result.error.cause).toBe(reason);
|
||||
expect(result.error.cleanupIncomplete).toBe(false);
|
||||
await vi.waitFor(() => expect(browser.activeConnectionCount()).toBe(0));
|
||||
} finally {
|
||||
await provider.close();
|
||||
await browser.close();
|
||||
}
|
||||
});
|
||||
@@ -1,9 +1,14 @@
|
||||
import { createHash } from "node:crypto";
|
||||
import { createServer } from "node:http";
|
||||
import { createServer as createNetServer, type Socket } from "node:net";
|
||||
|
||||
import { afterEach, describe, expect, it, vi } from "vitest";
|
||||
|
||||
import { CdpConnection } from "../../../src/browser/CdpConnection.js";
|
||||
import {
|
||||
AnalysisCancelledError,
|
||||
AnalysisOutputError,
|
||||
} from "../../../src/domain/analysisErrorCore.js";
|
||||
import { WEB_RUNTIME_LIMITS } from "../../../src/domain/webRuntime.js";
|
||||
import {
|
||||
startFakeCdpBrowser,
|
||||
@@ -73,7 +78,31 @@ afterEach(async () => {
|
||||
await Promise.all(browsers.splice(0).map(async (browser) => browser.close()));
|
||||
});
|
||||
|
||||
describe("CDP connection", () => {
|
||||
describe("CDP connection admission cancellation", () => {
|
||||
it("preserves an AnalysisError reason when admission is already cancelled", async () => {
|
||||
const controller = new AbortController();
|
||||
const reason = new AnalysisOutputError(
|
||||
"inspect_web_page",
|
||||
"caller supplied cancellation reason",
|
||||
);
|
||||
controller.abort(reason);
|
||||
|
||||
const failure = await CdpConnection.connect(
|
||||
"ws://127.0.0.1:1/unused",
|
||||
"inspect_web_page",
|
||||
controller.signal,
|
||||
).then(
|
||||
() => undefined,
|
||||
(cause: unknown) => cause,
|
||||
);
|
||||
expect(failure).toBeInstanceOf(AnalysisCancelledError);
|
||||
if (!(failure instanceof AnalysisCancelledError))
|
||||
throw new Error("Expected admission cancellation");
|
||||
expect(failure.cause).toBe(reason);
|
||||
});
|
||||
});
|
||||
|
||||
describe("CDP connection response correlation", () => {
|
||||
it("correlates concurrent command responses over a real WebSocket", async () => {
|
||||
const browser = await startFakeCdpBrowser();
|
||||
browsers.push(browser);
|
||||
@@ -96,6 +125,35 @@ describe("CDP connection", () => {
|
||||
await connection.close();
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
describe("CDP command cancellation causes", () => {
|
||||
it("preserves a supplied reason when a command starts with an aborted signal", async () => {
|
||||
const browser = await startFakeCdpBrowser();
|
||||
browsers.push(browser);
|
||||
const connection = await CdpConnection.connect(
|
||||
browser.browserWebSocketUrl,
|
||||
"observe_web_session",
|
||||
);
|
||||
const controller = new AbortController();
|
||||
const reason = new Error("command was cancelled before send");
|
||||
controller.abort(reason);
|
||||
try {
|
||||
const failure = await connection
|
||||
.send("Page.enable", {}, undefined, controller.signal)
|
||||
.then(
|
||||
() => undefined,
|
||||
(cause: unknown) => cause,
|
||||
);
|
||||
expect(failure).toBeInstanceOf(AnalysisCancelledError);
|
||||
if (!(failure instanceof AnalysisCancelledError))
|
||||
throw new Error("Expected pre-send command cancellation");
|
||||
expect(failure.cause).toBe(reason);
|
||||
expect(browser.commands).toHaveLength(0);
|
||||
} finally {
|
||||
await connection.close();
|
||||
}
|
||||
});
|
||||
|
||||
it("waits for an unresponsive command until caller cancellation", async () => {
|
||||
const browser = await startFakeCdpBrowser({ hangOnMethod: "Page.enable" });
|
||||
@@ -112,17 +170,134 @@ describe("CDP connection", () => {
|
||||
undefined,
|
||||
controller.signal,
|
||||
);
|
||||
const assertion = expect(pending).rejects.toMatchObject({
|
||||
_tag: "AnalysisCancelledError",
|
||||
operation: "observe_web_session",
|
||||
});
|
||||
controller.abort();
|
||||
await assertion;
|
||||
await vi.waitFor(() => expect(browser.commands).toHaveLength(1));
|
||||
const reason = new Error("cancel the pending CDP command");
|
||||
controller.abort(reason);
|
||||
const failure = await pending.then(
|
||||
() => undefined,
|
||||
(cause: unknown) => cause,
|
||||
);
|
||||
expect(failure).toBeInstanceOf(AnalysisCancelledError);
|
||||
if (!(failure instanceof AnalysisCancelledError))
|
||||
throw new Error("Expected command cancellation");
|
||||
expect(failure.operation).toBe("observe_web_session");
|
||||
expect(failure.cause).toBe(reason);
|
||||
} finally {
|
||||
await connection.close();
|
||||
}
|
||||
});
|
||||
|
||||
it("preserves AbortController's default reason for an in-flight command", async () => {
|
||||
const browser = await startFakeCdpBrowser({ hangOnMethod: "Page.enable" });
|
||||
browsers.push(browser);
|
||||
const connection = await CdpConnection.connect(
|
||||
browser.browserWebSocketUrl,
|
||||
"observe_web_session",
|
||||
);
|
||||
const controller = new AbortController();
|
||||
try {
|
||||
const pending = connection.send(
|
||||
"Page.enable",
|
||||
{},
|
||||
undefined,
|
||||
controller.signal,
|
||||
);
|
||||
await vi.waitFor(() => expect(browser.commands).toHaveLength(1));
|
||||
controller.abort();
|
||||
const failure = await pending.then(
|
||||
() => undefined,
|
||||
(cause: unknown) => cause,
|
||||
);
|
||||
expect(failure).toBeInstanceOf(AnalysisCancelledError);
|
||||
if (!(failure instanceof AnalysisCancelledError))
|
||||
throw new Error("Expected command cancellation");
|
||||
expect(failure.cause).toBe(controller.signal.reason);
|
||||
expect(failure.cause).toMatchObject({ name: "AbortError" });
|
||||
} finally {
|
||||
await connection.close();
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
describe("CDP handshake cancellation", () => {
|
||||
it("preserves handshake cancellation and closes the pending socket", async () => {
|
||||
const sockets = new Set<Socket>();
|
||||
const server = createNetServer((socket) => {
|
||||
sockets.add(socket);
|
||||
socket.once("close", () => sockets.delete(socket));
|
||||
socket.resume();
|
||||
});
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
server.once("error", reject);
|
||||
server.listen(0, "127.0.0.1", resolve);
|
||||
});
|
||||
const address = server.address();
|
||||
if (address === null || typeof address === "string")
|
||||
throw new Error("Expected a TCP listener address");
|
||||
const controller = new AbortController();
|
||||
const reason = new Error("cancel pending CDP handshake");
|
||||
const connecting = CdpConnection.connect(
|
||||
`ws://127.0.0.1:${String(address.port)}/target`,
|
||||
"inspect_web_page",
|
||||
controller.signal,
|
||||
);
|
||||
try {
|
||||
await vi.waitFor(() => expect(sockets.size).toBe(1));
|
||||
controller.abort(reason);
|
||||
const failure = await connecting.then(
|
||||
() => undefined,
|
||||
(cause: unknown) => cause,
|
||||
);
|
||||
expect(failure).toBeInstanceOf(AnalysisCancelledError);
|
||||
if (!(failure instanceof AnalysisCancelledError))
|
||||
throw new Error("Expected handshake cancellation");
|
||||
expect(failure.cause).toBe(reason);
|
||||
await vi.waitFor(() => expect(sockets.size).toBe(0));
|
||||
} finally {
|
||||
if (sockets.size > 0) for (const socket of sockets) socket.destroy();
|
||||
await new Promise<void>((resolve, reject) =>
|
||||
server.close((error) =>
|
||||
error === undefined ? resolve() : reject(error),
|
||||
),
|
||||
);
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
describe("CDP transport failure precedence", () => {
|
||||
it("keeps a completed transport failure primary over a later abort", async () => {
|
||||
const browser = await startFakeCdpBrowser({ closeOnMethod: "Page.enable" });
|
||||
browsers.push(browser);
|
||||
const controller = new AbortController();
|
||||
const connection = await CdpConnection.connect(
|
||||
browser.browserWebSocketUrl,
|
||||
"observe_web_session",
|
||||
);
|
||||
try {
|
||||
const pending = connection
|
||||
.send("Page.enable", {}, undefined, controller.signal)
|
||||
.then(
|
||||
() => undefined,
|
||||
(cause: unknown) => cause,
|
||||
);
|
||||
await vi.waitFor(() => expect(browser.commands).toHaveLength(1));
|
||||
const failure = await pending;
|
||||
expect(failure).toMatchObject({
|
||||
_tag: "BrowserObservationError",
|
||||
reason: "disconnected",
|
||||
});
|
||||
controller.abort(new Error("abort after transport failure"));
|
||||
expect(failure).toMatchObject({
|
||||
_tag: "BrowserObservationError",
|
||||
reason: "disconnected",
|
||||
});
|
||||
} finally {
|
||||
await connection.close();
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
describe("CDP connection command responses", () => {
|
||||
it("surfaces an unmodeled fake command as a CDP method rejection", async () => {
|
||||
const browser = await startFakeCdpBrowser();
|
||||
browsers.push(browser);
|
||||
|
||||
Reference in new issue
Block a user