fix: keep stale session cleanup from closing a replacement range

lib/app/session-end.js

@@ -1,7 +1,7 @@
 // SPDX-License-Identifier: MIT
 // Copyright (c) 2026 EvoMap
 
-import { withClosedRange } from '../distill/range.js';
+import { rangeOf, withClosedRange } from '../distill/range.js';
 import { endedSession, isEarlierTurn, markedReported, pendingReportsOf, sessionOf, unreportedOf } from '../session/state.js';
 import { TASKS } from './ports.js';
 import { closeUnfinishedTurn, settledSessionOf, withoutPendingControls } from './turn-end.js';
@@ -30,6 +30,7 @@ export async function sessionEnded(ports, { sessionId, projectDir, turnId = null
   const due = pending ? pending.map(({ asset }) => asset) : unreportedOf(ending, sessionId, Infinity);
   const queued = pending ? pending.map(({ entry }) => entry) : turnReportsOf(ports.identity, due, { reason: 'aborted', checks: [] });
   const controls = ending.pendingControls;
+  const endingRange = rangeOf(await ports.candidates.read(), sessionId);
   await ports.reports.enqueue([...queued, ...controls]);
   // A prompt uses the same fence to open its turn after this session and its range have ended together.
   const ended = await ports.sessions.withSession(sessionId, async () => {
@@ -40,7 +41,7 @@ export async function sessionEnded(ports, { sessionId, projectDir, turnId = null
       return endedSession(withoutPendingControls(marked, controls));
     });
     if (!saved) return false;
-    await ports.candidates.update((pending) => withClosedRange(pending, sessionId));
+    await ports.candidates.update((pending) => withClosedRange(pending, sessionId, endingRange));
     await ports.notes.clear(sessionId, session);
     return true;
   });

lib/distill/range.js

@@ -126,9 +126,13 @@ export function withSessionActivity(pending, sessionId, now) {
   return { ...pending, [sessionId]: withActivity(range, now) };
 }
 
-/** A session that ended keeps its range due, so a distillation that failed at its end is tried again. */
-export const withClosedRange = (pending, sessionId) =>
-  (pending?.[sessionId] ? { ...pending, [sessionId]: { ...rangeOf(pending, sessionId), closed: true } } : undefined);
+/** A session end keeps the range it found due, leaving a replacement range alone. */
+export function withClosedRange(pending, sessionId, expected) {
+  const range = rangeOf(pending, sessionId);
+  if (!range || !expected || range.at !== expected.at || range.sessionEpoch !== expected.sessionEpoch
+    || !expected.turns.some((read) => range.turns.some((turn) => turn.turn === read.turn))) return undefined;
+  return { ...pending, [sessionId]: { ...range, closed: true } };
+}
 
 const isHeld = (range, isGone) => Boolean(range.claimedBy) && !isGone(range.claimedBy);
 const isOver = (range, isGone) => Boolean(range.closed) || isGone(range.pid);

test/app/turn-lifecycle.test.js

@@ -6,7 +6,7 @@ import { test } from 'node:test';
 
 import { runBackgroundTasks } from '../../lib/app/background.js';
 import { TASKS } from '../../lib/app/ports.js';
-import { MIN_WAKE_DELAY_MS } from '../../lib/app/distill-session.js';
+import { MIN_WAKE_DELAY_MS, distillSession } from '../../lib/app/distill-session.js';
 import { lastTurnOf, promptSubmitted } from '../../lib/app/prompt.js';
 import { sessionEnded } from '../../lib/app/session-end.js';
 import { sessionStarted } from '../../lib/app/session-start.js';
@@ -17,7 +17,7 @@ import { EPISODE_QUIET_MS } from '../../lib/distill/range.js';
 import { sessionOf as sessionOfStored } from '../../lib/session/state.js';
 import { contentHashOf, fileStoresOf } from '../../lib/runtime/store/file-stores.js';
 import { ASSET, PROMPT, memoryPorts } from './memory-ports.js';
-import { fixTurn } from './work-turns.js';
+import { DRAFTED, IR, fixTurn, modelAnswering } from './work-turns.js';
 
 const WAIT = { waitMs: 5_000 };
 const nativePassingCheck = JSON.parse(readFileSync(new URL('../fixtures/node-spec-passing-errors.json', import.meta.url), 'utf8'));
@@ -643,6 +643,42 @@ test('a prompt waits until session and range cleanup finish before opening a new
   assert.deepEqual([session.open, session.turn, session.turnId], [true, resumed.turn, 'next']);
 });
 
+test('late session cleanup cannot close a replacement range after a prompt bypasses the held lifecycle lock', async (t) => {
+  const { ports } = memoryPorts({ model: modelAnswering(IR), distillAnswer: { status: 'stored', gene: DRAFTED.gene } });
+  const dir = mkdtempSync(join(tmpdir(), 'evolver-session-end-fallback-'));
+  t.after(() => rmSync(dir, { recursive: true, force: true }));
+  const { sessions, candidates } = fileStoresOf({ dir, agent: 'testagent' });
+  Object.assign(ports, { sessions, candidates });
+  await verifiedTurn(ports, 's1');
+  await end(ports, 's1', { model: 'model-x' });
+  const closingRange = Promise.withResolvers();
+  const resume = Promise.withResolvers();
+  const updateCandidates = ports.candidates.update;
+  ports.candidates.update = async (...args) => {
+    ports.candidates.update = updateCandidates;
+    closingRange.resolve();
+    await resume.promise;
+    return updateCandidates(...args);
+  };
+  const ending = sessionEnded(ports, { sessionId: 's1', projectDir: '/proj' });
+  await closingRange.promise;
+  try {
+    const next = await promptSubmitted(ports, { sessionId: 's1', prompt: 'continue the repair', turnId: 'next', ...WAIT });
+    assert.equal(next.skipped.code, 'ELOCKED');
+    assert.equal((await distillSession(ports, { sessionId: 's1' })).status, 'stored');
+    await toolFinished(ports, { sessionId: 's1', turnId: 'next', note: { kind: 'edit', paths: ['/proj/lib/next.js'] } });
+    await toolFinished(ports, { sessionId: 's1', turnId: 'next', note: { kind: 'shell', command: 'npm test', exit: 0 } });
+    await end(ports, 's1', { turnId: 'next', model: 'model-x' });
+    assert.deepEqual((await ports.candidates.read()).s1.turns.map(({ turn }) => turn), [next.turn]);
+  } finally {
+    resume.resolve();
+    await ending;
+  }
+  const current = (await ports.candidates.read()).s1;
+  assert.notEqual(current.closed, true);
+  assert.equal((await ports.sessions.read('s1')).turnId, 'next');
+});
+
 test('a note that raced past the open-turn check is still kept out of the next turn', async () => {
   const { ports } = memoryPorts();
   await promptSubmitted(ports, { sessionId: 's1', prompt: PROMPT, turnId: 'g2', ...WAIT });

test/distill.test.js

@@ -2,7 +2,7 @@ import assert from 'node:assert/strict';
 import { test } from 'node:test';
 
 import { draftNamed, draftOf, draftsListText, skipReasonOf, submitRequestOf, withDraft, withoutDraft } from '../lib/distill.js';
-import { baseFor, isCurrentRange, settledRange, withCorrection, withoutExpiredLeftovers, withoutWorkTurn, withWorkTurn } from '../lib/distill/range.js';
+import { baseFor, isCurrentRange, settledRange, withClosedRange, withCorrection, withoutExpiredLeftovers, withoutWorkTurn, withWorkTurn } from '../lib/distill/range.js';
 import { corpusEntryOf } from '../lib/distill/corpus.js';
 
 test('settling what a distillation read cannot touch a range made at the same time or in another session epoch', () => {
@@ -19,6 +19,21 @@ test('settling what a distillation read cannot touch a range made at the same ti
   }
 });
 
+test('session cleanup closes only the range it observed, including legacy session epochs', () => {
+  const read = { at: 1_000, sessionEpoch: 1, turns: [{ turn: 1, prompt: 'p', steps: [], answer: '', at: 1_000 }] };
+  assert.equal(withClosedRange({ s: read }, 's', read).s.closed, true);
+  for (const replacement of [
+    { ...read, at: 2_000 },
+    { ...read, sessionEpoch: 2 },
+    { ...read, turns: [{ ...read.turns[0], turn: 2 }] },
+  ]) {
+    assert.equal(withClosedRange({ s: replacement }, 's', read), undefined);
+  }
+  assert.equal(withClosedRange({ s: read }, 's', null), undefined);
+  const { sessionEpoch, ...legacy } = read;
+  assert.equal(withClosedRange({ s: legacy }, 's', legacy).s.closed, true);
+});
+
 const RESULT = {
   ok: true,
   distill_id: 'abc123',