From 0666a0f4b5bcff8381d348aa47536c07d0445df3 Mon Sep 17 00:00:00 2001 From: Mike Benner <36419818+mikebenner@users.noreply.github.com> Date: Sat, 25 Jul 2026 18:37:48 -0700 Subject: [PATCH 1/2] =?UTF-8?q?feat:=20Windows=20support=20for=20the=20bri?= =?UTF-8?q?dge=20=E2=80=94=20dial=20herdr's=20named=20pipe?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit herdr's Windows beta does not expose AF_UNIX: the .sock path on disk is a : pointer file, and the real 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 the two dial sites in herdr-client.ts now route through bridge/dial.ts, which adapts node:net to the same write/flush/end handler shape on win32 and stays on Bun.connect everywhere else. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01VutEkP9A8oY72ZQBfFJdSw --- bridge/dial.ts | 65 ++++++++++++++++++++++++++ bridge/herdr-client.ts | 103 ++++++++++++++++++++--------------------- 2 files changed, 114 insertions(+), 54 deletions(-) create mode 100644 bridge/dial.ts diff --git a/bridge/dial.ts b/bridge/dial.ts new file mode 100644 index 0000000..2790bf0 --- /dev/null +++ b/bridge/dial.ts @@ -0,0 +1,65 @@ +// 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; + 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; +}; + +export function dialHerdr(socketPath: string, handlers: DialHandlers): Promise { + if (process.platform !== "win32") { + return 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; + } + + const pipeName = "\\\\.\\pipe\\" + socketPath; + return new Promise((resolve, reject) => { + const sock = net.connect(pipeName); + const handle: SockHandle = { + write: (data) => sock.write(data), + flush: () => {}, + // destroy(), not end(): herdr's one-shot RPC is done once we have the reply line, and the + // callers rely on close being prompt and re-entrant-safe (they guard with a settled flag). + end: () => sock.destroy(), + }; + let opened = false; + 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", () => handlers.close?.(handle)); + }); +} diff --git a/bridge/herdr-client.ts b/bridge/herdr-client.ts index dfaeda7..da1b2b8 100644 --- a/bridge/herdr-client.ts +++ b/bridge/herdr-client.ts @@ -1,4 +1,5 @@ import type { AgentStatus } from "./types.ts"; +import { dialHerdr, type SockHandle } from "./dial.ts"; import { decodeReplyLine, decodeStreamLine } from "./wire.ts"; // ───────────────────────────────────────────────────────────────────────────── @@ -102,9 +103,9 @@ export class HerdrClient { return new Promise((resolve, reject) => { let buf = ""; let settled = false; - // The live socket, once Bun.connect opens one. Hoisted so EVERY terminal path (timeout + // The live socket, once the dial opens one. Hoisted so EVERY terminal path (timeout // included) can close it — otherwise a timeout leaves the FD dangling. - let socket: Bun.Socket | null = null; + let socket: SockHandle | null = null; // Stream-decode so a multi-byte UTF-8 codepoint split across chunk boundaries isn't // corrupted into replacement characters. const decoder = new TextDecoder("utf-8"); @@ -129,32 +130,29 @@ export class HerdrClient { this.timeoutMs, ); - Bun.connect({ - unix: this.socketPath, - socket: { - open(s) { - socket = s; - }, - data(s, chunk) { - socket = s; - buf += decoder.decode(chunk, { stream: true }); - const nl = buf.indexOf("\n"); - if (nl < 0) return; - const line = buf.slice(0, nl); - finish(() => { - try { - resolve(decodeReplyLine(line, method)); - } catch (e) { - reject(e as Error); - } - }); - }, - error(_s, err) { - finish(() => reject(err)); - }, - close() { - finish(() => reject(new Error(`herdr ${method}: connection closed before reply`))); - }, + dialHerdr(this.socketPath, { + open(s) { + socket = s; + }, + data(s, chunk) { + socket = s; + buf += decoder.decode(chunk, { stream: true }); + const nl = buf.indexOf("\n"); + if (nl < 0) return; + const line = buf.slice(0, nl); + finish(() => { + try { + resolve(decodeReplyLine(line, method)); + } catch (e) { + reject(e as Error); + } + }); + }, + error(_s, err) { + finish(() => reject(err)); + }, + close() { + finish(() => reject(new Error(`herdr ${method}: connection closed before reply`))); }, }) .then((s) => { @@ -219,7 +217,7 @@ export class HerdrClient { const id = `es${++idCounter}`; const decoder = new TextDecoder("utf-8"); let buf = ""; - let socket: Bun.Socket | null = null; + let socket: SockHandle | null = null; let down = false; let acked = false; @@ -265,31 +263,28 @@ export class HerdrClient { opts.onEvent(decoded.event, decoded.data); }; - Bun.connect({ - unix: this.socketPath, - socket: { - open(s) { - socket = s; - }, - // Multiple lines can arrive per chunk (bursty events); drain ALL complete lines and keep the - // stream open. Stream-decode so a multi-byte codepoint split across chunks isn't corrupted. - data(s, chunk) { - socket = s; - buf += decoder.decode(chunk, { stream: true }); - let nl = buf.indexOf("\n"); - while (nl >= 0 && !down) { - const line = buf.slice(0, nl); - buf = buf.slice(nl + 1); - handleLine(line); - nl = buf.indexOf("\n"); - } - }, - error(_s, err) { - fireDown(err.message || "socket error"); - }, - close() { - fireDown("connection closed"); - }, + dialHerdr(this.socketPath, { + open(s) { + socket = s; + }, + // Multiple lines can arrive per chunk (bursty events); drain ALL complete lines and keep the + // stream open. Stream-decode so a multi-byte codepoint split across chunks isn't corrupted. + data(s, chunk) { + socket = s; + buf += decoder.decode(chunk, { stream: true }); + let nl = buf.indexOf("\n"); + while (nl >= 0 && !down) { + const line = buf.slice(0, nl); + buf = buf.slice(nl + 1); + handleLine(line); + nl = buf.indexOf("\n"); + } + }, + error(_s, err) { + fireDown(err.message || "socket error"); + }, + close() { + fireDown("connection closed"); }, }) .then((s) => { From 85a7411941ebfa3cee1dfdcd03ed0426d07617f7 Mon Sep 17 00:00:00 2001 From: Mike Benner <36419818+mikebenner@users.noreply.github.com> Date: Sun, 26 Jul 2026 11:31:27 -0700 Subject: [PATCH 2/2] =?UTF-8?q?fix:=20harden=20the=20Windows=20pipe=20dial?= =?UTF-8?q?=20=E2=80=94=20cancellable=20connects,=20path=20normalization,?= =?UTF-8?q?=20APPDATA=20default,=20tests?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review follow-ups from #25/#27 cross-review: - dialHerdr exposes onDial(cancel) so a caller timeout that fires mid-connect aborts the pending dial instead of leaking the OS handle (both request() and subscribeEvents() wire it up); a dial destroyed before connect now settles its promise instead of pending forever. - toPipeName() passes through already-prefixed pipe names (either slash direction) instead of double-prefixing. - defaultSocketPath() resolves %APPDATA%\herdr\herdr.sock on win32 so HERDR_SOCKET_PATH is no longer required there; pure and unit-tested. - New dial.test.ts: pure toPipeName cases everywhere, plus live named- pipe round-trip, mid-codepoint chunk split, connect-error, and cancel coverage on win32 (skipped elsewhere). - SockHandle.end() documented as immediate destroy, not a drained close. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01VutEkP9A8oY72ZQBfFJdSw --- bridge/config.test.ts | 24 +++++++- bridge/config.ts | 19 ++++++- bridge/dial.test.ts | 125 +++++++++++++++++++++++++++++++++++++++++ bridge/dial.ts | 55 ++++++++++++++++-- bridge/herdr-client.ts | 27 +++++++++ 5 files changed, 243 insertions(+), 7 deletions(-) create mode 100644 bridge/dial.test.ts diff --git a/bridge/config.test.ts b/bridge/config.test.ts index 209c610..8192efd 100644 --- a/bridge/config.test.ts +++ b/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 /AppData/Roaming when APPDATA is unset", () => { + expect(defaultSocketPath("win32", {}, "C:\\Users\\u")).toBe( + join("C:\\Users\\u", "AppData", "Roaming", "herdr", "herdr.sock"), + ); + }); +}); diff --git a/bridge/config.ts b/bridge/config.ts index 221dbe1..e01eb85 100644 --- a/bridge/config.ts +++ b/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 = 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 }), diff --git a/bridge/dial.test.ts b/bridge/dial.test.ts new file mode 100644 index 0000000..327d6cf --- /dev/null +++ b/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 => { + 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((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 => + new Promise((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(); + }); +}); diff --git a/bridge/dial.ts b/bridge/dial.ts index 2790bf0..56fc380 100644 --- a/bridge/dial.ts +++ b/bridge/dial.ts @@ -8,6 +8,12 @@ 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; }; @@ -16,11 +22,29 @@ export type DialHandlers = { 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 { if (process.platform !== "win32") { - return Bun.connect({ + const promise = Bun.connect({ unix: socketPath, socket: { open(s) { @@ -37,19 +61,34 @@ export function dialHerdr(socketPath: string, handlers: DialHandlers): Promise; + // 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 = "\\\\.\\pipe\\" + socketPath; + const pipeName = toPipeName(socketPath); return new Promise((resolve, reject) => { const sock = net.connect(pipeName); const handle: SockHandle = { write: (data) => sock.write(data), flush: () => {}, - // destroy(), not end(): herdr's one-shot RPC is done once we have the reply line, and the - // callers rely on close being prompt and re-entrant-safe (they guard with a settled flag). end: () => sock.destroy(), }; let opened = false; + handlers.onDial?.(() => sock.destroy()); sock.on("connect", () => { opened = true; handlers.open?.(handle); @@ -60,6 +99,12 @@ export function dialHerdr(socketPath: string, handlers: DialHandlers): Promise handlers.close?.(handle)); + 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); + }); }); } diff --git a/bridge/herdr-client.ts b/bridge/herdr-client.ts index da1b2b8..08eb1a8 100644 --- a/bridge/herdr-client.ts +++ b/bridge/herdr-client.ts @@ -106,6 +106,9 @@ export class HerdrClient { // The live socket, once the dial opens one. Hoisted so EVERY terminal path (timeout // included) can close it — otherwise a timeout leaves the FD dangling. let socket: SockHandle | null = null; + // Aborts a dial that is still connecting — a timeout that fires mid-connect has no socket + // to end() yet, and without this the pending OS handle lives until the connect settles. + let cancelDial: (() => void) | null = null; // Stream-decode so a multi-byte UTF-8 codepoint split across chunk boundaries isn't // corrupted into replacement characters. const decoder = new TextDecoder("utf-8"); @@ -123,7 +126,15 @@ export class HerdrClient { /* ignore */ } socket = null; + } else if (cancelDial) { + // Timed out (or failed) while still connecting — abort the in-flight dial. + try { + cancelDial(); + } catch { + /* ignore */ + } } + cancelDial = null; }; const timer = setTimeout( () => finish(() => reject(new Error(`herdr ${method}: timed out after ${this.timeoutMs}ms`))), @@ -131,6 +142,9 @@ export class HerdrClient { ); dialHerdr(this.socketPath, { + onDial(cancel) { + cancelDial = cancel; + }, open(s) { socket = s; }, @@ -218,6 +232,7 @@ export class HerdrClient { const decoder = new TextDecoder("utf-8"); let buf = ""; let socket: SockHandle | null = null; + let cancelDial: (() => void) | null = null; let down = false; let acked = false; @@ -233,7 +248,16 @@ export class HerdrClient { /* ignore */ } socket = null; + } else if (cancelDial) { + // Ack timeout (or close()) while the dial was still connecting — abort it so repeated + // reconnect attempts can't stack pending OS handles. + try { + cancelDial(); + } catch { + /* ignore */ + } } + cancelDial = null; opts.onDown(reason); }; @@ -264,6 +288,9 @@ export class HerdrClient { }; dialHerdr(this.socketPath, { + onDial(cancel) { + cancelDial = cancel; + }, open(s) { socket = s; },