feat: Windows support for the bridge via herdr's named pipe

bridge/config.test.ts

@@ -1,6 +1,7 @@
 import { afterEach, beforeEach, describe, expect, test } from "bun:test";
+import { join } from "node:path";
 
-import { loadConfig } from "./config.ts";
+import { defaultSocketPath, loadConfig } from "./config.ts";
 
 // loadConfig is the deployment contract — env vars in, a resolved Config out. Pure (just reads
 // process.env + homedir), so we drive it by mutating the environment and restoring it after.
@@ -178,3 +179,24 @@ describe("loadConfig", () => {
     expect(cfg.host).toBe("0.0.0.0");
   });
 });
+
+// Pure — both platform branches are testable from any host (expectations use join() so the
+// host's separator never leaks into the assertion).
+describe("defaultSocketPath", () => {
+  test("unix default lives under ~/.config/herdr", () => {
+    expect(defaultSocketPath("linux", {}, "/home/u")).toBe(join("/home/u", ".config", "herdr", "herdr.sock"));
+    expect(defaultSocketPath("darwin", {}, "/Users/u")).toBe(join("/Users/u", ".config", "herdr", "herdr.sock"));
+  });
+
+  test("win32 default honours APPDATA", () => {
+    expect(defaultSocketPath("win32", { APPDATA: "C:\\Users\\u\\AppData\\Roaming" }, "C:\\Users\\u")).toBe(
+      join("C:\\Users\\u\\AppData\\Roaming", "herdr", "herdr.sock"),
+    );
+  });
+
+  test("win32 falls back to <home>/AppData/Roaming when APPDATA is unset", () => {
+    expect(defaultSocketPath("win32", {}, "C:\\Users\\u")).toBe(
+      join("C:\\Users\\u", "AppData", "Roaming", "herdr", "herdr.sock"),
+    );
+  });
+});

bridge/config.ts

@@ -139,6 +139,23 @@ export interface Config {
   skipServe: boolean;
 }
 
