fix: preserve shared state on read and replacement failures

lib/agent/driver.js

@@ -171,7 +171,8 @@ export function createDriver({ profile, ports, transcripts, env = process.env, l
         const at = { sessionId, projectDir: profile.payload.projectDir(payload, context) };
         const waitLeft = (budgetOf = 'prompt') => waitBudgetOf(profile.budgets, budgetOf, { lifetime, startedAt });
         return await HANDLERS[event]({ profile, ports, hostEvent, payload, context, at, waitLeft, startedAt, options });
-      } catch {
+      } catch (error) {
+        options.onError?.(error);
         return profile.reply.empty(hostEvent, payload);
       }
     },

lib/channels/hook/runner.js

@@ -13,7 +13,7 @@ import { readPayload } from './stdin.js';
  * One command-hook invocation: the host's payload on stdin, the agent's answer as JSON on stdout, exit 0. Inside
  * this plugin's own headless distillation the hook answers the agent's empty reply and does nothing else.
  */
-export async function runHookProcess(agent, hostEvent, { stdin = process.stdin, stdout = process.stdout, env = process.env, exit = process.exit, profileOf = profileNamed } = {}) {
+export async function runHookProcess(agent, hostEvent, { stdin = process.stdin, stdout = process.stdout, stderr = process.stderr, env = process.env, exit = process.exit, profileOf = profileNamed } = {}) {
   const startedAt = Date.now();
   const write = (answer) => stdout.write(JSON.stringify(answer ?? {}), () => exit(0));
   const profile = await profileOf(agent).catch(() => null);
@@ -22,5 +22,14 @@ export async function runHookProcess(agent, hostEvent, { stdin = process.stdin,
   if (isHeadlessRun(env)) return write(profile.reply.empty(hostEvent));
   const { ports } = environmentFor({ profile, env, lifetime: EPHEMERAL, runBackground: runBackgroundTasks });
   const driver = createDriver({ profile, ports, transcripts, env, lifetime: EPHEMERAL, startedAt });
-  return write(await driver.handle(hostEvent, payload));
+  let diagnostic;
+  const answer = await driver.handle(hostEvent, payload, {
+    onError(error) {
+      if (['EINVALIDSTATE', 'ESTATEIO', 'ELOCKED'].includes(error?.code)) {
+        diagnostic = `Evolver hook skipped a state operation (${error.code}). Check the plugin state directory and retry.\n`;
+      }
+    },
+  });
+  if (diagnostic) return stderr.write(diagnostic, () => write(answer));
+  return write(answer);
 }

lib/runtime/store/file-lock.js

@@ -15,22 +15,41 @@ const CLAIM_STALE_MS = 60 * 1000;
 export function readState(target, fallback) {
   try {
     const parsed = JSON.parse(fs.readFileSync(target, 'utf8'));
-    return parsed && typeof parsed === 'object' && Array.isArray(parsed) === Array.isArray(fallback) ? parsed : fallback;
-  } catch {
-    return fallback;
+    if (!parsed || typeof parsed !== 'object' || Array.isArray(parsed) !== Array.isArray(fallback)) throw Object.assign(new Error(`Invalid state shape in ${target}; repair it before retrying.`), { code: 'EINVALIDSTATE' });
+    return parsed;
+  } catch (error) {
+    if (error?.code === 'ENOENT') return fallback;
+    if (error instanceof SyntaxError) throw Object.assign(new Error(`Invalid JSON in ${target}; repair it before retrying.`), { code: 'EINVALIDSTATE', cause: error });
+    if (error?.code === 'EINVALIDSTATE') throw error;
+    throw Object.assign(new Error(`Could not read state in ${target}; check access and retry.`), { code: 'ESTATEIO', cause: error });
   }
 }
 
 export function writeState(target, value) {
   fs.mkdirSync(path.dirname(target), { recursive: true, mode: 0o700 });
   const temporary = `${target}.${process.pid}.${crypto.randomBytes(6).toString('hex')}.tmp`;
-  fs.writeFileSync(temporary, JSON.stringify(value), { mode: 0o600, flag: 'wx' });
+  const deadline = Date.now() + LOCK_WAIT_MS;
+  let descriptor;
   try {
-    fs.renameSync(temporary, target);
-  } catch (error) {
-    if (!['EEXIST', 'EPERM'].includes(error?.code)) throw error;
-    fs.rmSync(target, { force: true });
-    fs.renameSync(temporary, target);
+    descriptor = fs.openSync(temporary, 'wx', 0o600);
+    try {
+      fs.writeFileSync(descriptor, JSON.stringify(value));
+    } catch (error) {
+      try { fs.closeSync(descriptor); } catch {}
+      throw error;
+    }
+    fs.closeSync(descriptor);
+    for (;;) {
+      try {
+        fs.renameSync(temporary, target);
+        return;
+      } catch (error) {
+        if (!['EEXIST', 'EACCES', 'EPERM', 'EBUSY'].includes(error?.code) || Date.now() >= deadline) throw error;
+        pause(LOCK_RETRY_MS);
+      }
+    }
+  } finally {
+    if (descriptor !== undefined) removeState(temporary);
   }
 }
 
@@ -92,23 +111,34 @@ function acquire(lockPath, deadline) {
 }
 
 function release(lockPath, token) {
-  if (contentOf(lockPath) === token) removeState(lockPath);
+  const deadline = Date.now() + LOCK_WAIT_MS;
+  for (;;) {
+    try {
+      if (fs.readFileSync(lockPath, 'utf8') === token) fs.unlinkSync(lockPath);
+      return;
+    } catch (error) {
+      if (!['EACCES', 'EPERM', 'EBUSY'].includes(error?.code) || Date.now() >= deadline) return;
+      pause(LOCK_RETRY_MS);
+    }
+  }
 }
 
 /**
  * Applies `mutate` to a state file under an exclusive lock, so hook processes sharing it cannot overwrite each
- * other's entries. A dead owner's lock is taken over; one that cannot be had within LOCK_WAIT_MS is given up on
- * and the update runs unguarded, which never blocks a session. `mutate` returns undefined to leave the file.
+ * other's entries. A dead owner's lock is taken over; an unavailable lock refuses the update before reading
+ * or mutating state. `mutate` returns undefined to leave the file.
  */
 export function updateState(target, fallback, mutate) {
   const lockPath = `${target}.lock`;
-  const token = acquire(lockPath, Date.now() + LOCK_WAIT_MS);
+  let token, cause;
+  try { token = acquire(lockPath, Date.now() + LOCK_WAIT_MS); } catch (error) { cause = error; }
+  if (!token) throw Object.assign(new Error('State is busy or unavailable; no update was made', { cause }), { code: 'ELOCKED' });
   try {
     const next = mutate(readState(target, fallback));
     if (next !== undefined) writeState(target, next);
     return next;
   } finally {
-    if (token) release(lockPath, token);
+    release(lockPath, token);
   }
 }
 

test/channels/hook.test.js

@@ -1,4 +1,7 @@
 import assert from 'node:assert/strict';
+import { createHash } from 'node:crypto';
+import { mkdirSync, readFileSync, writeFileSync } from 'node:fs';
+import { join } from 'node:path';
 import { Readable } from 'node:stream';
 import { after, before, test } from 'node:test';
 
@@ -19,12 +22,12 @@ let proxy;
 before(async () => { proxy = await fakeProxy(); });
 after(() => proxy.close());
 
-function invoke(hostEvent, payload, { env = proxy.env, raw } = {}) {
+function invoke(hostEvent, payload, { env = proxy.env, raw, stderr } = {}) {
   return new Promise((resolve) => {
     let written = '';
     const stdout = { write: (text, done) => { written += text; done(); } };
     const stdin = Readable.from([raw ?? JSON.stringify(payload)]);
-    runHookProcess('testhook', hostEvent, { stdin, stdout, env, exit: () => resolve(JSON.parse(written)), profileOf: async () => profile });
+    runHookProcess('testhook', hostEvent, { stdin, stdout, stderr, env, exit: () => resolve(JSON.parse(written)), profileOf: async () => profile });
   });
 }
 
@@ -44,3 +47,18 @@ test('inside the plugin\'s own headless run, a hook answers the empty reply and
 test('a payload that is not JSON still gets an answer', async () => {
   assert.deepEqual(await invoke('UserPromptSubmit', null, { raw: '{not json' }), { empty: 'UserPromptSubmit' });
 });
+
+test('corrupt hook state stays intact and emits a diagnostic alongside the neutral reply', async () => {
+  const dir = profile.stateDir(proxy.env);
+  const key = createHash('sha256').update('s-corrupt').digest('hex').slice(0, 16);
+  const file = join(dir, `testhook-session-${key}.json`);
+  mkdirSync(dir, { recursive: true });
+  writeFileSync(file, '{"private text":');
+  let diagnostic = '';
+  const stderr = { write: (text, done) => { diagnostic += text; done(); } };
+
+  assert.deepEqual(await invoke('UserPromptSubmit', { session_id: 's-corrupt', cwd: '/proj', prompt: PROMPT }, { stderr }), { empty: 'UserPromptSubmit' });
+  assert.match(diagnostic, /EINVALIDSTATE/);
+  assert.doesNotMatch(diagnostic, /private text/);
+  assert.equal(readFileSync(file, 'utf8'), '{"private text":');
+});