fix(browser): harden waitFor event handling

packages/bcode-browser/skills/browser-execute/SKILL.md

@@ -115,8 +115,10 @@ Common moves:
 ```js
 // Navigate.
 await session.Page.enable()
-await session.Page.navigate({ url: "https://example.com" })
-await session.waitFor("Page.loadEventFired")
+await Promise.all([
+  session.waitFor("Page.loadEventFired", { timeoutMs: 15_000 }),
+  session.Page.navigate({ url: "https://example.com" }),
+])
 
 // Evaluate JS in the page.
 const r = await session.Runtime.evaluate({
@@ -153,8 +155,10 @@ export async function scrapeTitles(session: any, urls: string[]) {
   const titles: string[] = []
   await session.Page.enable()
   for (const url of urls) {
-    await session.Page.navigate({ url })
-    await session.waitFor("Page.loadEventFired")
+    await Promise.all([
+      session.waitFor("Page.loadEventFired", { timeoutMs: 15_000 }),
+      session.Page.navigate({ url }),
+    ])
     const r = await session.Runtime.evaluate({ expression: "document.title", returnByValue: true })
     titles.push(r.result.value)
   }

packages/bcode-browser/src/cdp/session.ts

@@ -25,6 +25,13 @@ export type ConnectOptions = {
   timeoutMs?: number;
 };
 
+export type WaitForOptions<T> = {
+  /** Only resolve when the event payload matches this predicate. */
+  predicate?: (params: T) => boolean;
+  /** Maximum wait in ms. Default 30000. */
+  timeoutMs?: number;
+};
+
 /** A Chromium-based browser detected as running on this machine. */
 export type DetectedBrowser = {
   /** Short label, e.g. 'Google Chrome', 'Brave', 'Comet'. */
@@ -189,15 +196,47 @@ export class Session implements Transport {
   }
 
   /** Wait for the next event matching `method` (and optional predicate). */
-  waitFor<T = unknown>(method: string, predicate?: (params: T) => boolean, timeoutMs = 30_000): Promise<T> {
+  waitFor<T = unknown>(method: string, options?: WaitForOptions<T>): Promise<T>;
+  waitFor<T = unknown>(method: string, predicate?: (params: T) => boolean, timeoutMs?: number): Promise<T>;
+  waitFor<T = unknown>(
+    method: string,
+    predicateOrOptions?: ((params: T) => boolean) | WaitForOptions<T>,
+    positionalTimeoutMs?: number,
+  ): Promise<T> {
+    if (
+      predicateOrOptions !== undefined &&
+      typeof predicateOrOptions !== 'function' &&
+      (typeof predicateOrOptions !== 'object' || predicateOrOptions === null || Array.isArray(predicateOrOptions))
+    ) {
+      throw new TypeError('waitFor expects a predicate function or an options object');
+    }
+    const options = typeof predicateOrOptions === 'object' ? predicateOrOptions : undefined;
+    if (options?.predicate !== undefined && typeof options.predicate !== 'function') {
+      throw new TypeError('waitFor options.predicate must be a function');
+    }
+    const predicate = typeof predicateOrOptions === 'function' ? predicateOrOptions : options?.predicate;
+    const timeoutMs = options?.timeoutMs ?? positionalTimeoutMs ?? 30_000;
+    const sessionId = isBrowserLevel(method) ? undefined : this.activeSessionId;
+    if (!Number.isFinite(timeoutMs) || timeoutMs < 0) {
+      throw new TypeError('waitFor timeoutMs must be a non-negative finite number');
+    }
+
     return new Promise((resolve, reject) => {
       const timer = setTimeout(() => {
         unsub();
         reject(new Error(`Timeout waiting for ${method}`));
       }, timeoutMs);
-      const unsub = this.onEvent((m, params) => {
+      const unsub = this.onEvent((m, params, eventSessionId) => {
         if (m !== method) return;
-        if (predicate && !predicate(params as T)) return;
+        if (sessionId !== undefined && eventSessionId !== sessionId) return;
+        try {
+          if (predicate && !predicate(params as T)) return;
+        } catch (error) {
+          clearTimeout(timer);
+          unsub();
+          reject(error);
+          return;
+        }
         clearTimeout(timer);
         unsub();
         resolve(params as T);
@@ -423,4 +462,3 @@ async function tryReadDevToolsActivePort(
     return undefined;
   }
 }
-

packages/bcode-browser/test/browser-execute.test.ts

@@ -87,8 +87,10 @@ test.skipIf(!enabled)("workspace import inside a snippet", async () => {
          await session.use(page.targetId)
        }
        await session.Page.enable()
-       await session.Page.navigate({ url: "data:text/html,<title>bcode-be</title>" })
-       await session.waitFor("Page.loadEventFired", undefined, 5000)
+       await Promise.all([
+         session.waitFor("Page.loadEventFired", { timeoutMs: 5000 }),
+         session.Page.navigate({ url: "data:text/html,<title>bcode-be</title>" }),
+       ])
        const r = await session.Runtime.evaluate({ expression: "document.title", returnByValue: true })
        return r.result.value
      }`,