+/**
+ * herdr's default socket location: `~/.config/herdr/herdr.sock` on Unix, `%APPDATA%\herdr\herdr.sock`
+ * on Windows (the Windows beta keeps its config root under AppData\Roaming). Pure so both branches
+ * are unit-testable on any platform.
+ */
+export function defaultSocketPath(
+  platform: NodeJS.Platform = process.platform,
+  env: Record<string, string | undefined> = process.env,
+  home: string = homedir(),
+): string {
+  if (platform === "win32") {
+    const appData = env.APPDATA ?? join(home, "AppData", "Roaming");
+    return join(appData, "herdr", "herdr.sock");
+  }
+  return join(home, ".config", "herdr", "herdr.sock");
+}
+
 export function loadConfig(): Config {
   const stateDir =
     process.env.HERDR_PLUGIN_STATE_DIR ??
@@ -148,7 +165,7 @@ export function loadConfig(): Config {
   const submitKeys = envList("COLLIE_SUBMIT_KEYS");
 
   return {
-    socketPath: process.env.HERDR_SOCKET_PATH ?? join(homedir(), ".config", "herdr", "herdr.sock"),
+    socketPath: process.env.HERDR_SOCKET_PATH ?? defaultSocketPath(),
     port: envInt("COLLIE_PORT", 8787, { min: 1, max: 65535 }),
     host: process.env.COLLIE_HOST ?? "127.0.0.1",
     pollMs: envInt("COLLIE_POLL_MS", 1500, { min: 250 }),

bridge/dial.test.ts

@@ -0,0 +1,125 @@
+import { describe, expect, test } from "bun:test";
+import net from "node:net";
+
+import { dialHerdr, toPipeName } from "./dial.ts";
+
+// toPipeName is pure and runs everywhere. The live-pipe suite needs a real Windows named pipe, so
+// it is skipped off win32 — mirroring the repo convention that transport code is exercised where
+// it actually runs (the Unix Bun.connect path stays untested here for the same reason).
+
+describe("toPipeName", () => {
+  test("prefixes a plain socket path with the pipe namespace", () => {
+    expect(toPipeName("C:\\Users\\u\\AppData\\Roaming\\herdr\\herdr.sock")).toBe(
+      "\\\\.\\pipe\\C:\\Users\\u\\AppData\\Roaming\\herdr\\herdr.sock",
+    );
+  });
+
+  test("passes an already-prefixed pipe name through unchanged", () => {
+    expect(toPipeName("\\\\.\\pipe\\already-a-pipe")).toBe("\\\\.\\pipe\\already-a-pipe");
+    expect(toPipeName("//./pipe/already-a-pipe")).toBe("//./pipe/already-a-pipe");
+  });
+});
+
+const describeWin = process.platform === "win32" ? describe : describe.skip;
+
+describeWin("dialHerdr over a live named pipe (win32 only)", () => {
+  const pipeFor = (tag: string) => `\\\\.\\pipe\\collie-dial-test-${process.pid}-${tag}`;
+
+  /** One-connection line server: waits for a request line, replies with `chunks`, then closes. */
+  const serveOnce = async (pipe: string, chunks: Buffer[], gapMs = 0): Promise<net.Server> => {
+    const server = net.createServer((conn) => {
+      conn.once("data", () => {
+        let i = 0;
+        const writeNext = () => {
+          if (i >= chunks.length) {
+            conn.end();
+            return;
+          }
+          conn.write(chunks[i++]!);
+          setTimeout(writeNext, gapMs);
+        };
+        writeNext();
+      });
+    });
+    await new Promise<void>((resolve, reject) => {
+      server.once("error", reject);
+      server.listen(pipe, resolve);
+    });
+    return server;
+  };
+
+  /** Dial, send one request line, resolve with everything received up to the first newline. */
+  const requestLine = (pipe: string): Promise<string> =>
+    new Promise<string>((resolve, reject) => {
+      const received: Buffer[] = [];
+      let settled = false;
+      const once = (fn: () => void) => {
+        if (settled) return;
+        settled = true;
+        fn();
+      };
+      dialHerdr(pipe, {
+        data(_s, chunk) {
+          received.push(Buffer.from(chunk));
+          const text = Buffer.concat(received).toString("utf-8");
+          if (text.includes("\n")) once(() => resolve(text.slice(0, text.indexOf("\n"))));
+        },
+        error(_s, err) {
+          once(() => reject(err));
+        },
+        close() {
+          once(() => reject(new Error("closed before a full reply line")));
+        },
+      })
+        .then((s) => s.write('{"id":"t","method":"probe","params":{}}\n'))
+        .catch((err) => once(() => reject(err)));
+    });
+
+  test("one-shot request/reply round-trips, accepting an already-prefixed pipe name", async () => {
+    const pipe = pipeFor("roundtrip");
+    const server = await serveOnce(pipe, [Buffer.from('{"ok":true}\n', "utf-8")]);
+    try {
+      expect(await requestLine(pipe)).toBe('{"ok":true}');
+    } finally {
+      server.close();
+    }
+  });
+
+  test("a reply split mid-codepoint across chunks reassembles byte-perfect", async () => {
+    const pipe = pipeFor("split");
+    const payload = Buffer.from('{"emoji":"🐕🦮"}\n', "utf-8");
+    const cut = 12; // inside the first emoji's 4-byte sequence
+    const server = await serveOnce(pipe, [payload.subarray(0, cut), payload.subarray(cut)], 15);
+    try {
+      expect(await requestLine(pipe)).toBe('{"emoji":"🐕🦮"}');
+    } finally {
+      server.close();
+    }
+  });
+
+  test("dialing a nonexistent pipe rejects and fires the error handler", async () => {
+    let sawError = false;
+    await expect(
+      dialHerdr(pipeFor("nonexistent"), {
+        error() {
+          sawError = true;
+        },
+      }),
+    ).rejects.toBeDefined();
+    expect(sawError).toBe(true);
+  });
+
+  test("onDial cancel settles the promise instead of leaving it pending", async () => {
+    let cancel: (() => void) | null = null;
+    const p = dialHerdr(pipeFor("cancelled"), {
+      onDial(c) {
+        cancel = c;
+      },
+    });
+    expect(cancel).not.toBeNull();
+    cancel!();
+    // Whether the abort lands as "closed before connect" or the connect error races first,
+    // the promise must settle — a caller that already timed out must not leak a pending dial.
+    await expect(p).rejects.toBeDefined();
+  });
+});

bridge/dial.ts

@@ -0,0 +1,110 @@
+// Platform dial shim. On Unix the herdr API socket is a real AF_UNIX socket and Bun.connect
+// handles it. On Windows (herdr Windows beta) the ".sock" path is a pointer file — the actual
+// transport is a named pipe whose name is the full socket path (\\.\pipe\C:\...\herdr.sock).
+// Bun.connect({unix}) cannot open named pipes, but Bun's node:net can, so we adapt it to the
+// same handler shape the two call sites in herdr-client.ts use (write/flush/end only).
+import net from "node:net";
+
+export type SockHandle = {
+  write(data: string): unknown;
+  flush(): void;
+  /**
+   * Closes the connection IMMEDIATELY — on win32 this is destroy(), not a graceful half-close,
+   * so queued-but-unflushed data may be dropped. That is correct for both current consumers
+   * (one-shot RPCs close only after the reply line arrives; the event stream uses it to cancel),
+   * but do not reuse this handle anywhere that needs written data drained on close.
+   */
+  end(): void;
+};
+
+export type DialHandlers = {
+  open?(s: SockHandle): void;
+  data?(s: SockHandle, chunk: Uint8Array): void;
+  error?(s: SockHandle, err: Error): void;
+  close?(s: SockHandle): void;
+  /**
+   * Invoked synchronously during dialHerdr() with a canceller that aborts the dial even while it
+   * is still connecting. The returned promise alone cannot do that — there is no handle to close
+   * until `open` — so a caller-side timeout that fires mid-connect would otherwise leak the
+   * pending OS handle until the connect itself resolves or fails.
+   */
+  onDial?(cancel: () => void): void;
+};
+
+/**
+ * herdr's Windows beta names its pipe after the full socket path. Pass through a value that is
+ * already a pipe name (either slash direction) so an explicit HERDR_SOCKET_PATH=\\.\pipe\… works.
+ */
+export function toPipeName(socketPath: string): string {
+  if (socketPath.startsWith("\\\\.\\pipe\\") || socketPath.startsWith("//./pipe/")) {
+    return socketPath;
+  }
+  return "\\\\.\\pipe\\" + socketPath;
+}
+
+export function dialHerdr(socketPath: string, handlers: DialHandlers): Promise<SockHandle> {
+  if (process.platform !== "win32") {
+    const promise = Bun.connect({
+      unix: socketPath,
+      socket: {
+        open(s) {
+          handlers.open?.(s as unknown as SockHandle);
+        },
+        data(s, chunk) {
+          handlers.data?.(s as unknown as SockHandle, chunk);
+        },
+        error(s, err) {
+          handlers.error?.(s as unknown as SockHandle, err);
+        },
+        close(s) {
+          handlers.close?.(s as unknown as SockHandle);
+        },
+      },
+    }) as unknown as Promise<SockHandle>;
+    // Bun.connect exposes no pre-open handle; best available cancellation is closing the socket
+    // the moment the connect resolves. Callers already tolerate a late open followed by close.
+    handlers.onDial?.(() => {
+      promise
+        .then((s) => {
+          try {
+            s.end();
+          } catch {
+            /* ignore */
+          }
+        })
+        .catch(() => {
+          /* connect failed — nothing to close */
+        });
+    });
+    return promise;
+  }
+
+  const pipeName = toPipeName(socketPath);
+  return new Promise<SockHandle>((resolve, reject) => {
+    const sock = net.connect(pipeName);
+    const handle: SockHandle = {
+      write: (data) => sock.write(data),
+      flush: () => {},
+      end: () => sock.destroy(),
+    };
+    let opened = false;
+    handlers.onDial?.(() => sock.destroy());
+    sock.on("connect", () => {
+      opened = true;
+      handlers.open?.(handle);
+      resolve(handle);
+    });
+    sock.on("data", (chunk: Buffer) => handlers.data?.(handle, chunk));
+    sock.on("error", (err) => {
+      if (!opened) reject(err);
+      handlers.error?.(handle, err as Error);
+    });
+    sock.on("close", () => {
+      // A dial destroyed while still connecting emits close without error; settle the promise so
+      // no caller is left awaiting forever. Guarded rejects/finishes upstream make this a no-op
+      // when the caller already timed out.
+      if (!opened) reject(new Error("dial closed before connect"));
+      handlers.close?.(handle);
+    });
+  });
+}