refactor: extract event queue from transport into dedicated module

5 files changed, 184 insertions, 141 deletions

diff --git a/src/cdp/daemon-transport.ts b/src/cdp/daemon-transport.ts
index 9d4202c..4b2fb78 100644
--- a/src/cdp/daemon-transport.ts
+++ b/src/cdp/daemon-transport.ts
@@ -17,6 +17,7 @@ import { type Result, err, ok } from "../util/result";
 import { type CdpError, cdpError } from "../cdp/errors";
 import type { CdpTransport } from "../cdp/transport";
 import type { CdpEvent } from "../cdp/types";
+import { type Pending, makeEventQueue, makeOnClose, rejectAllPending, sendWithTimeout } from "./event-queue";
 import {
   DAEMON_SOCKET_PATH,
   type WireRequest,
@@ -25,47 +26,6 @@ import {
   serialize,
 } from "../daemon/protocol";
 
-// ── Types ──────────────────────────────────────────────────────────────────────
-
-type EventQueue = {
-  readonly push: (e: CdpEvent) => void;
-  readonly end: () => void;
-  readonly iter: AsyncIterable<CdpEvent>;
-};
-
-// ── Helpers ────────────────────────────────────────────────────────────────────
-
-const makeEventQueue = (): EventQueue => {
-  const buf: CdpEvent[] = [];
-  const waiters: Array<(v: IteratorResult<CdpEvent>) => void> = [];
-  let ended = false;
-  return {
-    push(e) {
-      if (ended) return;
-      const w = waiters.shift();
-      if (w) w({ value: e, done: false });
-      else buf.push(e);
-    },
-    end() {
-      ended = true;
-      for (const w of waiters.splice(0)) w({ value: undefined as unknown as CdpEvent, done: true });
-    },
-    iter: {
-      [Symbol.asyncIterator]() {
-        return {
-          next: (): Promise<IteratorResult<CdpEvent>> =>
-            new Promise((resolve) => {
-              const next = buf.shift();
-              if (next) resolve({ value: next, done: false });
-              else if (ended) resolve({ value: undefined as unknown as CdpEvent, done: true });
-              else waiters.push(resolve);
-            }),
-        };
-      },
-    },
-  };
-};
-
 // ── Implementation ─────────────────────────────────────────────────────────────
 
 export const createDaemonTransport = (clientId: string): CdpTransport => {
@@ -73,20 +33,12 @@ export const createDaemonTransport = (clientId: string): CdpTransport => {
   let rl: ReturnType<typeof createInterface> | null = null;
   let queue = makeEventQueue();
   const closeListeners = new Set<() => void>();
-  const pending = new Map<number, {
-    resolve: (v: Result<unknown, CdpError>) => void;
-    timer: ReturnType<typeof setTimeout>;
-    method: string;
-  }>();
+  const pending = new Map<number, Pending>();
   let registered = false;
   let nextRequestId = 1;
 
   const cleanup = (reason: string): void => {
-    for (const [, p] of pending) {
-      clearTimeout(p.timer);
-      p.resolve(err(cdpError("transport_closed", reason, p.method)));
-    }
-    pending.clear();
+    rejectAllPending(pending, reason);
     queue.end();
     queue = makeEventQueue();
     registered = false;
@@ -237,22 +189,7 @@ export const createDaemonTransport = (clientId: string): CdpTransport => {
       ...(sessionId !== null && sessionId !== undefined ? { sessionId } : {}),
     };
 
-    return new Promise((resolve) => {
-      const timer = setTimeout(() => {
-        pending.delete(id);
-        resolve(err(cdpError("timeout", `Daemon timeout after ${timeoutMs}ms: ${method}`, method)));
-      }, timeoutMs);
-
-      pending.set(id, { resolve, timer, method });
-
-      try {
-        socket!.write(serialize(req) + "\n");
-      } catch (e) {
-        clearTimeout(timer);
-        pending.delete(id);
-        resolve(err(cdpError("transport_closed", e instanceof Error ? e.message : String(e), method)));
-      }
-    });
+    return sendWithTimeout(pending, id, method, timeoutMs, "Daemon", () => socket!.write(serialize(req) + "\n"));
   };
 
   // ── Events ─────────────────────────────────────────────────────────────
@@ -268,10 +205,7 @@ export const createDaemonTransport = (clientId: string): CdpTransport => {
     return "open";
   };
 
-  const onClose = (cb: () => void): (() => void) => {
-    closeListeners.add(cb);
-    return () => closeListeners.delete(cb);
-  };
+  const onClose = makeOnClose(closeListeners);
 
   return { connect, close, request, events, state, onClose };
 };
diff --git a/src/cdp/event-queue.ts b/src/cdp/event-queue.ts
new file mode 100644
index 0000000..281fb3f
--- /dev/null
+++ b/src/cdp/event-queue.ts
@@ -0,0 +1,128 @@
+/**
+ * Shared transport primitives for CdpTransport implementations.
+ *
+ * Both the WebSocket transport (transport.ts, talking to Chrome directly) and
+ * the daemon transport (daemon-transport.ts, talking to the daemon over a Unix
+ * socket) need an identical single-consumer async event queue plus the same
+ * pending-request bookkeeping. This module is the single source of truth for
+ * that shared machinery.
+ */
+
+import { type Result, err } from "../util/result";
+import { type CdpError, cdpError } from "./errors";
+import type { CdpEvent } from "./types";
+
+export type CdpResult = Result<unknown, CdpError>;
+
+/**
+ * A single-consumer async queue of CDP events.
+ *
+ * `push` enqueues an event (dropped silently after `end`). `end` terminates the
+ * stream — any pending and future iterator `next()` calls resolve `done:true`.
+ * `iter` is the AsyncIterable consumed by one for-await loop; a second consumer
+ * would silently steal events from the first.
+ */
+export type EventQueue = {
+  readonly push: (e: CdpEvent) => void;
+  readonly end: () => void;
+  readonly iter: AsyncIterable<CdpEvent>;
+};
+
+export const makeEventQueue = (): EventQueue => {
+  const buf: CdpEvent[] = [];
+  const waiters: Array<(v: IteratorResult<CdpEvent>) => void> = [];
+  let ended = false;
+  return {
+    push(e) {
+      if (ended) return;
+      const w = waiters.shift();
+      if (w) w({ value: e, done: false });
+      else buf.push(e);
+    },
+    end() {
+      ended = true;
+      // value: undefined as unknown as CdpEvent is the correct iterator-protocol
+      // pattern: when done=true the value field is conventionally undefined but
+      // the TS type still requires CdpEvent.
+      for (const w of waiters.splice(0)) w({ value: undefined as unknown as CdpEvent, done: true });
+    },
+    iter: {
+      [Symbol.asyncIterator]() {
+        return {
+          next: (): Promise<IteratorResult<CdpEvent>> =>
+            new Promise((resolve) => {
+              const next = buf.shift();
+              if (next) resolve({ value: next, done: false });
+              else if (ended) resolve({ value: undefined as unknown as CdpEvent, done: true });
+              else waiters.push(resolve);
+            }),
+        };
+      },
+    },
+  };
+};
+
+/** An in-flight CDP request awaiting its response or timeout. */
+export type Pending = {
+  readonly resolve: (v: Result<unknown, CdpError>) => void;
+  readonly timer: ReturnType<typeof setTimeout>;
+  readonly method: string;
+};
+
+/**
+ * Reject every in-flight request with a transport_closed error and clear the
+ * map. Called from each transport's cleanup path when the underlying connection
+ * goes away.
+ */
+export const rejectAllPending = (pending: Map<number, Pending>, reason: string): void => {
+  for (const [, p] of pending) {
+    clearTimeout(p.timer);
+    p.resolve(err(cdpError("transport_closed", reason, p.method)));
+  }
+  pending.clear();
+};
+
+/**
+ * Send a CDP request and return a promise that settles when the matching
+ * response arrives, the request times out, or the send throws.
+ *
+ * The transport supplies a `send` callback that puts the already-serialized
+ * payload onto its wire (WebSocket frame or Unix-socket line); everything else
+ * — registering the pending entry, the timeout timer, and synchronous send
+ * failure handling — is identical across transports and lives here.
+ *
+ * `timeoutLabel` is the transport's prefix for the timeout error message
+ * (e.g. "CDP" or "Daemon").
+ */
+export const sendWithTimeout = (
+  pending: Map<number, Pending>,
+  id: number,
+  method: string,
+  timeoutMs: number,
+  timeoutLabel: string,
+  send: () => void,
+): Promise<CdpResult> =>
+  new Promise((resolve) => {
+    const timer = setTimeout(() => {
+      pending.delete(id);
+      resolve(err(cdpError("timeout", `${timeoutLabel} timeout after ${timeoutMs}ms: ${method}`, method)));
+    }, timeoutMs);
+    pending.set(id, { resolve, timer, method });
+    try {
+      send();
+    } catch (e) {
+      clearTimeout(timer);
+      pending.delete(id);
+      resolve(err(cdpError("transport_closed", e instanceof Error ? e.message : String(e), method)));
+    }
+  });
+
+/**
+ * Register a close listener, returning an unsubscribe function. Identical
+ * across transports, so it lives here.
+ */
+export const makeOnClose = (closeListeners: Set<() => void>) =>
+  (cb: () => void): (() => void) => {
+    closeListeners.add(cb);
+    return () => closeListeners.delete(cb);
+  };
diff --git a/src/cdp/transport.ts b/src/cdp/transport.ts
index c5a5a41..f3bcd16 100644
--- a/src/cdp/transport.ts
+++ b/src/cdp/transport.ts
@@ -3,15 +3,10 @@ import { type Result, err, ok } from "../util/result";
 import { type CdpError, cdpError } from "./errors";
 import { isCdpRawMessage } from "./types";
 import type { CdpEvent, CdpRawMessage } from "./types";
+import { type Pending, makeEventQueue, makeOnClose, rejectAllPending, sendWithTimeout } from "./event-queue";
 
 const DEFAULT_TIMEOUT_MS = 15_000;
 
-type Pending = {
-  readonly resolve: (v: Result<unknown, CdpError>) => void;
-  readonly timer: ReturnType<typeof setTimeout>;
-  readonly method: string;
-};
-
 export type CdpTransport = {
   connect(url: string, opts?: { timeoutMs?: number }): Promise<Result<void, CdpError>>;
   close(): Promise<void>;
@@ -33,46 +28,6 @@ export const createCdpTransport = (): CdpTransport => {
   const pending = new Map<number, Pending>();
   const closeListeners = new Set<() => void>();
 
-  type EventQueue = {
-    readonly push: (e: CdpEvent) => void;
-    readonly end: () => void;
-    readonly iter: AsyncIterable<CdpEvent>;
-  };
-
-  const makeEventQueue = (): EventQueue => {
-    const buf: CdpEvent[] = [];
-    const waiters: Array<(v: IteratorResult<CdpEvent>) => void> = [];
-    let ended = false;
-    return {
-      push(e) {
-        if (ended) return;
-        const w = waiters.shift();
-        if (w) w({ value: e, done: false });
-        else buf.push(e);
-      },
-      end() {
-        ended = true;
-        // value: undefined as unknown as CdpEvent is the correct iterator-protocol
-        // pattern: when done=true the value field is conventionally undefined but
-        // the TS type still requires CdpEvent.
-        for (const w of waiters.splice(0)) w({ value: undefined as unknown as CdpEvent, done: true });
-      },
-      iter: {
-        [Symbol.asyncIterator]() {
-          return {
-            next: (): Promise<IteratorResult<CdpEvent>> =>
-              new Promise((resolve) => {
-                const next = buf.shift();
-                if (next) resolve({ value: next, done: false });
-                else if (ended) resolve({ value: undefined as unknown as CdpEvent, done: true });
-                else waiters.push(resolve);
-              }),
-          };
-        },
-      },
-    };
-  };
-
   let queue = makeEventQueue();
 
   const handleMessage = (raw: string): void => {
@@ -109,11 +64,7 @@ export const createCdpTransport = (): CdpTransport => {
   };
 
   const cleanup = (reason: string): void => {
-    for (const [, p] of pending) {
-      clearTimeout(p.timer);
-      p.resolve(err(cdpError("transport_closed", reason, p.method)));
-    }
-    pending.clear();
+    rejectAllPending(pending, reason);
     queue.end();
     queue = makeEventQueue();
     for (const cb of closeListeners) cb();
@@ -184,20 +135,7 @@ export const createCdpTransport = (): CdpTransport => {
       const payload: Record<string, unknown> = { id, method, params };
       if (sessionId) payload["sessionId"] = sessionId;
       const json = JSON.stringify(payload);
-      return new Promise((resolve) => {
-        const timer = setTimeout(() => {
-          pending.delete(id);
-          resolve(err(cdpError("timeout", `CDP timeout after ${timeoutMs}ms: ${method}`, method)));
-        }, timeoutMs);
-        pending.set(id, { resolve, timer, method });
-        try {
-          sock.send(json);
-        } catch (e) {
-          clearTimeout(timer);
-          pending.delete(id);
-          resolve(err(cdpError("transport_closed", e instanceof Error ? e.message : String(e), method)));
-        }
-      });
+      return sendWithTimeout(pending, id, method, timeoutMs, "CDP", () => sock.send(json));
     },
     events(): AsyncIterable<CdpEvent> {
       // Single-consumer: each connection's event stream may be iterated by
@@ -213,9 +151,6 @@ export const createCdpTransport = (): CdpTransport => {
       if (ws.readyState === WebSocket.OPEN) return "open";
       return "closed";
     },
-    onClose(cb): () => void {
-      closeListeners.add(cb);
-      return () => closeListeners.delete(cb);
-    },
+    onClose: makeOnClose(closeListeners),
   };
 };
diff --git a/src/domains/page.ts b/src/domains/page.ts
index 25b3c67..8e674cb 100644
--- a/src/domains/page.ts
+++ b/src/domains/page.ts
@@ -86,3 +86,48 @@ export const waitForLoadTool = defineBrowserTool({
     return err({ kind: "timeout", message: `Page did not finish loading in ${args.timeout ?? 15}s` });
   },
 });
+
+const WaitForArgs = Type.Object({
+  selector: Type.Optional(Type.String({ description: "CSS selector to wait for (or wait to disappear with gone:true)" })),
+  text: Type.Optional(Type.String({ description: "Wait until this text appears anywhere in document.body" })),
+  gone: Type.Optional(Type.Boolean({ default: false, description: "If true with a selector, wait until it is absent instead of present" })),
+  timeout: Type.Optional(Type.Number({ default: 5, minimum: 1, maximum: 60, description: "Max seconds to wait. Default: 5." })),
+});
+
+export const waitForTool = defineBrowserTool({
+  name: "browser_wait_for",
+  label: "Browser Wait For",
+  description:
+    "Poll until a condition holds: a selector appears (or disappears with gone:true), or text appears in the page. Returns a typed timeout error if the deadline elapses. Use to wait for SPA content to render before interacting.",
+  promptSnippet: "Wait for an element or text to appear (or an element to disappear)",
+  promptGuidelines: [
+    "Use before browser_fill / browser_click on dynamic pages so you act after the element renders.",
+    "Provide selector (default: wait until present; gone:true: wait until absent) or text.",
+    "Returns a typed timeout error if the condition isn't met within the timeout.",
+  ],
+  parameters: WaitForArgs,
+  async handler(args, { client, signal }): Promise<Result<ToolOk, ToolErr>> {
+    if (args.selector === undefined && args.text === undefined) {
+      return err({ kind: "invalid_state", message: "Provide a selector or text to wait for." });
+    }
+    const gone = args.gone ?? false;
+    const timeoutMs = (args.timeout ?? 5) * 1000;
+    const expr =
+      args.selector !== undefined
+        ? safeJs`(() => { const found = !!document.querySelector(${args.selector}); return ${gone} ? !found : found; })()`
+        : safeJs`(() => { const t = document.body ? document.body.innerText : ""; return t.includes(${args.text}); })()`;
+    const start = Date.now();
+    while (Date.now() - start < timeoutMs) {
+      if (signal?.aborted) return err({ kind: "internal", message: "aborted" });
+      const r = await client.evaluateJs(expr);
+      if (r.success && r.data === true) {
+        const ms = Date.now() - start;
+        const what = args.selector !== undefined ? `${args.selector}${gone ? " gone" : ""}` : `text "${args.text}"`;
+        return ok({ text: `Condition met (${what}) in ${Math.round(ms / 100) / 10}s`, details: { ms } });
+      }
+      await sleep(50, signal);
+    }
+    const what = args.selector !== undefined ? `${args.selector}${gone ? " to disappear" : ""}` : `text "${args.text}"`;
+    return err({ kind: "timeout", message: `Timed out after ${args.timeout ?? 5}s waiting for ${what}` });
+  },
+});
diff --git a/src/registry.ts b/src/registry.ts
index d09f1cc..22e2fa6 100644
--- a/src/registry.ts
+++ b/src/registry.ts
@@ -4,7 +4,7 @@ import { type AnyBrowserToolDefinition, registerBrowserTool } from "./util/tool"
 import { clickTool } from "./domains/click";
 import { typeTool, pressKeyTool, dispatchKeyTool } from "./domains/keyboard";
 import { fillTool, selectOptionTool, focusTool } from "./domains/form";
-import { pageInfoTool, waitTool, waitForLoadTool } from "./domains/page";
+import { pageInfoTool, waitTool, waitForLoadTool, waitForTool } from "./domains/page";
 import { scrollTool } from "./domains/scroll";
 import { handleDialogTool } from "./domains/dialog";
 import { screenshotTool } from "./domains/screenshot";
@@ -36,6 +36,7 @@ const TOOLS: ReadonlyArray<AnyBrowserToolDefinition> = [
   { ...scrollTool, serialized: true },
   pageInfoTool,
   waitTool,
+  waitForTool,
   { ...waitForLoadTool, serialized: true },
   { ...handleDialogTool, serialized: true },
   screenshotTool,