perf(native-chat): coalesce ready restart admissions

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,

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<T>(
+  writer: JournalRowWriter,
+  planChanges: (changes: readonly JournalQueueCardChange[]) => JournalQueuePlan,
+  plan: () => { changes: readonly JournalQueueCardChange[]; result: T },
+  receipt?: JournalOperationReceipt,
+  admission?: JournalWriteAdmission
+): Promise<T> {
+  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
+    })
+}

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<QueuedMessageRow> {
     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<T>(
     plan: () => { changes: readonly JournalQueueCardChange[]; result: T },
-    receipt?: JournalOperationReceipt
+    receipt?: JournalOperationReceipt,
+    admission?: JournalWriteAdmission
   ): Promise<T> {
-    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
+    )
   }
 }

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: <T>(run: JournalWriteBody<T>) => Promise<T>
+  serialize: ConstructorParameters<typeof JournalRowWriter>[0]['serialize']
   database: () => JournalHostDatabase
   state: () => JournalReducerState
   readOnly: () => boolean

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<AgentJournalCursor> {
     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)
   }

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<T>(
 export type JournalRowWriterDeps = {
   sessionId: string
   now: () => number
-  serialize: <T>(run: JournalWriteBody<T>) => Promise<T>
+  serialize: <T>(
+    run: JournalWriteBody<T>,
+    admission?: JournalWriteAdmission,
+    published?: () => void
+  ) => Promise<T>
   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<JournalRow[]> {
-    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)
   }
 

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> = T extends PromiseLike<unknown> ? never : T
 export type JournalWriteBody<T> = () => JournalWriteResult<T>
+export type JournalWriteAdmission = <T>(write: JournalWriteBody<T>) => Promise<T>
 
 /** 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<T>(run: JournalWriteBody<T>): Promise<T> {
+  serialize<T>(
+    run: JournalWriteBody<T>,
+    admission?: JournalWriteAdmission,
+    published?: () => void
+  ): Promise<T> {
     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)
   }
 

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)

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

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) {

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<TValue> =
   | { ok: false; refusal: AgentSessionWireRefusal }
 
 export type StructuredAgentSessionSendAcceptance = (
-  apply: () => Promise<TurnOutcome<AgentSessionSendResult>>
+  apply: (admission: JournalWriteAdmission) => Promise<TurnOutcome<AgentSessionSendResult>>
 ) => Promise<TurnOutcome<AgentSessionSendResult>>
 
 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)

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<StructuredAgentSessionHostDeps, 'store' | 'recoveryCapsule'>
   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
   }
 }
 

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<AgentSessionStoreTransactions['commitReceipt']>) =>
+    this.transactions.commitReceipt(...args)
 
   readCommandReceipt = (...args: Parameters<AgentSessionStoreTransactions['readCommandReceipt']>) =>
     this.transactions.readCommandReceipt(...args)

src/main/runtime/agent-session/records/agent-session-recovery-capsule.test.ts

@@ -48,6 +48,38 @@ async function fileFailure(overrides: Partial<AgentSessionResumeFailureInput> =
 }
 
 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<unknown> | undefined
+    let dismissal: Promise<number> | 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)

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<T>(
     selection: AgentSessionResumeSelection,
     now: number,
-    accept: () => Promise<T>
+    accept: JournalWriteBody<T>
   ) {
     return this.selections.withSelectedResume(selection, now, accept)
   }

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<string, RecoveryEntry>,
+    failed: ReadonlyMap<string, AgentSessionResumeFailureRecord>
+  ) => () => 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: <T>(apply: () => Promise<T>) => Promise<T>
@@ -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<T>(
     selection: AgentSessionResumeSelection,
     now: number,
-    accept: () => Promise<T>
+    accept: JournalWriteBody<T>
   ): 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<void> {
+    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[],

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<void> {
-    const run = this.queue.then(() => {
+  commitReceipt(
+    receipt: JournalOperationReceipt,
+    admission?: JournalWriteAdmission
+  ): Promise<void> {
+    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
   }