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>

6 files changed, 204 insertions, 199 deletions

diff --git a/src/cdp/event-queue.ts b/src/cdp/event-queue.ts
index 330bdb8..08024b0 100644
--- a/src/cdp/event-queue.ts
+++ b/src/cdp/event-queue.ts
@@ -1,9 +1,5 @@
-import { type Result, err } from "../util/result";
-import { type CdpError, cdpError } from "./errors";
 import type { CdpEvent } from "./types";
 
-export type CdpResult = Result<unknown, CdpError>;
-
 export type EventQueue = {
   readonly push: (e: CdpEvent) => void;
   readonly end: () => void;
@@ -40,46 +36,3 @@ export const makeEventQueue = (): EventQueue => {
     },
   };
 };
-
-export type Pending = {
-  readonly resolve: (v: Result<unknown, CdpError>) => void;
-  readonly timer: ReturnType<typeof setTimeout>;
-  readonly method: string;
-};
-
-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();
-};
-
-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)));
-    }
-  });
-
-export const makeOnClose = (closeListeners: Set<() => void>) =>
-  (cb: () => void): (() => void) => {
-    closeListeners.add(cb);
-    return () => closeListeners.delete(cb);
-  };
diff --git a/src/cdp/pending-requests.ts b/src/cdp/pending-requests.ts
new file mode 100644
index 0000000..03b19cd
--- /dev/null
+++ b/src/cdp/pending-requests.ts
@@ -0,0 +1,47 @@
+import { type Result, err } from "../util/result";
+import { type CdpError, cdpError } from "./errors";
+
+export type CdpResult = Result<unknown, CdpError>;
+
+export type Pending = {
+  readonly resolve: (v: Result<unknown, CdpError>) => void;
+  readonly timer: ReturnType<typeof setTimeout>;
+  readonly method: string;
+};
+
+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();
+};
+
+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)));
+    }
+  });
+
+export const makeOnClose = (closeListeners: Set<() => void>) =>
+  (cb: () => void): (() => void) => {
+    closeListeners.add(cb);
+    return () => closeListeners.delete(cb);
+  };
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/cdp/daemon-transport.ts b/src/daemon/transport.ts
index eb2bbd5..4e75215 100644
--- a/src/cdp/daemon-transport.ts
+++ b/src/daemon/transport.ts
@@ -4,14 +4,15 @@ 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 { type Pending, makeEventQueue, makeOnClose, rejectAllPending, sendWithTimeout } from "./event-queue";
+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 "../daemon/protocol";
+} from "./protocol";
 
 export const createDaemonTransport = (clientId: string): CdpTransport => {
   let socket: Socket | null = null;
diff --git a/src/index.ts b/src/index.ts
index 6097881..0133f65 100644
--- a/src/index.ts
+++ b/src/index.ts
@@ -7,7 +7,7 @@ import { readPin } from "./profile/store";
 import { type BrowserState, defaultState, persistState, restoreState } from "./state";
 import { registerAllTools } from "./registry";
 import { cleanupTempDirs } from "./util/truncate";
-import { createDaemonTransport } from "./cdp/daemon-transport";
+import { createDaemonTransport } from "./daemon/transport";
 import { setDebugClicks } from "./util/debug";
 import { asBoolean, asString } from "./util/guards";