diff --git a/lib/app/ports.js b/lib/app/ports.js index 6aef5c0..792c879 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 6487c92..a96616f 100644 --- a/lib/app/prompt.js +++ b/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 }); diff --git a/lib/app/session-end.js b/lib/app/session-end.js index cfdd647..3d077c5 100644 --- a/lib/app/session-end.js +++ b/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,6 +23,7 @@ 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; @@ -29,11 +31,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/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/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/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/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..44a4e92 100644 --- a/test/app/turn-lifecycle.test.js +++ b/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 }); 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;