From f057bf72e49d931fa88c7e72eb44d0e0b76c4aa0 Mon Sep 17 00:00:00 2001 From: "xyanua." Date: Wed, 11 Nov 2026 08:16:12 +0700 Subject: [PATCH 1/3] fix: release unrun distillation claims after a batch failure --- lib/app/distill-session.js | 15 ++++++++++++++- test/app/distill-session.test.js | 21 +++++++++++++++++++++ 2 files changed, 35 insertions(+), 1 deletion(-) diff --git a/lib/app/distill-session.js b/lib/app/distill-session.js index 4c1bc3d..e59bb28 100644 --- a/lib/app/distill-session.js +++ b/lib/app/distill-session.js @@ -99,7 +99,20 @@ export async function distillSession(ports, { sessionId: asked, focus = '', anyS /** 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 outcomes = []; - for (const claimed of (await claim(ports, claimedDue)) ?? []) outcomes.push(await run(ports, claimed)); + const claims = (await claim(ports, claimedDue)) ?? []; + try { + for (const claimed of claims) outcomes.push(await run(ports, claimed)); + } catch (error) { + // A failed run must also return the batch's claims that never reached run's settlement. + await ports.candidates.update((pending) => { + let next; + for (const claimed of claims) { + next = settledRange(next ?? pending, claimed.sessionId, claimed.range, { now: ports.clock.now() }) ?? next; + } + return next; + }); + throw error; + } return confirmDraftOutcomes(ports, outcomes); } diff --git a/test/app/distill-session.test.js b/test/app/distill-session.test.js index eaf45db..8615196 100644 --- a/test/app/distill-session.test.js +++ b/test/app/distill-session.test.js @@ -848,6 +848,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' }); From 5060eefd07666694af9bc02005ba2b1cb5c8cb67 Mon Sep 17 00:00:00 2001 From: "xyanua." Date: Wed, 11 Nov 2026 08:18:51 +0700 Subject: [PATCH 2/3] 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 cb3beaa..cf5c216 100644 --- a/lib/app/turn-end.js +++ b/lib/app/turn-end.js @@ -179,8 +179,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 b196e26035656f76a98ee0bd79850134d56d1d32 Mon Sep 17 00:00:00 2001 From: forrestlinfeng Date: Wed, 11 Nov 2026 10:10:29 +0800 Subject: [PATCH 3/3] Release only the batch claims that never ran A run that threw already settled its own claim, and earlier runs settled theirs; releasing those again relied on the claim guard alone and could drop a claim this process took again meanwhile. Co-Authored-By: Claude Opus 5.5 --- lib/app/distill-session.js | 30 +++++++++++++++++------------- 1 file changed, 17 insertions(+), 13 deletions(-) diff --git a/lib/app/distill-session.js b/lib/app/distill-session.js index 9d28f17..0df489f 100644 --- a/lib/app/distill-session.js +++ b/lib/app/distill-session.js @@ -97,22 +97,26 @@ 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 outcomes = []; const claims = (await claim(ports, claimedDue)) ?? []; - try { - for (const claimed of claims) outcomes.push(await run(ports, claimed)); - } catch (error) { - // A failed run must also return the batch's claims that never reached run's settlement. - await ports.candidates.update((pending) => { - let next; - for (const claimed of claims) { - next = settledRange(next ?? pending, claimed.sessionId, claimed.range, { now: ports.clock.now() }) ?? next; - } - return next; - }); - throw error; + const outcomes = []; + 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); }