fix: retain native turn work across transient persistence failures

lib/app/ports.js

@@ -23,7 +23,7 @@
  *                                     epoch and turn counter; calls without one keep the legacy session namespace.
  * @property {(sessionId: string, note: object, scope?: {turn: number, sessionEpoch?: number}) => Promise<void>} append
  * @property {(sessionId: string, scope?: {turn: number, sessionEpoch?: number}) => Promise<object[]>} read
- * @property {(sessionId: string, scope?: {turn: number, sessionEpoch?: number}) => Promise<{ notes: object[], commit: () => Promise<void> }>} take
+ * @property {(sessionId: string, scope?: {turn: number, sessionEpoch?: number}) => Promise<{ notes: object[], commit: () => Promise<void>, release: () => void }>} take
  * @property {(sessionId: string, scope?: {turn: number, sessionEpoch?: number}) => Promise<void>} clear
  *
  * @typedef {object} EditBaseline       Each file's content hash before the turn first edited it.

lib/app/turn-end.js

@@ -22,9 +22,12 @@ const SETTLING_WAIT_MS = 2_000;
 const sameTurn = (session, expected) => session.turnId === expected.turnId && session.turn === expected.turn && session.sessionEpoch === expected.sessionEpoch;
 
 // A turn end that began later (a re-judge of the same turn) owns `settling` from then on; an earlier one leaves it.
