diff --git a/src/browser/CdpBrowserProvider.admission.test.ts b/src/browser/CdpBrowserProvider.admission.test.ts index d09c44aba0..c47e72c1bc 100644 --- a/src/browser/CdpBrowserProvider.admission.test.ts +++ b/src/browser/CdpBrowserProvider.admission.test.ts @@ -1,9 +1,18 @@ import { expect, it } from "vitest"; +import { createServer, type ServerResponse } from "node:http"; import { createBrowserObservationProvider } from "../composition/browserObservation.js"; import { projectAnalysisError } from "../domain/analysisErrorProjection.js"; import { analyzeWebBundleInputSchema } from "../domain/webBundleAnalysis.js"; import { WebBundleAnalysisAdmission } from "./IsolatedWebBundleAnalyzer.js"; +import { CdpBrowserProvider } from "./CdpBrowserProvider.js"; +import { + inspectWebPageInputSchema, + listBrowserTargetsInputSchema, +} from "../domain/browserObservation.js"; +import { observeWebSessionInputSchema } from "../domain/browserSession.js"; +import { discoverWebMcpToolsInputSchema } from "../domain/webMcpDiscovery.js"; +import { captureWebScreenshotInputSchema } from "../domain/webScreenshot.js"; const input = analyzeWebBundleInputSchema.parse({ cdp_endpoint: "http://127.0.0.1:9", @@ -65,3 +74,103 @@ it("does not share a capture limit across browser providers", async () => { assertSettled(settled); } }); + +it.each(["complete", "cancelled"] as const)( + "drains admitted HTTP discovery (%s) and refuses new work once provider shutdown starts", + async (outcome) => { + let requests = 0; + let held: ServerResponse | undefined; + let observed!: () => void; + const seen = new Promise((resolve) => { + observed = resolve; + }); + const server = createServer((request, response) => { + requests += 1; + if (request.url === "/json/list" && held === undefined) { + held = response; + observed(); + } else if (request.url === "/json/list") response.end("[]"); + else + response.end( + JSON.stringify({ + Browser: "fixture", + "Protocol-Version": "1.3", + "User-Agent": "fixture", + "V8-Version": "fixture", + "WebKit-Version": "fixture", + webSocketDebuggerUrl: + endpoint.replace("http:", "ws:") + "/devtools/browser/fixture", + }), + ); + }); + await new Promise((resolve) => + server.listen(0, "127.0.0.1", resolve), + ); + const address = server.address(); + if (address === null || typeof address === "string") + throw new Error("Missing fixture endpoint"); + const endpoint = `http://127.0.0.1:${address.port}`; + const provider = new CdpBrowserProvider(); + const selected = listBrowserTargetsInputSchema.parse({ + cdp_endpoint: endpoint, + }); + const target = { cdp_endpoint: endpoint, target_id: "page-1" }; + const lateRequests = [ + () => provider.listTargets(selected), + () => provider.inspectPage(inspectWebPageInputSchema.parse(target)), + () => provider.analyzeBundle(analyzeWebBundleInputSchema.parse(target)), + () => provider.observeSession(observeWebSessionInputSchema.parse(target)), + () => + provider.discoverWebMcpTools( + discoverWebMcpToolsInputSchema.parse(target), + ), + () => + provider.captureScreenshot( + captureWebScreenshotInputSchema.parse(target), + ), + ]; + const controller = new AbortController(); + const reason = { caller: "cancel discovery during provider shutdown" }; + const pending = provider.listTargets(selected, { + signal: controller.signal, + }); + let closing: Promise | undefined; + try { + await seen; + let closed = false; + closing = provider.close().then(() => { + closed = true; + }); + await new Promise((resolve) => setImmediate(resolve)); + expect(closed).toBe(false); + for (const request of lateRequests) + expect((await request()).ok).toBe(false); + expect(requests).toBe(2); + if (outcome === "cancelled") controller.abort(reason); + else held?.end("[]"); + const result = await pending; + if (outcome === "complete") + expect(result).toMatchObject({ ok: true, value: { targets: [] } }); + else { + if (result.ok) throw new Error("Expected caller cancellation"); + expect(result.error).toMatchObject({ _tag: "AnalysisCancelledError" }); + expect(result.error.cause).toBe(reason); + } + await closing; + for (const request of lateRequests) + expect((await request()).ok).toBe(false); + expect(requests).toBe(2); + } finally { + held?.end("[]"); + await pending; + await closing; + await provider.close(); + server.closeAllConnections(); + await new Promise((resolve, reject) => + server.close((error) => + error === undefined ? resolve() : reject(error), + ), + ); + } + }, +); diff --git a/src/browser/CdpBrowserProvider.ts b/src/browser/CdpBrowserProvider.ts index b925063861..0b8bd9c48b 100644 --- a/src/browser/CdpBrowserProvider.ts +++ b/src/browser/CdpBrowserProvider.ts @@ -104,14 +104,27 @@ const CLEANUP_DOMAINS = [ export class CdpBrowserProvider implements BrowserObservationPort { readonly #owner = new CdpTargetSessionOwner(); readonly #bundleAnalyzer: IsolatedWebBundleAnalyzer; + #activeOperations = 0; + #closing = false; + #closePromise: Promise | undefined; + readonly #idleWaiters = new Set<() => void>(); constructor(admission?: WebBundleAnalysisAdmission) { this.#bundleAnalyzer = new IsolatedWebBundleAnalyzer(undefined, admission); } async close(): Promise { + this.#closing = true; + this.#closePromise ??= this.#closeResources().catch((cause: unknown) => { + this.#closePromise = undefined; + throw cause; + }); + return await this.#closePromise; + } + + async #closeResources(): Promise { const outcomes = await Promise.allSettled([ - this.#owner.close(), + this.#closeTransports(), this.#bundleAnalyzer.close(), ]); const failures = outcomes.flatMap((outcome) => @@ -134,6 +147,35 @@ export class CdpBrowserProvider implements BrowserObservationPort { } } + async #closeTransports(): Promise { + while (this.#activeOperations > 0) + await new Promise((resolve) => this.#idleWaiters.add(resolve)); + await this.#owner.close(); + } + + async #withAdmission( + operation: BrowserObservationOperation, + work: () => Promise>, + ): Promise> { + if (this.#closing) + return err( + new ProviderAdapterError(CDP_BROWSER_PROVIDER_IDENTITY.id, operation, { + diagnostics: { reason: "provider_closed" }, + userMessage: "Browser provider is closed to new observations.", + }), + ); + this.#activeOperations += 1; + try { + return await work(); + } finally { + this.#activeOperations -= 1; + if (this.#activeOperations === 0) { + for (const resolve of this.#idleWaiters) resolve(); + this.#idleWaiters.clear(); + } + } + } + identity(): ProviderIdentity { return CDP_BROWSER_PROVIDER_IDENTITY; } @@ -141,6 +183,15 @@ export class CdpBrowserProvider implements BrowserObservationPort { async listTargets( input: ListBrowserTargetsInput, options: ExecutionOptions = {}, + ): Promise> { + return await this.#withAdmission("list_browser_targets", () => + this.#listTargets(input, options), + ); + } + + async #listTargets( + input: ListBrowserTargetsInput, + options: ExecutionOptions, ): Promise> { try { const endpoint = await openCdpEndpoint( @@ -195,6 +246,15 @@ export class CdpBrowserProvider implements BrowserObservationPort { async inspectPage( input: InspectWebPageInput, options: ExecutionOptions = {}, + ): Promise> { + return await this.#withAdmission("inspect_web_page", () => + this.#inspectPage(input, options), + ); + } + + async #inspectPage( + input: InspectWebPageInput, + options: ExecutionOptions, ): Promise> { try { const captured = await this.#capture(input, options, "inspect_web_page"); @@ -207,6 +267,15 @@ export class CdpBrowserProvider implements BrowserObservationPort { async analyzeBundle( input: AnalyzeWebBundleInput, options: ExecutionOptions = {}, + ): Promise> { + return await this.#withAdmission("analyze_web_bundle", () => + this.#analyzeBundle(input, options), + ); + } + + async #analyzeBundle( + input: AnalyzeWebBundleInput, + options: ExecutionOptions, ): Promise> { const admission = await this.#bundleAnalyzer.reserve(); if (!admission.ok) return admission; @@ -282,6 +351,15 @@ export class CdpBrowserProvider implements BrowserObservationPort { async observeSession( input: ObserveWebSessionInput, options: ExecutionOptions = {}, + ): Promise> { + return await this.#withAdmission("observe_web_session", () => + this.#observeSession(input, options), + ); + } + + async #observeSession( + input: ObserveWebSessionInput, + options: ExecutionOptions, ): Promise> { try { return ok( @@ -303,6 +381,15 @@ export class CdpBrowserProvider implements BrowserObservationPort { async discoverWebMcpTools( input: DiscoverWebMcpToolsInput, options: ExecutionOptions = {}, + ): Promise> { + return await this.#withAdmission("discover_webmcp_tools", () => + this.#discoverWebMcpTools(input, options), + ); + } + + async #discoverWebMcpTools( + input: DiscoverWebMcpToolsInput, + options: ExecutionOptions, ): Promise> { try { return ok( @@ -336,6 +423,15 @@ export class CdpBrowserProvider implements BrowserObservationPort { async captureScreenshot( input: CaptureWebScreenshotInput, options: ExecutionOptions = {}, + ): Promise> { + return await this.#withAdmission("capture_web_screenshot", () => + this.#captureScreenshot(input, options), + ); + } + + async #captureScreenshot( + input: CaptureWebScreenshotInput, + options: ExecutionOptions, ): Promise> { try { return ok(