fix: Windows support - skip directory fsync, green test suite

src/outbox.ts

@@ -52,8 +52,12 @@ export function Outbox(opts: { storageDir: string }): OutboxInstance {
     }
     chmodSync(temp, 0o600)
     renameSync(temp, target)
-    const dirFd = openSync(directory, "r")
-    try { fsyncSync(dirFd) } finally { closeSync(dirFd) }
+    // Directory fsync is POSIX-only (EPERM on Windows); file fsync above is
+    // the best durability available there.
+    if (process.platform !== "win32") {
+      const dirFd = openSync(directory, "r")
+      try { fsyncSync(dirFd) } finally { closeSync(dirFd) }
+    }
   }
 
   function update(endpointId: string, messageId: string, values: Partial<OutboxRecord>): OutboxRecord | null {

src/queue.ts

@@ -167,6 +167,9 @@ function migrationStateFiles(spoolDir: string, state: SpoolState): string[] {
 }
 
 function syncMigrationDirectory(directory: string): void {
+  // Directory fsync is POSIX-only; on Windows it raises EPERM. The fsync'd
+  // file plus rename is the best durability available there.
+  if (process.platform === "win32") return
   const fd = openSync(directory, "r")
   try {
     fsyncSync(fd)
@@ -605,6 +608,7 @@ export function MessageQueue(opts: QueueOptions): QueueInstance {
   }
 
   function syncDirectory(directory: string): void {
+    if (process.platform === "win32") return
     const dirFd = openSync(directory, "r")
     try {
       fsyncSync(dirFd)

tests/ack-integration.test.mjs

@@ -10,22 +10,25 @@ import { Outbox } from "../dist/outbox.js"
 import { Sender } from "../dist/sender.js"
 
 const noopLogger = async () => {}
+// Exercise the UDS listener where the platform has one; Windows gets the
+// loopback TCP listener (UDS binds are unsupported there).
+const listenerPlatform = process.platform === "win32" ? "win32" : "darwin"
 
 test("held accept/drop/expiry outcomes round-trip as durable final ACKs", async () => {
   const dir = await mkdtemp(join(tmpdir(), "peers-ack-e2e-"))
   const outbox = Outbox({ storageDir: join(dir, "sender") })
   const senderEndpoint = "session-sender"
   const receiverEndpoint = "session-receiver"
   const senderListener = InboxListener({
-    token: "sender-token", maxBodyBytes: 20_000, runtimeDir: join(dir, "runtime"), processId: "sender", platform: "darwin",
+    token: "sender-token", maxBodyBytes: 20_000, runtimeDir: join(dir, "runtime"), processId: "sender", platform: listenerPlatform,
     resolveEndpoint: ({ toEndpointId }) => toEndpointId === senderEndpoint ? senderEndpoint : null,
     onMessage: async () => "refused",
     onAcknowledgement: async (ack) => { await outbox.applyAcknowledgement(ack) },
     logger: noopLogger,
   })
   const queue = MessageQueue({ endpointId: receiverEndpoint, maxQueue: 10, maxHeld: 10, heldExpiryMs: 1_000, inboxFile: join(dir, "receiver", "inbox.json"), logger: noopLogger })
   const receiverListener = InboxListener({
-    token: "receiver-token", maxBodyBytes: 20_000, runtimeDir: join(dir, "runtime"), processId: "receiver", platform: "darwin",
+    token: "receiver-token", maxBodyBytes: 20_000, runtimeDir: join(dir, "runtime"), processId: "receiver", platform: listenerPlatform,
     resolveEndpoint: ({ toEndpointId }) => toEndpointId === receiverEndpoint ? receiverEndpoint : null,
     onMessage: async (message) => await queue.hold(message) ? "held" : "duplicate",
     logger: noopLogger,

tests/config.test.mjs

@@ -1,18 +1,19 @@
 import { test } from "node:test"
 import assert from "node:assert/strict"
+import { join } from "node:path"
 import { resolveConfig, defaultDataDir, validateName, defaultPeerName } from "../dist/config.js"
 
 test("defaultDataDir honors XDG_DATA_HOME and falls back to ~/.local/share", () => {
   assert.equal(defaultDataDir({ XDG_DATA_HOME: "/xdg" }), "/xdg")
-  assert.match(defaultDataDir({}), /\.local\/share$/)
+  assert.ok(defaultDataDir({}).endsWith(join(".local", "share")))
 })
 
 test("resolveConfig applies defaults", () => {
   const cfg = resolveConfig(undefined, { XDG_DATA_HOME: "/xdg" })
-  assert.equal(cfg.storageDir, "/xdg/opencode-plugin-peers")
-  assert.equal(cfg.peersDir, "/xdg/opencode-plugin-peers/peers.d")
-  assert.equal(cfg.inboxFile, "/xdg/opencode-plugin-peers/inbox.json")
-  assert.equal(cfg.spoolDir, "/xdg/opencode-plugin-peers/spool")
+  assert.equal(cfg.storageDir, join("/xdg", "opencode-plugin-peers"))
+  assert.equal(cfg.peersDir, join("/xdg", "opencode-plugin-peers", "peers.d"))
+  assert.equal(cfg.inboxFile, join("/xdg", "opencode-plugin-peers", "inbox.json"))
+  assert.equal(cfg.spoolDir, join("/xdg", "opencode-plugin-peers", "spool"))
   assert.equal(cfg.inboundPolicy, "accept")
   assert.equal(cfg.peerPermissions, "allow")
   assert.equal(cfg.heartbeatMs, 10_000)
@@ -41,7 +42,7 @@ test("resolveConfig merges user options", () => {
   assert.equal(cfg.peerPermissions, "ask")
   assert.equal(cfg.name, "frontend")
   assert.equal(cfg.maxQueue, 5)
-  assert.equal(cfg.peersDir, "/custom/peers.d")
+  assert.equal(cfg.peersDir, join("/custom", "peers.d"))
 })
 
 test("validateName accepts safe names and rejects dangerous ones", () => {

tests/outbox.test.mjs

@@ -36,7 +36,8 @@ test("outbox durably separates transport receipt from final acknowledgement", as
       acknowledgedAt: 200,
     }), true)
     assert.equal(restarted.get("session-from", "m-1").finalStatus, "delivered")
-    assert.equal((await stat(join(dir, "outbox"))).mode & 0o777, 0o700)
+    // Windows cannot represent POSIX permission bits; stat always reports 0o666.
+    if (process.platform !== "win32") assert.equal((await stat(join(dir, "outbox"))).mode & 0o777, 0o700)
   } finally {
     await rm(dir, { recursive: true, force: true })
   }

tests/queue-concurrency.test.mjs

@@ -5,13 +5,14 @@ import { mkdtemp, readdir, rm, stat, utimes, writeFile } from "node:fs/promises"
 import { tmpdir } from "node:os"
 import { join } from "node:path"
 import { setTimeout as delay } from "node:timers/promises"
+import { fileURLToPath } from "node:url"
 
-const worker = new URL("./fixtures/queue-worker.mjs", import.meta.url)
-const staleRaceWorker = new URL("./fixtures/stale-lock-race-worker.mjs", import.meta.url)
+const worker = fileURLToPath(new URL("./fixtures/queue-worker.mjs", import.meta.url))
+const staleRaceWorker = fileURLToPath(new URL("./fixtures/stale-lock-race-worker.mjs", import.meta.url))
 
 function runWorker(dir, name, id, text, maxQueue, operation = "enqueue") {
   return new Promise((resolve, reject) => {
-    const child = spawn(process.execPath, [worker.pathname, dir, name, id, text, String(maxQueue), operation], {
+    const child = spawn(process.execPath, [worker, dir, name, id, text, String(maxQueue), operation], {
       stdio: ["ignore", "pipe", "pipe"],
     })
     let stdout = ""
@@ -39,7 +40,7 @@ async function releaseWorkers(dir, count) {
 }
 
 function startStaleRaceWorker(dir, role, messageId, payloadBytes = 0) {
-  const child = spawn(process.execPath, [staleRaceWorker.pathname, dir, role, messageId, String(payloadBytes)], {
+  const child = spawn(process.execPath, [staleRaceWorker, dir, role, messageId, String(payloadBytes)], {
     stdio: ["ignore", "pipe", "pipe"],
   })
   let stdout = ""

tests/queue.test.mjs

@@ -7,6 +7,8 @@ import { MessageQueue, RateLimiter } from "../dist/queue.js"
 import { gateMessage, isLoopMessage } from "../dist/gating.js"
 
 const noopLogger = async () => {}
+// Windows cannot represent POSIX permission bits; stat always reports 0o666.
+const isPosix = process.platform !== "win32"
 
 const msg = (id) => ({
   id,
@@ -62,7 +64,7 @@ test("queue: stores queued messages in an endpoint spool", async () => {
     const [file] = await readdir(queuedDir)
     const entry = join(queuedDir, file)
     assert.deepEqual(JSON.parse(await readFile(entry, "utf8")).message.id, "durable")
-    assert.equal((await stat(entry)).mode & 0o777, 0o600)
+    if (isPosix) assert.equal((await stat(entry)).mode & 0o777, 0o600)
   } finally {
     await rm(dir, { recursive: true, force: true })
   }
@@ -148,7 +150,7 @@ test("queue: repairs permissions on an existing endpoint directory", async () =>
 
     const q = MessageQueue({ endpointId: "endpoint-a", maxQueue: 1, maxHeld: 1, inboxFile: join(dir, "inbox.json"), logger: noopLogger })
     assert.equal(q.enqueue(msg("secure-endpoint")), true)
-    assert.equal((await stat(endpointDir)).mode & 0o777, 0o700)
+    if (isPosix) assert.equal((await stat(endpointDir)).mode & 0o777, 0o700)
   } finally {
     await rm(dir, { recursive: true, force: true })
   }

tests/real-opencode.test.mjs

@@ -144,6 +144,7 @@ function v2Message(from, to, id, text) {
 }
 
 test("real OpenCode hosts provide busy exact injection, restart recovery, permission boundaries, v1 interop, and held command ACKs", { timeout: 60_000 }, async (t) => {
+  if (process.platform === "win32") return t.skip("transport liveness assertions are UDS-specific; POSIX-only for now")
   const found = spawnSync("which", ["opencode"], { encoding: "utf8" })
   if (found.status !== 0 || !found.stdout.trim()) return t.skip("opencode binary is unavailable")
   const binary = found.stdout.trim()

tests/registry.test.mjs

@@ -6,6 +6,8 @@ import { join } from "node:path"
 import { Registry, newInstanceId, pidAlive, uniqueName } from "../dist/registry.js"
 
 const noopLogger = async () => {}
+// Windows cannot represent POSIX permission bits; stat always reports 0o666.
+const isPosix = process.platform !== "win32"
 
 async function makeDir() {
   return mkdtemp(join(tmpdir(), "peers-registry-"))
@@ -43,8 +45,10 @@ test("start writes a 0600 entry file with expected fields", async () => {
     assert.equal(entry.inboundPolicy, "accept")
     assert.equal(entry.inboxUrl, "http://127.0.0.1:5000")
     assert.ok(entry.heartbeatAt > 0)
-    const mode = (await stat(join(dir, files[0]))).mode & 0o777
-    assert.equal(mode, 0o600)
+    if (isPosix) {
+      const mode = (await stat(join(dir, files[0]))).mode & 0o777
+      assert.equal(mode, 0o600)
+    }
     await reg.stop()
     assert.deepEqual(await readdir(dir), [])
   } finally {
@@ -192,7 +196,7 @@ test("v2 registry publishes one endpoint per session plus the most-recent v1 com
       entry: JSON.parse(await readFile(join(dir, file), "utf8")),
       mode: (await stat(join(dir, file))).mode & 0o777,
     })))
-    assert.ok(entries.every(({ mode }) => mode === 0o600))
+    if (isPosix) assert.ok(entries.every(({ mode }) => mode === 0o600))
 
     const v2 = entries.filter(({ entry }) => entry.version === 2).map(({ entry }) => entry)
     assert.deepEqual(v2.map((entry) => entry.endpointId).sort(), ["session-alpha", "session-beta"])

tests/transport.test.mjs

@@ -10,6 +10,8 @@ import { Sender } from "../dist/sender.js"
 import { LocalTransport } from "../dist/transport.js"
 
 const noopLogger = async () => {}
+// Node has no Unix-domain socket support on Windows; the UDS tests are POSIX-only.
+const udsSkip = process.platform === "win32" ? "Unix-domain sockets are unavailable on Windows" : false
 
 const v2Message = (toEndpointId) => ({
   version: 2,
@@ -22,7 +24,7 @@ const v2Message = (toEndpointId) => ({
   sentAt: Date.now(),
 })
 
-test("local UDS transport authenticates and routes v2 exactly while v1 uses compatibility endpoint", async () => {
+test("local UDS transport authenticates and routes v2 exactly while v1 uses compatibility endpoint", { skip: udsSkip }, async () => {
   const runtimeDir = await mkdtemp(join(tmpdir(), "peers-transport-"))
   const routed = []
   const acknowledgements = []
@@ -110,7 +112,7 @@ test("local UDS transport authenticates and routes v2 exactly while v1 uses comp
   }
 })
 
-test("UDS startup rejects a live process-id collision without unlinking the owner", async () => {
+test("UDS startup rejects a live process-id collision without unlinking the owner", { skip: udsSkip }, async () => {
   const runtimeDir = await mkdtemp(join(tmpdir(), "peers-transport-collision-"))
   const received = []
   const first = InboxListener({
@@ -150,7 +152,7 @@ test("UDS startup rejects a live process-id collision without unlinking the owne
   }
 })
 
-test("UDS startup removes a stale socket left by a crashed owner", async () => {
+test("UDS startup removes a stale socket left by a crashed owner", { skip: udsSkip }, async () => {
   const runtimeDir = await mkdtemp(join(tmpdir(), "peers-transport-stale-"))
   const socketPath = join(runtimeDir, "stale-process.sock")
   const child = spawn(process.execPath, [
@@ -187,7 +189,7 @@ test("UDS startup removes a stale socket left by a crashed owner", async () => {
   }
 })
 
-test("UDS start failures close the server and leave no socket artifact", async () => {
+test("UDS start failures close the server and leave no socket artifact", { skip: udsSkip }, async () => {
   const runtimeDir = await mkdtemp(join(tmpdir(), "peers-transport-failure-"))
   const listener = InboxListener({
     token: "secret",