diff --git a/lib/app/ports.js b/lib/app/ports.js index 6aef5c0..d3a71aa 100644 --- a/lib/app/ports.js +++ b/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} append * @property {(sessionId: string, scope?: {turn: number, sessionEpoch?: number}) => Promise} read - * @property {(sessionId: string, scope?: {turn: number, sessionEpoch?: number}) => Promise<{ notes: object[], commit: () => Promise }>} take + * @property {(sessionId: string, scope?: {turn: number, sessionEpoch?: number}) => Promise<{ notes: object[], commit: () => Promise, release: () => void }>} take * @property {(sessionId: string, scope?: {turn: number, sessionEpoch?: number}) => Promise} clear * * @typedef {object} EditBaseline Each file's content hash before the turn first edited it. diff --git a/lib/app/turn-end.js b/lib/app/turn-end.js index 5d4b8fa..d2c6551 100644 --- a/lib/app/turn-end.js +++ b/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; } diff --git a/lib/runtime/store/file-lock.js b/lib/runtime/store/file-lock.js index 7743c6e..771677d 100644 --- a/lib/runtime/store/file-lock.js +++ b/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) { diff --git a/lib/runtime/store/file-stores.js b/lib/runtime/store/file-stores.js index be247a5..fc00fcc 100644 --- a/lib/runtime/store/file-stores.js +++ b/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(), }; diff --git a/test/app/memory-ports.js b/test/app/memory-ports.js index 6176d31..c9af5cb 100644 --- a/test/app/memory-ports.js +++ b/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)), }, diff --git a/test/app/turn-lifecycle.test.js b/test/app/turn-lifecycle.test.js index e174f16..8af281a 100644 --- a/test/app/turn-lifecycle.test.js +++ b/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 }); diff --git a/test/runtime/file-stores.test.js b/test/runtime/file-stores.test.js index a12f6c8..825b9eb 100644 --- a/test/runtime/file-stores.test.js +++ b/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 };