fix(browser): bound CDP handshakes and command responses (#3555)

* fix(browser): bound CDP handshakes and command responses

* test(browser): expire stalled deadlines after fixture readiness

* docs(browser): explain handshake deadline ownership

docs/browser-observation.md

@@ -343,8 +343,10 @@ an invalid supplied size remains unknown with incomplete resource coverage while
 the resource's other facts are retained. Zero and nonnegative fractional sizes
 are preserved.
 
-CDP discovery, connection, and commands wait for the peer or caller cancellation;
-a stalled peer can leave an operation waiting. WebSocket messages remain subject
+CDP discovery has a 10-second deadline. Each WebSocket handshake and command
+response has its own 20-second deadline, separate from the caller-selected
+observation window. A stalled handshake or command reports `provider_timeout`;
+caller cancellation still reports cancellation. WebSocket messages remain subject
 to the [64 MiB limit](#request-boundary). REA does not impose a message-count or
 pending-command quota. On cancellation, it closes only its own WebSocket,
 waiting up to one second before terminating it. Malformed unsolicited events

docs/javascript-runtime-observation.md

@@ -95,6 +95,11 @@ pauses or resumes execution, calls `Runtime.runIfWaitingForDebugger`, launches
 the target, instruments JavaScript, or invokes Electron APIs. Closing a call
 closes only REA's WebSocket.
 
+Each WebSocket handshake and command response has its own 20-second deadline,
+separate from the caller-selected observation window. A stalled handshake or
+command reports `provider_timeout` and retains observations already received;
+caller cancellation still reports cancellation.
+
 The direct observations are:
 
 - `Debugger.scriptParsed`, establishing that a bounded script location was

src/browser/CdpConnection.ts

@@ -23,7 +23,7 @@ interface PendingCommand {
   readonly sessionId?: string;
   readonly resolve: (value: unknown) => void;
   readonly reject: (error: AnalysisError) => void;
-  readonly removeAbort: () => void;
+  readonly dispose: () => void;
 }
 
 /** Correlated JSON-RPC transport for one CDP target operation. */
@@ -79,6 +79,8 @@ export class CdpConnection {
         `CDP maxPayloadBytes must be a positive safe integer no greater than ${String(MAX_WEBSOCKET_PAYLOAD_BYTES)}.`,
       );
     const socket = new WebSocket(url, {
+      // waitForOpen owns the deadline, cancellation, cleanup, and typed errors;
+      // keep ws's timer disabled so handshake failures have one owner.
       handshakeTimeout: 0,
       maxPayload: maxPayloadBytes,
       perMessageDeflate: false,
@@ -115,7 +117,7 @@ export class CdpConnection {
     return () => this.#protocolFailureListeners.delete(listener);
   }
 
-  /** Execute one command, optionally within a flat target session. */
+  /** Bound each response separately from the caller-selected observation window. */
   async send(
     method: string,
     params: Readonly<Record<string, unknown>> = {},
@@ -130,41 +132,51 @@ export class CdpConnection {
       throw new AnalysisCancelledError(this.operation);
     const id = this.#nextId;
     this.#nextId += 1;
+    const message = JSON.stringify({
+      id,
+      method,
+      params,
+      ...(sessionId === undefined ? {} : { sessionId }),
+    });
     return await new Promise((resolve, reject) => {
       const onAbort = (): void => {
         const pending = this.#pending.get(id);
         if (pending === undefined) return;
-        pending.removeAbort();
-        this.#pending.delete(id);
+        this.#complete(id, pending);
         reject(new AnalysisCancelledError(this.operation));
       };
+      const timer = setTimeout(() => {
+        const pending = this.#pending.get(id);
+        if (pending === undefined) return;
+        this.#complete(id, pending);
+        reject(
+          new BrowserObservationError(this.operation, "timeout", {
+            detail: `CDP command ${method} exceeded its ${String(WEB_RUNTIME_LIMITS.commandTimeoutMs)} ms response deadline.`,
+          }),
+        );
+      }, WEB_RUNTIME_LIMITS.commandTimeoutMs);
       signal?.addEventListener("abort", onAbort, { once: true });
       this.#pending.set(id, {
         method,
         ...(sessionId === undefined ? {} : { sessionId }),
         resolve,
         reject,
-        removeAbort: () => signal?.removeEventListener("abort", onAbort),
-      });
-      this.socket.send(
-        JSON.stringify({
-          id,
-          method,
-          params,
-          ...(sessionId === undefined ? {} : { sessionId }),
-        }),
-        (error) => {
-          if (error === undefined || error === null) return;
-          const pending = this.#pending.get(id);
-          if (pending === undefined) return;
-          this.#complete(id, pending);
-          reject(
-            new BrowserObservationError(this.operation, "disconnected", {
-              cause: error,
-            }),
-          );
+        dispose: () => {
+          clearTimeout(timer);
+          signal?.removeEventListener("abort", onAbort);
         },
-      );
+      });
+      this.socket.send(message, (error) => {
+        if (error === undefined || error === null) return;
+        const pending = this.#pending.get(id);
+        if (pending === undefined) return;
+        this.#complete(id, pending);
+        reject(
+          new BrowserObservationError(this.operation, "disconnected", {
+            cause: error,
+          }),
+        );
+      });
     });
   }
 
@@ -282,7 +294,7 @@ export class CdpConnection {
   }
 
   #complete(id: number, pending: PendingCommand): void {
-    pending.removeAbort();
+    pending.dispose();
     this.#pending.delete(id);
   }
 
@@ -370,11 +382,21 @@ const waitForOpen = async (
       onFailure(new AnalysisCancelledError(operation));
     };
     const cleanup = (): void => {
+      clearTimeout(timer);
       socket.off("open", onOpen);
       socket.off("error", onFailure);
       socket.off("close", onClose);
       signal?.removeEventListener("abort", onAbort);
     };
+    const timer = setTimeout(() => {
+      terminateSocket(socket);
+      cleanup();
+      reject(
+        new BrowserObservationError(operation, "timeout", {
+          detail: `CDP WebSocket handshake exceeded its ${String(WEB_RUNTIME_LIMITS.commandTimeoutMs)} ms deadline.`,
+        }),
+      );
+    }, WEB_RUNTIME_LIMITS.commandTimeoutMs);
     socket.once("open", onOpen);
     socket.once("error", onFailure);
     socket.once("close", onClose);

tests/boundary/browser/cdpTargetDeadlines.test.ts

@@ -0,0 +1,277 @@
+import { createServer, type Socket } from "node:net";
+
+import { afterEach, beforeEach, expect, it, vi } from "vitest";
+
+import { CdpBrowserProvider } from "../../../src/browser/CdpBrowserProvider.js";
+import { CdpConnection } from "../../../src/browser/CdpConnection.js";
+import { projectAnalysisError } from "../../../src/domain/analysisErrorProjection.js";
+import { inspectWebPageInputSchema } from "../../../src/domain/browserObservation.js";
+import { WEB_RUNTIME_LIMITS } from "../../../src/domain/webRuntime.js";
+import { V8InspectorProvider } from "../../../src/inspector/V8InspectorProvider.js";
+import { startFakeCdpBrowser } from "../../fixtures/fakeCdpBrowser.js";
+import { startFakeV8Inspector } from "../../fixtures/inspector/fakeV8Inspector.js";
+
+// Keep real socket scheduling; expire only a timer whose owning phase is ready.
+beforeEach(() => {
+  vi.spyOn(globalThis, "setTimeout");
+  vi.spyOn(globalThis, "clearTimeout");
+});
+
+afterEach(() => vi.restoreAllMocks());
+
+const activeTimers = (milliseconds: number) => {
+  const scheduled = vi.mocked(globalThis.setTimeout);
+  const cleared = vi.mocked(globalThis.clearTimeout);
+  return scheduled.mock.calls.flatMap(([callback, delay], index) => {
+    const result = scheduled.mock.results[index];
+    if (
+      delay !== milliseconds ||
+      result?.type !== "return" ||
+      cleared.mock.calls.some(([timer]) => timer === result.value)
+    )
+      return [];
+    return [{ callback, timer: result.value }];
+  });
+};
+
+const expireTimer = (milliseconds: number): void => {
+  const timers = activeTimers(milliseconds);
+  expect(timers).toHaveLength(1);
+  const timer = timers[0];
+  if (timer === undefined) throw new Error("Expected one active timer");
+  try {
+    timer.callback();
+    if (milliseconds === WEB_RUNTIME_LIMITS.commandTimeoutMs)
+      expect(globalThis.clearTimeout).toHaveBeenCalledWith(timer.timer);
+  } finally {
+    // The observation callback resolves naturally without cancelling its timer.
+    clearTimeout(timer.timer);
+  }
+};
+
+it("bounds a stalled WebSocket handshake and releases its socket", async () => {
+  const sockets = new Set<Socket>();
+  const server = createServer((socket) => {
+    sockets.add(socket);
+    socket.on("close", () => sockets.delete(socket));
+    socket.resume();
+  });
+  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("Expected a TCP address");
+  try {
+    const rejection = expect(
+      CdpConnection.connect(
+        `ws://127.0.0.1:${String(address.port)}/target`,
+        "inspect_web_page",
+        AbortSignal.timeout(10_000),
+      ),
+    ).rejects.toMatchObject({ reason: "timeout" });
+    await vi.waitFor(() => expect(sockets.size).toBe(1), { timeout: 5_000 });
+    expireTimer(WEB_RUNTIME_LIMITS.commandTimeoutMs);
+    await rejection;
+    await vi.waitFor(() => expect(sockets.size).toBe(0));
+  } finally {
+    for (const socket of sockets) socket.destroy();
+    await new Promise<void>((resolve) => server.close(() => resolve()));
+  }
+});
+
+it.each(["Target.attachToTarget", "Page.enable", "Runtime.enable"])(
+  "times out %s after HTTP discovery and closes the target session",
+  async (hangOnMethod) => {
+    const browser = await startFakeCdpBrowser({ hangOnMethod });
+    try {
+      const pending = new CdpBrowserProvider().inspectPage(
+        inspectWebPageInputSchema.parse({
+          cdp_endpoint: browser.endpoint,
+          target_id: "allowed-page",
+          observation_ms: 0,
+        }),
+        { signal: AbortSignal.timeout(10_000) },
+      );
+      await vi.waitFor(
+        () =>
+          expect(browser.commands.map(({ method }) => method)).toContain(
+            hangOnMethod,
+          ),
+        { timeout: 5_000 },
+      );
+      expireTimer(WEB_RUNTIME_LIMITS.commandTimeoutMs);
+      const result = await pending;
+      if (result.ok) throw new Error("Expected a target command timeout");
+      expect(projectAnalysisError(result.error)).toMatchObject({
+        code: "provider_timeout",
+        category: "timeout",
+      });
+      expect(browser.commands.map(({ method }) => method)).toContain(
+        hangOnMethod,
+      );
+      await vi.waitFor(() => expect(browser.activeConnectionCount()).toBe(0));
+    } finally {
+      await browser.close();
+    }
+  },
+);
+
+it("ignores a late timed-out reply and keeps following commands correlated", async () => {
+  const browser = await startFakeCdpBrowser({ hangOnMethod: "Page.enable" });
+  const connection = await CdpConnection.connect(
+    browser.browserWebSocketUrl,
+    "inspect_web_page",
+  );
+  try {
+    const rejection = expect(
+      connection.send(
+        "Page.enable",
+        {},
+        undefined,
+        AbortSignal.timeout(10_000),
+      ),
+    ).rejects.toMatchObject({ reason: "timeout" });
+    await vi.waitFor(() => expect(browser.commands).toHaveLength(1), {
+      timeout: 5_000,
+    });
+    expireTimer(WEB_RUNTIME_LIMITS.commandTimeoutMs);
+    await rejection;
+    const command = browser.commands[0];
+    if (command === undefined)
+      throw new Error("Expected the timed-out command");
+    browser.emitRawMessage(JSON.stringify({ id: command.id, result: {} }));
+    await expect(connection.send("Runtime.enable")).resolves.toEqual({});
+  } finally {
+    await connection.close();
+    await browser.close();
+  }
+});
+
+it.each(["success", "rejection", "abort", "disconnect"])(
+  "releases handshake and command timers on %s",
+  async (outcome) => {
+    const cleared = vi.mocked(globalThis.clearTimeout);
+    const browser = await startFakeCdpBrowser({
+      ...(outcome === "abort" || outcome === "success"
+        ? { hangOnMethod: "Runtime.enable" }
+        : {}),
+      ...(outcome === "disconnect" ? { closeOnMethod: "Runtime.enable" } : {}),
+    });
+    const controller = new AbortController();
+    const removeAbort = vi.spyOn(controller.signal, "removeEventListener");
+    const connection = await CdpConnection.connect(
+      browser.browserWebSocketUrl,
+      "inspect_web_page",
+      controller.signal,
+    );
+    try {
+      const pending = connection.send(
+        outcome === "rejection" ? "Fixture.unmodeledMethod" : "Runtime.enable",
+        {},
+        undefined,
+        controller.signal,
+      );
+      const assertion =
+        outcome === "success"
+          ? expect(pending).resolves.toEqual({})
+          : expect(pending).rejects.toMatchObject({
+              _tag:
+                outcome === "abort"
+                  ? "AnalysisCancelledError"
+                  : "BrowserObservationError",
+            });
+      if (outcome === "success") {
+        await vi.waitFor(() => expect(browser.commands).toHaveLength(1), {
+          timeout: 5_000,
+        });
+        // A healthy reply may take longer than the former 100 ms test budget.
+        await new Promise<void>((resolve) => setTimeout(resolve, 250));
+        const command = browser.commands[0];
+        if (command === undefined)
+          throw new Error("Expected a pending command");
+        browser.emitRawMessage(JSON.stringify({ id: command.id, result: {} }));
+      }
+      if (outcome === "abort") controller.abort();
+      await assertion;
+      const scheduled = vi.mocked(globalThis.setTimeout);
+      const deadlines = scheduled.mock.calls.flatMap((call, index) =>
+        call[1] === WEB_RUNTIME_LIMITS.commandTimeoutMs
+          ? [scheduled.mock.results[index]?.value]
+          : [],
+      );
+      expect(deadlines).toHaveLength(2);
+      for (const timer of deadlines)
+        expect(cleared).toHaveBeenCalledWith(timer);
+      expect(removeAbort).toHaveBeenCalledTimes(2);
+    } finally {
+      await connection.close();
+      await browser.close();
+    }
+  },
+);
+
+it.each(["Runtime.enable", "Debugger.enable"])(
+  "times out Inspector %s and preserves observations received before expiry",
+  async (hangOnMethod) => {
+    const inspector = await startFakeV8Inspector({
+      targetUrl: "https://fixture.example/app.js",
+      hangOnMethod,
+    });
+    try {
+      const pending = new V8InspectorProvider().observe(
+        {
+          inspector_endpoint: inspector.endpoint,
+          target_id: inspector.targetId,
+          observation_ms: 0,
+        },
+        { signal: AbortSignal.timeout(10_000) },
+      );
+      await vi.waitFor(
+        () =>
+          expect(inspector.commands.map(({ method }) => method)).toContain(
+            hangOnMethod,
+          ),
+        { timeout: 5_000 },
+      );
+      expireTimer(WEB_RUNTIME_LIMITS.commandTimeoutMs);
+      const result = await pending;
+      if (result.ok) throw new Error("Expected an Inspector command timeout");
+      expect(projectAnalysisError(result.error)).toMatchObject({
+        code: "provider_timeout",
+        category: "timeout",
+      });
+      expect(result.error.partialObservation).toMatchObject({
+        capture: {
+          events_observed: hangOnMethod === "Debugger.enable" ? 1 : 0,
+        },
+      });
+    } finally {
+      await inspector.close();
+    }
+  },
+);
+
+it("keeps the chosen observation window independent of command deadlines", async () => {
+  const browser = await startFakeCdpBrowser();
+  const observationMs = WEB_RUNTIME_LIMITS.commandTimeoutMs + 250;
+  try {
+    const pending = new CdpBrowserProvider().inspectPage(
+      inspectWebPageInputSchema.parse({
+        cdp_endpoint: browser.endpoint,
+        target_id: "allowed-page",
+        observation_ms: observationMs,
+      }),
+    );
+    await vi.waitFor(
+      () => expect(activeTimers(observationMs)).toHaveLength(1),
+      {
+        timeout: 5_000,
+      },
+    );
+    expect(activeTimers(WEB_RUNTIME_LIMITS.commandTimeoutMs)).toHaveLength(0);
+    expireTimer(observationMs);
+    const result = await pending;
+    expect(result.ok).toBe(true);
+  } finally {
+    await browser.close();
+  }
+});

tests/fixtures/inspector/fakeV8Inspector.ts

@@ -28,6 +28,7 @@ interface FakeV8InspectorOptions {
   readonly additionalTargetUrl?: string;
   readonly additionalTargetCount?: number;
   readonly closeOnMethod?: string;
+  readonly hangOnMethod?: string;
 }
 
 export interface FakeV8Inspector {
@@ -147,6 +148,7 @@ const registerInspectorSocket = (
       socket.close();
       return;
     }
+    if (options.hangOnMethod === command.method) return;
     respondToInspectorCommand(socket, command, options);
     socket.send(JSON.stringify({ id: command.id, result: {} }));
   });