@@ -125,8 +127,10 @@ test.skipIf(!enabled)("Page.captureScreenshot is collected into result.screensho
           {
             description: "Capture two screenshots",
             code: `await session.Page.enable();
-                   await session.Page.navigate({ url: "data:text/html,<title>shot</title><body>hi" });
-                   await session.waitFor("Page.loadEventFired", undefined, 5000);
+                   await Promise.all([
+                     session.waitFor("Page.loadEventFired", { timeoutMs: 5000 }),
+                     session.Page.navigate({ url: "data:text/html,<title>shot</title><body>hi" }),
+                   ]);
                    const a = await session.Page.captureScreenshot({ format: "png" });
                    const b = await session.Page.captureScreenshot({ format: "jpeg", quality: 50 });
                    return { aLen: a.data.length, bLen: b.data.length };`,

packages/bcode-browser/test/cdp-session.test.ts

@@ -0,0 +1,107 @@
+import { afterAll, beforeAll, expect, test } from "bun:test"
+import { Session } from "../src/cdp/session"
+
+const channel = "cdp-events"
+const server = Bun.serve({
+  port: 0,
+  fetch(req, srv) {
+    return srv.upgrade(req) ? undefined : new Response("nope", { status: 400 })
+  },
+  websocket: {
+    open(ws) {
+      ws.subscribe(channel)
+    },
+    message(ws, raw) {
+      const message: unknown = JSON.parse(String(raw))
+      if (typeof message !== "object" || message === null) return
+      const method = Reflect.get(message, "method")
+      const id = Reflect.get(message, "id")
+      if (method !== "Page.navigate" || typeof id !== "number") return
+      ws.send(JSON.stringify({ method: "Page.loadEventFired", params: { timestamp: 1 } }))
+      ws.send(JSON.stringify({ id, result: { frameId: "frame" } }))
+    },
+    close() {},
+  },
+})
+const session = new Session()
+
+beforeAll(async () => {
+  await session.connect({ wsUrl: `ws://127.0.0.1:${server.port}/` })
+})
+
+afterAll(() => {
+  session.close()
+  server.stop(true)
+})
+
+const emit = (method: string, params: unknown, sessionId?: string) => {
+  server.publish(channel, JSON.stringify({ method, params, sessionId }))
+}
+
+test("waitFor accepts predicate and timeout options", async () => {
+  const waiting = session.waitFor<{ ready: boolean }>("Test.options", {
+    predicate: (params) => params.ready,
+    timeoutMs: 1_000,
+  })
+  emit("Test.options", { ready: false })
+  emit("Test.options", { ready: true })
+  expect(await waiting).toEqual({ ready: true })
+})
+
+test("waitFor options timeout is honored", async () => {
+  const started = performance.now()
+  await expect(session.waitFor("Test.timeout", { timeoutMs: 20 })).rejects.toThrow("Timeout waiting for Test.timeout")
+  expect(performance.now() - started).toBeLessThan(500)
+})
+
+test("waitFor rejects and unsubscribes when a predicate throws", async () => {
+  let calls = 0
+  const waiting = session.waitFor("Test.predicate-error", {
+    predicate: () => {
+      calls++
+      throw new Error("predicate failed")
+    },
+    timeoutMs: 1_000,
+  })
+  emit("Test.predicate-error", {})
+  await expect(waiting).rejects.toThrow("predicate failed")
+  emit("Test.predicate-error", {})
+  await Bun.sleep(10)
+  expect(calls).toBe(1)
+})
+
+test("waitFor retains the positional signature", async () => {
+  const waiting = session.waitFor<{ ready: boolean }>("Test.positional", (params) => params.ready, 1_000)
+  emit("Test.positional", { ready: true })
+  expect(await waiting).toEqual({ ready: true })
+})
+
+test("a waiter registered before navigation catches an event emitted before the navigation response", async () => {
+  const [loaded] = await Promise.all([
+    session.waitFor<{ timestamp: number }>("Page.loadEventFired", { timeoutMs: 1_000 }),
+    session.domains.Page.navigate({ url: "https://example.com" }),
+  ])
+  expect(loaded).toEqual({ timestamp: 1 })
+})
+
+test("waitFor ignores matching events from another attached session", async () => {
+  session.setActiveSession("session-active")
+  try {
+    const waiting = session.waitFor<{ source: string }>("Page.loadEventFired", { timeoutMs: 1_000 })
+    emit("Page.loadEventFired", { source: "background" }, "session-background")
+    emit("Page.loadEventFired", { source: "active" }, "session-active")
+    expect(await waiting).toEqual({ source: "active" })
+  } finally {
+    session.setActiveSession(undefined)
+  }
+})
+
+test("waitFor rejects invalid runtime arguments immediately", () => {
+  expect(() =>
+    // @ts-expect-error Runtime callers can still pass invalid JavaScript.
+    session.waitFor("Test.invalid-predicate", { predicate: "not a function" }),
+  ).toThrow("waitFor options.predicate must be a function")
+  expect(() =>
+    session.waitFor("Test.invalid-timeout", { timeoutMs: Number.NaN }),
+  ).toThrow("waitFor timeoutMs must be a non-negative finite number")
+})

packages/bcode-browser/test/cdp-smoke.test.ts

@@ -32,8 +32,10 @@ test.skipIf(!enabled)("Session connects, navigates, reads title", async () => {
     }
 
     await session.domains.Page.enable()
-    await session.domains.Page.navigate({ url: "data:text/html,<title>bcode-smoke</title>" })
-    await session.waitFor("Page.loadEventFired", undefined, 5000)
+    await Promise.all([
+      session.waitFor("Page.loadEventFired", { timeoutMs: 5000 }),
+      session.domains.Page.navigate({ url: "data:text/html,<title>bcode-smoke</title>" }),
+    ])
 
     const r = (await session.domains.Runtime.evaluate({
       expression: "document.title",