-const settleTurn = (ports, sessionId, opened) => ports.sessions.update(sessionId, (stored) => {
+// A native turn whose evidence could not be read reopens, so the host's next turn end or recovery delivers it again.
+const settleTurn = (ports, sessionId, opened, reopen = false) => ports.sessions.update(sessionId, (stored) => {
   const current = sessionOf(stored);
-  return sameTurn(current, opened) && current.settling === opened.settling ? settledSession(current) : undefined;
+  if (!sameTurn(current, opened) || current.settling !== opened.settling) return undefined;
+  const settled = settledSession(current);
+  return reopen ? { ...settled, open: true, waiting: opened.waiting } : settled;
 });
 
 async function closeTurn({ sessions, clock }, sessionId, turnId, expected) {
@@ -51,6 +54,7 @@ async function reportTurn({ identity, sessions, reports, clock, confirmTurnCompl
   const queued = turnReportsOf(identity, due, { reason, checks, source });
   const retries = confirmTurnCompletion ? pendingReportsOf(session, sessionId, opened.turn - 1) : [];
   const controls = [...session.pendingControls, ...(verified ? trialControlsOf(identity, session, sessionId, opened, checks) : [])];
+  const result = { queued, controls: controls.length + retries.length, strategies: due.map((asset) => asset.strategyText) };
   try {
     await reports.enqueue([...retries.map(({ entry }) => entry), ...queued, ...controls]);
   } catch (error) {
@@ -60,7 +64,9 @@ async function reportTurn({ identity, sessions, reports, clock, confirmTurnCompl
       const pending = confirmTurnCompletion ? withPendingReports(current, due, queued) : current;
       return withPendingControls(pending, controls);
     });
-    throw error;
+    // A native turn's reports wait in the session with their evidence, so its work is still recorded before the error.
+    if (!confirmTurnCompletion) throw error;
+    return { ...result, failure: error };
   }
   await sessions.update(sessionId, (stored) => {
     const current = sessionOf(stored);
@@ -70,7 +76,7 @@ async function reportTurn({ identity, sessions, reports, clock, confirmTurnCompl
     const labelable = opened.isCommand ? session.labelable : queued.map(({ report }) => reportedReuseOf(report));
     return { ...withoutPendingControls(session, controls), labelable };
   });
-  return { queued, controls: controls.length + retries.length, strategies: due.map((asset) => asset.strategyText) };
+  return result;
 }
 
 // Keyed by withholding, so turn ends that overlap neither drop each other's pending controls nor queue one twice.
@@ -167,6 +173,20 @@ async function rejudgeWorkTurn(ports, sessionId, turnId, { answer, model, projec
   }
 }
 
+/** The closed turn's evidence, read before anything is written; null once its session epoch has moved on. */
+async function readTurnEvidence(ports, sessionId, opened, recordOf) {
+  if ((await ports.sessions.read(sessionId)).sessionEpoch !== opened.sessionEpoch) return null;
+  const observed = await recordOf?.(opened);
+  if ((await ports.sessions.read(sessionId)).sessionEpoch !== opened.sessionEpoch) return null;
+  const notes = await ports.notes.take(sessionId, opened);
+  try {
+    return { observed, notes, edits: await ports.baseline.take(sessionId, opened.turnId, opened) };
+  } catch (error) {
+    notes.release();
+    throw error;
+  }
+}
+
 /**
  * Closes the open turn (only `turnId`, when given) with how it ended — 'answer', 'aborted', 'refusal' or 'error' —
  * and what it did: `recordOf(openedTurn)` when the host keeps its own record (a transcript), else the tool notes. Every
@@ -175,19 +195,13 @@ async function rejudgeWorkTurn(ports, sessionId, turnId, { answer, model, projec
  * turn joins the session's range for distillation; `isDistillable` says the range passes the gate now. `strategies`
  * holds the text of each strategy it reported. Null when no such turn was open.
  */
-async function recordTurn(ports, sessionId, opened, { reason, answer, model, projectDir, recordOf, source }) {
-  if ((await ports.sessions.read(sessionId)).sessionEpoch !== opened.sessionEpoch) return null;
-  const [notes, edits] = await Promise.all([
-    ports.notes.take(sessionId, opened), ports.baseline.take(sessionId, opened.turnId, opened),
-  ]);
-  const observed = await recordOf?.(opened);
-  if ((await ports.sessions.read(sessionId)).sessionEpoch !== opened.sessionEpoch) return null;
+async function recordTurn(ports, sessionId, opened, { reason, answer, model, projectDir, source }, { observed, notes, edits }) {
   const ofThisTurn = (entry) => !entry.turnId || !opened.turnId || entry.turnId === opened.turnId;
   const record = observed ?? turnRecordOf(notes.notes.filter(ofThisTurn), { pairShellCalls: ports.confirmTurnCompletion });
   const finished = { ...record, answer: String(answer || record.answer || '').trim() };
   const verified = !ports.confirmTurnCompletion || (reason === 'answer' && finished.answer
     && finished.checks.length > 0 && finished.checks.every(check => verdictOf(check) !== 'unknown'));
-  const { queued, controls, strategies } = await reportTurn(ports, sessionId, opened, { reason, checks: finished.checks, source, verified });
+  const { queued, controls, strategies, failure } = await reportTurn(ports, sessionId, opened, { reason, checks: finished.checks, source, verified });
   if ((await ports.sessions.read(sessionId)).sessionEpoch !== opened.sessionEpoch) return null;
   const reuses = queued.map(({ report }) => reportedReuseOf(report));
   if (verified) await ports.retention.remember({ sessionId, projectDir, reuses, editedBefore: edits.before });
@@ -197,7 +211,7 @@ async function recordTurn(ports, sessionId, opened, { reason, answer, model, pro
   if (!isWork) await ports.candidates.update((pending) => rangeInEpoch(pending, sessionId, opened.sessionEpoch)
     ? withSessionActivity(pending, sessionId, ports.clock.now()) : undefined);
   const isDistillable = isWork && gateOf(rangeInEpoch(await ports.candidates.read(), sessionId, opened.sessionEpoch)?.turns ?? []).passed;
-  return { turn: opened.turn, record: finished, reports: queued.map(({ report }) => report), strategies, isDistillable, controls };
+  return { turn: opened.turn, record: finished, reports: queued.map(({ report }) => report), strategies, isDistillable, controls, failure };
 }
 
 export async function turnEnded(ports, { sessionId, turnId, reason, answer = '', model = null, projectDir, recordOf = null, source = OBSERVED, expectedTurn }) {
@@ -208,15 +222,18 @@ export async function turnEnded(ports, { sessionId, turnId, reason, answer = '',
     await scheduleDueDistillation(ports);
     return null;
   }
-  let ended;
+  let ended, evidenceRead = false;
   try {
-    ended = await recordTurn(ports, sessionId, opened, { reason, answer, model, projectDir, recordOf, source });
+    const evidence = await readTurnEvidence(ports, sessionId, opened, recordOf);
+    evidenceRead = true;
+    if (evidence) ended = await recordTurn(ports, sessionId, opened, { reason, answer, model, projectDir, source }, evidence);
   } finally {
-    await settleTurn(ports, sessionId, opened);
+    await settleTurn(ports, sessionId, opened, ports.confirmTurnCompletion && !evidenceRead);
   }
   if (!ended) return null;
-  const { controls, ...result } = ended;
+  const { controls, failure, ...result } = ended;
   await scheduleDueDistillation(ports, result.reports.length > 0 || controls > 0 ? [TASKS.DRAIN] : []);
+  if (failure) throw failure;
   return result;
 }
 

lib/runtime/store/file-lock.js

@@ -223,12 +223,23 @@ export function readStateLines(target) {
 
 export function takeStateLines(target, now = Date.now()) {
   const claims = [...staleClaimsOf(target, now), target].map((source) => claimed(source, target)).filter(Boolean);
-  return {
-    lines: claims.flatMap(linesOf),
-    commit: () => {
-      for (const claim of claims) fs.rmSync(claim, { force: true });
-    },
+  // 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 = () => {
+    const releasedAt = new Date(Date.now() - 2 * CLAIM_STALE_MS);
+    for (const claim of claims) {
+      try { fs.utimesSync(claim, releasedAt, releasedAt); } catch { /* Ordinary stale recovery can retry an inaccessible claim. */ }
+    }
+  };
+  const commit = () => {
+    for (const claim of claims) fs.rmSync(claim, { force: true });
   };
+  try {
+    return { lines: claims.flatMap(linesOf), commit, release };
+  } catch (error) {
+    release();
+    throw error;
+  }
 }
 
 function isStale(lockDir, now) {

lib/runtime/store/file-stores.js

@@ -42,8 +42,8 @@ function linesStore(target) {
     append: async (value) => appendStateLine(target, value),
     read: async () => readStateLines(target),
     take: async () => {
-      const { lines, commit } = takeStateLines(target);
-      return { lines, commit: async () => commit() };
+      const { lines, commit, release } = takeStateLines(target);
+      return { lines, commit: async () => commit(), release };
     },
     clear: async () => removeState(target),
   };
@@ -151,8 +151,8 @@ export function fileStoresOf({ dir, agent }) {
     append: (sessionId, note, scope) => linesStore(evidenceFile('turn', sessionId, scope)).append(note),
     read: (sessionId, scope) => linesStore(evidenceFile('turn', sessionId, scope)).read(),
     take: async (sessionId, scope) => {
-      const { lines, commit } = await linesStore(evidenceFile('turn', sessionId, scope)).take();
-      return { notes: lines, commit };
+      const { lines, commit, release } = await linesStore(evidenceFile('turn', sessionId, scope)).take();
+      return { notes: lines, commit, release };
     },
     clear: (sessionId, scope) => linesStore(evidenceFile('turn', sessionId, scope)).clear(),
   };

test/app/memory-ports.js

@@ -59,7 +59,8 @@ const lines = () => {
     take: async (key) => {
       const taken = byKey.get(key) ?? [];
       byKey.delete(key);
-      return { lines: taken, commit: async () => {} };
+      const release = () => byKey.set(key, [...taken, ...(byKey.get(key) ?? [])]);
+      return { lines: taken, commit: async () => {}, release };
     },
     clear: async (key) => byKey.delete(key),
   };
@@ -101,8 +102,8 @@ export function memoryPorts({ assets = [ASSET], recipes = [], stepAssets = [], d
       append: (id, note, scope) => notes.append(evidenceKey(id, scope), note),
       read: (id, scope) => notes.read(evidenceKey(id, scope)),
       take: async (id, scope) => {
-        const { lines: taken, commit } = await notes.take(evidenceKey(id, scope));
-        return { notes: taken, commit };
+        const { lines: taken, commit, release } = await notes.take(evidenceKey(id, scope));
+        return { notes: taken, commit, release };
       },
       clear: (id, scope) => notes.clear(evidenceKey(id, scope)),
     },

test/app/turn-lifecycle.test.js

@@ -1,5 +1,5 @@
 import assert from 'node:assert/strict';
-import { mkdtempSync, readdirSync, readFileSync, rmSync, utimesSync, writeFileSync } from 'node:fs';
+import fs, { mkdtempSync, readdirSync, readFileSync, rmSync, utimesSync, writeFileSync } from 'node:fs';
 import { tmpdir } from 'node:os';
 import { join } from 'node:path';
 import { test } from 'node:test';
@@ -458,6 +458,17 @@ test('a turn end that cannot queue its reports leaves the injections unreported
   assert.equal(sessionOfStored(stored).settling, null, 'the next prompt does not wait for an end that failed');
 });
 
+test('a turn end that cannot queue its reports keeps no retention for reuses a later end reports again', async () => {
+  const { ports, retained } = memoryPorts();
+  await verifiedTurn(ports, 's1');
+  await beforeEdit(ports, { sessionId: 's1', file: '/proj/lib/redis.js' });
+  const enqueue = ports.reports.enqueue;
+  ports.reports.enqueue = async () => { throw new Error('killed'); };
+  await assert.rejects(end(ports, 's1'), /killed/);
+  ports.reports.enqueue = enqueue;
+  assert.deepEqual(retained, []);
+});
+
 test('a turn whose work tree cannot be read still joins the range, without its project', async () => {
   for (const project of [{ snapshot: async () => { throw new Error('git failed'); } }, undefined]) {
     const { ports } = memoryPorts();
@@ -706,15 +717,92 @@ test('the current native epoch replaces an older range with its own snapshot bas
   assert.deepEqual(range.turns.map(({ prompt }) => prompt), ['New project']);
 });
 
-test('a verified native report retries with its original evidence after a local queue failure', async () => {
+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) => {
+    const { ports, queued, retained } = memoryPorts();
+    const dir = mkdtempSync(join(tmpdir(), 'evolver-native-read-retry-'));
+    t.after(() => rmSync(dir, { recursive: true, force: true }));
+    const { sessions, candidates, notes, baseline } = fileStoresOf({ dir, agent: 'testagent' });
+    Object.assign(ports, { sessions, candidates, notes, baseline, confirmTurnCompletion: true, modelForTurn: () => NATIVE_ROUTE });
+    const sessionId = 's1', turnId = 't1';
+    await promptSubmitted(ports, { sessionId, turnId, prompt: PROMPT, ...WAIT });
+    const expectedTurn = await sessions.read(sessionId);
+    let reads = 0;
+    const recordOf = read === 'terminal record' ? () => {
+      if (++reads === 1) throw new Error('temporary read failure');
+      return NATIVE_PASS;
+    } : null;
+    let editedFile;
+    if (!recordOf) {
+      editedFile = join(dir, 'fixture.js');
+      await baseline.remember(sessionId, editedFile, turnId, expectedTurn);
+      const call = { kind: 'shell', command: 'npm test', callId: 'check' };
+      await toolFinished(ports, { sessionId, turnId, note: { ...call, running: true } });
+      await toolFinished(ports, { sessionId, turnId, note: { ...call, exit: 0 } });
+      await toolFinished(ports, { sessionId, turnId, note: { kind: 'answer', text: NATIVE_PASS.answer } });
+      const readFile = fs.readFileSync;
+      const stem = read === 'saved tool notes' ? 'testagent-turn-' : 'testagent-edits-';
+      t.mock.method(fs, 'readFileSync', (file, ...args) => {
+        if (typeof file === 'string' && file.startsWith(dir) && file.includes(stem) && file.endsWith('.taking') && ++reads === 1) {
+          throw Object.assign(new Error('temporary read failure'), { code: 'EACCES' });
+        }
+        return readFile(file, ...args);
+      });
+    }
+    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(), {});
+
+    const result = await deliver();
+    assert.deepEqual(result?.record.checks, NATIVE_PASS.checks);
+    assert.equal(await deliver(), null);
+    assert.deepEqual(queued.map(({ report }) => [report.task_id, report.outcome]), [['testagent:s1:1', 'success']]);
+    const range = (await candidates.read())[sessionId];
+    assert.deepEqual(range?.model, NATIVE_ROUTE);
+    assert.deepEqual(range?.turns.map(({ turn }) => turn), [1]);
+    assert.equal(retained.length, 1);
+    if (editedFile) assert.equal(retained[0].editedBefore.has(editedFile), true);
+  });
+}
+
+test('a native evidence read failure cannot reopen a turn after a newer prompt', async () => {
+  const { ports, queued } = memoryPorts();
+  ports.confirmTurnCompletion = true;
+  await promptSubmitted(ports, { sessionId: 's1', turnId: 'old', prompt: PROMPT, ...WAIT });
+  const expectedTurn = await ports.sessions.read('s1');
+  await assert.rejects(end(ports, 's1', { turnId: 'old', expectedTurn, recordOf: async () => {
+    await promptSubmitted(ports, { sessionId: 's1', turnId: 'new', prompt: 'A new task', ...WAIT });
+    throw new Error('temporary read failure');
+  } }), /temporary read failure/);
+
+  assert.equal(await end(ports, 's1', { turnId: 'old', expectedTurn, recordOf: () => NATIVE_PASS }), null);
+  const current = await ports.sessions.read('s1');
+  assert.equal(current.turnId, 'new');
+  assert.equal(current.open, true);
+  assert.deepEqual(queued, []);
+  assert.deepEqual(await ports.candidates.read(), {});
+});
+
+test('a verified native report retries with its original evidence and keeps its work after a local queue failure', async (t) => {
   const second = { ...ASSET, asset_id: 'sha256:' + 'b'.repeat(64), signals_match: ['next-fixture'] };
-  const { ports, queued } = memoryPorts({ assets: [ASSET, second] });
+  const { ports, queued, retained } = memoryPorts({ assets: [ASSET, second] });
+  const dir = mkdtempSync(join(tmpdir(), 'evolver-native-queue-retry-'));
+  t.after(() => rmSync(dir, { recursive: true, force: true }));
+  const { sessions, candidates, notes, baseline } = fileStoresOf({ dir, agent: 'testagent' });
+  Object.assign(ports, { sessions, candidates, notes, baseline, modelForTurn: () => NATIVE_ROUTE });
   ports.confirmTurnCompletion = true;
   await promptSubmitted(ports, { sessionId: 's1', turnId: 't1', prompt: PROMPT, ...WAIT });
   const enqueue = ports.reports.enqueue;
   ports.reports.enqueue = async () => { throw new Error('local queue unavailable'); };
   await assert.rejects(end(ports, 's1', { turnId: 't1', recordOf: () => NATIVE_PASS }), /local queue unavailable/);
   ports.reports.enqueue = enqueue;
+  const range = (await ports.candidates.read()).s1;
+  assert.deepEqual(range?.turns.map(({ turn }) => turn), [1]);
+  assert.deepEqual(range.model, NATIVE_ROUTE);
+  assert.equal((await sessions.read('s1')).open, false);
+  assert.equal(retained.length, 1);
 
   await promptSubmitted(ports, { sessionId: 's1', turnId: 't2', prompt: 'fix the next-fixture retry policy', ...WAIT });
   await end(ports, 's1', { turnId: 't2', reason: 'aborted', recordOf: () => NATIVE_PASS });

test/runtime/file-stores.test.js

@@ -60,6 +60,21 @@ test('evidence generations keep turns and epochs apart without changing legacy c
   assert.deepEqual([...(await cursor.baseline.take('s1')).before.keys()], ['/legacy.js']);
 });
 
+test('released turn evidence is immediately recoverable alongside newer appends, and pruning keeps it', 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: 'arrived after take' }, scope);
+  first.release();
+
+  cursor.prune();
+  const recovered = await cursor.notes.take('s1', scope);
+  assert.deepEqual(recovered.notes, [{ command: 'first' }, { command: 'arrived after take' }]);
+  await recovered.commit();
+  assert.deepEqual((await cursor.notes.take('s1', scope)).notes, []);
+});
+
 test('pruning expires generation evidence and its claims without touching report claims', async () => {
   const { dir, stores: cursor } = stores();
   const scope = { sessionEpoch: 10, turn: 1 };