diff --git a/lib/app/distill-session.js b/lib/app/distill-session.js index e56b6d6..0df489f 100644 --- a/lib/app/distill-session.js +++ b/lib/app/distill-session.js @@ -97,10 +97,27 @@ export async function distillSession(ports, { sessionId: asked, focus = '', anyS return gate.passed ? { ...outcome, agentGene: { writtenFor: ports.candidates.referenceOf(sessionId, range), prompt } } : outcome; } +async function releaseClaims({ candidates, clock }, claims) { + if (claims.length === 0) return; + await candidates.update((pending) => { + const release = (table, { sessionId, range }) => settledRange(table, sessionId, range, { now: clock.now() }) ?? table; + const released = claims.reduce(release, pending); + return released === pending ? undefined : released; + }); +} + /** Distills the ranges that are due: their session is over (ended, or killed with its owner), quiet, or fixed an error. */ export async function distillDue(ports) { + const claims = (await claim(ports, claimedDue)) ?? []; const outcomes = []; - for (const claimed of (await claim(ports, claimedDue)) ?? []) outcomes.push(await run(ports, claimed)); + for (const [index, claimed] of claims.entries()) { + try { + outcomes.push(await run(ports, claimed)); + } catch (error) { + await releaseClaims(ports, claims.slice(index + 1)); + throw error; + } + } return confirmDraftOutcomes(ports, outcomes); } diff --git a/test/app/distill-session.test.js b/test/app/distill-session.test.js index fbe8d38..0b01792 100644 --- a/test/app/distill-session.test.js +++ b/test/app/distill-session.test.js @@ -870,6 +870,27 @@ test('with no headless model, distilling now hands the agent the IR prompt, and assert.equal((await ports.drafts.read()).length, 0); }); +test('a failed background range releases the unrun ranges so the same process can retry', async () => { + const made = await fixedSession({ model: modelAnswering(IR), distillAnswer: { status: 'stored', gene: DRAFTED.gene } }); + const { ports, requests } = made; + await fixTurn(ports, 's2'); + await sessionEnded(ports, { sessionId: 's1', projectDir: '/proj' }); + await sessionEnded(ports, { sessionId: 's2', projectDir: '/proj' }); + const updateCorpus = ports.corpus.update; + ports.corpus.update = async () => { throw new Error('corpus write denied'); }; + + await assert.rejects(distillDue(ports), /corpus write denied/); + ports.corpus.update = updateCorpus; + + const pending = await ports.candidates.read(); + assert.deepEqual([pending.s1.claimedBy, pending.s2.claimedBy], [null, null]); + assert.deepEqual([pending.s1.attempts, pending.s2.attempts], [0, 0]); + assert.deepEqual((await distillDue(ports)).map(({ status }) => status), ['stored', 'stored']); + assert.deepEqual(await ports.candidates.read(), {}); + assert.deepEqual(distills(requests).map(({ body }) => body.session_id), ['testagent:s1', 'testagent:s2']); + assert.deepEqual(await distillDue(ports), []); +}); + test('a draft write failure releases the claim so the same process can retry', async () => { const { ports } = await fixedSession({ distillAnswer: DRAFTED }); const asked = await distillSession(ports, { sessionId: 's1' });