From 42f1048c66f34eee787eec38e92eef8486caa16d Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Fri, 13 Nov 2026 20:52:36 -0800 Subject: [PATCH] perf(native-chat): coalesce ready restart admissions --- .../agent-session-journal/journal-store.ts | 2 +- .../journal-queue-change-writer.ts | 41 ++++++ .../journal-queued-messages.ts | 44 +++---- .../journal-store-collaborators.ts | 3 +- .../submissions/journal-submission-writer.ts | 7 +- .../writing/journal-row-writer.ts | 38 +++++- .../writing/journal-write-queue.ts | 16 ++- ...tructured-agent-session-queued-messages.ts | 4 +- ...ent-session-restart-failure-filing.test.ts | 3 + ...structured-agent-session-host-mutations.ts | 13 +- .../structured-agent-session-turns.ts | 8 +- .../structured-agent-session-restart-send.ts | 120 +++++++++++------- .../agent-session-record-publication.ts | 5 +- .../agent-session-recovery-capsule.test.ts | 32 +++++ .../records/agent-session-recovery-capsule.ts | 3 +- .../agent-session-resume-selections.ts | 78 ++++++++++-- .../agent-session-store-transactions.ts | 11 +- 17 files changed, 313 insertions(+), 115 deletions(-) create mode 100644 src/main/native-chat/agent-session-journal/queued-messages/journal-queue-change-writer.ts diff --git a/src/main/native-chat/agent-session-journal/journal-store.ts b/src/main/native-chat/agent-session-journal/journal-store.ts index 54a90d3b1ed..42da802e14d 100644 --- a/src/main/native-chat/agent-session-journal/journal-store.ts +++ b/src/main/native-chat/agent-session-journal/journal-store.ts @@ -91,7 +91,7 @@ export class AgentSessionJournal { identity: this.identity, now: options.now ?? (() => Date.now()), mintEpoch: options.mintEpoch ?? randomUUID, - serialize: (run) => this.queue.serialize(run), + serialize: (run, admission, published) => this.queue.serialize(run, admission, published), database: () => this.database, state: () => this.state, readOnly: () => this.database.readOnly, diff --git a/src/main/native-chat/agent-session-journal/queued-messages/journal-queue-change-writer.ts b/src/main/native-chat/agent-session-journal/queued-messages/journal-queue-change-writer.ts new file mode 100644 index 00000000000..22191e2e621 --- /dev/null +++ b/src/main/native-chat/agent-session-journal/queued-messages/journal-queue-change-writer.ts @@ -0,0 +1,41 @@ +import type { JournalWriteAdmission } from '../writing/journal-write-queue' +import type { JournalOperationReceipt, JournalRowWriter } from '../writing/journal-row-writer' +import type { JournalQueueCardChange } from './journal-queue-events' +import type { JournalQueuePlan } from './journal-queued-messages' + +/** Plan and accept card changes on the journal lane, then publish before releasing it. */ +export function writeJournalQueueChanges( + writer: JournalRowWriter, + planChanges: (changes: readonly JournalQueueCardChange[]) => JournalQueuePlan, + plan: () => { changes: readonly JournalQueueCardChange[]; result: T }, + receipt?: JournalOperationReceipt, + admission?: JournalWriteAdmission +): Promise { + let planned: { result: T; plan: JournalQueuePlan } | undefined + return writer + .enqueueRows( + () => { + const next = plan() + planned = { result: next.result, plan: planChanges(next.changes) } + return planned.plan.rows + }, + { + write: (db, row) => { + planned?.plan.receipt.write(db, row) + receipt?.write(db, row) + }, + committed: () => { + planned?.plan.receipt.committed() + receipt?.committed() + } + }, + undefined, + admission + ) + .then(() => { + if (!planned) { + throw new Error('agent_session_queue_write_unplanned') + } + return planned.result + }) +} diff --git a/src/main/native-chat/agent-session-journal/queued-messages/journal-queued-messages.ts b/src/main/native-chat/agent-session-journal/queued-messages/journal-queued-messages.ts index 68b15fa56a9..142f37f551a 100644 --- a/src/main/native-chat/agent-session-journal/queued-messages/journal-queued-messages.ts +++ b/src/main/native-chat/agent-session-journal/queued-messages/journal-queued-messages.ts @@ -1,3 +1,5 @@ +import { writeJournalQueueChanges } from './journal-queue-change-writer' +import type { JournalWriteAdmission } from '../writing/journal-write-queue' import type { JournalReducerState } from '../projection/journal-reducer' import { journalInputAttemptEnded, @@ -129,7 +131,8 @@ export class JournalQueuedMessages { holdReason?: QueuedMessageHoldReason options?: AgentSessionExecutionOptions }, - receipt?: JournalOperationReceipt + receipt?: JournalOperationReceipt, + admission?: JournalWriteAdmission ): Promise { const { messageId, body } = input const claims = journalAttachmentClaimReceipt((db) => @@ -155,7 +158,8 @@ export class JournalQueuedMessages { } return { changes: [{ messageId, change }], result: undefined } }, - composeJournalOperationReceipts(claims, ...(receipt ? [receipt] : [])) + composeJournalOperationReceipts(claims, ...(receipt ? [receipt] : [])), + admission ).then(() => { const card = this.get(messageId) if (!card) { @@ -296,35 +300,17 @@ export class JournalQueuedMessages { ) } - /** Resolve after preceding writes; commit receipts with changed rows and skip empty plans. */ private write( plan: () => { changes: readonly JournalQueueCardChange[]; result: T }, - receipt?: JournalOperationReceipt + receipt?: JournalOperationReceipt, + admission?: JournalWriteAdmission ): Promise { - let planned: { result: T; plan: JournalQueuePlan } | undefined - return this.deps.writer - .enqueueRows( - () => { - const next = plan() - planned = { result: next.result, plan: this.plan(next.changes) } - return planned.plan.rows - }, - { - write: (db, row) => { - planned?.plan.receipt.write(db, row) - receipt?.write(db, row) - }, - committed: () => { - planned?.plan.receipt.committed() - receipt?.committed() - } - } - ) - .then(() => { - if (!planned) { - throw new Error('agent_session_queue_write_unplanned') - } - return planned.result - }) + return writeJournalQueueChanges( + this.deps.writer, + (changes) => this.plan(changes), + plan, + receipt, + admission + ) } } diff --git a/src/main/native-chat/agent-session-journal/store-lifecycle/journal-store-collaborators.ts b/src/main/native-chat/agent-session-journal/store-lifecycle/journal-store-collaborators.ts index 98f978de069..e827b91371d 100644 --- a/src/main/native-chat/agent-session-journal/store-lifecycle/journal-store-collaborators.ts +++ b/src/main/native-chat/agent-session-journal/store-lifecycle/journal-store-collaborators.ts @@ -24,7 +24,6 @@ import { JournalSubmissionWriter } from '../submissions/journal-submission-write import { validateJournalDormantAnswer } from '../submissions/journal-dormant-answer' import type { JournalRow } from '../durable-rows/journal-row-schema' import type { AgentSessionJournal } from '../journal-store' -import type { JournalWriteBody } from '../writing/journal-write-queue' import { journalInputClaimsReceipt, journalAttachmentClaimsReceipt, @@ -40,7 +39,7 @@ export type JournalStoreHost = { identity: AgentSessionJournalIdentity now: () => number mintEpoch: () => string - serialize: (run: JournalWriteBody) => Promise + serialize: ConstructorParameters[0]['serialize'] database: () => JournalHostDatabase state: () => JournalReducerState readOnly: () => boolean diff --git a/src/main/native-chat/agent-session-journal/submissions/journal-submission-writer.ts b/src/main/native-chat/agent-session-journal/submissions/journal-submission-writer.ts index c685881121f..43d29c4beb7 100644 --- a/src/main/native-chat/agent-session-journal/submissions/journal-submission-writer.ts +++ b/src/main/native-chat/agent-session-journal/submissions/journal-submission-writer.ts @@ -19,6 +19,7 @@ import { } from '../writing/journal-row-builders' import type { JournalRow, JournalSubmissionRow } from '../durable-rows/journal-row-schema' import type { JournalOperationReceipt, JournalRowWriter } from '../writing/journal-row-writer' +import type { JournalWriteAdmission } from '../writing/journal-write-queue' import type { JournalSubmissionConsume, JournalSubmissionInput, @@ -53,7 +54,8 @@ export class JournalSubmissionWriter { itemId: string expectedRevision: number contextEpoch: string - } + }, + admission?: JournalWriteAdmission ): Promise { const { claimAttachments, identity, queuedMessages, rowWriter, state } = this.deps const buildSubmission = journalSubmissionRowBuilder(state, identity, input, consume) @@ -125,7 +127,8 @@ export class JournalSubmissionWriter { receipt?.committed() } }, - (rows) => rows.at(-1) + (rows) => rows.at(-1), + admission ) .then(rowCursor) } diff --git a/src/main/native-chat/agent-session-journal/writing/journal-row-writer.ts b/src/main/native-chat/agent-session-journal/writing/journal-row-writer.ts index 128c3dab950..23e0fd2ae3f 100644 --- a/src/main/native-chat/agent-session-journal/writing/journal-row-writer.ts +++ b/src/main/native-chat/agent-session-journal/writing/journal-row-writer.ts @@ -4,7 +4,7 @@ import type { JournalHostDatabase } from '../database/journal-host-database' import type { AgentJournalCursor } from '../../../../shared/agent-session/journal/agent-session-journal-identity' import type { JournalRow, JournalSubmissionRow } from '../durable-rows/journal-row-schema' import { assertJournalFence, assertJournalWritable } from './journal-write-guards' -import type { JournalWriteBody } from './journal-write-queue' +import type { JournalWriteBody, JournalWriteAdmission } from './journal-write-queue' import type Database from '../../../sqlite/sync-database' /** Write receipts in the row transaction; notify synchronously after COMMIT, never rollback. */ @@ -48,7 +48,11 @@ export function writeReceiptIfChanged( export type JournalRowWriterDeps = { sessionId: string now: () => number - serialize: (run: JournalWriteBody) => Promise + serialize: ( + run: JournalWriteBody, + admission?: JournalWriteAdmission, + published?: () => void + ) => Promise database: () => JournalHostDatabase readOnly: () => boolean highestFence: () => number @@ -100,16 +104,33 @@ export class JournalRowWriter { enqueueRows( plan: () => readonly ((seq: number, ts: number) => JournalRow)[], receipt?: JournalReceiptPreparation, - receiptRow?: (rows: readonly JournalRow[]) => JournalRow | undefined + receiptRow?: (rows: readonly JournalRow[]) => JournalRow | undefined, + admission?: JournalWriteAdmission ): Promise { - return this.deps.serialize(() => this.writeRows(plan, receipt, receiptRow)) + let publish: (() => void) | undefined + return this.deps.serialize( + () => + this.writeRows( + plan, + receipt, + receiptRow, + admission + ? (run) => { + publish = run + } + : undefined + ), + admission, + () => publish?.() + ) } /** Caller must already own the write lane. */ writeRows( plan: () => readonly ((seq: number, ts: number) => JournalRow)[], prepareReceipt?: JournalReceiptPreparation, - receiptRow?: (rows: readonly JournalRow[]) => JournalRow | undefined + receiptRow?: (rows: readonly JournalRow[]) => JournalRow | undefined, + afterAdmission?: (publish: () => void) => void ): JournalRow[] { assertJournalWritable(this.deps.readOnly(), this.deps.sessionId) const first = this.deps.nextSequence() @@ -138,7 +159,12 @@ export class JournalRowWriter { (adopt) => adopt?.() ) receipt?.committed() - this.deps.commit(rows, ts, inputChanged) + const publish = () => this.deps.commit(rows, ts, inputChanged) + if (afterAdmission) { + afterAdmission(publish) + } else { + publish() + } return this.project(rows) } diff --git a/src/main/native-chat/agent-session-journal/writing/journal-write-queue.ts b/src/main/native-chat/agent-session-journal/writing/journal-write-queue.ts index 3ddf2482f29..ef3f7140f14 100644 --- a/src/main/native-chat/agent-session-journal/writing/journal-write-queue.ts +++ b/src/main/native-chat/agent-session-journal/writing/journal-write-queue.ts @@ -5,6 +5,7 @@ import { AgentSessionJournalError } from './journal-write-guards' /** Async bodies could interleave sequence assignment and commits. */ export type JournalWriteResult = T extends PromiseLike ? never : T export type JournalWriteBody = () => JournalWriteResult +export type JournalWriteAdmission = (write: JournalWriteBody) => Promise /** Idle writes commit synchronously; reentrant/waiting writes queue. Admission remains valid after close. */ export class JournalWriteQueue { @@ -20,10 +21,23 @@ export class JournalWriteQueue { this.closed = true } - serialize(run: JournalWriteBody): Promise { + serialize( + run: JournalWriteBody, + admission?: JournalWriteAdmission, + published?: () => void + ): Promise { if (this.closed) { return Promise.reject(this.closedError()) } + if (admission) { + // Own the FIFO position before admission; no SQLite transaction spans its await. + return this.join(() => + admission(run).then((value) => { + published?.() + return value + }) + ) + } return this.lineBusy ? this.join(run) : this.runNow(run) } diff --git a/src/main/native-chat/agent-session-wire/delivery/structured-agent-session-queued-messages.ts b/src/main/native-chat/agent-session-wire/delivery/structured-agent-session-queued-messages.ts index 1682c42de81..e73df18c008 100644 --- a/src/main/native-chat/agent-session-wire/delivery/structured-agent-session-queued-messages.ts +++ b/src/main/native-chat/agent-session-wire/delivery/structured-agent-session-queued-messages.ts @@ -47,6 +47,7 @@ export async function maybeQueueStructuredAgentSessionSend( | 'journal' | 'fence' | 'operationReceipt' + | 'sendAdmission' | 'currentWork' | 'agents' | 'agent' @@ -111,7 +112,8 @@ export async function maybeQueueStructuredAgentSessionSend( options: captured.options, ...(params.userSend ? { requireAttachments: true } : {}) }, - ctx.operationReceipt + ctx.operationReceipt, + ctx.sendAdmission ) } catch (error) { return agentSessionAdmissionWriteRefusal(ctx, error) diff --git a/src/main/native-chat/agent-session-wire/host/testing/structured-agent-session-restart-failure-filing.test.ts b/src/main/native-chat/agent-session-wire/host/testing/structured-agent-session-restart-failure-filing.test.ts index 27aa423f696..0f02dd82f5b 100644 --- a/src/main/native-chat/agent-session-wire/host/testing/structured-agent-session-restart-failure-filing.test.ts +++ b/src/main/native-chat/agent-session-wire/host/testing/structured-agent-session-restart-failure-filing.test.ts @@ -26,6 +26,7 @@ it('keeps an accepted restart offer when its settlement cannot establish Stop hi const settle = AgentSessionRecoveryCapsule.prototype.settleResume let retained: string[] | undefined let failures: unknown + let historyUnavailable = false vi.spyOn(AgentSessionRecoveryCapsule.prototype, 'settleResume').mockImplementationOnce( async function (this: AgentSessionRecoveryCapsule, ...args) { // A historical correction after admission requires fresh retirement evidence. @@ -37,6 +38,7 @@ it('keeps an accepted restart offer when its settlement cannot establish Stop hi const reading = vi.spyOn(database.db, 'prepare').mockImplementation((sql) => { if (!failed && sql.includes('seq >= ? AND seq <= ?') && sql.includes('stopEvent')) { failed = true + historyUnavailable = true throw new Error('retirement history temporarily unavailable') } return prepare(sql) @@ -54,6 +56,7 @@ it('keeps an accepted restart offer when its settlement cannot establish Stop hi ) await host.restartResume.continueAfterRestart([SESSION], 'modal') + expect(historyUnavailable).toBe(true) expect(retained).toEqual([SESSION]) expect(failures).toEqual([]) const submissions = (await host.journalSnapshot(SESSION)).submissions diff --git a/src/main/native-chat/agent-session-wire/operations/structured-agent-session-host-mutations.ts b/src/main/native-chat/agent-session-wire/operations/structured-agent-session-host-mutations.ts index 5af0aaaf75b..7f26eb6ce52 100644 --- a/src/main/native-chat/agent-session-wire/operations/structured-agent-session-host-mutations.ts +++ b/src/main/native-chat/agent-session-wire/operations/structured-agent-session-host-mutations.ts @@ -1,3 +1,4 @@ +import type { JournalWriteAdmission } from '../../agent-session-journal/writing/journal-write-queue' import type { StructuredAgentSessionSendAcceptance } from './structured-agent-session-turns' // Share mutation admission/planning/publication. Conversation writes need no child lease; // provider-only effects start their child before admission, sends start through delivery. @@ -121,12 +122,16 @@ export function sendStructuredAgentSessionTurn( ) } try { - const apply = () => - runQueueableStructuredAgentSessionSend(context, ctx, params, () => plan.run(ctx)) + const apply = (admission?: JournalWriteAdmission) => { + const sending = admission ? { ...ctx, sendAdmission: admission } : ctx + return runQueueableStructuredAgentSessionSend(context, sending, params, () => + plan.run(sending) + ) + } const sent = await (params.accept - ? params.accept(() => { + ? params.accept((admission) => { accepted = true - return apply() + return apply(admission) }) : apply()) if (!sent.ok) { diff --git a/src/main/native-chat/agent-session-wire/operations/structured-agent-session-turns.ts b/src/main/native-chat/agent-session-wire/operations/structured-agent-session-turns.ts index 3cd9b713808..5dbd95aad66 100644 --- a/src/main/native-chat/agent-session-wire/operations/structured-agent-session-turns.ts +++ b/src/main/native-chat/agent-session-wire/operations/structured-agent-session-turns.ts @@ -22,6 +22,7 @@ import type { AgentSessionSendResult } from '../../../../shared/agent-session/ag import type { StructuredAgentSessionCurrentWork } from '../observations/structured-agent-session-current-work' import type { AgentSessionJournal } from '../../agent-session-journal/journal-store' import type { JournalOperationReceipt } from '../../agent-session-journal/writing/journal-row-writer' +import type { JournalWriteAdmission } from '../../agent-session-journal/writing/journal-write-queue' import type { StructuredAgentSessionAdapter } from '../providers/structured-agent-session-adapter' import { classifyJournalOpenFailure, @@ -58,6 +59,7 @@ export type AgentSessionTurnContext = { failureTextContext?: AgentSessionFailureWordsContext /** Commit success alongside its accepting write or standalone receipt. */ operationReceipt?: AgentSessionOperationReceipt + sendAdmission?: JournalWriteAdmission now: () => number } @@ -72,7 +74,7 @@ export type TurnOutcome = | { ok: false; refusal: AgentSessionWireRefusal } export type StructuredAgentSessionSendAcceptance = ( - apply: () => Promise> + apply: (admission: JournalWriteAdmission) => Promise> ) => Promise> function invalid( @@ -171,7 +173,9 @@ export async function performSend( await ctx.journal.appendSubmission( { ...input, options: captured.options, fence: ctx.fence, handoverRecorded: true }, undefined, - ctx.operationReceipt + ctx.operationReceipt, + undefined, + ctx.sendAdmission ) } catch (error) { forgetNativeChatPrompt(ctx.journal, input.clientMessageId) diff --git a/src/main/native-chat/agent-session-wire/recovery/structured-agent-session-restart-send.ts b/src/main/native-chat/agent-session-wire/recovery/structured-agent-session-restart-send.ts index 7686c69c119..d4c5e419688 100644 --- a/src/main/native-chat/agent-session-wire/recovery/structured-agent-session-restart-send.ts +++ b/src/main/native-chat/agent-session-wire/recovery/structured-agent-session-restart-send.ts @@ -1,3 +1,4 @@ +import type { JournalWriteAdmission } from '../../agent-session-journal/writing/journal-write-queue' import type { StructuredAgentSessionSendAcceptance } from '../operations/structured-agent-session-turns' import type { StructuredAgentSessionHostDeps } from '../host/structured-agent-session-host-types' import type { AgentSessionResumeSelection } from '../../../runtime/agent-session/records/agent-session-recovery-capsule' @@ -14,7 +15,7 @@ import { RestartContinuationSupersededError } from './structured-agent-session-restart-continuation' -/** Take the capsule lock only after the send owns its session lane and has finished preparation. */ +/** Join the finite capsule frontier only after preparation and the accepting write queue are ready. */ export function restartSendAcceptance(input: { deps: Pick selection: AgentSessionResumeSelection @@ -27,56 +28,83 @@ export function restartSendAcceptance(input: { const { deps, selection, attempt, stillWanted } = input const { marker } = selection return async (apply) => { - let reached = false - const result = await deps - .recoveryCapsule!.withSelectedResume(selection, input.now(), async () => { - reached = true - if (input.stopping()) { - throw new RestartContinuationDeferredError() - } - if (!stillWanted()) { + let admitted = false + let controlError: unknown + const admission: JournalWriteAdmission = async (write) => { + let reached = false + try { + const result = await deps.recoveryCapsule!.withSelectedResume( + selection, + input.now(), + () => { + reached = true + if (input.stopping()) { + throw new RestartContinuationDeferredError() + } + if (!stillWanted()) { + throw new RestartContinuationSupersededError() + } + admitted = true + return write() + } + ) + if (!result.selected) { throw new RestartContinuationSupersededError() } - const outcome = await apply() - if (!outcome.ok && conclusiveRestartRefusal(outcome.refusal)) { - const read = deps.store.readCommandReceipt({ kind: 'global' }, attempt.id) - if (read.verdict === 'absent') { - const prepared = restartContinuationEnvelope( - marker.sessionId, - input.fence()!, - marker, - attempt.id, - attempt.body - ) - await deps.store.commitOperationReceipt( - buildCommandReceiptTransaction({ kind: 'global' }, () => ({ - operationId: attempt.id, - sessionId: marker.sessionId, - callerKey: STRUCTURED_AGENT_SESSION_RESTART_CONTINUATION_CALLER, - method: 'agentSession.send', - fingerprint: prepared.envelope.payloadFingerprint, - status: 'rejected', - acceptedAt: input.now(), - rejection: { - reference: agentSessionRefusalReference(outcome.refusal), - message: outcome.refusal.message - } - })) - ) - } - } - return outcome - }) - .catch((error: unknown) => { - if (!reached) { - throw new RestartContinuationDeferredError() + return result.value + } catch (error) { + if ( + !reached || + error instanceof RestartContinuationDeferredError || + error instanceof RestartContinuationSupersededError + ) { + // Journal adapters translate write errors; preserve the recovery control outcome instead. + controlError = + error instanceof RestartContinuationSupersededError + ? error + : new RestartContinuationDeferredError() + throw controlError } throw error - }) - if (!result.selected) { - throw new RestartContinuationSupersededError() + } } - return result.value + const outcome = await apply(admission) + if (controlError) { + throw controlError + } + if (!outcome.ok && conclusiveRestartRefusal(outcome.refusal)) { + const read = deps.store.readCommandReceipt({ kind: 'global' }, attempt.id) + if (read.verdict === 'absent') { + const prepared = restartContinuationEnvelope( + marker.sessionId, + input.fence()!, + marker, + attempt.id, + attempt.body + ) + await deps.store.commitOperationReceipt( + buildCommandReceiptTransaction({ kind: 'global' }, () => ({ + operationId: attempt.id, + sessionId: marker.sessionId, + callerKey: STRUCTURED_AGENT_SESSION_RESTART_CONTINUATION_CALLER, + method: 'agentSession.send', + fingerprint: prepared.envelope.payloadFingerprint, + status: 'rejected', + acceptedAt: input.now(), + rejection: { + reference: agentSessionRefusalReference(outcome.refusal), + message: outcome.refusal.message + } + })), + admission + ) + } + } + if (!admitted) { + // Replay and refusals without a write still validate the exact offer at this boundary. + await admission(() => undefined) + } + return outcome } } diff --git a/src/main/runtime/agent-session/records/agent-session-record-publication.ts b/src/main/runtime/agent-session/records/agent-session-record-publication.ts index 40fd7954bbd..1cb08bf88fb 100644 --- a/src/main/runtime/agent-session/records/agent-session-record-publication.ts +++ b/src/main/runtime/agent-session/records/agent-session-record-publication.ts @@ -1,4 +1,3 @@ -import type { JournalOperationReceipt } from '../../../native-chat/agent-session-journal/writing/journal-row-writer' import type { AgentSessionStoreState } from './agent-session-store-state' import type { AgentSessionStoreTransactions } from './agent-session-store-transactions' @@ -30,8 +29,8 @@ export abstract class AgentSessionRecordPublication { admittedSendCount = (sessionId: string) => this.transactions.admittedSendCount(sessionId) - commitOperationReceipt = (receipt: JournalOperationReceipt) => - this.transactions.commitReceipt(receipt) + commitOperationReceipt = (...args: Parameters) => + this.transactions.commitReceipt(...args) readCommandReceipt = (...args: Parameters) => this.transactions.readCommandReceipt(...args) diff --git a/src/main/runtime/agent-session/records/agent-session-recovery-capsule.test.ts b/src/main/runtime/agent-session/records/agent-session-recovery-capsule.test.ts index 1add738c9f9..729d43b1477 100644 --- a/src/main/runtime/agent-session/records/agent-session-recovery-capsule.test.ts +++ b/src/main/runtime/agent-session/records/agent-session-recovery-capsule.test.ts @@ -48,6 +48,38 @@ async function fileFailure(overrides: Partial = } describe('durable restart offers', () => { + it('keeps healthy admissions independent and honors a queued dismissal before a later admission', async () => { + await capsule.record( + [marker(), marker({ sessionId: 'second' }), marker({ sessionId: 'later' })], + NOW + ) + const selections = await capsule.selectResume(NOW) + const first = selections.find((selection) => selection.marker.sessionId === SESSION)! + const second = selections.find((selection) => selection.marker.sessionId === 'second')! + const later = selections.find((selection) => selection.marker.sessionId === 'later')! + const other = new AgentSessionRecoveryCapsule(directory) + const accepted: string[] = [] + let lateAdmission: Promise | undefined + let dismissal: Promise | undefined + const results = await Promise.allSettled([ + capsule.withSelectedResume(first, NOW, () => { + lateAdmission = capsule.withSelectedResume(later, NOW, () => accepted.push('later')) + dismissal = other.dismiss(['later'], NOW) + throw new Error('first acceptance failed') + }), + capsule.withSelectedResume(second, NOW, () => accepted.push('second')) + ]) + + expect(results).toMatchObject([ + { status: 'rejected', reason: { message: 'first acceptance failed' } }, + { status: 'fulfilled', value: { selected: true, value: 1 } } + ]) + expect(await dismissal).toBe(1) + expect(await lateAdmission).toEqual({ selected: false }) + expect(accepted).toEqual(['second']) + expect((await other.list(NOW)).map((offer) => offer.sessionId)).toEqual([SESSION, 'second']) + }) + it('reserves only one copy of a selected session across competing owners', async () => { await capsule.record([marker(), marker({ sessionId: 'second' })], NOW) const other = new AgentSessionRecoveryCapsule(directory) diff --git a/src/main/runtime/agent-session/records/agent-session-recovery-capsule.ts b/src/main/runtime/agent-session/records/agent-session-recovery-capsule.ts index 2608d162f4a..0c7419d8f50 100644 --- a/src/main/runtime/agent-session/records/agent-session-recovery-capsule.ts +++ b/src/main/runtime/agent-session/records/agent-session-recovery-capsule.ts @@ -10,6 +10,7 @@ import { join } from 'node:path' import type { AgentSessionResumeMarker } from '../../../../shared/agent-session/lifecycle/agent-session-resume-marker' import { readRecoveryCapsule, publishRecoveryCapsule } from './agent-session-recovery-capsule-file' import { withFileTransactionLock } from '../../../filesystem/file-transaction-lock' +import type { JournalWriteBody } from '../../../native-chat/agent-session-journal/writing/journal-write-queue' import { MAX_FAILURE_FIELD_LENGTH, normalizeState, @@ -47,7 +48,7 @@ export class AgentSessionRecoveryCapsule { withSelectedResume( selection: AgentSessionResumeSelection, now: number, - accept: () => Promise + accept: JournalWriteBody ) { return this.selections.withSelectedResume(selection, now, accept) } diff --git a/src/main/runtime/agent-session/records/agent-session-resume-selections.ts b/src/main/runtime/agent-session/records/agent-session-resume-selections.ts index 8371019dc9d..aa37a7bcd45 100644 --- a/src/main/runtime/agent-session/records/agent-session-resume-selections.ts +++ b/src/main/runtime/agent-session/records/agent-session-resume-selections.ts @@ -10,6 +10,16 @@ import { type RecoveryCapsuleState, type RecoveryEntry } from './agent-session-recovery-capsule-entries' +import type { JournalWriteBody } from '../../../native-chat/agent-session-journal/writing/journal-write-queue' + +type ResumeFrontier = { + now: number + run: ( + entries: ReadonlyMap, + failed: ReadonlyMap + ) => () => void + reject: (error: unknown) => void +} export type AgentSessionResumeSelection = { marker: AgentSessionResumeMarker @@ -39,6 +49,8 @@ function selectedResume( } export class AgentSessionResumeSelections { + private readonly ready: ResumeFrontier[] = [] + private draining = false constructor( private readonly deps: { withLock: (apply: () => Promise) => Promise @@ -116,28 +128,66 @@ export class AgentSessionResumeSelections { }) } - /** The callback is only the send's short acceptance transaction, never provider startup. */ + /** Callers own their write queue; each callback is one synchronous acceptance transaction. */ withSelectedResume( selection: AgentSessionResumeSelection, now: number, - accept: () => Promise + accept: JournalWriteBody ): Promise<{ selected: false } | { selected: true; value: T }> { - return this.deps.withLock(async () => { - const state = normalizeState(await this.deps.readState(), now) - const id = selection.marker.sessionId - if ( - !selectedResume( - state.entries.find((entry) => entry.marker.sessionId === id), - state.failed.find((failure) => failure.marker.sessionId === id), - selection - ) - ) { - return { selected: false } + return new Promise((resolve, reject) => { + this.ready.push({ + now, + reject, + run: (entries, failed) => { + try { + const id = selection.marker.sessionId + const result = selectedResume(entries.get(id), failed.get(id), selection) + ? { selected: true as const, value: accept() } + : { selected: false as const } + return () => resolve(result) + } catch (error) { + return () => reject(error) + } + } + }) + if (!this.draining) { + this.draining = true + void this.drainReady() } - return { selected: true, value: await accept() } }) } + private async drainReady(): Promise { + let frontier: ResumeFrontier[] = [] + try { + const settle = await this.deps.withLock(async () => { + // Arrivals after this capture belong to the next finite drain. + frontier = this.ready.splice(0) + const now = frontier.reduce((latest, peer) => Math.max(latest, peer.now), 0) + const state = normalizeState(await this.deps.readState(), now) + const entries = new Map(state.entries.map((entry) => [entry.marker.sessionId, entry])) + const failed = new Map(state.failed.map((failure) => [failure.marker.sessionId, failure])) + return frontier.map((peer) => peer.run(entries, failed)) + }) + settle.forEach((finish) => finish()) + } catch (error) { + // Lock acquisition may fail before a frontier is captured. + if (frontier.length === 0) { + frontier = this.ready.splice(0) + } + frontier.forEach((peer) => peer.reject(error)) + } finally { + if (this.ready.length) { + // Publication and queued dismissals get a turn before another fresh inventory read. + setImmediate(() => { + void this.drainReady() + }) + } else { + this.draining = false + } + } + } + /** Reconcile receipts synchronously under the lock; retire only the captured witness/version. */ settleResume( selections: readonly AgentSessionResumeSelection[], diff --git a/src/main/runtime/agent-session/records/agent-session-store-transactions.ts b/src/main/runtime/agent-session/records/agent-session-store-transactions.ts index 0ad9c9062f3..f5fe4bbf0e5 100644 --- a/src/main/runtime/agent-session/records/agent-session-store-transactions.ts +++ b/src/main/runtime/agent-session/records/agent-session-store-transactions.ts @@ -1,6 +1,7 @@ import { sessionCrewChanges } from '../../orchestration/crew/crew-changes' import type { JournalHostDatabase } from '../../../native-chat/agent-session-journal/database/journal-host-database' import type { JournalOperationReceipt } from '../../../native-chat/agent-session-journal/writing/journal-row-writer' +import type { JournalWriteAdmission } from '../../../native-chat/agent-session-journal/writing/journal-write-queue' import { journalOpenRefusalError } from '../../../native-chat/agent-session-journal/replay/journal-open-failure' import { AgentSessionJournalError } from '../../../native-chat/agent-session-journal/writing/journal-write-guards' import { @@ -229,11 +230,15 @@ export class AgentSessionStoreTransactions { } /** Commit a receipt separately when its command made no journal write. */ - commitReceipt(receipt: JournalOperationReceipt): Promise { - const run = this.queue.then(() => { + commitReceipt( + receipt: JournalOperationReceipt, + admission?: JournalWriteAdmission + ): Promise { + const write = () => { this.journalDatabase.transaction((db) => receipt.write(db)) receipt.committed() - }) + } + const run = this.queue.then(() => (admission ? admission(write) : write())) this.queue = run.catch(() => {}) return run }