Fix service worker upstream routing

Author: pirate

go/modcdp/client/ModCDPClient.go

@@ -707,6 +707,7 @@ func (c *ModCDPClient) connectUpstreamTransport() error {
 				(c.Config.ServerConfig.Upstream.UpstreamWSCDPURL == "" ||
 					c.Config.ServerConfig.Upstream.UpstreamWSCDPURL == initialCDPURL ||
 					c.Config.ServerConfig.Upstream.UpstreamWSCDPURL == launchedCDPURL) {
+				c.Config.ServerConfig.Upstream.UpstreamMode = "ws"
 				c.Config.ServerConfig.Upstream.UpstreamWSCDPURL = loopbackCDPURL
 			}
 		}

go/modcdp/injector/extension.zip

@@ -107,7 +107,7 @@ func (l BrowserLauncher) ConfigForServer(upstreamConfig UpstreamTransportConfig)
 		launcherLocalLoopbackCDPURL = upstreamConfig.UpstreamWSCDPURL
 	}
 	if launcherLocalLoopbackCDPURL != "" {
-		return map[string]any{"upstream": map[string]any{"upstream_ws_cdp_url": launcherLocalLoopbackCDPURL}}
+		return map[string]any{"upstream": map[string]any{"upstream_mode": "ws", "upstream_ws_cdp_url": launcherLocalLoopbackCDPURL}}
 	}
 	return map[string]any{}
 }

go/modcdp/launcher/BrowserLauncher.go

