fix: release unrun distillation claims after a batch failure

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);
 }
 

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' });