From 914ba565ce624fd4df7673f47cd38a5e4af450e2 Mon Sep 17 00:00:00 2001 From: "xyanua." Date: Wed, 11 Nov 2026 08:28:32 +0700 Subject: [PATCH 1/4] fix: keep resumed turns safe from late session cleanup --- lib/app/ports.js | 1 + lib/app/prompt.js | 19 ++++--- lib/app/session-end.js | 23 +++++--- lib/runtime/store/file-stores.js | 29 +++++----- test/app/memory-ports.js | 1 + test/app/turn-lifecycle.test.js | 90 ++++++++++++++++++++++++++++++++ 6 files changed, 135 insertions(+), 28 deletions(-) diff --git a/lib/app/ports.js b/lib/app/ports.js index e5307d0..a95c93c 100644 --- a/lib/app/ports.js +++ b/lib/app/ports.js @@ -14,6 +14,7 @@ * @property {string} label * * @typedef {object} SessionStore + * @property {(sessionId: string, work: () => T|Promise) => Promise} withSession Serializes prompt opening and session cleanup across stores. * @property {(sessionId: string) => Promise} read * @property {(visit: (session: object, sessionId: string) => void) => Promise} readEach * @property {(sessionId: string, mutate: Function) => Promise} update diff --git a/lib/app/prompt.js b/lib/app/prompt.js index 5d13b07..2765abb 100644 --- a/lib/app/prompt.js +++ b/lib/app/prompt.js @@ -55,10 +55,10 @@ async function recordedCorrection(ports, sessionId, session, prompt) { /** * A correction marks the turn and changes its range before removing drafts from the prior version. Its reuses are - * reported again under `:correction`. Every prompt resets the quiet period for a range that is not yet due. + * reported again under `:correction`. The caller opens the turn and records its activity afterwards. */ async function labelFollowUp(ports, sessionId, prompt) { - const { sessions, candidates, drafts, reports, clock } = ports; + const { sessions, drafts, reports } = ports; const label = followUpLabelOf(prompt); let skipped; if (label === 'correction') { @@ -67,13 +67,11 @@ async function labelFollowUp(ports, sessionId, prompt) { skipped = correction.skipped; 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 }; } /** A recall's match when it brings a strategy (`{ ids, text, stepTexts? }`, a Recipe's steps or one asset), else null. */ @@ -161,9 +159,14 @@ async function rememberTurnHead({ project, sessions }, sessionId, { turn, sessio * 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 { skipped, reportsQueued } = isCommand ? {} : await labelFollowUp(ports, sessionId, prompt); + const pid = ports.owner.current(); + const opened = await ports.sessions.withSession(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() })); + }); + if (!isCommand) await scheduleDueDistillation(ports, 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 }); diff --git a/lib/app/session-end.js b/lib/app/session-end.js index 410c628..fc23f88 100644 --- a/lib/app/session-end.js +++ b/lib/app/session-end.js @@ -8,6 +8,7 @@ import { settledSessionOf, turnEnded, withoutPendingControls } from './turn-end. 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 @@ -19,8 +20,9 @@ const NOTHING_DONE = Object.freeze({ checks: [], commands: [], paths: [], edits: export async function sessionEnded(ports, { sessionId, projectDir, turnId = null, settleMs = 0, openTurnReason = 'aborted' }) { const session = await settledSessionOf(ports, sessionId, settleMs); if (isEarlierTurn(session, turnId)) return; - await turnEnded(ports, { sessionId, reason: openTurnReason, projectDir, recordOf: () => NOTHING_DONE }); + await turnEnded(ports, { sessionId, reason: openTurnReason, projectDir, recordOf: () => NOTHING_DONE, expectedTurn: session }); 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; @@ -28,11 +30,18 @@ export async function sessionEnded(ports, { sessionId, projectDir, turnId = null 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]); } diff --git a/lib/runtime/store/file-stores.js b/lib/runtime/store/file-stores.js index be247a5..2804974 100644 --- a/lib/runtime/store/file-stores.js +++ b/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, {}); diff --git a/test/app/memory-ports.js b/test/app/memory-ports.js index 6176d31..e30ad52 100644 --- a/test/app/memory-ports.js +++ b/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); diff --git a/test/app/turn-lifecycle.test.js b/test/app/turn-lifecycle.test.js index e174f16..38ecd6f 100644 --- a/test/app/turn-lifecycle.test.js +++ b/test/app/turn-lifecycle.test.js @@ -519,6 +519,96 @@ 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 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 }); From c9860118c0f1ca1d088defa3cf44ac6b57d67c93 Mon Sep 17 00:00:00 2001 From: "xyanua." Date: Wed, 11 Nov 2026 08:18:51 +0700 Subject: [PATCH 2/4] 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 ded5bcd746f50a37dd15b16a62983ee0b819ab6f Mon Sep 17 00:00:00 2001 From: forrestlinfeng Date: Wed, 11 Nov 2026 10:13:05 +0800 Subject: [PATCH 3/4] Close only the turn session end found open A prompt resumed while a session end was closing its open turn had its new turn closed instead. Co-Authored-By: Claude Opus 5.5 --- lib/app/turn-end.js | 2 +- test/app/turn-lifecycle.test.js | 23 +++++++++++++++++++++++ 2 files changed, 24 insertions(+), 1 deletion(-) diff --git a/lib/app/turn-end.js b/lib/app/turn-end.js index 5d4b8fa..b903e32 100644 --- a/lib/app/turn-end.js +++ b/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, { diff --git a/test/app/turn-lifecycle.test.js b/test/app/turn-lifecycle.test.js index 38ecd6f..44a4e92 100644 --- a/test/app/turn-lifecycle.test.js +++ b/test/app/turn-lifecycle.test.js @@ -569,6 +569,29 @@ test('a session end delayed before closing its turn cannot close a resumed turn' 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-')); From 9cadb2bd592fe78a32bc53758b04dd987abb06d6 Mon Sep 17 00:00:00 2001 From: forrestlinfeng Date: Wed, 11 Nov 2026 10:14:15 +0800 Subject: [PATCH 4/4] Open a prompt's turn when the session lifecycle lock stays held A session-end hook killed while holding the lock would otherwise fail every prompt of the resumed session until the lock goes stale, as the draft correction lock already avoids. Co-Authored-By: Claude Opus 5.5 --- lib/app/prompt.js | 23 ++++++++++++++++++----- test/agent/driver.test.js | 14 ++++++++++++++ test/runtime/file-stores.test.js | 10 ++++++++++ 3 files changed, 42 insertions(+), 5 deletions(-) diff --git a/lib/app/prompt.js b/lib/app/prompt.js index ff26d55..a96616f 100644 --- a/lib/app/prompt.js +++ b/lib/app/prompt.js @@ -73,6 +73,17 @@ async function labelFollowUp(ports, sessionId, prompt) { 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. */ const strategyOf = (match) => (match?.ids?.length && match.text ? match : null); @@ -154,18 +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, reportsQueued } = isCommand ? {} : await labelFollowUp(ports, sessionId, prompt); + const labelled = isCommand ? {} : await labelFollowUp(ports, sessionId, prompt); const pid = ports.owner.current(); - const opened = await ports.sessions.withSession(sessionId, async () => { + 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() })); }); - if (!isCommand) await scheduleDueDistillation(ports, reportsQueued ? [TASKS.DRAIN] : []); + 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 }); diff --git a/test/agent/driver.test.js b/test/agent/driver.test.js index b2010fd..24fa39a 100644 --- a/test/agent/driver.test.js +++ b/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); diff --git a/test/runtime/file-stores.test.js b/test/runtime/file-stores.test.js index a12f6c8..cc858a0 100644 --- a/test/runtime/file-stores.test.js +++ b/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;