From 7f4e7d5577adf2f4ea6edf869cbb35a43469b8d3 Mon Sep 17 00:00:00 2001 From: morluto <76467478+morluto@users.noreply.github.com> Date: Mon, 19 Oct 2026 14:00:01 +0800 Subject: [PATCH] fix(browser): preserve caller cancellation through CDP observations --- .../BrowserObservationService.test.ts | 18 ++ src/application/BrowserObservationService.ts | 36 ++- src/application/WebRuntimeService.test.ts | 7 +- src/application/WebRuntimeService.ts | 4 +- src/browser/CdpCaptureValues.test.ts | 30 +++ src/browser/CdpCaptureValues.ts | 7 +- src/browser/CdpConnection.ts | 24 +- src/browser/CdpEndpoint.test.ts | 223 ++++++++++++++++++ src/browser/CdpEndpoint.ts | 56 ++++- src/browser/CdpObservationSession.ts | 7 +- src/browser/execution/CdpExecutionCapture.ts | 4 +- .../execution/CdpExecutionWindow.test.ts | 16 +- src/browser/execution/CdpExecutionWindow.ts | 10 +- .../browser/callerCancellationReason.test.ts | 95 ++++++++ tests/boundary/browser/cdpConnection.test.ts | 189 ++++++++++++++- 15 files changed, 678 insertions(+), 48 deletions(-) create mode 100644 src/browser/CdpEndpoint.test.ts create mode 100644 tests/boundary/browser/callerCancellationReason.test.ts diff --git a/src/application/BrowserObservationService.test.ts b/src/application/BrowserObservationService.test.ts index 20b40c10c..d2cdde1c2 100644 --- a/src/application/BrowserObservationService.test.ts +++ b/src/application/BrowserObservationService.test.ts @@ -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); +}); diff --git a/src/application/BrowserObservationService.ts b/src/application/BrowserObservationService.ts index 0c1f2a036..6251e9d4d 100644 --- a/src/application/BrowserObservationService.ts +++ b/src/application/BrowserObservationService.ts @@ -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( diff --git a/src/application/WebRuntimeService.test.ts b/src/application/WebRuntimeService.test.ts index b0160d4d0..7e7c65148 100644 --- a/src/application/WebRuntimeService.test.ts +++ b/src/application/WebRuntimeService.test.ts @@ -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 () => { diff --git a/src/application/WebRuntimeService.ts b/src/application/WebRuntimeService.ts index d7180b397..38d293454 100644 --- a/src/application/WebRuntimeService.ts +++ b/src/application/WebRuntimeService.ts @@ -187,7 +187,9 @@ const parseInput = ( options: ExecutionOptions | undefined, ): Result => { 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) diff --git a/src/browser/CdpCaptureValues.test.ts b/src/browser/CdpCaptureValues.test.ts index 7095bb954..83d5b5adf 100644 --- a/src/browser/CdpCaptureValues.test.ts +++ b/src/browser/CdpCaptureValues.test.ts @@ -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); + }); }); diff --git a/src/browser/CdpCaptureValues.ts b/src/browser/CdpCaptureValues.ts index c7585040a..c5010c6b6 100644 --- a/src/browser/CdpCaptureValues.ts +++ b/src/browser/CdpCaptureValues.ts @@ -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((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(); diff --git a/src/browser/CdpConnection.ts b/src/browser/CdpConnection.ts index fa2f4f22a..2c5512c15 100644 --- a/src/browser/CdpConnection.ts +++ b/src/browser/CdpConnection.ts @@ -66,7 +66,8 @@ export class CdpConnection { limits?: { readonly maxPayloadBytes: number }, onConnected?: (connection: CdpConnection) => void, ): Promise { - 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); diff --git a/src/browser/CdpEndpoint.test.ts b/src/browser/CdpEndpoint.test.ts new file mode 100644 index 000000000..bf96c1914 --- /dev/null +++ b/src/browser/CdpEndpoint.test.ts @@ -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((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); +}); diff --git a/src/browser/CdpEndpoint.ts b/src/browser/CdpEndpoint.ts index 3db49d28c..8aff0945d 100644 --- a/src/browser/CdpEndpoint.ts +++ b/src/browser/CdpEndpoint.ts @@ -157,21 +157,50 @@ export const withCdpDiscoveryDeadline = async ( timeoutMs: number = CDP_DISCOVERY_RESPONSE_LIMITS.timeoutMs, ): Promise => { 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 => 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, }); diff --git a/src/browser/CdpObservationSession.ts b/src/browser/CdpObservationSession.ts index b256c4fa4..61b65b399 100644 --- a/src/browser/CdpObservationSession.ts +++ b/src/browser/CdpObservationSession.ts @@ -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); diff --git a/src/browser/execution/CdpExecutionCapture.ts b/src/browser/execution/CdpExecutionCapture.ts index 023c3f8e2..14c5bb641 100644 --- a/src/browser/execution/CdpExecutionCapture.ts +++ b/src/browser/execution/CdpExecutionCapture.ts @@ -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; diff --git a/src/browser/execution/CdpExecutionWindow.test.ts b/src/browser/execution/CdpExecutionWindow.test.ts index f6cf40878..e906be92a 100644 --- a/src/browser/execution/CdpExecutionWindow.test.ts +++ b/src/browser/execution/CdpExecutionWindow.test.ts @@ -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 { diff --git a/src/browser/execution/CdpExecutionWindow.ts b/src/browser/execution/CdpExecutionWindow.ts index 18d09e573..f8fa4b7ef 100644 --- a/src/browser/execution/CdpExecutionWindow.ts +++ b/src/browser/execution/CdpExecutionWindow.ts @@ -63,18 +63,20 @@ export class CdpExecutionWindow { /** The timer starts at arming, before progress delivery can delay the external action. */ start(duration: number): Promise { - 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 }); diff --git a/tests/boundary/browser/callerCancellationReason.test.ts b/tests/boundary/browser/callerCancellationReason.test.ts new file mode 100644 index 000000000..e28fa26bf --- /dev/null +++ b/tests/boundary/browser/callerCancellationReason.test.ts @@ -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(); + } +}); diff --git a/tests/boundary/browser/cdpConnection.test.ts b/tests/boundary/browser/cdpConnection.test.ts index 69ec74a9e..592fe350d 100644 --- a/tests/boundary/browser/cdpConnection.test.ts +++ b/tests/boundary/browser/cdpConnection.test.ts @@ -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(); + const server = createNetServer((socket) => { + sockets.add(socket); + socket.once("close", () => sockets.delete(socket)); + socket.resume(); + }); + await new Promise((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((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);