refactor: one CDP transport, not two createCdpTransport was a second full WebSocket-to-Chrome implementation reachable only through client.ts's `?? createCdpTransport()` fallback, which production never took — every session and every automated test passes the daemon transport. transport.ts now holds only the CdpTransport interface, and the client requires a transport instead of inventing one. The "Session with given id not found" match that both transports carried becomes classifyRemoteError in cdp/errors.ts, next to the error kind it produces. test/manual/profile-concurrency-test.ts was the sole caller of the fallback: it drives its own Chrome on a private port. It now spawns a daemon per scenario with BU_CDP_WS set, and refuses to run when a daemon is already up, since that one is bound to a different browser. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
amankumarsingh77 amankumarsingh7702@gmail.com
Tue, 28 Jul 2026 21:50:38 +0530
5 files changed,
50 insertions(+),
147 deletions(-)
M
src/cdp/daemon-transport.ts
→
src/cdp/daemon-transport.ts
@@ -1,8 +1,8 @@
import { createConnection, type Socket } from "node:net"; import { createInterface } from "node:readline"; import { type Result, err, ok } from "../util/result"; -import { type CdpError, cdpError } from "../cdp/errors"; -import type { CdpTransport } from "../cdp/transport"; +import { type CdpError, cdpError, classifyRemoteError } from "../cdp/errors"; +import { DEFAULT_TIMEOUT_MS, type CdpTransport } from "../cdp/transport"; import type { CdpEvent } from "../cdp/types"; import { type Pending, makeEventQueue, makeOnClose, rejectAllPending, sendWithTimeout } from "./event-queue"; import {@@ -91,13 +91,7 @@ pending.delete(msg.id);
clearTimeout(p.timer); if (msg.error) { - p.resolve(err(cdpError( - msg.error.message.includes("Session with given id not found") - ? "session_not_found" - : "remote_error", - msg.error.message, - p.method, - ))); + p.resolve(err(cdpError(classifyRemoteError(msg.error.message), msg.error.message, p.method))); } else { p.resolve(ok(msg.result)); }@@ -149,7 +143,7 @@ method: string,
params: Record<string, unknown>, opts?: { sessionId?: string | null; timeoutMs?: number }, ): Promise<Result<unknown, CdpError>> => { - const timeoutMs = opts?.timeoutMs ?? 15_000; + const timeoutMs = opts?.timeoutMs ?? DEFAULT_TIMEOUT_MS; const sessionId = opts?.sessionId ?? null; const sock = socket;
M
src/cdp/errors.ts
→
src/cdp/errors.ts
@@ -12,6 +12,10 @@ readonly message: string;
readonly method?: string; }; +// Chrome reports a dead session as a generic protocol error, so the message text is the only signal that a reattach — not a retry — is what will fix it. +export const classifyRemoteError = (message: string): CdpErrorKind => + message.includes("Session with given id not found") ? "session_not_found" : "remote_error"; + export const cdpError = ( kind: CdpErrorKind, message: string,
M
src/cdp/transport.ts
→
src/cdp/transport.ts
@@ -1,11 +1,8 @@
-import WebSocket from "ws"; -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"; +import type { Result } from "../util/result"; +import type { CdpError } from "./errors"; +import type { CdpEvent } from "./types"; -const DEFAULT_TIMEOUT_MS = 15_000; +export const DEFAULT_TIMEOUT_MS = 15_000; export type CdpTransport = { connect(url: string, opts?: { timeoutMs?: number }): Promise<Result<void, CdpError>>;@@ -19,128 +16,3 @@ events(): AsyncIterable<CdpEvent>;
state(): "open" | "closed" | "connecting"; onClose(cb: () => void): () => void; }; - -export const createCdpTransport = (): CdpTransport => { - let ws: WebSocket | null = null; - let nextId = 1; - const pending = new Map<number, Pending>(); - const closeListeners = new Set<() => void>(); - - let queue = makeEventQueue(); - - const handleMessage = (raw: string): void => { - let parsed: unknown; - try { - parsed = JSON.parse(raw); - } catch { - return; - } - if (!isCdpRawMessage(parsed)) return; - const msg: CdpRawMessage = parsed; - if (msg.id === undefined) { - if (msg.method) { - queue.push({ - method: msg.method, - params: msg.params, - ...(msg.sessionId !== undefined ? { sessionId: msg.sessionId } : {}), - }); - } - return; - } - const p = pending.get(msg.id); - if (!p) return; - pending.delete(msg.id); - clearTimeout(p.timer); - if (msg.error) { - const kind = msg.error.message.includes("Session with given id not found") - ? "session_not_found" - : "remote_error"; - p.resolve(err(cdpError(kind, msg.error.message, p.method))); - return; - } - p.resolve(ok(msg.result)); - }; - - const cleanup = (reason: string): void => { - rejectAllPending(pending, reason); - queue.end(); - queue = makeEventQueue(); - for (const cb of closeListeners) cb(); - }; - - return { - connect(url, opts = {}): Promise<Result<void, CdpError>> { - const timeoutMs = opts.timeoutMs ?? 10_000; - return new Promise((resolve) => { - let settled = false; - const settle = (r: Result<void, CdpError>): void => { - if (settled) return; - settled = true; - resolve(r); - }; - try { - ws = new WebSocket(url); - } catch (e) { - settle(err(cdpError("transport_closed", e instanceof Error ? e.message : String(e)))); - return; - } - const timer = setTimeout(() => { - ws?.close(); - ws = null; - settle(err(cdpError("timeout", `CDP WebSocket connection timed out after ${timeoutMs}ms`))); - }, timeoutMs); - ws.onopen = () => { - clearTimeout(timer); - settle(ok(undefined)); - }; - ws.onmessage = (ev: WebSocket.MessageEvent) => { - const data = ev.data; - handleMessage(typeof data === "string" ? data : data.toString()); - }; - ws.onerror = () => { - clearTimeout(timer); - settle(err(cdpError("transport_closed", "CDP WebSocket error during connection"))); - }; - ws.onclose = () => { - clearTimeout(timer); - ws = null; - cleanup("WebSocket closed"); - settle(err(cdpError("transport_closed", "CDP WebSocket closed"))); - }; - }); - }, - close(): Promise<void> { - if (ws) { - try { ws.close(1000, "Shutdown"); } catch {} - ws = null; - } - cleanup("close() called"); - return Promise.resolve(); - }, - request(method, params, opts = {}): Promise<Result<unknown, CdpError>> { - const timeoutMs = opts.timeoutMs ?? DEFAULT_TIMEOUT_MS; - const sessionId = opts.sessionId ?? null; - if (!ws || ws.readyState !== WebSocket.OPEN) { - return Promise.resolve(err(cdpError("transport_closed", "Browser not connected. Is Chrome running?", method))); - } - // Bind to a local const so the non-null holds across microtasks between the readyState check and send(). - const sock = ws; - const id = nextId++; - const payload: Record<string, unknown> = { id, method, params }; - if (sessionId) payload["sessionId"] = sessionId; - const json = JSON.stringify(payload); - return sendWithTimeout(pending, id, method, timeoutMs, "CDP", () => sock.send(json)); - }, - events(): AsyncIterable<CdpEvent> { - // Single-consumer: a second for-await loop over this iterable silently steals events from the first. - return queue.iter; - }, - state(): "open" | "closed" | "connecting" { - if (!ws) return "closed"; - if (ws.readyState === WebSocket.CONNECTING) return "connecting"; - if (ws.readyState === WebSocket.OPEN) return "open"; - return "closed"; - }, - onClose: makeOnClose(closeListeners), - }; -};
M
src/client.ts
→
src/client.ts
@@ -7,7 +7,6 @@ import { type Mutex, createMutex } from "./util/mutex";
import { discoverEndpoint } from "./cdp/discovery"; import { type CdpError, cdpError } from "./cdp/errors"; import type { CdpTransport } from "./cdp/transport"; -import { createCdpTransport } from "./cdp/transport"; import { type CdpSession, createCdpSession } from "./cdp/session"; import { type OwnershipRegistry, createOwnershipRegistry } from "./cdp/ownership"; import { ensureHarnessWindow, openHarnessTab } from "./cdp/target-factory";@@ -17,7 +16,7 @@ import type { DaemonStatus, DialogInfo, PageInfo, TabInfo } from "./cdp/types";
export type BrowserClientOptions = { readonly namespace: string; - readonly transport?: CdpTransport; + readonly transport: CdpTransport; readonly remote?: { readonly cdpUrl: string; readonly browserId: string }; readonly initialOwnership?: { readonly ownedTargetIds?: ReadonlyArray<string>;@@ -96,7 +95,7 @@ });
}; export const createBrowserClient = (opts: BrowserClientOptions): BrowserClient => { - const transport = opts.transport ?? createCdpTransport(); + const transport = opts.transport; const ownershipInit: { ownedTargetIds?: ReadonlyArray<string>; harnessWindowTargetId?: string; harnessWindowId?: number } = {}; if (opts.initialOwnership?.ownedTargetIds !== undefined) { ownershipInit.ownedTargetIds = opts.initialOwnership.ownedTargetIds;
M
test/manual/profile-concurrency-test.ts
→
test/manual/profile-concurrency-test.ts
@@ -3,6 +3,8 @@ import { existsSync, mkdtempSync, rmSync } from "node:fs";
import { tmpdir } from "node:os"; import { join } from "node:path"; import { createBrowserClient } from "../../src/client"; +import { createDaemonTransport } from "../../src/cdp/daemon-transport"; +import { isDaemonRunning, spawnDaemon } from "../../src/daemon/spawn"; import { closeIsolatedTab, openIsolatedTab } from "../../src/domains/isolated-tab"; const PORT = Number(process.env["PROFILE_CONC_PORT"] ?? 9488);@@ -56,6 +58,25 @@ } catch {}
} }; +// The daemon owns the only socket to Chrome and reads BU_CDP_WS once, at spawn — so each scenario needs its own daemon pointed at its own browser. +const startDaemonFor = async (endpoint: string): Promise<{ readonly pid: number | undefined } | null> => { + process.env["BU_CDP_WS"] = endpoint; + const child = spawnDaemon(); + if (!child) return null; + for (let i = 0; i < 50; i++) { + if (await isDaemonRunning()) return { pid: child.pid }; + await sleep(200); + } + killTree(child.pid); + return null; +}; + +const stopDaemon = async (pid: number | undefined): Promise<void> => { + killTree(pid); + for (let i = 0; i < 50 && (await isDaemonRunning()); i++) await sleep(200); + delete process.env["BU_CDP_WS"]; +}; + async function scenario(exePath: string, label: string, pinned: boolean): Promise<void> { const udd = mkdtempSync(join(tmpdir(), "pi-profile-conc-")); const base = [@@ -72,6 +93,7 @@ stdio: "ignore",
}); browser.unref(); + let daemonPid: number | undefined; try { let endpoint: string | undefined; for (let i = 0; i < 60 && !endpoint; i++) {@@ -94,9 +116,15 @@ }).unref();
await sleep(6_000); } - process.env["BU_CDP_WS"] = endpoint; + const daemon = await startDaemonFor(endpoint); + check(daemon !== null, `${label}: daemon started against the test browser`); + if (!daemon) return; + daemonPid = daemon.pid; + + const namespace = `conc-${pinned ? "pinned" : "plain"}`; const client = createBrowserClient({ - namespace: `conc-${pinned ? "pinned" : "plain"}`, + namespace, + transport: createDaemonTransport(namespace), ...(pinned ? { profilePin: { userDataDir: udd, profileDir: "Profile 2", label: "pinned", savedAt: "" } } : {}),@@ -138,6 +166,7 @@ for (const tab of tabs) await closeIsolatedTab(client, tab);
await client.closeOwnedTabs(); await client.stop(); } finally { + await stopDaemon(daemonPid); killTree(browser.pid); try { rmSync(udd, { recursive: true, force: true, maxRetries: 5, retryDelay: 200 });@@ -150,6 +179,11 @@ const exePath = findBrowser();
if (!exePath) { console.error("No Chrome/Chromium found — set CHROME_BIN. Skipping."); process.exit(process.env["CI"] ? 1 : 0); + } + // A daemon that is already up is bound to a different Chrome, and the socket path is fixed — this test would silently drive the wrong browser. + if (await isDaemonRunning()) { + console.error("A pi browser daemon is already running. Stop it before running this test."); + process.exit(1); } console.log(`browser: ${exePath}\n`);