From 362b2517c28750465926bb3ed136e77cf5881a4d Mon Sep 17 00:00:00 2001 From: "xyanua." Date: Wed, 11 Nov 2026 10:35:23 +0700 Subject: [PATCH] fix: keep native evidence complete after a failed claim release --- lib/runtime/store/file-lock.js | 41 +++++++++++++++++++++++++------- lib/runtime/store/file-stores.js | 2 +- test/app/turn-lifecycle.test.js | 41 ++++++++++++++++++++++++++++++-- test/runtime/file-stores.test.js | 29 +++++++++++++++++++++- 4 files changed, 101 insertions(+), 12 deletions(-) diff --git a/lib/runtime/store/file-lock.js b/lib/runtime/store/file-lock.js index 771677d..b5c9b20 100644 --- a/lib/runtime/store/file-lock.js +++ b/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(); diff --git a/lib/runtime/store/file-stores.js b/lib/runtime/store/file-stores.js index 5f341ea..5987003 100644 --- a/lib/runtime/store/file-stores.js +++ b/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), diff --git a/test/app/turn-lifecycle.test.js b/test/app/turn-lifecycle.test.js index ec03441..7423a2a 100644 --- a/test/app/turn-lifecycle.test.js +++ b/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; diff --git a/test/runtime/file-stores.test.js b/test/runtime/file-stores.test.js index c078e6a..fb128a4 100644 --- a/test/runtime/file-stores.test.js +++ b/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 };