fix(browser): drain admitted operations before closing provider (#10033)

fix(browser): drain complete CDP operations during shutdown

Co-authored-by: Cursor Agent <cursoragent@cursor.com>

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<void>((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<void>((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<void> | undefined;
+    try {
+      await seen;
+      let closed = false;
+      closing = provider.close().then(() => {
+        closed = true;
+      });
+      await new Promise<void>((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<void>((resolve, reject) =>
+        server.close((error) =>
+          error === undefined ? resolve() : reject(error),
+        ),
+      );
+    }
+  },
+);

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<void> | undefined;
+  readonly #idleWaiters = new Set<() => void>();
 
   constructor(admission?: WebBundleAnalysisAdmission) {
     this.#bundleAnalyzer = new IsolatedWebBundleAnalyzer(undefined, admission);
   }
 
   async close(): Promise<void> {
+    this.#closing = true;
+    this.#closePromise ??= this.#closeResources().catch((cause: unknown) => {
+      this.#closePromise = undefined;
+      throw cause;
+    });
+    return await this.#closePromise;
+  }
+
+  async #closeResources(): Promise<void> {
     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<void> {
+    while (this.#activeOperations > 0)
+      await new Promise<void>((resolve) => this.#idleWaiters.add(resolve));
+    await this.#owner.close();
+  }
+
+  async #withAdmission<Value>(
+    operation: BrowserObservationOperation,
+    work: () => Promise<Result<Value, AnalysisError>>,
+  ): Promise<Result<Value, AnalysisError>> {
+    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<Result<BrowserTargetList, AnalysisError>> {
+    return await this.#withAdmission("list_browser_targets", () =>
+      this.#listTargets(input, options),
+    );
+  }
+
+  async #listTargets(
+    input: ListBrowserTargetsInput,
+    options: ExecutionOptions,
   ): Promise<Result<BrowserTargetList, AnalysisError>> {
     try {
       const endpoint = await openCdpEndpoint(
@@ -195,6 +246,15 @@ export class CdpBrowserProvider implements BrowserObservationPort {
   async inspectPage(
     input: InspectWebPageInput,
     options: ExecutionOptions = {},
+  ): Promise<Result<WebPageInspection, AnalysisError>> {
+    return await this.#withAdmission("inspect_web_page", () =>
+      this.#inspectPage(input, options),
+    );
+  }
+
+  async #inspectPage(
+    input: InspectWebPageInput,
+    options: ExecutionOptions,
   ): Promise<Result<WebPageInspection, AnalysisError>> {
     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<Result<WebBundleAnalysis, AnalysisError>> {
+    return await this.#withAdmission("analyze_web_bundle", () =>
+      this.#analyzeBundle(input, options),
+    );
+  }
+
+  async #analyzeBundle(
+    input: AnalyzeWebBundleInput,
+    options: ExecutionOptions,
   ): Promise<Result<WebBundleAnalysis, AnalysisError>> {
     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<Result<WebObservationSession, AnalysisError>> {
+    return await this.#withAdmission("observe_web_session", () =>
+      this.#observeSession(input, options),
+    );
+  }
+
+  async #observeSession(
+    input: ObserveWebSessionInput,
+    options: ExecutionOptions,
   ): Promise<Result<WebObservationSession, AnalysisError>> {
     try {
       return ok(
@@ -303,6 +381,15 @@ export class CdpBrowserProvider implements BrowserObservationPort {
   async discoverWebMcpTools(
     input: DiscoverWebMcpToolsInput,
     options: ExecutionOptions = {},
+  ): Promise<Result<WebMcpDiscovery, AnalysisError>> {
+    return await this.#withAdmission("discover_webmcp_tools", () =>
+      this.#discoverWebMcpTools(input, options),
+    );
+  }
+
+  async #discoverWebMcpTools(
+    input: DiscoverWebMcpToolsInput,
+    options: ExecutionOptions,
   ): Promise<Result<WebMcpDiscovery, AnalysisError>> {
     try {
       return ok(
@@ -336,6 +423,15 @@ export class CdpBrowserProvider implements BrowserObservationPort {
   async captureScreenshot(
     input: CaptureWebScreenshotInput,
     options: ExecutionOptions = {},
+  ): Promise<Result<WebScreenshot, AnalysisError>> {
+    return await this.#withAdmission("capture_web_screenshot", () =>
+      this.#captureScreenshot(input, options),
+    );
+  }
+
+  async #captureScreenshot(
+    input: CaptureWebScreenshotInput,
+    options: ExecutionOptions,
   ): Promise<Result<WebScreenshot, AnalysisError>> {
     try {
       return ok(