fix(state): serialize concurrent updateState/saveState with file lock and atomic write

plugins/codex/scripts/lib/state.mjs

@@ -12,6 +12,10 @@ const STATE_FILE_NAME = "state.json";
 const JOBS_DIR_NAME = "jobs";
 const MAX_JOBS = 50;
 
+const LOCK_FILE_NAME = "state.lock";
+const LOCK_RETRY_DELAY_MS = 50;
+const LOCK_TIMEOUT_MS = 10000;
+
 function nowIso() {
   return new Date().toISOString();
 }
@@ -55,6 +59,84 @@ export function ensureStateDir(cwd) {
   fs.mkdirSync(resolveJobsDir(cwd), { recursive: true });
 }
 
+// ---------------------------------------------------------------------------
+// File locking — serializes concurrent updateState / saveState calls so the
+// read-modify-write cycle in updateState() is atomic across processes.
+// Uses O_EXCL ("wx") for atomic lock-file creation and records the PID so
+// stale locks from crashed processes can be reclaimed.
+// ---------------------------------------------------------------------------
+
+function resolveLockFile(cwd) {
+  return path.join(resolveStateDir(cwd), LOCK_FILE_NAME);
+}
+
+function isProcessAlive(pid) {
+  try {
+    process.kill(pid, 0);
+    return true;
+  } catch {
+    return false;
+  }
+}
+
+function acquireLock(cwd) {
+  ensureStateDir(cwd);
+  const lockFile = resolveLockFile(cwd);
+  const deadline = Date.now() + LOCK_TIMEOUT_MS;
+  while (Date.now() < deadline) {
+    try {
+      const fd = fs.openSync(lockFile, "wx");
+      fs.writeFileSync(fd, String(process.pid));
+      fs.closeSync(fd);
+      return true;
+    } catch (err) {
+      if (err.code !== "EEXIST") throw err;
+      // Lock exists — check if the owner is still alive.
+      try {
+        const content = fs.readFileSync(lockFile, "utf8").trim();
+        const pid = parseInt(content, 10);
+        // Treat NaN, 0, or dead PID as stale/corrupt.
+        if (!pid || !isProcessAlive(pid)) {
+          // Re-read before unlinking: another process may have already
+          // reclaimed this lock with a fresh PID since our first read.
+          const recheck = fs.readFileSync(lockFile, "utf8").trim();
+          if (recheck === content) {
+            fs.unlinkSync(lockFile);
+          }
+          continue;
+        }
+      } catch {
+        // Lock file was removed between checks — retry immediately.
+        continue;
+      }
+      const sleepEnd = Date.now() + LOCK_RETRY_DELAY_MS;
+      while (Date.now() < sleepEnd) {
+        /* spin-wait; short duration, avoids timer overhead */
+      }
+    }
+  }
+  return false;
+}
+
+function releaseLock(cwd) {
+  try {
+    fs.unlinkSync(resolveLockFile(cwd));
+  } catch {
+    // Lock already removed — nothing to do.
+  }
+}
+
+function withStateLock(cwd, fn) {
+  if (!acquireLock(cwd)) {
+    throw new Error(`Timed out waiting for state lock: ${resolveLockFile(cwd)}`);
+  }
+  try {
+    return fn();
+  } finally {
+    releaseLock(cwd);
+  }
+}
+
 export function loadState(cwd) {
   const stateFile = resolveStateFile(cwd);
   if (!fs.existsSync(stateFile)) {
@@ -89,7 +171,12 @@ function removeFileIfExists(filePath) {
   }
 }
 
-export function saveState(cwd, state) {
+/**
+ * Internal save — caller must already hold the state lock.
+ * Writes atomically (temp file + rename) so concurrent readers never see a
+ * partially-written state.json.
+ */
+function saveStateLocked(cwd, state) {
   const previousJobs = loadState(cwd).jobs;
   ensureStateDir(cwd);
   const nextJobs = pruneJobs(state.jobs ?? []);
@@ -111,14 +198,27 @@ export function saveState(cwd, state) {
     removeFileIfExists(job.logFile);
   }
 
-  fs.writeFileSync(resolveStateFile(cwd), `${JSON.stringify(nextState, null, 2)}\n`, "utf8");
+  const stateFile = resolveStateFile(cwd);
+  const tempFile = `${stateFile}.tmp.${process.pid}`;
+  fs.writeFileSync(tempFile, `${JSON.stringify(nextState, null, 2)}\n`, "utf8");
+  fs.renameSync(tempFile, stateFile);
   return nextState;
 }
 
+/**
+ * Persist state, acquiring the lock first to avoid racing with concurrent
+ * updateState / saveState callers.
+ */
+export function saveState(cwd, state) {
+  return withStateLock(cwd, () => saveStateLocked(cwd, state));
+}
+
 export function updateState(cwd, mutate) {
-  const state = loadState(cwd);
-  mutate(state);
-  return saveState(cwd, state);
+  return withStateLock(cwd, () => {
+    const state = loadState(cwd);
+    mutate(state);
+    return saveStateLocked(cwd, state);
+  });
 }
 
 export function generateJobId(prefix = "job") {

plugins/codex/scripts/session-lifecycle-hook.mjs

@@ -13,7 +13,7 @@ import {
   sendBrokerShutdown,
   teardownBrokerSession
 } from "./lib/broker-lifecycle.mjs";
-import { loadState, resolveStateFile, saveState } from "./lib/state.mjs";
+import { loadState, resolveStateFile, updateState } from "./lib/state.mjs";
 import { TRANSCRIPT_PATH_ENV } from "./lib/claude-session-transfer.mjs";
 import { resolveWorkspaceRoot } from "./lib/workspace.mjs";
 
@@ -68,9 +68,11 @@ function cleanupSessionJobs(cwd, sessionId) {
     }
   }
 
-  saveState(workspaceRoot, {
-    ...state,
-    jobs: state.jobs.filter((job) => job.sessionId !== sessionId)
+  // Use updateState so the read-modify-write is atomic under the state lock.
+  // This prevents a concurrent upsertJob from being silently dropped (and its
+  // artifacts deleted) when the stale snapshot loaded above is saved back.
+  updateState(workspaceRoot, (currentState) => {
+    currentState.jobs = currentState.jobs.filter((job) => job.sessionId !== sessionId);
   });
 }