From df53629cafaef6460c80b8b63d0e62fd953d6bb1 Mon Sep 17 00:00:00 2001 From: "xyanua." Date: Wed, 11 Nov 2026 08:20:57 +0700 Subject: [PATCH 1/5] fix: retain native turn work across transient persistence failures --- lib/app/ports.js | 2 +- lib/app/turn-end.js | 63 +++++++++++++++--------- lib/runtime/store/file-lock.js | 19 ++++++-- lib/runtime/store/file-stores.js | 8 +-- test/app/turn-lifecycle.test.js | 83 ++++++++++++++++++++++++++++++-- test/runtime/file-stores.test.js | 14 ++++++ 6 files changed, 153 insertions(+), 36 deletions(-) diff --git a/lib/app/ports.js b/lib/app/ports.js index e5307d0..9c32988 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 cb3beaa..f9780ef 100644 --- a/lib/app/turn-end.js +++ b/lib/app/turn-end.js @@ -57,9 +57,11 @@ async function closeTurn({ sessions, clock }, sessionId, turnId, { expectedTurn, return opened; } -const settleTurn = (ports, sessionId, opened) => ports.sessions.update(sessionId, (stored) => { +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; }); // Reports are queued before their injections are marked reported, so a hook killed in between leaves an injection @@ -79,6 +81,7 @@ async function reportTurn(ports, sessionId, opened, { reason, checks, source, ve 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) { @@ -88,7 +91,8 @@ async function reportTurn(ports, sessionId, opened, { reason, checks, source, ve const pending = confirmTurnCompletion ? withPendingReports(current, due, queued) : current; return withPendingControls(pending, controls); }); - throw error; + // Finish the verified turn's bookkeeping before surfacing the queue failure. + return { ...result, failure: error }; } await sessions.update(sessionId, (stored) => { const current = sessionOf(stored); @@ -99,7 +103,7 @@ async function reportTurn(ports, sessionId, opened, { reason, checks, source, ve 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. @@ -222,27 +226,27 @@ async function rejudgeWorkTurn(ports, sessionId, turnId, { reason, answer, model } } -/** - * 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 - * injection of the turn is reported with that evidence, a trial Gene the Proxy withheld from it is reported as a - * control, its reuses wait for the next prompt's label, its edits for the next session's retention check, and a work - * 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, expected, allowNewCandidate, finalizationPending }) { +/** Reads the closed turn's evidence before any reports or candidate writes. */ +async function readTurnEvidence(ports, sessionId, opened, recordOf) { 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; + 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; + } +} + +async function recordTurn(ports, sessionId, opened, { reason, answer, model, projectDir, source, expected, allowNewCandidate, finalizationPending }, { 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, expected }); + const { queued, controls, strategies, failure } = await reportTurn(ports, sessionId, opened, { reason, checks: finished.checks, source, verified, expected }); 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 }); @@ -257,13 +261,13 @@ async function recordTurn(ports, sessionId, opened, { reason, answer, model, pro && gateOf(rangeInEpoch(await ports.candidates.read(), sessionId, opened.sessionEpoch)?.turns ?? []).passed; return { turn: opened.turn, record: finished, reports: queued.map(({ report }) => report), strategies, isDistillable, - ...(expected && { isCandidate: isWork }), controls, + ...(expected && { isCandidate: isWork }), controls, failure, }; } /** A native record for a retained turn uses only its own evidence and captured route. */ export async function finishDeferredNativeTurn(ports, sessionId, opened, { reason, record, model }) { - const { queued, controls } = await reportTurn(ports, sessionId, opened, { + const { queued, controls, failure } = await reportTurn(ports, sessionId, opened, { reason, checks: record.checks, source: OBSERVED, verified: true, expected: opened, deferred: true, }); const isCandidate = await rememberWorkTurn(ports, sessionId, opened, { @@ -271,9 +275,18 @@ export async function finishDeferredNativeTurn(ports, sessionId, opened, { reaso deferred: true, allowNewCandidate: !opened.native.candidateMade, finalizationPending: true, }); if (queued.length || controls) await scheduleDueDistillation(ports, [TASKS.DRAIN]); + if (failure) throw failure; return { turn: opened.turn, reports: queued.map(({ report }) => report), isCandidate }; } +/** + * 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 + * injection of the turn is reported with that evidence, a trial Gene the Proxy withheld from it is reported as a + * control, its reuses wait for the next prompt's label, its edits for the next session's retention check, and a work + * 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. + */ export async function turnEnded(ports, { sessionId, turnId, reason, answer = '', model = null, projectDir, recordOf = null, source = OBSERVED, expectedTurn, ...native }) { const { expected, allowNewCandidate, finalizationPending } = native; const opened = await closeTurn(ports, sessionId, turnId, { expectedTurn, expected }); @@ -283,15 +296,19 @@ 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, expected, allowNewCandidate, finalizationPending }); + const evidence = await readTurnEvidence(ports, sessionId, opened, recordOf); + evidenceRead = true; + if (evidence) ended = await recordTurn(ports, sessionId, opened, { reason, answer, model, projectDir, source, expected, allowNewCandidate, finalizationPending }, evidence); } finally { - await settleTurn(ports, sessionId, opened); + // Only a failure before writes can reopen the same native turn for immediate redelivery. + 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..266ae36 100644 --- a/lib/runtime/store/file-lock.js +++ b/lib/runtime/store/file-lock.js @@ -223,12 +223,21 @@ 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 }); - }, + // Released claims stay separate from new appends and become available to the next take immediately. + const release = () => { + for (const claim of claims) { + try { fs.utimesSync(claim, 0, 0); } 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/turn-lifecycle.test.js b/test/app/turn-lifecycle.test.js index e174f16..84d72ec 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'; @@ -706,15 +706,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..f6837b5 100644 --- a/test/runtime/file-stores.test.js +++ b/test/runtime/file-stores.test.js @@ -60,6 +60,20 @@ 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', 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(); + + 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 }; From cef26f362b4d2feec0269851db45e0f5f65d2225 Mon Sep 17 00:00:00 2001 From: "xyanua." Date: Wed, 11 Nov 2026 08:18:51 +0700 Subject: [PATCH 2/5] fix: retain unknown native owners during finalization --- lib/app/turn-end.js | 4 ++-- test/app/native-turn.test.js | 9 +++++++++ 2 files changed, 11 insertions(+), 2 deletions(-) diff --git a/lib/app/turn-end.js b/lib/app/turn-end.js index f9780ef..2085f9c 100644 --- a/lib/app/turn-end.js +++ b/lib/app/turn-end.js @@ -183,8 +183,8 @@ async function rememberWorkTurn(ports, sessionId, opened, options) { const planned = matching ? baseFor(pending, sessionId, workTurn) : workTurn.head ?? null; const snapshot = projectDir && !deferred ? await snapshotIfReadable(project, projectDir, planned) : null; const base = planned ?? (snapshot?.head ? { root: snapshot.root, head: snapshot.head } : null); - // A detached observer's parent is a hook process, while the open turn carries the native owner. - const pid = expected?.pid ?? (confirmTurnCompletion ? opened.pid : null) ?? owner.current(); + // A detached observer must retain the captured native owner, including an unknown one. + const pid = expected?.pid !== undefined ? expected.pid : (confirmTurnCompletion ? opened.pid : null) ?? owner.current(); const context = { base, project: snapshot, projectDir: projectDir ?? null, model, sessionEpoch: opened.sessionEpoch, pid, native: Boolean(native || expected), source, now: clock.now() }; // If the turn moved to deferred storage during the snapshot, its recovery uses native evidence alone. const canKeep = (stored) => canKeepTurn(sessionOf(stored), opened) && Boolean(hasDeferredTurn(sessionOf(stored), opened)) === deferred; diff --git a/test/app/native-turn.test.js b/test/app/native-turn.test.js index d5b9268..05c2fc4 100644 --- a/test/app/native-turn.test.js +++ b/test/app/native-turn.test.js @@ -61,6 +61,15 @@ test('native prompt binds turn zero before terminal; Stop hints cannot close it assert.deepEqual(made.sessions.get('session').nativeCursor.mark, { offset: 500, fileId: 'wire' }); }); +test('native finalization preserves an unknown captured owner without probing its own ancestry', async () => { + const made = await pending({ owner: null }); + made.ports.owner.current = () => assert.fail('finalization must use the captured owner'); + const ended = await reconcileNativeTurn(made.ports, at, facts()); + assert.equal(ended.ended.isCandidate, true); + assert.equal((await made.ports.candidates.read()).session.pid, null); + assert.equal(made.queued.length, 1); +}); + test('read errors, partial wire and a native prompt before the pending mark cannot settle a turn', async () => { for (const bad of [{ ...facts(), status: 'partial' }, { ...facts(), bind: { ...facts().bind, mark: 50 } }, { ...facts(), mark: { offset: 500, fileId: 'rotated' } }]) { const made = await pending(); From 6337dc2f5a3bea4d19393e28b732189fa76eaaca Mon Sep 17 00:00:00 2001 From: forrestlinfeng Date: Wed, 11 Nov 2026 10:19:33 +0800 Subject: [PATCH 3/5] Keep finishing a turn after a queue failure to native hosts Only a native turn caches its reports with their evidence when the queue fails. Elsewhere the injections stay unreported for a later end, which remembers their retention with its own edits, so this end must not as well. Co-Authored-By: Claude Opus 5.5 --- lib/app/turn-end.js | 3 ++- test/app/turn-lifecycle.test.js | 11 +++++++++++ 2 files changed, 13 insertions(+), 1 deletion(-) diff --git a/lib/app/turn-end.js b/lib/app/turn-end.js index 86eb7dc..e8943af 100644 --- a/lib/app/turn-end.js +++ b/lib/app/turn-end.js @@ -64,7 +64,8 @@ async function reportTurn({ identity, sessions, reports, clock, confirmTurnCompl const pending = confirmTurnCompletion ? withPendingReports(current, due, queued) : current; return withPendingControls(pending, controls); }); - // Finish the verified turn's bookkeeping before surfacing the queue failure. + // 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) => { diff --git a/test/app/turn-lifecycle.test.js b/test/app/turn-lifecycle.test.js index 84d72ec..8af281a 100644 --- a/test/app/turn-lifecycle.test.js +++ b/test/app/turn-lifecycle.test.js @@ -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(); From 266ce4fa267cd648de05419be11cc10be3bf4927 Mon Sep 17 00:00:00 2001 From: forrestlinfeng Date: Wed, 11 Nov 2026 10:19:56 +0800 Subject: [PATCH 4/5] Keep released turn evidence through pruning A claim released at the epoch looked days past the evidence lifetime, so the next background run could delete evidence the turn's redelivery was about to take. Co-Authored-By: Claude Opus 5.5 --- lib/runtime/store/file-lock.js | 6 ++++-- test/runtime/file-stores.test.js | 3 ++- 2 files changed, 6 insertions(+), 3 deletions(-) diff --git a/lib/runtime/store/file-lock.js b/lib/runtime/store/file-lock.js index 266ae36..771677d 100644 --- a/lib/runtime/store/file-lock.js +++ b/lib/runtime/store/file-lock.js @@ -223,10 +223,12 @@ export function readStateLines(target) { export function takeStateLines(target, now = Date.now()) { const claims = [...staleClaimsOf(target, now), target].map((source) => claimed(source, target)).filter(Boolean); - // Released claims stay separate from new appends and become available to the next take immediately. + // 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, 0, 0); } catch { /* Ordinary stale recovery can retry an inaccessible claim. */ } + try { fs.utimesSync(claim, releasedAt, releasedAt); } catch { /* Ordinary stale recovery can retry an inaccessible claim. */ } } }; const commit = () => { diff --git a/test/runtime/file-stores.test.js b/test/runtime/file-stores.test.js index f6837b5..825b9eb 100644 --- a/test/runtime/file-stores.test.js +++ b/test/runtime/file-stores.test.js @@ -60,7 +60,7 @@ 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', async () => { +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); @@ -68,6 +68,7 @@ test('released turn evidence is immediately recoverable alongside newer appends' 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(); From d580d73229983aefdfee340a749bf7fda295ca76 Mon Sep 17 00:00:00 2001 From: forrestlinfeng Date: Wed, 11 Nov 2026 10:20:05 +0800 Subject: [PATCH 5/5] Require every notes store to release a take Co-Authored-By: Claude Opus 5.5 --- lib/app/ports.js | 2 +- lib/app/turn-end.js | 2 +- test/app/memory-ports.js | 7 ++++--- 3 files changed, 6 insertions(+), 5 deletions(-) diff --git a/lib/app/ports.js b/lib/app/ports.js index 874a086..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, release?: () => void }>} 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 e8943af..d2c6551 100644 --- a/lib/app/turn-end.js +++ b/lib/app/turn-end.js @@ -182,7 +182,7 @@ async function readTurnEvidence(ports, sessionId, opened, recordOf) { try { return { observed, notes, edits: await ports.baseline.take(sessionId, opened.turnId, opened) }; } catch (error) { - notes.release?.(); + notes.release(); throw error; } } 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)), },