fix: keep native evidence complete after a failed claim release

lib/runtime/store/file-lock.js

@@ -187,25 +187,41 @@ function linesOf(file) {
   });
 }
 
-function staleClaimsOf(target, now) {
+function claimFilesOf(target) {
   const prefix = `${path.basename(target)}.`;
   try {
     return fs.readdirSync(path.dirname(target))
       .filter((name) => name.startsWith(prefix) && name.endsWith('.taking'))
-      .map((name) => path.join(path.dirname(target), name))
-      .filter((file) => now - fs.statSync(file).mtimeMs > CLAIM_STALE_MS);
-  } catch {
+      .map((name) => path.join(path.dirname(target), name));
+  } catch (error) {
+    if (error?.code === 'ENOENT') return [];
+    throw error;
+  }
+}
+
+function staleClaimsOf(target, now, requireComplete) {
+  try {
+    return claimFilesOf(target).filter((file) => {
+      try { return now - fs.statSync(file).mtimeMs > CLAIM_STALE_MS; }
+      catch (error) {
+        if (error?.code === 'ENOENT') return false;
+        throw error;
+      }
+    });
+  } catch (error) {
+    if (requireComplete) throw error;
     return [];
   }
 }
 
-function claimed(source, target) {
+function claimed(source, target, requireComplete) {
   const claim = `${target}.${process.pid}.${crypto.randomBytes(6).toString('hex')}.taking`;
   try {
     fs.renameSync(source, claim);
     fs.utimesSync(claim, new Date(), new Date());
     return claim;
-  } catch {
+  } catch (error) {
+    if (requireComplete && error?.code !== 'ENOENT') throw error;
     return null;
   }
 }
@@ -221,8 +237,9 @@ export function readStateLines(target) {
   }
 }
 
-export function takeStateLines(target, now = Date.now()) {
-  const claims = [...staleClaimsOf(target, now), target].map((source) => claimed(source, target)).filter(Boolean);
+export function takeStateLines(target, now = Date.now(), { requireComplete = false } = {}) {
+  const sources = [...staleClaimsOf(target, now, requireComplete), target];
+  const claims = [];
   // A released claim is dated just past stale, so the next take recovers it at once while pruning, which expires
   // evidence days old, keeps it.
   const release = () => {
@@ -235,6 +252,14 @@ export function takeStateLines(target, now = Date.now()) {
     for (const claim of claims) fs.rmSync(claim, { force: true });
   };
   try {
+    for (const source of sources) {
+      const claim = claimed(source, target, requireComplete);
+      if (claim) claims.push(claim);
+    }
+    // Native evidence must wait for every earlier claim, even if returning that claim failed.
+    if (requireComplete && claimFilesOf(target).some((file) => !claims.includes(file))) {
+      throw Object.assign(new Error('Turn evidence is still claimed; retry after its reader finishes.'), { code: 'ELOCKED' });
+    }
     return { lines: claims.flatMap(linesOf), commit, release };
   } catch (error) {
     release();

lib/runtime/store/file-stores.js

@@ -53,7 +53,7 @@ function linesStore(target) {
     append: async (value) => appendStateLine(target, value),
     read: async () => readStateLines(target),
     take: async () => {
-      const { lines, commit, release } = takeStateLines(target);
+      const { lines, commit, release } = takeStateLines(target, Date.now(), { requireComplete: true });
       return { lines, commit: async () => commit(), release };
     },
     clear: async () => removeState(target),

test/app/turn-lifecycle.test.js

@@ -830,8 +830,8 @@ test('the current native epoch replaces an older range with its own snapshot bas
   assert.deepEqual(range.turns.map(({ prompt }) => prompt), ['New project']);
 });
 
-for (const read of ['terminal record', 'saved tool notes', 'edit baseline']) {
-  test(`redelivering a confirmed native turn recovers from a transient ${read} read failure`, async (t) => {
+for (const [read, releaseDenied = false] of [['terminal record'], ['saved tool notes'], ['edit baseline'], ['saved tool notes', true]]) {
+  test(`redelivering a confirmed native turn recovers from a transient ${read} read failure${releaseDenied ? ' with denied claim release' : ''}`, async (t) => {
     const { ports, queued, retained } = memoryPorts();
     const dir = mkdtempSync(join(tmpdir(), 'evolver-native-read-retry-'));
     t.after(() => rmSync(dir, { recursive: true, force: true }));
@@ -861,12 +861,29 @@ for (const read of ['terminal record', 'saved tool notes', 'edit baseline']) {
         }
         return readFile(file, ...args);
       });
+      if (releaseDenied) {
+        const touch = fs.utimesSync;
+        t.mock.method(fs, 'utimesSync', (file, atime, mtime) => {
+          if (typeof file === 'string' && file.startsWith(dir) && file.endsWith('.taking') && Number(atime) < Date.now() - 60_000) {
+            throw Object.assign(new Error('claim release denied'), { code: 'EACCES' });
+          }
+          return touch(file, atime, mtime);
+        });
+      }
     }
     const deliver = () => turnEnded(ports, { sessionId, turnId, reason: 'answer', expectedTurn, recordOf });
     await assert.rejects(deliver, /temporary read failure/);
     assert.equal((await sessions.read(sessionId)).open, true);
     assert.deepEqual(queued, []);
     assert.deepEqual(await candidates.read(), {});
+    if (releaseDenied) {
+      await assert.rejects(deliver, { code: 'ELOCKED' });
+      assert.equal((await sessions.read(sessionId)).open, true);
+      assert.deepEqual(queued, []);
+      assert.deepEqual(await candidates.read(), {});
+      t.mock.restoreAll();
+      for (const name of readdirSync(dir)) if (name.endsWith('.taking')) utimesSync(join(dir, name), new Date(Date.now() - 120_000), new Date(Date.now() - 120_000));
+    }
 
     const result = await deliver();
     assert.deepEqual(result?.record.checks, NATIVE_PASS.checks);
@@ -880,6 +897,26 @@ for (const read of ['terminal record', 'saved tool notes', 'edit baseline']) {
   });
 }
 
+test('an epoch change while taking native notes cannot queue a replacement session report', async () => {
+  const { ports, queued, advance } = memoryPorts();
+  ports.confirmTurnCompletion = true;
+  await promptSubmitted(ports, { sessionId: 's1', turnId: 'old', prompt: PROMPT, ...WAIT });
+  const take = ports.notes.take;
+  ports.notes.take = async (...args) => {
+    const notes = await take(...args);
+    advance(1);
+    await ports.sessions.update('s1', () => ({ turn: 0 }));
+    await promptSubmitted(ports, { sessionId: 's1', turnId: 'new', prompt: PROMPT, ...WAIT });
+    return notes;
+  };
+  assert.equal(await end(ports, 's1', { turnId: 'old', recordOf: () => NATIVE_PASS }), null);
+  assert.deepEqual(queued, []);
+  assert.deepEqual(await ports.candidates.read(), {});
+  const current = await ports.sessions.read('s1');
+  assert.equal(current.open, true);
+  assert.equal(current.turnId, 'new');
+});
+
 test('a native evidence read failure cannot reopen a turn after a newer prompt', async () => {
   const { ports, queued } = memoryPorts();
   ports.confirmTurnCompletion = true;

test/runtime/file-stores.test.js

@@ -8,7 +8,7 @@ import { tmpdir } from 'node:os';
 import { join } from 'node:path';
 import { after, test } from 'node:test';
 
-import { readState, updateState } from '../../lib/runtime/store/file-lock.js';
+import { appendStateLine, readState, takeStateLines, updateState } from '../../lib/runtime/store/file-lock.js';
 import { fileStoresOf, rangeReferenceOf } from '../../lib/runtime/store/file-stores.js';
 
 const dirs = [];
@@ -75,6 +75,33 @@ test('released turn evidence is immediately recoverable alongside newer appends,
   assert.deepEqual((await cursor.notes.take('s1', scope)).notes, []);
 });
 
+test('turn evidence waits for an earlier take without dropping new appends', async () => {
+  const { stores: cursor } = stores();
+  const scope = { sessionEpoch: 10, turn: 1 };
+  await cursor.notes.append('s1', { command: 'first' }, scope);
+  const first = await cursor.notes.take('s1', scope);
+  await cursor.notes.append('s1', { command: 'new' }, scope);
+  await assert.rejects(cursor.notes.take('s1', scope), { code: 'ELOCKED' });
+  await first.commit();
+  const next = await cursor.notes.take('s1', scope);
+  assert.deepEqual(next.notes, [{ command: 'new' }]);
+  await next.commit();
+});
+
+test('queue takes still consume new appends while an earlier take awaits commit', () => {
+  const { dir } = stores();
+  const target = join(dir, 'queue.jsonl');
+  appendStateLine(target, { report: 'first' });
+  const first = takeStateLines(target);
+  appendStateLine(target, { report: 'next' });
+  const next = takeStateLines(target);
+  assert.deepEqual(first.lines, [{ report: 'first' }]);
+  assert.deepEqual(next.lines, [{ report: 'next' }]);
+  first.commit();
+  next.commit();
+  assert.deepEqual(takeStateLines(target).lines, []);
+});
+
 test('pruning expires generation evidence and its claims without touching report claims', async () => {
   const { dir, stores: cursor } = stores();
   const scope = { sessionEpoch: 10, turn: 1 };