@@ -258,7 +258,28 @@ export class ModCDPClient<
       upstream_cdp_send_timeout_ms: this.config.client_cdp_send_timeout_ms,
     });
     if (upstream !== undefined) {
-      this.upstream.update(upstream);
+      const upstream_mode =
+        typeof (upstream as { upstream_mode?: unknown }).upstream_mode === "string"
+          ? (upstream as { upstream_mode: string }).upstream_mode
+          : this.upstream.config.upstream_mode;
+      if (upstream_mode !== this.upstream.config.upstream_mode) {
+        const Upstream = upstream_transport_constructors.get(upstream_mode);
+        if (!Upstream) throw new Error(`unknown upstream_mode=${upstream_mode}`);
+        const previous_upstream = this.upstream;
+        this.upstream = new Upstream(upstream);
+        this.upstream.update({
+          upstream_cdp_send_timeout_ms: this.config.client_cdp_send_timeout_ms,
+        });
+        this.router.stop();
+        this.router = new AutoSessionRouter({
+          ...this.router.config,
+          upstream: this.upstream,
+          types: this.types,
+        });
+        void previous_upstream.close();
+      } else {
+        this.upstream.update(upstream);
+      }
     }
     if (router !== undefined) {
       this.router.config = ModCDPRouterConfigSchema.parse({

js/src/client/ModCDPClient.ts

@@ -9,13 +9,20 @@ import type { z } from "zod";
 
 type LauncherConfig = z.input<typeof ModCDPLauncherConfigSchema>;
 type LauncherMode = ReturnType<typeof ModCDPLauncherConfigSchema.parse>["launcher_mode"];
+type LauncherUpstreamConfig = UpstreamTransportConfig & {
+  upstream_pipe_read?: NodeJS.ReadableStream;
+  upstream_pipe_write?: NodeJS.WritableStream;
+};
 
 type LaunchedBrowser = {
   proc?: unknown;
   cdp_listen_port?: number;
+  // Browser websocket CDP endpoint when one exists. Pipe transports expose pipe handles instead.
   cdp_url: string | null;
   // Extension-dialable loopback CDP endpoint when it differs from cdp_url (usually they are the same unless public-facing cdp url differs from intranet/localhost equivalent).
   loopback_cdp_url?: string | null;
+  pipe_read?: NodeJS.ReadableStream | null;
+  pipe_write?: NodeJS.WritableStream | null;
   profile_dir?: string | null;
   browserbase_session_id?: string | null;
   browserbase_session_url?: string | null;
@@ -78,9 +85,11 @@ class BrowserLauncher {
   }
 
   configForUpstream(): UpstreamTransportConfig {
-    const config: UpstreamTransportConfig = {};
+    const config: LauncherUpstreamConfig = {};
     const upstream_ws_cdp_url = this.launched?.cdp_url ?? this.config.launcher_remote_cdp_url;
     if (upstream_ws_cdp_url) config.upstream_ws_cdp_url = upstream_ws_cdp_url;
+    if (this.launched?.pipe_read) config.upstream_pipe_read = this.launched.pipe_read;
+    if (this.launched?.pipe_write) config.upstream_pipe_write = this.launched.pipe_write;
     return config;
   }
 
@@ -89,9 +98,11 @@ class BrowserLauncher {
       this.launched?.loopback_cdp_url ??
       (upstream.config.upstream_mode === "ws" && upstream.config.upstream_ws_cdp_url
         ? upstream.config.upstream_ws_cdp_url
-        : null);
+        : upstream.config.upstream_mode !== "ws" && upstream.config.upstream_mode !== "pipe" && this.launched?.cdp_url
+          ? this.launched.cdp_url
+          : null);
     return launcher_local_loopback_cdp_url
-      ? { upstream: { upstream_ws_cdp_url: launcher_local_loopback_cdp_url } }
+      ? { upstream: { upstream_mode: "ws", upstream_ws_cdp_url: launcher_local_loopback_cdp_url } }
       : {};
   }
 

js/src/launcher/BrowserLauncher.ts

@@ -10,7 +10,20 @@ import net from "node:net";
 import type { AddressInfo } from "node:net";
 import { homedir, platform, tmpdir } from "node:os";
 import path from "node:path";
-import { BrowserLauncher, type LauncherConfig, type LaunchedBrowser } from "./BrowserLauncher.js";
+import { z } from "zod";
+import {
+  BrowserLauncher,
+  resolveCdpWebSocketUrl,
+  type LauncherConfig,
+  type LaunchedBrowser,
+} from "./BrowserLauncher.js";
+import { ModCDPLauncherConfigSchema } from "../types/modcdp.js";
+
+const LocalBrowserLauncherConfigSchema = ModCDPLauncherConfigSchema.extend({
+  launcher_local_cdp_transport: z.enum(["ws", "pipe"]).default("ws"),
+});
+type LocalBrowserLauncherConfig = z.infer<typeof LocalBrowserLauncherConfigSchema>;
+type LocalBrowserLauncherInput = z.input<typeof LocalBrowserLauncherConfigSchema>;
 
 function wildcardToRegExp(value: string) {
   return new RegExp(`^${value.replace(/[.+^${}()|[\]\\]/g, "\\$&").replace(/\*/g, ".*")}$`);
@@ -145,6 +158,28 @@ function delay(ms: number) {
   return new Promise((resolve) => setTimeout(resolve, ms));
 }
 
+function mergeLocalChromeArgs(existing: string[] = [], incoming: string[] = []) {
+  const args = [...existing, ...incoming];
+  const load_extension_paths: string[] = [];
+  const merged: string[] = [];
+  for (const arg of args) {
+    if (!arg.startsWith("--load-extension=")) {
+      merged.push(arg);
+      continue;
+    }
+    for (const extension_path of arg.slice("--load-extension=".length).split(",")) {
+      if (extension_path && !load_extension_paths.includes(extension_path)) load_extension_paths.push(extension_path);
+    }
+  }
+  if (load_extension_paths.length > 0) {
+    const first_url_index = merged.findIndex((arg) => !arg.startsWith("-"));
+    const load_extension_arg = `--load-extension=${load_extension_paths.join(",")}`;
+    if (first_url_index === -1) merged.push(load_extension_arg);
+    else merged.splice(first_url_index, 0, load_extension_arg);
+  }
+  return merged;
+}
+
 async function terminateProcess(proc: ChildProcess, timeoutMs = 2_000) {
   if (proc.exitCode !== null || proc.signalCode !== null) return;
   const signalProcess = (signal: NodeJS.Signals) => {
@@ -176,6 +211,98 @@ async function removeProfileDir(profile_dir: string) {
   } catch {}
 }
 
+async function waitForPipeReady(
+  pipe_read: NodeJS.ReadableStream,
+  pipe_write: NodeJS.WritableStream,
+  timeoutMs: number,
+) {
+  let buffer = "";
+  const readyId = 1;
+  await new Promise<void>((resolve, reject) => {
+    const cleanup = () => {
+      clearTimeout(timeout);
+      pipe_read.off("data", onData);
+      pipe_read.off("error", onError);
+      pipe_write.off("error", onError);
+    };
+    const onError = (error: Error) => {
+      cleanup();
+      reject(error);
+    };
+    const onData = (chunk: Buffer | string) => {
+      buffer += chunk.toString();
+      while (buffer.includes("\0")) {
+        const [raw, ...rest] = buffer.split("\0");
+        buffer = rest.join("\0");
+        if (!raw) continue;
+        const message = JSON.parse(raw);
+        if (message.id !== readyId) continue;
+        cleanup();
+        if (message.error) reject(new Error(message.error.message ?? "Browser.getVersion failed over pipe"));
+        else resolve();
+      }
+    };
+    const timeout = setTimeout(() => {
+      cleanup();
+      reject(new Error(`Chrome remote-debugging pipe did not respond within ${timeoutMs}ms`));
+    }, timeoutMs);
+    pipe_read.on("data", onData);
+    pipe_read.on("error", onError);
+    pipe_write.on("error", onError);
+    pipe_write.write(`${JSON.stringify({ id: readyId, method: "Browser.getVersion" })}\0`);
+  });
+}
+
+async function waitForCdpWebSocketUrl(cdp_url: string, timeout_ms: number, poll_interval_ms: number) {
+  const deadline = Date.now() + timeout_ms;
+  let lastError: unknown = null;
+  while (Date.now() < deadline) {
+    try {
+      return await resolveCdpWebSocketUrl(cdp_url);
+    } catch (error) {
+      lastError = error;
+      await delay(poll_interval_ms);
+    }
+  }
+  if (lastError instanceof Error) {
+    throw new Error(
+      `Chrome at ${cdp_url} did not expose a WebSocket CDP URL within ${timeout_ms}ms: ${lastError.message}`,
+    );
+  }
+  throw new Error(`Chrome at ${cdp_url} did not expose a WebSocket CDP URL within ${timeout_ms}ms`);
+}
+
+async function waitForBrowserSelectedCdpWebSocketUrl(
+  profile_dir: string,
+  timeout_ms: number,
+  poll_interval_ms: number,
+  assertChromeRunning: () => void,
+) {
+  const deadline = Date.now() + timeout_ms;
+  let lastError: unknown = null;
+  while (Date.now() < deadline) {
+    assertChromeRunning();
+    const activePort = await readDevToolsActivePort(profile_dir);
+    if (activePort) {
+      try {
+        return {
+          cdp_listen_port: activePort.cdp_listen_port,
+          cdp_url: await resolveCdpWebSocketUrl(activePort.cdp_url),
+        };
+      } catch (error) {
+        lastError = error;
+      }
+    }
+    await delay(poll_interval_ms);
+  }
+  if (lastError instanceof Error) {
+    throw new Error(
+      `Chrome did not expose DevToolsActivePort from ${profile_dir} within ${timeout_ms}ms: ${lastError.message}`,
+    );
+  }
+  throw new Error(`Chrome did not expose DevToolsActivePort from ${profile_dir} within ${timeout_ms}ms`);
+}
+
 async function readDevToolsActivePort(profile_dir: string) {
   const activePortPath = path.join(profile_dir, "DevToolsActivePort");
   let body: string;
@@ -193,8 +320,27 @@ async function readDevToolsActivePort(profile_dir: string) {
 }
 
 class LocalBrowserLauncher extends BrowserLauncher {
-  constructor(config: LauncherConfig = {}) {
-    super({ ...config, launcher_mode: "local" });
+  declare config: LocalBrowserLauncherConfig;
+
+  constructor(config: LocalBrowserLauncherInput = {}) {
+    const { launcher_local_cdp_transport: _launcher_local_cdp_transport, ...base_config } = config;
+    super({ ...base_config, launcher_mode: "local" } as LauncherConfig);
+    this.config = LocalBrowserLauncherConfigSchema.parse({ ...config, launcher_mode: "local" });
+  }
+
+  override update(config: LocalBrowserLauncherInput = {}) {
+    const next_config = LocalBrowserLauncherConfigSchema.parse({ ...this.config, ...config, launcher_mode: "local" });
+    if (config.launcher_local_args) {
+      next_config.launcher_local_args = mergeLocalChromeArgs(this.config.launcher_local_args, config.launcher_local_args);
+    }
+    if (config.launcher_local_extra_args) {
+      next_config.launcher_local_extra_args = mergeLocalChromeArgs(
+        this.config.launcher_local_extra_args,
+        config.launcher_local_extra_args,
+      );
+    }
+    this.config = next_config;
+    return this;
   }
 
   static findChromeBinary(explicit?: string | null) {
@@ -219,9 +365,12 @@ class LocalBrowserLauncher extends BrowserLauncher {
   }
 
   async launch(config: LauncherConfig = {}): Promise<LaunchedBrowser> {
-    const launch_config = { ...this.config, ...config };
+    const launch_config = LocalBrowserLauncherConfigSchema.parse({ ...this.config, ...config, launcher_mode: "local" });
     const exe = LocalBrowserLauncher.findChromeBinary(launch_config.launcher_local_executable_path);
-    const usePort = launch_config.launcher_local_cdp_listen_port ?? 0;
+    const usePipe = launch_config.launcher_local_cdp_transport === "pipe";
+    const useLoopbackCdp =
+      !usePipe || launch_config.launcher_local_loopback_cdp || launch_config.launcher_local_cdp_listen_port != null;
+    const usePort = useLoopbackCdp ? (launch_config.launcher_local_cdp_listen_port ?? 0) : null;
     const profile_dir = launch_config.launcher_local_user_data_dir || (await mkdtemp(path.join(tmpdir(), "modcdp.")));
     const default_headless = process.platform === "linux" && !process.env.DISPLAY;
     const headless = launch_config.launcher_local_headless ?? default_headless;
@@ -232,15 +381,21 @@ class LocalBrowserLauncher extends BrowserLauncher {
       "--disable-gpu",
       sandbox === false ? "--no-sandbox" : null,
       `--user-data-dir=${profile_dir}`,
-      "--remote-debugging-address=127.0.0.1",
-      `--remote-debugging-port=${usePort}`,
+      useLoopbackCdp ? "--remote-debugging-address=127.0.0.1" : null,
+      useLoopbackCdp ? `--remote-debugging-port=${usePort}` : null,
+      usePipe ? "--remote-debugging-pipe" : null,
       ...launch_config.launcher_local_args,
       ...launch_config.launcher_local_extra_args,
       "about:blank",
     ].filter(Boolean);
 
+    const stdio = (usePipe ? ["ignore", "ignore", "ignore", "pipe", "pipe"] : "ignore") as
+      | "ignore"
+      | "inherit"
+      | "pipe"
+      | import("node:child_process").StdioOptions;
     const proc = spawn(exe, flags, {
-      stdio: "ignore",
+      stdio,
       detached: process.platform !== "win32",
     });
     let spawnError: Error | null = null;
@@ -262,6 +417,46 @@ class LocalBrowserLauncher extends BrowserLauncher {
       }
     };
 
+    if (usePipe) {
+      const pipe_write = proc.stdio[3] as NodeJS.WritableStream | null;
+      const pipe_read = proc.stdio[4] as NodeJS.ReadableStream | null;
+      if (!pipe_write || !pipe_read) {
+        await close();
+        throw new Error("Chrome remote-debugging pipe stdio handles were not created.");
+      }
+      assertChromeRunning();
+      await waitForPipeReady(pipe_read, pipe_write, launch_config.launcher_local_chrome_ready_timeout_ms);
+      const loopback =
+        usePort == null
+          ? null
+          : usePort === 0
+            ? await waitForBrowserSelectedCdpWebSocketUrl(
+                profile_dir,
+                launch_config.launcher_local_chrome_ready_timeout_ms,
+                launch_config.launcher_local_chrome_ready_poll_interval_ms,
+                assertChromeRunning,
+              )
+            : {
+                cdp_listen_port: usePort,
+                cdp_url: await waitForCdpWebSocketUrl(
+                  `http://127.0.0.1:${usePort}`,
+                  launch_config.launcher_local_chrome_ready_timeout_ms,
+                  launch_config.launcher_local_chrome_ready_poll_interval_ms,
+                ),
+              };
+      this.launched = {
+        proc,
+        ...(loopback == null ? {} : { cdp_listen_port: loopback.cdp_listen_port }),
+        cdp_url: null,
+        ...(loopback == null ? {} : { loopback_cdp_url: loopback.cdp_url }),
+        pipe_read,
+        pipe_write,
+        profile_dir,
+        close,
+      };
+      return this.launched;
+    }
+
     const deadline = Date.now() + launch_config.launcher_local_chrome_ready_timeout_ms;
     while (Date.now() < deadline) {
       try {

js/src/launcher/LocalBrowserLauncher.ts

@@ -8,9 +8,10 @@
 import * as Browser from "../types/generated/zod/Browser.js";
 import * as Runtime from "../types/generated/zod/Runtime.js";
 import type { z } from "zod";
-import { ModCDPClient } from "../client/ModCDPClient.js";
+import { ModCDPClient, upstream_transport_constructors } from "../client/ModCDPClient.js";
 import { ModCDPConfigureParamsSchema, ModCDPServerConfigSchema } from "../types/modcdp.js";
 import { routeFor } from "../translate/translate.js";
+import { ChromeDebuggerUpstreamTransport } from "../transport/ChromeDebuggerUpstreamTransport.js";
 import { DownstreamTransportSet } from "../transport/DownstreamTransportSet.js";
 import { NativeMessagingDownstreamTransport } from "../transport/NativeMessagingDownstreamTransport.js";
 import { NATSDownstreamTransport } from "../transport/NATSDownstreamTransport.js";
@@ -45,6 +46,8 @@ const DEFAULT_ROUTES = {
 const OFFSCREEN_KEEP_ALIVE_PORT_NAME = "ModCDPOffscreenKeepAlive";
 const OFFSCREEN_KEEP_ALIVE_PATH = "offscreen/keepalive.html";
 
+upstream_transport_constructors.set("chromedebugger", ChromeDebuggerUpstreamTransport);
+
 /**
  * Extension-side ModCDP server.
  *

js/src/server/ModCDPServer.ts

@@ -75,6 +75,15 @@ class ChromeDebuggerUpstreamTransport extends UpstreamTransport {
     this.config = ChromeDebuggerUpstreamTransportConfigSchema.parse({ ...config, upstream_mode: "chromedebugger" });
   }
 
+  override update(config: z.input<typeof ChromeDebuggerUpstreamTransportConfigSchema> = {}) {
+    this.config = ChromeDebuggerUpstreamTransportConfigSchema.parse({
+      ...this.config,
+      ...config,
+      upstream_mode: "chromedebugger",
+    });
+    return this;
+  }
+
   /** Install chrome.debugger listeners for this service-worker lifetime. */
   override async connect() {
     this.installEventListener();