fix: keep resumed turns open during stale session cleanup

lib/app/ports.js

@@ -14,6 +14,7 @@
  * @property {string} label
  *
  * @typedef {object} SessionStore
+ * @property {<T>(sessionId: string, work: () => T|Promise<T>) => Promise<T>} withSession  Serializes prompt opening and session cleanup across stores.
  * @property {(sessionId: string) => Promise<object>} read
  * @property {(visit: (session: object, sessionId: string) => void) => Promise<void>} readEach
  * @property {(sessionId: string, mutate: Function) => Promise<object|undefined>} update

lib/app/prompt.js

@@ -52,8 +52,8 @@ async function recordedCorrection(ports, sessionId, session) {
  * turn in the session's range, for distillation to read, takes back the drafts distilled from that turn before it,
  * and the last turn's reuses are reported again under `:correction`. The turn is marked corrected first, so a draft's
  * confirmation or publication sees the mark, and the range changes before drafts are removed, so a draft written
- * after the removal still sees the turn it read changed. Every prompt is the session's activity, which a range waits
- * out before it is distilled. Answers the error of a correction lock it had to go without, if any.
+ * after the removal still sees the turn it read changed. The caller records the prompt as the session's activity
+ * when it opens the turn. Answers the error of a correction lock it had to go without, and whether reports were queued.
  */
 async function labelFollowUp(ports, sessionId, prompt) {
   const { sessions, candidates, drafts, reports, clock } = ports;
@@ -66,13 +66,22 @@ async function labelFollowUp(ports, sessionId, prompt) {
     await candidates.update((pending) => withCorrection(pending, sessionId, session.turn, correctionTextOf(promptTextOf(prompt)), clock.now(), correction));
     if (correction.turn > 0) await drafts.update((list) => withoutDraftsForTurn(list, sessionId, correction.turn, correction.sessionEpoch));
   }
-  await candidates.update((pending) => withSessionActivity(pending, sessionId, clock.now()));
   // Read again: a turn end that overlapped the waits above may have left newer reuses to label.
   const { labelable } = sessionOf(await sessions.read(sessionId));
   const queued = correctionReportsOf(labelable, label, sessionId);
   if (queued.length > 0) await reports.enqueue(queued);
-  await scheduleDueDistillation(ports, queued.length > 0 ? [TASKS.DRAIN] : []);
-  return skipped;
+  return { skipped, reportsQueued: queued.length > 0 };
+}
+
+// The lifecycle lock only orders a prompt against a session end's last writes. One held past its wait, as by a killed
+// session-end hook, must not cost the person their turn, so the turn then opens without it and the hook reports it.
+async function fencedOpen(ports, sessionId, open) {
+  try {
+    return { opened: await ports.sessions.withSession(sessionId, open) };
+  } catch (error) {
+    if (error?.code !== 'ELOCKED') throw error;
+    return { opened: await open(), skipped: error };
+  }
 }
 
 /** A recall's match when it brings a strategy (`{ ids, text, stepTexts? }`, a Recipe's steps or one asset), else null. */
@@ -156,13 +165,20 @@ async function rememberTurnHead({ project, sessions }, sessionId, { turn, sessio
  * only while the turn is still open, once a model step does. With `holdForToolResult`, for a host whose prompt hook
  * cannot add context, a strategy that lands in time waits for the turn's first tool result instead (see
  * `takeWaitingStrategy`). A plugin command's turn keeps the last real turn's reuses for the next prompt to label, is
- * recalled for nothing and never joins the session's range of work turns. `skipped` is the error of a correction lock
- * the prompt went without.
+ * recalled for nothing and never joins the session's range of work turns. `skipped` is the error of a correction or
+ * session lifecycle lock the prompt went without.
  */
 export async function promptSubmitted(ports, { sessionId, prompt, projectDir = null, turnId = null, turnStart = null, hostTurn = null, isCommand = false, waitMs, keepLate = false, holdForToolResult = false }) {
-  const skipped = isCommand ? undefined : await labelFollowUp(ports, sessionId, prompt);
-  const opened = await ports.sessions.update(sessionId, (stored) =>
-    openedTurn(sessionOf(stored), { turnId, turnStart, hostTurn, prompt: prompt ? promptTextOf(prompt) : null, task: taskTextOf(prompt), pid: ports.owner.current(), isCommand, now: ports.clock.now() }));
+  const labelled = isCommand ? {} : await labelFollowUp(ports, sessionId, prompt);
+  const pid = ports.owner.current();
+  const fenced = await fencedOpen(ports, sessionId, async () => {
+    if (!isCommand) await ports.candidates.update((pending) => withSessionActivity(pending, sessionId, ports.clock.now()));
+    return ports.sessions.update(sessionId, (stored) =>
+      openedTurn(sessionOf(stored), { turnId, turnStart, hostTurn, prompt: prompt ? promptTextOf(prompt) : null, task: taskTextOf(prompt), pid, isCommand, now: ports.clock.now() }));
+  });
+  const { opened } = fenced;
+  const skipped = labelled.skipped ?? fenced.skipped;
+  if (!isCommand) await scheduleDueDistillation(ports, labelled.reportsQueued ? [TASKS.DRAIN] : []);
   const wantsRecall = !isCommand && ports.settings?.recall !== false;
   const recall = wantsRecall ? recalled(ports, sessionId, { turn: opened.turn, prompt, projectDir, waitMs, keepLate, holdForToolResult }) : { injection: '' };
   const headRead = isCommand || !projectDir ? null : rememberTurnHead(ports, sessionId, { turn: opened.turn, sessionEpoch: opened.sessionEpoch, projectDir });

lib/app/session-end.js

@@ -8,6 +8,7 @@ import { closeUnfinishedTurn, settledSessionOf, withoutPendingControls } from '.
 import { turnReportsOf } from './turn-reports.js';
 
 const NOTHING_DONE = Object.freeze({ checks: [], commands: [], paths: [], edits: [], answer: '' });
+const isSameTurn = (session, expected) => session.turn === expected.turn && session.sessionEpoch === expected.sessionEpoch;
 
 /**
  * Only writes, since hosts bound a session-end hook tightly: once the last turn's end had its chance to arrive
@@ -22,18 +23,26 @@ export async function sessionEnded(ports, { sessionId, projectDir, turnId = null
   if (isEarlierTurn(session, turnId)) return;
   await closeUnfinishedTurn(ports, { sessionId, as: openTurnReason, projectDir, recordOf: () => NOTHING_DONE });
   const ending = sessionOf(await ports.sessions.read(sessionId));
+  if (!isSameTurn(ending, session)) return;
   // Native mode only replays cached reports; session start settles explicitly adopted legacy injections.
   // An uncached native entry has no safe legacy provenance, so shutdown cannot classify it.
   const pending = ports.confirmTurnCompletion ? pendingReportsOf(ending, sessionId) : 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;
   await ports.reports.enqueue([...queued, ...controls]);
-  await ports.sessions.update(sessionId, (stored) => {
-    const marked = markedReported(sessionOf(stored), due, ports.clock.now()) ?? sessionOf(stored);
-    return endedSession(withoutPendingControls(marked, 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 () => {
+    const saved = await ports.sessions.update(sessionId, (stored) => {
+      const current = sessionOf(stored);
+      if (!isSameTurn(current, ending)) return undefined;
+      const marked = markedReported(current, due, ports.clock.now()) ?? current;
+      return endedSession(withoutPendingControls(marked, controls));
+    });
+    if (!saved) return false;
+    await ports.candidates.update((pending) => withClosedRange(pending, sessionId));
+    await ports.notes.clear(sessionId, session);
+    return true;
   });
-  await ports.candidates.update((pending) => withClosedRange(pending, sessionId));
-  await ports.notes.clear(sessionId, session);
-  ports.scheduler.schedule([TASKS.DRAIN, TASKS.DISTILL]);
+  ports.scheduler.schedule(ended ? [TASKS.DRAIN, TASKS.DISTILL] : [TASKS.DRAIN]);
 }

lib/app/turn-end.js

@@ -236,7 +236,7 @@ export const recordOfRow = (row, recordOf = null) => (row?.record ? () => row.re
 export async function closeUnfinishedTurn(ports, { sessionId, settleMs = 0, as = 'error', projectDir, recordOf = null }) {
   const session = await settledSessionOf(ports, sessionId, settleMs);
   if (!session.open) return;
-  if (!ports.confirmTurnCompletion) return turnEnded(ports, { sessionId, reason: as, projectDir, recordOf });
+  if (!ports.confirmTurnCompletion) return turnEnded(ports, { sessionId, reason: as, projectDir, recordOf, expectedTurn: session });
   const row = ports.turnStatus?.(sessionId, session.turnId);
   const reason = reasonOfNativeStatus(row?.status, true);
   const ended = await turnEnded(ports, {

lib/runtime/store/file-stores.js

@@ -19,10 +19,21 @@ import { adoptLegacyIn } from './legacy.js';
 const MAX_DRAIN_ROUNDS = 5;
 // A session file holds the turn counter, which a resumed session needs for as long as it may still be resumed.
 const SESSION_TTL_MS = 30 * 24 * 60 * 60 * 1000;
-const CORRECTION_LOCK_WAIT_MS = 2_000;
-const CORRECTION_LOCK_RETRY_MS = 20;
+const SESSION_LOCK_WAIT_MS = 2_000;
+const SESSION_LOCK_RETRY_MS = 20;
 const digestOf = (key) => crypto.createHash('sha256').update(String(key)).digest('hex').slice(0, 16);
 
+async function withSessionLock(lock, work, signal, name) {
+  const deadline = Date.now() + SESSION_LOCK_WAIT_MS;
+  let result;
+  for (;;) {
+    signal?.throwIfAborted();
+    if (await withLock(lock, async () => { result = await work(); })) return result;
+    if (Date.now() >= deadline) throw Object.assign(new Error(`Timed out waiting for the ${name} lock. Please retry.`), { code: 'ELOCKED' });
+    await delay(SESSION_LOCK_RETRY_MS, undefined, { signal });
+  }
+}
+
 // A range's turn numbers name it apart from one created again in the same millisecond after its turns were distilled.
 export function rangeReferenceOf(sessionId, { at, changedAt, turns }) {
   const sources = turns.map(({ turn, sessionEpoch }) => sessionEpoch === undefined ? turn : [turn, sessionEpoch]);
@@ -105,17 +116,8 @@ export function fileStoresOf({ dir, agent }) {
   const correctionName = (sessionId, turn, sessionEpoch) => digestOf(JSON.stringify([sessionId, turn, sessionEpoch]));
   const correctionFile = (...source) => path.join(corrections, correctionName(...source));
   const draftCorrections = {
-    withSession: async (sessionId, work, signal) => {
-      const lock = path.join(corrections, `${digestOf(sessionId)}.lock`);
-      const deadline = Date.now() + CORRECTION_LOCK_WAIT_MS;
-      let result;
-      for (;;) {
-        signal?.throwIfAborted();
-        if (await withLock(lock, async () => { result = await work(); })) return result;
-        if (Date.now() >= deadline) throw Object.assign(new Error('Timed out waiting for the draft correction lock. Please retry.'), { code: 'ELOCKED' });
-        await delay(CORRECTION_LOCK_RETRY_MS, undefined, { signal });
-      }
-    },
+    withSession: (sessionId, work, signal) =>
+      withSessionLock(path.join(corrections, `${digestOf(sessionId)}.lock`), work, signal, 'draft correction'),
     mark: async (sessionId, turn, sessionEpoch) => {
       fs.mkdirSync(corrections, { recursive: true, mode: 0o700 });
       fs.closeSync(fs.openSync(correctionFile(sessionId, turn, sessionEpoch), 'a', 0o600));
@@ -136,6 +138,7 @@ export function fileStoresOf({ dir, agent }) {
     if (failure) throw failure;
   };
   const sessions = {
+    withSession: (sessionId, work) => withSessionLock(sessionFile('lifecycle', sessionId, 'lock'), work, undefined, 'session lifecycle'),
     read: async (sessionId) => readState(sessionFile('session', sessionId), {}),
     readEach: async (visit) => eachSession((target) => {
       const stored = readState(target, {});

test/agent/driver.test.js

@@ -129,6 +129,20 @@ test('a correction whose draft lock stays held still opens its turn and recalls,
   assert.equal(await made.ports.draftCorrections.has('s1', 1, session.sessionEpoch), true);
 });
 
+test('a prompt whose session lifecycle lock stays held still opens its turn and recalls, and reports the lock', async () => {
+  const made = memoryPorts();
+  const { driver } = driverOf(hookProfile(), made);
+  const held = Object.assign(new Error('Timed out waiting for the session lifecycle lock. Please retry.'), { code: 'ELOCKED' });
+  made.ports.sessions.withSession = async () => { throw held; };
+  const errors = [];
+
+  await driver.handle('UserPromptSubmit', { ...base, prompt: PROMPT }, { onError: (error) => errors.push(error) });
+  assert.deepEqual(errors, [held]);
+  assert.ok(made.requests.length > 0);
+  const session = made.sessions.get('s1');
+  assert.deepEqual([session.turn, session.open], [1, true]);
+});
+
 test('a host whose injection lasts one call gets the open turn\'s strategy again, without a second recall', async () => {
   const made = memoryPorts();
   const { driver, requests } = driverOf(hookProfile({ reply: { lifetime: 'call' }, payload: { turnId: (payload) => payload.turn } }), made);

test/app/memory-ports.js

@@ -81,6 +81,7 @@ export function memoryPorts({ assets = [ASSET], recipes = [], stepAssets = [], d
   const ports = {
     identity: { agent: 'testagent', label: 'Test Agent' },
     sessions: {
+      withSession: async (_id, work) => work(),
       read: async (id) => structuredClone(sessions.get(id) ?? {}),
       readEach: async (visit) => {
         for (const [id, stored] of sessions) visit(structuredClone(stored), id);

test/app/turn-lifecycle.test.js

@@ -519,6 +519,119 @@ test('a session end that names an earlier turn is stale even once the later turn
   assert.equal((await ports.sessions.read('s1')).turnId, 'g2', 'the stale end changed nothing');
 });
 
+test('a session end waiting on report storage leaves a resumed turn and its evidence open', async () => {
+  const { ports } = memoryPorts();
+  await verifiedTurn(ports, 's1');
+  await end(ports, 's1');
+  const queued = Promise.withResolvers();
+  const resume = Promise.withResolvers();
+  const enqueue = ports.reports.enqueue;
+  ports.reports.enqueue = async (entries) => {
+    queued.resolve();
+    await resume.promise;
+    return enqueue(entries);
+  };
+  const ending = sessionEnded(ports, { sessionId: 's1', projectDir: '/proj' });
+  await queued.promise;
+  ports.reports.enqueue = enqueue;
+  const next = await promptSubmitted(ports, { sessionId: 's1', prompt: 'continue the repair', turnId: 'next', ...WAIT });
+  await toolFinished(ports, { sessionId: 's1', turnId: 'next', note: { kind: 'shell', command: 'npm test', exit: 0, output: 'ok' } });
+  resume.resolve();
+  await ending;
+
+  const session = await ports.sessions.read('s1');
+  assert.deepEqual([session.open, session.turn, session.turnId], [true, next.turn, 'next']);
+  assert.notEqual((await ports.candidates.read()).s1.closed, true);
+  assert.equal((await ports.notes.read('s1', session)).some(({ command }) => command === 'npm test'), true);
+});
+
+test('a session end delayed before closing its turn cannot close a resumed turn', async () => {
+  const { ports } = memoryPorts();
+  await verifiedTurn(ports, 's1');
+  await end(ports, 's1');
+  const closing = Promise.withResolvers();
+  const resume = Promise.withResolvers();
+  const updateSession = ports.sessions.update;
+  ports.sessions.update = async (...args) => {
+    ports.sessions.update = updateSession;
+    closing.resolve();
+    await resume.promise;
+    return updateSession(...args);
+  };
+  const ending = sessionEnded(ports, { sessionId: 's1', projectDir: '/proj' });
+  await closing.promise;
+  const next = await promptSubmitted(ports, { sessionId: 's1', prompt: 'continue the repair', turnId: 'next', ...WAIT });
+  resume.resolve();
+  await ending;
+
+  const session = await ports.sessions.read('s1');
+  assert.deepEqual([session.open, session.turn, session.turnId], [true, next.turn, 'next']);
+  assert.notEqual((await ports.candidates.read()).s1.closed, true);
+});
+
+test('a session end delayed while closing an open turn cannot close the turn a resumed prompt opened', async () => {
+  const { ports } = memoryPorts();
+  await verifiedTurn(ports, 's1');
+  const closing = Promise.withResolvers();
+  const resume = Promise.withResolvers();
+  const updateSession = ports.sessions.update;
+  ports.sessions.update = async (...args) => {
+    ports.sessions.update = updateSession;
+    closing.resolve();
+    await resume.promise;
+    return updateSession(...args);
+  };
+  const ending = sessionEnded(ports, { sessionId: 's1', projectDir: '/proj' });
+  await closing.promise;
+  await closeUnfinishedTurn(ports, { sessionId: 's1', projectDir: '/proj' });
+  const next = await promptSubmitted(ports, { sessionId: 's1', prompt: 'continue the repair', turnId: 'next', ...WAIT });
+  resume.resolve();
+  await ending;
+
+  const session = await ports.sessions.read('s1');
+  assert.deepEqual([session.open, session.turn, session.turnId], [true, next.turn, 'next']);
+});
+
+test('a prompt waits until session and range cleanup finish before opening a new turn', async (t) => {
+  const { ports } = memoryPorts();
+  const dir = mkdtempSync(join(tmpdir(), 'evolver-session-end-fence-'));
+  t.after(() => rmSync(dir, { recursive: true, force: true }));
+  ports.sessions.withSession = fileStoresOf({ dir, agent: 'testagent' }).sessions.withSession;
+  await verifiedTurn(ports, 's1');
+  await end(ports, 's1');
+  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;
+  const opening = Promise.withResolvers();
+  const withSession = fileStoresOf({ dir, agent: 'testagent' }).sessions.withSession;
+  ports.sessions.withSession = (...args) => {
+    opening.resolve();
+    return withSession(...args);
+  };
+  let opened = false;
+  const next = promptSubmitted(ports, { sessionId: 's1', prompt: 'continue the repair', turnId: 'next', ...WAIT })
+    .then((result) => { opened = true; return result; });
+  try {
+    await opening.promise;
+    assert.equal(opened, false);
+    assert.equal((await ports.sessions.read('s1')).open, false);
+  } finally {
+    resume.resolve();
+    await ending;
+  }
+  const resumed = await next;
+  const session = await ports.sessions.read('s1');
+  assert.deepEqual([session.open, session.turn, session.turnId], [true, resumed.turn, '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/runtime/file-stores.test.js

@@ -533,6 +533,16 @@ process.disconnect();`;
   assert.deepEqual(await exited, [0, null]);
 });
 
+test('a session lifecycle lock that stays held refuses with ELOCKED, so a prompt can open without it', async () => {
+  const { stores: cursor } = stores();
+  let release;
+  const holder = cursor.sessions.withSession('s1', () => new Promise((resolve) => { release = resolve; }));
+  await assert.rejects(cursor.sessions.withSession('s1', () => assert.fail('work must not run')), { code: 'ELOCKED' });
+  assert.equal(await cursor.sessions.withSession('s2', () => 'other session'), 'other session');
+  release();
+  await holder;
+});
+
 test('a correction lock that stays held refuses with ELOCKED, so a hook reports it', async () => {
   const { stores: cursor } = stores();
   let release;