fix: preserve state when atomic replacement fails

docs/API_REFERENCE.md

@@ -186,7 +186,7 @@ Session helpers (read-only): `markStale(state, sessionId)`, `clearStale(state, s
 | `class StateStore` | `constructor(projectDir:string)` | `dir`, `file`, private `lockPath` |
 | `.withLock(fn)` | `<T>(fn:()=>T)=>T` | Exclusive-create `.state.lock` (`<pid>@<ts>`); LOCK_TIMEOUT_MS=5s; locks older than LOCK_STALE_MS=15s broken. R5 tradeoff documented in ARCHITECTURE.md |
 | `.load()` | `()=>State` | validateState gate (schema_version, shapes, key/name agreement) → reject to fresh + error; backfillState for legacy files |
-| `.save(state)` | `(state:State)=>void` | Atomic tmp+rename; blocking-sleep retry; direct-write fallback |
+| `.save(state)` | `(state:State)=>void` | Flushed tmp+rename; bounded retries; preserves previous file on failure |
 | `.update(fn)` | `(mutate:(state:State)=>void)=>State` | `withLock(load → mutate → save)` |
 | `LOCK_TIMEOUT_MS` / `LOCK_STALE_MS` | `5000` / `15000` | Exported for tests |
 | `stateDirFor` / `isInsideStateDir` | path helpers | |

docs/ARCHITECTURE.md

@@ -168,7 +168,7 @@ What exists instead, honestly: `stale` (offline proxy, host-verified), `delivery
 - **Locking:** every load→mutate→save runs under `withLock(fn)` (`src/core/store.ts` withLock) — exclusive-create `.state.lock` carrying `<pid>@<ts>`; acquisition retries (10ms busy waits) until `LOCK_TIMEOUT_MS=5s`, locks older than `LOCK_STALE_MS=15s` are presumed abandoned and broken. Bare reads stay lock-free because saves are atomic renames.
   - *Known tradeoff (R5):* a live-but-slow holder whose critical section exceeds 15s can get its lock broken mid-fn, briefly allowing concurrent mutation. All current critical sections are ms-scale synchronous blocks, so this is accepted rather than heartbeat-managed; revisit if async work ever moves inside the lock.
 - `load()` (`src/core/store.ts`): if missing → `emptyState()`. Parses then runs `validateState`: wrong `schema_version`, non-object roots, malformed member rows (bad role label / non-string role_prompt), channel key≠name aliasing, malformed messages/queues/delivered_to — ALL rejected to `emptyState()` + recovery error. Survivors run `backfillState` (`src/core/store.ts`): clamp `max_members∈[2,8]`, default timers for pre-timer files, migrate legacy ROLE-keyed timers to session-id keys (remapping the active role's elapsed onto its member).
-- `save(state)` (`src/core/store.ts`): `mkdirSync(recursive)`, `writeFileSync(tmp)` where tmp = `.state.<pid>.<rand>.tmp`, `renameSync(tmp, file)`. On error: blocking 50ms sleep + retry rename; on second failure direct `writeFileSync(file)`; on third throw with combined message.
+- `save(state)` (`src/core/store.ts`): `mkdirSync(recursive)`, `writeFileSync(tmp, payload, { flush: true })` where tmp = `.state.<pid>.<rand>.tmp`, `renameSync(tmp, file)`. On replacement failure: retry after bounded 50ms and 100ms blocking waits. If all three attempts fail, throw a persistence error with the original cause and preserve the previous file. Temporary files are cleaned up best-effort. Both stores share `replaceStateFile` in `src/core/atomic-file.ts`.
 - `update(fn)` (`src/core/store.ts`): `withLock(load → mutate → save)`.
 
 ## Hooks & Tools (plugin.ts)

src/core/atomic-file.ts

@@ -0,0 +1,27 @@
+import { renameSync, rmSync, writeFileSync } from "node:fs"
+
+/** Keep the last valid file intact if an atomic replacement is unavailable. */
+export function replaceStateFile(file: string, temporary: string, payload: string): void {
+  try {
+    writeFileSync(temporary, payload, { encoding: "utf8", flush: true })
+    const pause = new Int32Array(new SharedArrayBuffer(4))
+    for (const delay of [0, 50, 100]) {
+      // Persistence is synchronous; only this bounded failure path blocks.
+      if (delay) Atomics.wait(pause, 0, 0, delay)
+      try {
+        renameSync(temporary, file)
+        return
+      } catch (error) {
+        if (delay === 100) throw error
+      }
+    }
+  } catch (cause) {
+    throw new Error(`OpenComms: failed to persist state at ${file}`, { cause })
+  } finally {
+    try {
+      rmSync(temporary, { force: true })
+    } catch {
+      // Cleanup is best-effort and must not hide the persistence error.
+    }
+  }
+}

src/core/store.ts

@@ -5,7 +5,7 @@
  * atomic: we serialize to a temp file in the same directory, flush it, then
  * rename over the target. On Windows, `rename` over an existing file is
  * supported by Node's fs.rename (it maps to MoveFileEx with REPLACE_EXISTING),
- * but we defensively retry once after a short delay because antivirus or
+ * but we defensively retry with bounded delays because antivirus or
  * OneDrive can briefly hold a handle.
  *
  * Concurrency: multiple host sessions share one state.json. A bare
@@ -19,7 +19,6 @@
 import {
   mkdirSync,
   readFileSync,
-  renameSync,
   writeFileSync,
   existsSync,
   copyFileSync,
@@ -30,6 +29,7 @@ import {
   type PathOrFileDescriptor,
 } from "node:fs"
 import { dirname, join } from "node:path"
+import { replaceStateFile } from "./atomic-file.js"
 import { randomBytes } from "node:crypto"
 import {
   DEFAULT_MAX_MEMBERS,
@@ -541,30 +541,7 @@ export class StateStore {
     mkdirSync(this.dir, { recursive: true })
     const tmp = join(this.dir, `.state.${process.pid}.${randomBytes(4).toString("hex")}.tmp`)
     const payload = JSON.stringify(state, null, 2)
-    writeFileSync(tmp, payload, "utf8")
-    try {
-      renameSync(tmp, this.file)
-    } catch (error) {
-      // Windows: retry once after a short yielding pause (AV/OneDrive races).
-      // A blocking 50ms stall is acceptable here: the write already happened,
-      // and this path is rare; the atomic-rename guarantee is what matters.
-      try {
-        // save() is intentionally synchronous (callers rely on durability
-        // when it returns), so this rare retry path uses a bounded 50ms
-        // blocking wait rather than async. LOCK acquisition — the path that
-        // can wait seconds — uses yielding sleepAsync instead.
-        Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, 50)
-        renameSync(tmp, this.file)
-      } catch (second) {
-        try {
-          writeFileSync(this.file, payload, "utf8")
-        } catch {
-          throw new Error(
-            `OpenComms: failed to persist state (${(error as Error).message}; ${(second as Error).message})`,
-          )
-        }
-      }
-    }
+    replaceStateFile(this.file, tmp, payload)
   }
 
   /** Convenience: load, mutate, save — all under the cross-process lock. */

src/orchestrator/state.ts

@@ -20,18 +20,9 @@
  * from agent-facing tool arguments.
  */
 
-import {
-  existsSync,
-  mkdirSync,
-  readFileSync,
-  renameSync,
-  writeFileSync,
-  openSync,
-  closeSync,
-  unlinkSync,
-  statSync,
-} from "node:fs"
+import { existsSync, mkdirSync, readFileSync, writeFileSync, openSync, closeSync, unlinkSync, statSync } from "node:fs"
 import { join } from "node:path"
+import { replaceStateFile } from "../core/atomic-file.js"
 import { randomBytes, createHash } from "node:crypto"
 
 export const ORCHESTRATOR_FILE = "orchestrator.json"
@@ -475,23 +466,7 @@ export class OrchestratorStore {
   save(state: OrchestratorState): void {
     mkdirSync(this.dir, { recursive: true })
     const tmp = join(this.dir, `.orchestrator.${process.pid}.${randomBytes(4).toString("hex")}.tmp`)
-    writeFileSync(tmp, JSON.stringify(state, null, 2), "utf8")
-    try {
-      renameSync(tmp, this.file)
-    } catch {
-      // Windows AV/OneDrive races: bounded blocking retry (same discipline as
-      // StateStore.save), then direct write as a last resort.
-      try {
-        Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, 50)
-        renameSync(tmp, this.file)
-      } catch (second) {
-        try {
-          writeFileSync(this.file, JSON.stringify(state, null, 2), "utf8")
-        } catch {
-          throw new Error(`OpenComms: failed to persist orchestrator state (${(second as Error).message})`)
-        }
-      }
-    }
+    replaceStateFile(this.file, tmp, JSON.stringify(state, null, 2))
   }
 
   /** Load, mutate, save under the lock. The mutation runs synchronously. */

test/unit/atomic-save.test.ts

@@ -0,0 +1,115 @@
+import test from "node:test"
+import assert from "node:assert/strict"
+import fs from "node:fs"
+import { syncBuiltinESMExports } from "node:module"
+import { tmpdir } from "node:os"
+import { join } from "node:path"
+import { StateStore, emptyState } from "../../src/core/store.js"
+import { OrchestratorStore, emptyOrchestratorState } from "../../src/orchestrator/state.js"
+
+const stores = [
+  {
+    name: "StateStore",
+    create(dir: string) {
+      const store = new StateStore(dir)
+      return {
+        file: store.file,
+        dir: store.dir,
+        save(revision: number) {
+          const state = { ...emptyState(), revision }
+          store.save(state)
+        },
+      }
+    },
+  },
+  {
+    name: "OrchestratorStore",
+    create(dir: string) {
+      const store = new OrchestratorStore(dir)
+      return {
+        file: store.file,
+        dir: store.dir,
+        save(revision: number) {
+          const state = { ...emptyOrchestratorState(dir), revision }
+          store.save(state)
+        },
+      }
+    },
+  },
+]
+
+for (const fixture of stores) {
+  for (const succeeds of [true, false]) {
+    test(`${fixture.name}: rename ${succeeds ? "retry succeeds" : "failure preserves previous state"}`, (t) => {
+      const dir = fs.mkdtempSync(join(tmpdir(), "opencomms-atomic-"))
+      try {
+        const store = fixture.create(dir)
+        store.save(1)
+        const previous = fs.readFileSync(store.file, "utf8")
+        const rename = fs.renameSync
+        const error = Object.assign(new Error("replacement denied"), { code: "EPERM" })
+        let attempts = 0
+        t.mock.method(fs, "renameSync", (from: fs.PathLike, to: fs.PathLike) => {
+          attempts++
+          if (succeeds && attempts === 2) return rename(from, to)
+          throw error
+        })
+        syncBuiltinESMExports()
+        if (succeeds) {
+          store.save(2)
+          assert.equal(JSON.parse(fs.readFileSync(store.file, "utf8")).revision, 2)
+          assert.equal(attempts, 2)
+        } else {
+          assert.throws(() => store.save(2), /failed to persist/)
+          assert.equal(fs.readFileSync(store.file, "utf8"), previous)
+          assert.ok(attempts >= 2 && attempts <= 3)
+        }
+        assert.deepEqual(
+          fs.readdirSync(store.dir).filter((name) => name.endsWith(".tmp")),
+          [],
+        )
+      } finally {
+        t.mock.restoreAll()
+        syncBuiltinESMExports()
+        fs.rmSync(dir, { recursive: true, force: true })
+      }
+    })
+  }
+}
+for (const fixture of stores) {
+  test(`${fixture.name}: partial temporary write is cleaned up without touching state`, (t) => {
+    const dir = fs.mkdtempSync(join(tmpdir(), "opencomms-atomic-"))
+    try {
+      const store = fixture.create(dir)
+      store.save(1)
+      const previous = fs.readFileSync(store.file, "utf8")
+      const write = fs.writeFileSync
+      const error = Object.assign(new Error("disk full"), { code: "ENOSPC" })
+      t.mock.method(fs, "writeFileSync", (file: fs.PathOrFileDescriptor) => {
+        write(file, "partial")
+        throw error
+      })
+      const rename = t.mock.method(fs, "renameSync")
+      syncBuiltinESMExports()
+      assert.throws(
+        () => store.save(2),
+        (caught: unknown) => {
+          assert.ok(caught instanceof Error)
+          assert.match(caught.message, /failed to persist/)
+          assert.equal(caught.cause, error)
+          return true
+        },
+      )
+      assert.equal(rename.mock.callCount(), 0)
+      assert.equal(fs.readFileSync(store.file, "utf8"), previous)
+      assert.deepEqual(
+        fs.readdirSync(store.dir).filter((name) => name.endsWith(".tmp")),
+        [],
+      )
+    } finally {
+      t.mock.restoreAll()
+      syncBuiltinESMExports()
+      fs.rmSync(dir, { recursive: true, force: true })
+    }
+  })
+}