native chat: queued-message actions and question answers accept through their own receipt (#26988)

* feat(native-chat): add a chat-owned command receipt store

Adds the agent_session_command_receipts table and an insert-if-absent
receipt write that runs inside a command's own journal transaction. A
receipt records the command's identity, canonical hash, status and a
pointer to the row holding its result; it dies with its chat record.
Nothing uses it yet: the existing operation ledger stays authoritative.

* fix(native-chat): key command receipts like today's ledger ids

Rows are keyed by caller and operation id; a send looks across callers and
other methods look up only their caller's row, matching the existing ledger.
Bad rows and newer table shapes read as unreadable instead of throwing.

* fix(native-chat): order receipt lookups by insertion and key by operation id

Global lookups now take rows in insertion order, so a read and an insert
agree on which receipt answers an id regardless of the host clock. The
primary key leads with the operation id, so its own index serves both
lookups.

* fix(native-chat): accept sends through their own receipt

Send (direct, queued, steer) and /compact no longer commit a separate
ledger reservation before the message is saved. Their command receipt is
inserted-if-absent in the same transaction as the submission or queued
draft; a repeated id replays from it, a changed request conflicts.
Operation ids are checked for syntax only, so age, clock skew, pruning
and the reservation's own storage can no longer refuse a send.

The ledger success row they already co-committed stays as a temporary
insert-if-absent write so families not yet switched keep seeing these
ids; it is never read on this path.

* fix(native-chat): answer a send's receipt one way, share /compact's id space with Send, wake delivery once

- After its commit, a send whose publication fails answers through the same receipt replay a retry takes, so one id gets one answer; the in-memory copy and journal reload are gone.
- /compact ids are global like Send's: a retry under a new connection replays instead of needing a second acknowledgment write.
- The receipt route calls its runner directly; the runner wakes delivery when a new message row commits.
- The compatibility ledger row expires from host time; each temporary site is labelled.
- Receipt read errors and post-commit failures are logged.

* test(native-chat): expect receipt answers where tests assumed the send ledger

Restart-continuation tests now model the case Send cannot vouch for (saved but
not loaded into memory) and keep it unconfirmed; a later throw after the message
is in the chat answers continued, as a resend would. Drop /compact's unreachable
legacy replay branch and fix stale comments.

* test: run the queued tests that open the journal database in the Node runtime project

* fix(native-chat): queued-message actions and prompt answers accept through their own receipt

Send Now, Delete, Resume and prompt answers move onto the command-receipt route
Send uses: the receipt is inserted with the command's own write. A no-op answers
only once its receipt commits (otherwise outcome unknown), and a prompt answer's
replay reads the exact revision it was accepted at.

* fix(native-chat): replay a queued action from what its receipt recorded

Send Now's acknowledgement points at the submission it handed off; a question
answer's receipt keeps the accepted resolution; a no-op that cannot save its
receipt answers outcome unknown and holds nothing in memory; classified refusals
pass through.

* docs(native-chat): receipt comments say what a replay answers from

* fix(native-chat): refuse a prompt answer on a newer Orca's journal with the update words

A prompt answer now runs through the same newer-Orca refusal the queued-message actions use, so a read-only journal answers agent_session_journal_unreadable instead of throwing journal_read_only.

config/scripts/vitest-sqlite-runtime-files.mjs

@@ -120,11 +120,13 @@ export const SQLITE_RUNTIME_INCLUDE = [
   'src/main/native-chat/agent-session-wire/structured-agent-session-owed-work-release.test.ts',
   'src/main/native-chat/agent-session-wire/structured-agent-session-pre-spawn-first-answer.test.ts',
   'src/main/native-chat/agent-session-wire/structured-agent-session-processless-reservation.test.ts',
+  'src/main/native-chat/agent-session-wire/structured-agent-session-prompt-answer-receipts.test.ts',
   'src/main/native-chat/agent-session-wire/structured-agent-session-prompt-cancel.test.ts',
   'src/main/native-chat/agent-session-wire/structured-agent-session-provider-child-record.test.ts',
   'src/main/native-chat/agent-session-wire/structured-agent-session-provider-restore.test.ts',
   'src/main/native-chat/agent-session-wire/structured-agent-session-provider-started-catalog.test.ts',
   'src/main/native-chat/agent-session-wire/structured-agent-session-provider-started.test.ts',
+  'src/main/native-chat/agent-session-wire/structured-agent-session-queued-action-receipts.test.ts',
   'src/main/native-chat/agent-session-wire/structured-agent-session-queued-gate.test.ts',
   'src/main/native-chat/agent-session-wire/structured-agent-session-queued-newer-orca.test.ts',
   'src/main/native-chat/agent-session-wire/structured-agent-session-queued-pause-lift.test.ts',

src/main/native-chat/agent-session-journal/command-receipt-schema.ts

@@ -18,7 +18,7 @@ export function commandReceiptScope(
   return operationIdScope === 'global' ? { kind: 'global' } : { kind: 'caller', callerKey }
 }
 
-/** The pointer locates the result; if its row is gone, the command stays spent: read what remains, else unknown, never rerun. */
+/** What a replay answers from: a pointer (if its row is gone the command stays spent: read what remains, else unknown, never rerun), or the outcome itself. */
 export const commandReceiptResultSchema = z.discriminatedUnion('kind', [
   z.strictObject({
     kind: z.literal('journal-row'),
@@ -29,6 +29,28 @@ export const commandReceiptResultSchema = z.discriminatedUnion('kind', [
     kind: z.literal('queued-draft'),
     messageId: z.string().min(1)
   }),
+  /** An accepted prompt answer, kept whole: no rewind or later revision changes its replay. Its
+   *  values are the resolution the prompt's row holds, bounded as that row is. */
+  z.strictObject({
+    kind: z.literal('prompt-answer'),
+    itemId: z.string().min(1),
+    revision: z.int().positive(),
+    resolution: z.strictObject({
+      state: z.literal('resolved'),
+      selectedOptionId: z.string().nullable(),
+      answers: z
+        .array(
+          z.strictObject({
+            questionId: z.string(),
+            optionIds: z.array(z.string()),
+            other: z.string().optional()
+          })
+        )
+        .optional(),
+      resolvedBy: z.string().nullable(),
+      resolvedAt: z.number().nullable()
+    })
+  }),
   z.strictObject({
     kind: z.literal('no-op'),
     outcome: z.discriminatedUnion('kind', [
@@ -37,7 +59,12 @@ export const commandReceiptResultSchema = z.discriminatedUnion('kind', [
         cancelled: z.literal(false),
         turnId: z.string().optional()
       }),
-      z.strictObject({ kind: z.literal('queue-resume'), resumed: z.literal(false) })
+      z.strictObject({ kind: z.literal('queue-resume'), resumed: z.literal(false) }),
+      z.strictObject({
+        kind: z.literal('queue-delete'),
+        messageId: z.string().min(1),
+        disposition: z.enum(['dispatched', 'withdrawn', 'missing'])
+      })
     ])
   })
 ])

src/main/native-chat/agent-session-journal/journal-item-appender.ts

@@ -7,6 +7,7 @@ import { journalItemRowBuilder } from './journal-row-builders'
 import type { JournalReducerState } from './journal-reducer'
 import type { JournalAppendResult, JournalItemAppendOptions } from './journal-store-contracts'
 import type { JournalRow } from './journal-row-schema'
+import type { JournalOperationReceipt } from './journal-row-writer'
 
 // A module-level Error's lazy stack retains the bundle's startup code and read buffer.
 class NothingResolvedError extends Error {}
@@ -17,18 +18,22 @@ export class JournalItemAppender {
   constructor(
     private readonly deps: {
       state: () => JournalReducerState
-      enqueue: (build: (seq: number, ts: number) => JournalRow) => Promise<JournalRow>
+      enqueue: (
+        build: (seq: number, ts: number) => JournalRow,
+        receipt?: JournalOperationReceipt
+      ) => Promise<JournalRow>
     }
   ) {}
 
   append(
     identity: AgentJournalItemIdentity,
     body: AgentJournalItemBody,
-    options: JournalItemAppendOptions
+    options: JournalItemAppendOptions,
+    receipt?: JournalOperationReceipt
   ): Promise<JournalAppendResult> {
     const itemId = agentJournalItemKey(identity)
     return this.deps
-      .enqueue(journalItemRowBuilder(this.deps.state, identity, body, options))
+      .enqueue(journalItemRowBuilder(this.deps.state, identity, body, options), receipt)
       .then((row) => appendResult(row, itemId))
   }
 

src/main/native-chat/agent-session-journal/journal-open-failure.ts

@@ -115,6 +115,20 @@ export function journalOpenRefusal(error: unknown): AgentSessionWireRefusal {
   return refuse('agent_session_journal_unreadable', { reason }, message)
 }
 
+/** A newer Orca's journal refuses a write with the words a send gets there. */
+export async function refusingNewerOrcaJournal<TOutcome>(
+  run: Promise<TOutcome>
+): Promise<TOutcome | { ok: false; refusal: AgentSessionWireRefusal }> {
+  try {
+    return await run
+  } catch (error) {
+    if (isJournalWrittenByNewerOrca(error)) {
+      return { ok: false, refusal: journalOpenRefusal(error) }
+    }
+    throw error
+  }
+}
+
 /** The same refusal, thrown: for a host that could not open the journal at all. It keeps what
  *  failed as its cause, for the log. */
 export function journalOpenRefusalError(error: unknown): AgentSessionRefusalError {

src/main/native-chat/agent-session-journal/journal-queued-messages.ts

@@ -11,7 +11,7 @@ import {
   AGENT_SESSION_OPERATION_FUTURE_SKEW_MS
 } from '../../../shared/agent-session-host-authority'
 import type { JournalRow } from './journal-row-schema'
-import type { JournalOperationReceipt } from './journal-row-writer'
+import { writeReceiptIfChanged, type JournalOperationReceipt } from './journal-row-writer'
 import type { JournalQueuedMessagesDeps, JournalSubmissionConsume } from './journal-store-contracts'
 import { holdQueuedMessages } from './queued-message-holds'
 import {
@@ -92,6 +92,7 @@ export class JournalQueuedMessages {
     return this.transact(
       (db) => updateQueuedMessageText(db, this.deps.sessionId, input),
       (result) => result.status === 'updated',
+      undefined,
       () => this.editLeases.retire(input.messageId)
     )
   }
@@ -133,11 +134,10 @@ export class JournalQueuedMessages {
           queuedAt: { epoch, sequence: lastSequence },
           now: this.deps.now()
         })
-        receipt?.write(db)
         return row
       },
       () => inserted,
-      receipt?.committed
+      receipt
     )
   }
 
@@ -199,11 +199,11 @@ export class JournalQueuedMessages {
   }
 
   /** Compare-and-transition waiting ∪ returned rows to op-stamped tombstones,
-   *  kept only so a replay of the settling operation answers "spent". */
-  withdraw(input: {
-    messageIds: readonly string[]
-    settledByOp: string
-  }): Promise<QueuedMessageRow[]> {
+   *  kept only so a replay of the settling operation answers "spent". `receipt`: with a withdrawal. */
+  withdraw(
+    input: { messageIds: readonly string[]; settledByOp: string },
+    receipt?: JournalOperationReceipt
+  ): Promise<QueuedMessageRow[]> {
     if (input.messageIds.length === 0) {
       // Delete races and empty carries land here; neither may cost a write transaction.
       return Promise.resolve([])
@@ -215,7 +215,8 @@ export class JournalQueuedMessages {
           sessionId: this.deps.sessionId,
           now: this.deps.now()
         }),
-      (withdrawn) => withdrawn.length > 0
+      (withdrawn) => withdrawn.length > 0,
+      receipt
     )
   }
 
@@ -234,17 +235,19 @@ export class JournalQueuedMessages {
     }).length
   }
 
-  /** One standalone draft-table transaction on the journal's queue; one that
-   *  changed rows bumps the revision and notifies after COMMIT, `adopted` first. */
+  /** One standalone draft-table transaction on the journal's queue; one that changed rows commits
+   *  `receipt` with them, bumps the revision and notifies after COMMIT, receipt and `adopted` first. */
   private transact<T>(
     run: (db: Database.Database) => JournalWriteResult<T>,
     changed: (result: T) => boolean,
+    receipt?: JournalOperationReceipt,
     adopted?: () => void
   ): Promise<T> {
     return this.deps.serialize(() => {
       assertJournalWritable(this.deps.readOnly(), this.deps.sessionId)
-      const result = this.deps.database().transaction(run)
+      const result = this.deps.database().transaction(writeReceiptIfChanged(run, changed, receipt))
       if (changed(result)) {
+        receipt?.committed()
         adopted?.()
         this.changeRevision++
         this.deps.committed()

src/main/native-chat/agent-session-journal/journal-row-writer.ts

@@ -28,6 +28,21 @@ export function composeJournalOperationReceipts(
   }
 }
 
+/** `run`, with `receipt` written in its transaction only when `changed` says it wrote something. */
+export function writeReceiptIfChanged<T>(
+  run: (db: Database.Database) => T,
+  changed: (result: T) => boolean,
+  receipt?: JournalOperationReceipt
+): (db: Database.Database) => T {
+  return (db) => {
+    const result = run(db)
+    if (receipt && changed(result)) {
+      receipt.write(db)
+    }
+    return result
+  }
+}
+
 export type JournalRowWriterDeps = {
   sessionId: string
   now: () => number

src/main/native-chat/agent-session-journal/journal-store-collaborators.ts

@@ -16,7 +16,7 @@ import { JournalQueuedMessages } from './journal-queued-messages'
 import { JournalStopMarks } from './journal-stop-marks'
 import { journalQueuePauseRestatement } from './queued-message-pause'
 import type { JournalReducerState } from './journal-reducer'
-import { JournalRowWriter } from './journal-row-writer'
+import { JournalRowWriter, type JournalOperationReceipt } from './journal-row-writer'
 import { JournalStepWriter } from './journal-step-writer'
 import { restoreJournalStore } from './journal-store-restore'
 import { JournalSubmissionWriter } from './journal-submission-writer'
@@ -42,7 +42,10 @@ export type JournalStoreHost = {
   adopt: (loaded: JournalLoad) => void
   commit: (rows: readonly JournalRow[], savedAt: number) => void
   journal: () => AgentSessionJournal
-  enqueue: (build: (seq: number, ts: number) => JournalRow) => Promise<JournalRow>
+  enqueue: (
+    build: (seq: number, ts: number) => JournalRow,
+    receipt?: JournalOperationReceipt
+  ) => Promise<JournalRow>
 }
 
 export type JournalStoreCollaborators = {

src/main/native-chat/agent-session-journal/journal-store.ts

@@ -131,7 +131,7 @@ export class AgentSessionJournal {
       },
       notifyCommitted: () => this.onCommitted?.(),
       journal: () => this,
-      enqueue: (build) => this.rowWriter.enqueue(build)
+      enqueue: (build, receipt) => this.rowWriter.enqueue(build, undefined, receipt)
     })
     this.rowWriter = collaborators.rowWriter
     this.epochController = collaborators.epochController
@@ -285,9 +285,10 @@ export class AgentSessionJournal {
   appendItem(
     identity: AgentJournalItemIdentity,
     body: AgentJournalItemBody,
-    options: JournalItemAppendOptions
+    options: JournalItemAppendOptions,
+    receipt?: JournalOperationReceipt
   ): Promise<JournalAppendResult> {
-    return this.itemAppender.append(identity, body, options)
+    return this.itemAppender.append(identity, body, options, receipt)
   }
 
   /** An upsert whose row is chosen from the fold at its own turn in the queue; null writes nothing. */
@@ -296,12 +297,10 @@ export class AgentSessionJournal {
 
   appendTombstone(
     identity: AgentJournalItemIdentity,
-    options: JournalTombstoneInput
+    { fence }: JournalTombstoneInput
   ): Promise<AgentJournalCursor> {
     const itemId = agentJournalItemKey(identity)
-    return this.rowWriter.append(
-      journalTombstoneRowBuilder(() => this.state, itemId, options.fence)
-    )
+    return this.rowWriter.append(journalTombstoneRowBuilder(() => this.state, itemId, fence))
   }
 
   /** A Stop that took effect, timed by its row (`JournalStopEvent`). */
@@ -310,8 +309,9 @@ export class AgentSessionJournal {
   }
 
   /** A person's Resume of the queue. */
-  appendQueueResume(fence: number): Promise<AgentJournalCursor> {
-    return this.rowWriter.append(journalQueueResumeRowBuilder(() => this.state, fence))
+  appendQueueResume(fence: number, receipt?: JournalOperationReceipt): Promise<AgentJournalCursor> {
+    const build = journalQueueResumeRowBuilder(() => this.state, fence)
+    return this.rowWriter.append(build, undefined, receipt)
   }
 
   /** This open found waiting cards an earlier handle wrote (`queued-message-pause.ts`). */

src/main/native-chat/agent-session-wire/structured-agent-session-command-admission.ts

@@ -1,5 +1,8 @@
 import { parseAgentSessionOperationTimestamp } from '../../../shared/agent-session-host-authority'
-import { agentSessionLedgerRefusal } from '../../../shared/agent-session-mutation-envelope'
+import {
+  agentSessionLedgerRefusal,
+  refuseUnlessWriterAdmitted
+} from '../../../shared/agent-session-mutation-envelope'
 import {
   agentSessionRefusalFromReference,
   isAgentSessionRefusalError
@@ -15,6 +18,7 @@ import {
   refuseAgentSessionMutation
 } from './structured-agent-session-mutation-refusals'
 import type { AgentSessionMutationRequest } from './structured-agent-session-mutation-admission'
+import type { MutationCommandReceipt } from './structured-agent-session-mutation-plans'
 import { mutationTurnContext } from './structured-agent-session-mutation-turn-context'
 import { runCommandReceiptMutation } from './structured-agent-session-command-receipt'
 import {
@@ -22,10 +26,12 @@ import {
   resolveAgentSessionReplayOutcome
 } from './structured-agent-session-replay-outcome'
 
-/** Submission-accepting plans use their effect's receipt, never a ledger reservation. */
+/** A plan with a command receipt accepts through it, committed with its effect, never a ledger
+ *  reservation. */
 export async function admitCommandReceiptMutation<TValue>(
   request: AgentSessionMutationRequest<TValue>,
-  hostFingerprint: string
+  hostFingerprint: string,
+  commandReceipt: MutationCommandReceipt<TValue>
 ): Promise<AgentSessionMutationResult<TValue>> {
   const { store, envelope, plan, callerKey } = request
   if (parseAgentSessionOperationTimestamp(envelope.clientOperationId) === null) {
@@ -68,7 +74,11 @@ export async function admitCommandReceiptMutation<TValue>(
   if (!journal || !current) {
     return refuseAgentSessionMutation(AGENT_SESSION_NOT_ATTACHED)
   }
-  // These plans write only to the conversation, whose journal enforces the execution-host fence.
+  // A conversation write is fenced by its journal; anything the provider must carry needs the lease.
+  const leaseRefusal = plan.conversationWrite ? null : refuseUnlessWriterAdmitted(current.lease)
+  if (leaseRefusal) {
+    return refuseAgentSessionMutation(leaseRefusal)
+  }
   const context = mutationTurnContext(request, journal, current)
   try {
     const outcome = await runCommandReceiptMutation({
@@ -78,6 +88,7 @@ export async function admitCommandReceiptMutation<TValue>(
       wakeDelivery: request.wakeDelivery,
       envelope,
       plan,
+      commandReceipt,
       context
     })
     if ('committedReceipt' in outcome) {
@@ -151,7 +162,7 @@ async function answerCommandReceipt<TValue>(
   const replay = resolveAgentSessionReplayOutcome({
     operationId: envelope.clientOperationId,
     outcome,
-    reconstruct: () => plan.replay(context, outcome),
+    reconstruct: () => plan.replay(context, outcome, receipt.result),
     rerunWhenReplayMissing: plan.rerunWhenReplayMissing?.(context),
     recoverUnknownFromDurableState: plan.recoverUnknownFromDurableState
   })

src/main/native-chat/agent-session-wire/structured-agent-session-command-receipt.ts

@@ -1,60 +1,52 @@
 import type { AgentSessionMutationEnvelope } from '../../../shared/agent-session-wire'
+import { isAgentSessionRefusalError } from '../../../shared/agent-session-wire-refusals'
 import type { AgentSessionRecordStore } from '../../runtime/agent-session-record-store'
 import {
   commandReceiptScope,
-  type CommandReceipt
+  type CommandReceipt,
+  type CommandReceiptResult
 } from '../agent-session-journal/command-receipt-schema'
-import { buildCommandReceiptTransaction } from '../agent-session-journal/command-receipt-transaction'
-import { composeJournalOperationReceipts } from '../agent-session-journal/journal-row-writer'
-import type { MutationPlan } from './structured-agent-session-mutation-plans'
+import {
+  buildCommandReceiptTransaction,
+  CommandReceiptExistsError
+} from '../agent-session-journal/command-receipt-transaction'
+import {
+  composeJournalOperationReceipts,
+  type JournalOperationReceipt
+} from '../agent-session-journal/journal-row-writer'
+import type { JournalRow } from '../agent-session-journal/journal-row-schema'
+import type {
+  MutationCommandReceipt,
+  MutationPlan
+} from './structured-agent-session-mutation-plans'
+import { agentSessionOperationOutcomeUnknown } from './structured-agent-session-replay-outcome'
 import type { AgentSessionTurnContext, TurnOutcome } from './structured-agent-session-turns'
 
-/** Acceptance identity and compatibility bookkeeping share the submission or draft's commit. */
-export async function runCommandReceiptMutation<TValue>(input: {
+type CommandReceiptIdentity = {
   store: AgentSessionRecordStore
   operationCallerKey: string
   fingerprint: string
   envelope: AgentSessionMutationEnvelope
-  plan: MutationPlan<TValue>
+  plan: { method: string; operationIdScope?: 'global' }
   context: AgentSessionTurnContext
-  wakeDelivery?: (sessionId: string) => void
-}): Promise<TurnOutcome<TValue> | { committedReceipt: CommandReceipt }> {
-  const { store, operationCallerKey, fingerprint, envelope, plan, context, wakeDelivery } = input
+}
+
+/** Acceptance identity and compatibility bookkeeping share the commit of the command's effect. */
+export async function runCommandReceiptMutation<TValue>(
+  input: CommandReceiptIdentity & {
+    plan: MutationPlan<TValue>
+    commandReceipt: MutationCommandReceipt<TValue>
+    wakeDelivery?: (sessionId: string) => void
+  }
+): Promise<TurnOutcome<TValue> | { committedReceipt: CommandReceipt }> {
+  const { envelope, plan, context, commandReceipt, wakeDelivery } = input
   let acceptedReceipt: CommandReceipt | undefined
   let submissionWritten = false
-  const receipt = composeJournalOperationReceipts(
-    buildCommandReceiptTransaction(
-      commandReceiptScope(operationCallerKey, plan.operationIdScope),
-      (row) => {
-        if (row && row.kind !== 'submission') {
-          throw new Error('Send acceptance requires a submission')
-        }
-        submissionWritten = row !== undefined
-        acceptedReceipt = {
-          operationId: envelope.clientOperationId,
-          sessionId: envelope.sessionId,
-          callerKey: operationCallerKey,
-          method: plan.method,
-          fingerprint,
-          status: 'accepted',
-          acceptedAt: context.now(),
-          result: row
-            ? { kind: 'journal-row', epoch: row.epoch, sequence: row.seq }
-            : { kind: 'queued-draft', messageId: envelope.clientOperationId }
-        }
-        return acceptedReceipt
-      }
-    ),
-    // TEMPORARY: ledger co-write preserves cross-family identity until every mutation uses receipts.
-    store.operationOutcomeReceipt({
-      callerKey: operationCallerKey,
-      operationId: envelope.clientOperationId,
-      fingerprint,
-      now: context.now(),
-      ...(plan.operationIdScope ? { operationIdScope: plan.operationIdScope } : {}),
-      outcome: { status: 'succeeded', sessionId: envelope.sessionId }
-    })
-  )
+  const receipt = acceptanceReceipt(input, (row) => {
+    submissionWritten = row?.kind === 'submission'
+    acceptedReceipt = acceptedCommandReceipt(input, commandReceipt.result(row))
+    return acceptedReceipt
+  })
   let committed = false
   try {
     const ran = await plan.run({
@@ -67,7 +59,14 @@ export async function runCommandReceiptMutation<TValue>(input: {
         }
       }
     })
-    if (!committed || ran.ok) {
+    if (!committed) {
+      const unwritten = ran.ok ? commandReceipt.unwritten?.(ran.value, context) : null
+      return unwritten &&
+        !(await commitNoOpReceipt(input, acceptedCommandReceipt(input, unwritten)))
+        ? unknownOutcome(input)
+        : ran
+    }
+    if (ran.ok) {
       return ran
     }
     context.logger.warn('publishing an accepted command failed; replaying its receipt', {
@@ -104,3 +103,75 @@ export async function runCommandReceiptMutation<TValue>(input: {
   }
   return { committedReceipt: acceptedReceipt }
 }
+
+function acceptedCommandReceipt(
+  { envelope, operationCallerKey, plan, fingerprint, context }: CommandReceiptIdentity,
+  result: CommandReceiptResult
+): CommandReceipt {
+  return {
+    operationId: envelope.clientOperationId,
+    sessionId: envelope.sessionId,
+    callerKey: operationCallerKey,
+    method: plan.method,
+    fingerprint,
+    status: 'accepted',
+    acceptedAt: context.now(),
+    result
+  }
+}
+
+function acceptanceReceipt(
+  input: CommandReceiptIdentity,
+  receipt: (row?: JournalRow) => CommandReceipt
+): JournalOperationReceipt {
+  const { store, operationCallerKey, envelope, plan, fingerprint, context } = input
+  return composeJournalOperationReceipts(
+    buildCommandReceiptTransaction(
+      commandReceiptScope(operationCallerKey, plan.operationIdScope),
+      receipt
+    ),
+    // TEMPORARY: ledger co-write preserves cross-family identity until every mutation uses receipts.
+    store.operationOutcomeReceipt({
+      callerKey: operationCallerKey,
+      operationId: envelope.clientOperationId,
+      fingerprint,
+      now: context.now(),
+      ...(plan.operationIdScope ? { operationIdScope: plan.operationIdScope } : {}),
+      outcome: { status: 'succeeded', sessionId: envelope.sessionId }
+    })
+  )
+}
+
+function unknownOutcome(input: CommandReceiptIdentity): TurnOutcome<never> {
+  return {
+    ok: false,
+    refusal: agentSessionOperationOutcomeUnknown(input.envelope.clientOperationId)
+  }
+}
+
+/** A no-op is answered only once its receipt commits, alone. An identity verdict or a classified
+ *  refusal throws for admission to answer; an unclassified storage failure is logged and answered
+ *  unknown, with nothing recorded, so a retry decides afresh. */
+async function commitNoOpReceipt(
+  input: CommandReceiptIdentity,
+  receipt: CommandReceipt
+): Promise<boolean> {
+  try {
+    await input.store.commitOperationReceipt(acceptanceReceipt(input, () => receipt))
+    return true
+  } catch (error) {
+    if (error instanceof CommandReceiptExistsError || isAgentSessionRefusalError(error)) {
+      throw error
+    }
+    input.context.logger.warn(
+      'recording a command that changed nothing failed; answering unknown',
+      {
+        scope: 'command-receipt-unwritten',
+        sessionId: input.envelope.sessionId,
+        operationId: input.envelope.clientOperationId,
+        error
+      }
+    )
+    return false
+  }
+}

src/main/native-chat/agent-session-wire/structured-agent-session-host.test.ts

@@ -460,7 +460,7 @@ describe('respondToPrompt', () => {
     expect(answerPrompt).not.toHaveBeenCalled()
   })
 
-  it("does not turn a recorded refusal into another client's successful answer", async () => {
+  it("does not turn a refused answer into another client's successful answer", async () => {
     await attach()
     const prompt = await seedApproval()
     const rejectedFields = {
@@ -485,10 +485,12 @@ describe('respondToPrompt', () => {
       }
     )
 
+    // A refusal records nothing, so the retry is judged afresh: the prompt has moved on.
     expect(await host.respondToPrompt(CALLER, rejected)).toMatchObject({
       ok: false,
-      refusal: { code: 'agent_session_operation_invalid' }
+      refusal: { code: 'agent_session_item_revision_stale' }
     })
+    expect(answerPrompt).toHaveBeenCalledOnce()
   })
 
   it('keeps the answer and reports it undelivered when the provider callback throws', async () => {

src/main/native-chat/agent-session-wire/structured-agent-session-mutation-admission.ts

@@ -1,5 +1,5 @@
 // The one route every mutating agent-session call takes: recompute the fingerprint,
-// dispatch submission-accepting plans to their command receipt, otherwise admit
+// accept a plan that declares a command receipt through it, otherwise admit
 // through the ledger and check the lease. Methods cannot grow their own admission rules.
 //
 // Ledger-backed admission is two-phase when a call brings `prepareSession`. The ledger's
@@ -98,8 +98,8 @@ export async function admitAndRunAgentSessionMutation<TValue>(
   if (conflict) {
     return refuseAgentSessionMutation(conflict)
   }
-  if (plan.acceptsWithCommandReceipt) {
-    return admitCommandReceiptMutation(request, hostFingerprint)
+  if (plan.commandReceipt) {
+    return admitCommandReceiptMutation(request, hostFingerprint, plan.commandReceipt)
   }
   if (request.prepareSession) {
     const ledger = request.store.evaluateMutationOperation({

src/main/native-chat/agent-session-wire/structured-agent-session-mutation-plans.ts

@@ -1,8 +1,9 @@
 // One plan per mutating method: what it fingerprints, what it does, and how its
 // answer is rebuilt on a replay.
 //
-// Send and /compact prove acceptance by their command receipt, the rest by their ledger row; the
-// journal projects the current answer.
+// A plan with `commandReceipt` proves acceptance by that receipt, the rest by their ledger row. A
+// prompt answer's replay answers from the resolution its receipt keeps; the journal projects the
+// others' current answer.
 
 import type { AgentJournalMessageItem } from '../../../shared/agent-session-journal-types'
 import {
@@ -38,9 +39,17 @@ import {
   type AgentSessionTurnContext,
   type TurnOutcome
 } from './structured-agent-session-turns'
-import type { AgentSessionPromptRequest } from './structured-agent-session-turns-prompt'
+import {
+  acceptedPromptAnswer,
+  promptAnswerReceipt,
+  promptRowAnswer,
+  type AgentSessionPromptRequest
+} from './structured-agent-session-turns-prompt'
 import { queuedSendAnswer } from './structured-agent-session-queued-send-answer'
 import type { JournalOperationReceipt } from '../agent-session-journal/journal-row-writer'
+import type { CommandReceiptResult } from '../agent-session-journal/command-receipt-schema'
+import type { JournalRow } from '../agent-session-journal/journal-row-schema'
+import { refusingNewerOrcaJournal } from '../agent-session-journal/journal-open-failure'
 
 export type MutationPlan<TValue> = {
   method: string
@@ -51,13 +60,17 @@ export type MutationPlan<TValue> = {
   /** Still runs, decided from the committed ledger, when its ledger row cannot be written. */
   runsWithoutLedgerRow?: true
   run: (ctx: AgentSessionTurnContext) => Promise<TurnOutcome<TValue>>
-  replay: (ctx: AgentSessionTurnContext, outcome: AgentSessionOperationOutcome) => TValue | null
+  /** `receipt`: the accepted command receipt's result, for a plan with `commandReceipt`. */
+  replay: (
+    ctx: AgentSessionTurnContext,
+    outcome: AgentSessionOperationOutcome,
+    receipt?: CommandReceiptResult
+  ) => TValue | null
   rerunWhenReplayMissing?: (ctx: AgentSessionTurnContext) => boolean
   recoverUnknownFromDurableState?: boolean
 } & (
   | {
-      /** Accepts with a submission or draft; its command receipt commits in that write's transaction. */
-      acceptsWithCommandReceipt: true
+      commandReceipt: MutationCommandReceipt<TValue>
       settlesWithWrite?: never
       successReceipt?: never
       settledOutcome?: never
@@ -65,17 +78,47 @@ export type MutationPlan<TValue> = {
   | {
       /** Commits success with its row; paths without a committed receipt use fallback settlement. */
       settlesWithWrite: true
-      acceptsWithCommandReceipt?: never
+      commandReceipt?: never
       successReceipt?: () => JournalOperationReceipt
       settledOutcome?: (value: TValue) => AgentSessionOperationOutcome
     }
   | {
       settlesWithWrite?: never
-      acceptsWithCommandReceipt?: never
+      commandReceipt?: never
       settledOutcome?: (value: TValue) => AgentSessionOperationOutcome
     }
 )
 
+/** Accepted through its own command receipt, inserted if absent in the transaction of its effect. */
+export type MutationCommandReceipt<TValue> = {
+  /** Inside the effect's transaction: the journal row it wrote, or none for a draft-table write. */
+  result: (row: JournalRow | undefined) => CommandReceiptResult
+  /** A run that wrote nothing: the receipt committed alone, before the answer goes out (a no-op's
+   *  answer, or the earlier write it acknowledged), or null to record nothing. */
+  unwritten?: (value: TValue, ctx: AgentSessionTurnContext) => CommandReceiptResult | null
+}
+
+/** The journal row an accepted command wrote, which must be of the kind it accepts with. */
+export function journalRowReceiptResult(
+  row: JournalRow | undefined,
+  kind: JournalRow['kind']
+): CommandReceiptResult {
+  if (row?.kind !== kind) {
+    throw new Error(`an accepted command requires its ${kind} row`)
+  }
+  return { kind: 'journal-row', epoch: row.epoch, sequence: row.seq }
+}
+
+/** A send accepts with its submission, or a draft held while the agent works, keyed by its id. */
+function submissionOrDraftReceipt<TValue>(clientMessageId: string): MutationCommandReceipt<TValue> {
+  return {
+    result: (row) =>
+      row
+        ? journalRowReceiptResult(row, 'submission')
+        : { kind: 'queued-draft', messageId: clientMessageId }
+  }
+}
+
 /** Who a send is from, read off what it carries, the one place it is decided: the person's own
  *  send, another agent's message (its body names the sender), or neither, such as a dispatch
  *  preamble or a restart continuation. */
@@ -107,7 +150,7 @@ export function sendPlan(params: {
     method: 'agentSession.send',
     operationIdScope: 'global',
     conversationWrite: true,
-    acceptsWithCommandReceipt: true,
+    commandReceipt: submissionOrDraftReceipt(clientMessageId),
     // `delivery` joins the OPERATION fingerprint only; the submission row keeps
     // the body-only fingerprint the reducer's echo-aliasing recomputes.
     fields: {
@@ -192,7 +235,7 @@ export function conversationCommandPlan(params: {
     method: 'agentSession.conversationCommand',
     operationIdScope: 'global',
     conversationWrite: true,
-    acceptsWithCommandReceipt: true,
+    commandReceipt: submissionOrDraftReceipt(clientMessageId),
     fields: {
       command: STRUCTURED_AGENT_SESSION_COMPACT_COMMAND,
       ...(params.delivery ? { delivery: params.delivery } : {})
@@ -281,17 +324,13 @@ export function promptPlan(
       optionId: params.optionId,
       answers: params.answers
     },
-    run: (ctx) => performPrompt(ctx, params),
-    replay: (ctx) => {
-      const item = ctx.journal.snapshot().items.find((entry) => entry.itemId === params.itemId)
-      const body = item?.body
-      if (!item || !body || (body.kind !== 'approval' && body.kind !== 'question')) {
-        return null
-      }
-      return body.resolution.state === 'pending'
-        ? null
-        : { itemId: item.itemId, revision: item.revision, resolution: body.resolution }
-    }
+    // The resolved revision's answer, or the answer the prompt already held, kept in the receipt.
+    commandReceipt: {
+      result: (row) => promptAnswerReceipt(promptRowAnswer(row)),
+      unwritten: (value) => promptAnswerReceipt(value)
+    },
+    run: (ctx) => refusingNewerOrcaJournal(performPrompt(ctx, params)),
+    replay: (_ctx, _outcome, receipt) => acceptedPromptAnswer(receipt)
   }
 }
 

src/main/native-chat/agent-session-wire/structured-agent-session-operation-settlement.test.ts

@@ -16,12 +16,27 @@ import {
   HOST_TEST_THREAD as THREAD
 } from './structured-agent-session-host-test-data'
 import type { AgentSessionTurnContext } from './structured-agent-session-turns'
-import { sendPlan } from './structured-agent-session-mutation-plans'
+import {
+  sendPlan,
+  type MutationCommandReceipt,
+  type MutationPlan
+} from './structured-agent-session-mutation-plans'
 import { runCommandReceiptMutation } from './structured-agent-session-command-receipt'
 import { createStructuredAgentSessionLogger } from './structured-agent-session-logger'
 import { codexProviderHandle } from '../../../shared/agent-session-provider-handle-encoding'
 import { NO_STRUCTURED_AGENTS } from './structured-agent-session-adapter-router-test-support'
 
+const DRAFT_RECEIPT: MutationCommandReceipt<unknown> = {
+  result: () => ({ kind: 'queued-draft', messageId: 'draft' })
+}
+
+function withReceipt<TValue>(plan: MutationPlan<TValue>) {
+  if (!plan.commandReceipt) {
+    throw new Error('expected a plan that accepts through its command receipt')
+  }
+  return { plan, commandReceipt: plan.commandReceipt }
+}
+
 async function context(): Promise<AgentSessionTurnContext> {
   await attach()
   return {
@@ -68,12 +83,13 @@ it('rethrows a throw from a plan answered by its write, writing nothing and leav
     plan: {
       method: 'agentSession.send',
       fields: {},
-      settlesWithWrite: true,
+      commandReceipt: DRAFT_RECEIPT,
       run: async () => {
         throw refusal
       },
       replay: () => null
-    }
+    },
+    commandReceipt: DRAFT_RECEIPT
   }).catch((error: unknown) => error)
   expect(result).toBe(refusal)
   expect(writes).not.toHaveBeenCalled()
@@ -82,11 +98,11 @@ it('rethrows a throw from a plan answered by its write, writing nothing and leav
 })
 
 it.each([
-  { plan: 'answered by its write', settlesWithWrite: true as const, failures: 0 },
-  { plan: 'settled after its run', settlesWithWrite: undefined, failures: 2 }
+  { plan: 'answered by its write', acceptsWithReceipt: true, failures: 0 },
+  { plan: 'settled after its run', acceptsWithReceipt: false, failures: 2 }
 ])(
   'preserves a proven refusal of a plan $plan through $failures failed bookkeeping writes',
-  async ({ settlesWithWrite, failures }) => {
+  async ({ acceptsWithReceipt, failures }) => {
     const ctx = await context()
     const { store, dispatch } = hostTestState()
     const warning = vi.spyOn(console, 'warn').mockImplementation(() => {})
@@ -105,11 +121,12 @@ it.each([
       context: ctx,
       plan: { method: 'agentSession.send', fields: {}, run, replay: () => null }
     }
-    const result = settlesWithWrite
+    const result = acceptsWithReceipt
       ? await runCommandReceiptMutation({
           ...input,
           fingerprint: 'test-fingerprint',
-          plan: { ...input.plan, settlesWithWrite }
+          plan: { ...input.plan, commandReceipt: DRAFT_RECEIPT },
+          commandReceipt: DRAFT_RECEIPT
         })
       : await runSettledAgentSessionMutation(input)
     expect(result).toEqual(refusal)
@@ -135,7 +152,7 @@ it('accepts without touching the provider', async () => {
     fingerprint: 'test-fingerprint',
     envelope: operation,
     context: ctx,
-    plan: sendPlan({ envelope: operation, body, beforeRun })
+    ...withReceipt(sendPlan({ envelope: operation, body, beforeRun }))
   })
   expect(result).toMatchObject({ ok: true })
   expect(beforeRun).toHaveBeenCalledOnce()
@@ -162,7 +179,7 @@ it('refuses a superseded send at acceptance, recording and dispatching nothing',
     fingerprint: 'test-fingerprint',
     envelope: operation,
     context: ctx,
-    plan: sendPlan({ envelope: operation, body, beforeRun })
+    ...withReceipt(sendPlan({ envelope: operation, body, beforeRun }))
   }).catch((error: unknown) => error)
   expect(result).toBeInstanceOf(AgentSessionPreDispatchError)
   expect(writes).not.toHaveBeenCalled()
@@ -189,10 +206,10 @@ it('returns the pre-acceptance refusal without recording a receipt', async () =>
       fingerprint: 'test-fingerprint',
       envelope: operation,
       context: ctx,
-      plan: {
+      ...withReceipt({
         ...sendPlan({ envelope: operation, body }),
         run: async () => ({ ok: false, refusal })
-      }
+      })
     })
   ).toMatchObject({ ok: false, refusal })
   expect(writes).not.toHaveBeenCalled()

src/main/native-chat/agent-session-wire/structured-agent-session-operation-settlement.ts

@@ -16,7 +16,7 @@ export async function runSettledAgentSessionMutation<TValue>(input: {
   store: AgentSessionRecordStore
   operationCallerKey: string
   envelope: AgentSessionMutationEnvelope
-  plan: Exclude<MutationPlan<TValue>, { acceptsWithCommandReceipt: true }>
+  plan: Exclude<MutationPlan<TValue>, { commandReceipt: object }>
   context: AgentSessionTurnContext
 }): Promise<TurnOutcome<TValue>> {
   const operation = {

src/main/native-chat/agent-session-wire/structured-agent-session-prompt-answer-receipts.test.ts

@@ -0,0 +1,397 @@
+// A prompt answer accepts through its own command receipt, committed with the resolved revision
+// before the provider hears the answer: a retry answers from that receipt and the provider hears
+// each answer once.
+
+import { beforeEach, describe, expect, it, type Mock } from 'vitest'
+import { AGENT_SESSION_MAX_NEW_OPERATION_AGE_MS } from '../../../shared/agent-session-host-authority'
+import {
+  AGENT_JOURNAL_THREAD_SCOPE,
+  type AgentJournalItemBody
+} from '../../../shared/agent-session-journal-types'
+import { agentJournalItemKey } from '../../../shared/agent-session-journal-item-key'
+import type { AgentSessionRecordStore } from '../../runtime/agent-session-record-store'
+import {
+  openTestJournalHostDatabase,
+  SAVED_BY_NEWER_ORCA
+} from '../agent-session-journal/journal-host-database-test-support'
+import type { StructuredAgentSessionAdapter } from './structured-agent-session-adapter'
+import type { StructuredAgentSessionHost } from './structured-agent-session-host'
+import {
+  attach,
+  CALLER,
+  envelope,
+  hostTestState,
+  seedApproval
+} from './structured-agent-session-host-test-harness'
+import {
+  HOST_TEST_NOW as NOW,
+  HOST_TEST_SESSION as SESSION,
+  HOST_TEST_THREAD as THREAD
+} from './structured-agent-session-host-test-data'
+
+const OLD = NOW - AGENT_SESSION_MAX_NEW_OPERATION_AGE_MS - 60_000
+const FUTURE = NOW + 6 * 60_000
+const CONFLICT = {
+  ok: false,
+  refusal: { code: 'agent_session_operation_conflict', details: { reason: 'operationIdReused' } }
+}
+const UNKNOWN = { ok: false, refusal: { code: 'agent_session_operation_unknown' } }
+/** The provider identity `seedApproval` raises its prompt under. */
+const PROMPT = { provider: 'codex' as const, threadId: THREAD, turnId: 'turn-1', ordinal: 99 }
+
+let root: string
+let store: AgentSessionRecordStore
+let host: StructuredAgentSessionHost
+let answerPrompt: Mock<StructuredAgentSessionAdapter['answerPrompt']>
+let acquire: Mock<StructuredAgentSessionAdapter['acquire']>
+let serial = 0
+
+beforeEach(() => {
+  ;({ root, store, host, answerPrompt, acquire } = hostTestState())
+})
+
+function opId(at = NOW): string {
+  serial += 1
+  return `${at}-${serial.toString(16).padStart(32, 'd')}`
+}
+
+async function seeded() {
+  await attach()
+  return seedApproval()
+}
+
+function answer(
+  prompt: { itemId: string; revision: number },
+  clientOperationId: string,
+  optionId = 'allow'
+) {
+  const fields = { itemId: prompt.itemId, expectedRevision: prompt.revision, optionId }
+  return {
+    envelope: envelope('agentSession.respondTo:approval', fields, { clientOperationId }),
+    kind: 'approval' as const,
+    ...fields
+  }
+}
+
+function receipt(id: string, callerKey = CALLER.callerKey) {
+  return store.readCommandReceipt({ kind: 'caller', callerKey }, id)
+}
+
+function journal() {
+  const open = host.collaboratorsForTests().sessions.get(SESSION)?.journal
+  if (!open) {
+    throw new Error('expected an open conversation')
+  }
+  return open
+}
+
+function promptBody(itemId: string): AgentJournalItemBody {
+  const body = journal().itemBody(itemId)
+  if (!body) {
+    throw new Error(`no item ${itemId}`)
+  }
+  return body
+}
+
+/** The provider writes the answered prompt again, as a later echo of it would. */
+async function reviseFromProvider(itemId: string): Promise<void> {
+  const body = promptBody(itemId)
+  if (body.kind !== 'approval') {
+    throw new Error('expected an approval')
+  }
+  const events = acquire.mock.calls.at(-1)?.[0].events
+  events?.appendItem(
+    PROMPT,
+    {
+      ...body,
+      resolution: { ...body.resolution, resolvedBy: 'provider-echo', resolvedAt: NOW + 1 }
+    },
+    { turnScope: AGENT_JOURNAL_THREAD_SCOPE }
+  )
+  await host.flushStreamedEvents(SESSION)
+}
+
+/** What a rewind does to the conversation: a new epoch, the prompt rebuilt into it as it stands. */
+async function rewindKeepingPrompt(itemId: string): Promise<void> {
+  const fence = store.getRecord(SESSION)!.lease.runtimeFence
+  await journal().replaceEpochItems('handle_forked', fence, [
+    { identity: PROMPT, body: promptBody(itemId), turnScope: AGENT_JOURNAL_THREAD_SCOPE }
+  ])
+}
+
+function failReceiptWrites(): () => void {
+  const db = openTestJournalHostDatabase(root).db
+  db.exec(`CREATE TRIGGER fail_prompt_receipt BEFORE INSERT ON agent_session_command_receipts
+    BEGIN SELECT RAISE(ABORT, 'receipt unavailable'); END`)
+  return () => db.exec('DROP TRIGGER fail_prompt_receipt')
+}
+
+async function resolution(itemId: string) {
+  const page = await host.history({ sessionId: SESSION, direction: 'tail' })
+  const item = page.ok ? page.page.items.find((entry) => entry.itemId === itemId) : undefined
+  return item?.body.kind === 'approval' ? item.body.resolution.state : undefined
+}
+
+describe('a prompt answer', () => {
+  it.each([
+    ['older than a day', OLD],
+    ['six minutes ahead', FUTURE]
+  ])('accepts an id %s with its resolved revision', async (_case, at) => {
+    const prompt = await seeded()
+    const id = opId(at)
+    expect(await host.respondToPrompt(CALLER, answer(prompt, id))).toMatchObject({
+      ok: true,
+      replayed: false,
+      value: { itemId: prompt.itemId, resolution: { state: 'resolved' } }
+    })
+    expect(receipt(id)).toMatchObject({
+      verdict: 'readable',
+      receipt: { method: 'agentSession.respondTo:approval', result: { kind: 'prompt-answer' } }
+    })
+    expect(answerPrompt).toHaveBeenCalledOnce()
+  })
+
+  it('delivers once for the same id twice, in turn or at once', async () => {
+    const prompt = await seeded()
+    const params = answer(prompt, opId())
+    const [first, second] = await Promise.all([
+      host.respondToPrompt(CALLER, params),
+      host.respondToPrompt(CALLER, params)
+    ])
+    const third = await host.respondToPrompt(CALLER, params)
+    expect(first).toMatchObject({ ok: true, replayed: false })
+    expect(second).toMatchObject({
+      ok: true,
+      replayed: true,
+      value: { revision: prompt.revision + 1 }
+    })
+    expect(third).toMatchObject({
+      ok: true,
+      replayed: true,
+      value: { revision: prompt.revision + 1 }
+    })
+    expect(answerPrompt).toHaveBeenCalledOnce()
+  })
+
+  it('refuses the same id with another choice', async () => {
+    const prompt = await seeded()
+    const id = opId()
+    await host.respondToPrompt(CALLER, answer(prompt, id))
+    expect(await host.respondToPrompt(CALLER, answer(prompt, id, 'deny'))).toMatchObject(CONFLICT)
+    expect(answerPrompt).toHaveBeenCalledOnce()
+  })
+
+  it('points another id at the revision that already holds its choice and delivers nothing', async () => {
+    const prompt = await seeded()
+    await host.respondToPrompt(CALLER, answer(prompt, opId()))
+    const alias = opId()
+    const other = { callerKey: 'client-2' }
+    expect(await host.respondToPrompt(other, answer(prompt, alias))).toMatchObject({
+      ok: true,
+      replayed: false,
+      value: { revision: prompt.revision + 1 }
+    })
+    expect(receipt(alias, other.callerKey)).toMatchObject({
+      receipt: {
+        result: {
+          kind: 'prompt-answer',
+          itemId: prompt.itemId,
+          revision: prompt.revision + 1,
+          resolution: { state: 'resolved', selectedOptionId: 'allow' }
+        }
+      }
+    })
+    expect(await host.respondToPrompt(other, answer(prompt, alias))).toMatchObject({
+      ok: true,
+      replayed: true
+    })
+    expect(answerPrompt).toHaveBeenCalledOnce()
+  })
+
+  it('rolls the revision back when its receipt cannot be saved', async () => {
+    const prompt = await seeded()
+    const id = opId()
+    const restore = failReceiptWrites()
+    await expect(host.respondToPrompt(CALLER, answer(prompt, id))).rejects.toThrow(
+      'receipt unavailable'
+    )
+    restore()
+    expect(await resolution(prompt.itemId)).toBe('pending')
+    expect(receipt(id)).toEqual({ verdict: 'absent' })
+    expect(store.listOperationRows().find((row) => row.operationId === id)).toBeUndefined()
+  })
+
+  it('answers an unreadable receipt as unknown without asking the provider', async () => {
+    const prompt = await seeded()
+    const id = opId()
+    openTestJournalHostDatabase(root)
+      .db.prepare(
+        `INSERT INTO agent_session_command_receipts (operation_id, session_id, caller_key, method,
+          fingerprint, status, result_json, rejection_json, accepted_at)
+         VALUES (?, ?, ?, 'agentSession.respondTo:approval', 'x', 'accepted', '{', NULL, 0)`
+      )
+      .run(id, SESSION, CALLER.callerKey)
+    expect(await host.respondToPrompt(CALLER, answer(prompt, id))).toMatchObject({
+      ok: false,
+      refusal: { code: 'agent_session_operation_unknown' }
+    })
+    expect(answerPrompt).not.toHaveBeenCalled()
+    expect(await resolution(prompt.itemId)).toBe('pending')
+  })
+
+  it("refuses with the update words when a newer Orca's journal holds the chat", async () => {
+    const prompt = await seeded()
+    const id = opId()
+    Object.defineProperty(openTestJournalHostDatabase(root), 'readOnly', { value: true })
+    expect(await host.respondToPrompt(CALLER, answer(prompt, id))).toMatchObject({
+      ok: false,
+      ...SAVED_BY_NEWER_ORCA
+    })
+    expect(receipt(id)).toEqual({ verdict: 'absent' })
+  })
+
+  it('records nothing for a refusal before acceptance', async () => {
+    const prompt = await seeded()
+    const id = opId()
+    expect(await host.respondToPrompt(CALLER, answer(prompt, id, 'deny'))).toMatchObject({
+      ok: false,
+      refusal: { code: 'agent_session_operation_invalid' }
+    })
+    expect(receipt(id)).toEqual({ verdict: 'absent' })
+    expect(store.listOperationRows().find((row) => row.operationId === id)).toBeUndefined()
+  })
+
+  it('replays the resolution it was accepted with after the provider revises the prompt', async () => {
+    const prompt = await seeded()
+    const params = answer(prompt, opId())
+    const first = await host.respondToPrompt(CALLER, params)
+    expect(first).toMatchObject({ ok: true, replayed: false })
+    await reviseFromProvider(prompt.itemId)
+    expect(journal().item(prompt.itemId)?.revision).toBe(prompt.revision + 2)
+    const replay = await host.respondToPrompt(CALLER, params)
+    expect(replay).toEqual({ ...first, replayed: true, cursor: expect.anything() })
+    expect(answerPrompt).toHaveBeenCalledOnce()
+  })
+
+  it('replays the revision another id acknowledged after the provider revises the prompt', async () => {
+    const prompt = await seeded()
+    await host.respondToPrompt(CALLER, answer(prompt, opId()))
+    const other = { callerKey: 'client-2' }
+    const params = answer(prompt, opId())
+    const first = await host.respondToPrompt(other, params)
+    await reviseFromProvider(prompt.itemId)
+    const replay = await host.respondToPrompt(other, params)
+    expect(replay).toEqual({ ...first, replayed: true, cursor: expect.anything() })
+    expect(replay).toMatchObject({
+      value: { revision: prompt.revision + 1, resolution: { resolvedBy: CALLER.callerKey } }
+    })
+  })
+
+  it.each([
+    ['its own answer', CALLER],
+    ['the answer another id acknowledged', { callerKey: 'client-2' }]
+  ])(
+    'replays the accepted answer after a rewind, never answering again: %s',
+    async (_case, caller) => {
+      const prompt = await seeded()
+      const params = answer(prompt, opId())
+      if (caller !== CALLER) {
+        await host.respondToPrompt(CALLER, answer(prompt, opId()))
+      }
+      const first = await host.respondToPrompt(caller, params)
+      expect(first).toMatchObject({ ok: true, replayed: false })
+      await rewindKeepingPrompt(prompt.itemId)
+      expect(journal().item(prompt.itemId)?.revision).not.toBe(prompt.revision + 1)
+      expect(await host.respondToPrompt(caller, params)).toEqual({
+        ...first,
+        replayed: true,
+        cursor: expect.anything()
+      })
+      expect(answerPrompt).toHaveBeenCalledOnce()
+    }
+  )
+
+  it('replays a grouped answer whole after a rewind, free text included', async () => {
+    await attach()
+    const question = { ...PROMPT, ordinal: 100 }
+    acquire.mock.calls.at(-1)?.[0].events?.appendItem(
+      question,
+      {
+        kind: 'question',
+        question: '2 grouped questions',
+        options: [],
+        questions: [
+          {
+            id: 'q1',
+            question: 'Targets',
+            multiSelect: true,
+            options: [
+              { id: 'web', label: 'Web' },
+              { id: 'mobile', label: 'Mobile' }
+            ]
+          },
+          { id: 'q2', question: 'Host', multiSelect: false, options: [], freeTextQuestionId: 'q2' }
+        ],
+        resolution: { state: 'pending', selectedOptionId: null, resolvedBy: null, resolvedAt: null }
+      },
+      { turnScope: AGENT_JOURNAL_THREAD_SCOPE }
+    )
+    await host.flushStreamedEvents(SESSION)
+    const itemId = agentJournalItemKey(question)
+    const answers = [
+      { questionId: 'q1', optionIds: ['web', 'mobile'] },
+      { questionId: 'q2', optionIds: [], other: 'SSH host' }
+    ]
+    const fields = { itemId, expectedRevision: journal().item(itemId)!.revision, answers }
+    const params = {
+      envelope: envelope('agentSession.respondTo:question', fields, { clientOperationId: opId() }),
+      kind: 'question' as const,
+      ...fields
+    }
+    const first = await host.respondToPrompt(CALLER, params)
+    expect(first).toMatchObject({ ok: true, value: { resolution: { answers } } })
+    await journal().replaceEpochItems(
+      'handle_forked',
+      store.getRecord(SESSION)!.lease.runtimeFence,
+      []
+    )
+    expect(await host.respondToPrompt(CALLER, params)).toEqual({
+      ...first,
+      replayed: true,
+      cursor: expect.anything()
+    })
+    expect(answerPrompt).toHaveBeenCalledOnce()
+  })
+
+  it('answers unknown, never success, when acknowledging an answer cannot be recorded', async () => {
+    const prompt = await seeded()
+    await host.respondToPrompt(CALLER, answer(prompt, opId()))
+    const other = { callerKey: 'client-2' }
+    const id = opId()
+    const restore = failReceiptWrites()
+    expect(await host.respondToPrompt(other, answer(prompt, id))).toMatchObject(UNKNOWN)
+    restore()
+    expect(receipt(id, other.callerKey)).toEqual({ verdict: 'absent' })
+    expect(answerPrompt).toHaveBeenCalledOnce()
+  })
+
+  it('refuses a Stop reusing its id, and an answer reusing a Stop id', async () => {
+    const prompt = await seeded()
+    const answered = opId()
+    await host.respondToPrompt(CALLER, answer(prompt, answered))
+    expect(
+      await host.cancel(CALLER, {
+        envelope: envelope('agentSession.cancel', {}, { clientOperationId: answered })
+      })
+    ).toMatchObject(CONFLICT)
+    const second = await seedApproval('approve')
+    const stopped = opId()
+    await host.cancel(CALLER, {
+      envelope: envelope('agentSession.cancel', {}, { clientOperationId: stopped })
+    })
+    expect(await host.respondToPrompt(CALLER, answer(second, stopped, 'approve'))).toMatchObject(
+      CONFLICT
+    )
+    expect(receipt(stopped)).toEqual({ verdict: 'absent' })
+  })
+})

src/main/native-chat/agent-session-wire/structured-agent-session-queued-action-receipts.test.ts

@@ -0,0 +1,551 @@
+// Send now, Delete and Resume accept through their own command receipt, committed with the draft or
+// row they change: no ledger policy decides them, a retry answers from that receipt, and an id
+// another command already used is refused.
+
+import { afterEach, beforeEach, describe, expect, it } from 'vitest'
+import { AGENT_SESSION_MAX_NEW_OPERATION_AGE_MS } from '../../../shared/agent-session-host-authority'
+import { openTestJournalHostDatabase } from '../agent-session-journal/journal-host-database-test-support'
+import {
+  createQueuedMessageTestRig,
+  QUEUED_RIG_CALLER as CALLER,
+  type QueuedMessageTestRig
+} from './structured-agent-session-queued-message-rig.test-fixture'
+import {
+  HOST_TEST_NOW as NOW,
+  HOST_TEST_SESSION as SESSION
+} from './structured-agent-session-host-test-data'
+
+const CALLER_SCOPE = { kind: 'caller', callerKey: CALLER.callerKey } as const
+const OLD = NOW - AGENT_SESSION_MAX_NEW_OPERATION_AGE_MS - 60_000
+const FUTURE = NOW + 6 * 60_000
+
+let rig: QueuedMessageTestRig
+let serial = 0
+
+beforeEach(async () => {
+  rig = await createQueuedMessageTestRig()
+})
+
+afterEach(() => rig.dispose())
+
+function opId(at = NOW): string {
+  serial += 1
+  return `${at}-${serial.toString(16).padStart(32, 'e')}`
+}
+
+async function queuedDraft(text: string): Promise<string> {
+  const queued = await rig.send(text, 'queue-if-active').result
+  if (!queued.ok || !('queued' in queued.value)) {
+    throw new Error('expected a queued receipt')
+  }
+  return queued.value.queued.messageId
+}
+
+function receipt(id: string) {
+  return rig.store.readCommandReceipt(CALLER_SCOPE, id)
+}
+
+function ledgerRow(id: string) {
+  return rig.store.listOperationRows().find((row) => row.operationId === id)
+}
+
+function failReceiptWrites(): () => void {
+  const db = openTestJournalHostDatabase(rig.root).db
+  db.exec(`CREATE TRIGGER fail_queued_receipt BEFORE INSERT ON agent_session_command_receipts
+    BEGIN SELECT RAISE(ABORT, 'receipt unavailable'); END`)
+  return () => db.exec('DROP TRIGGER fail_queued_receipt')
+}
+
+function writeUnreadableReceipt(id: string, method: string): void {
+  openTestJournalHostDatabase(rig.root)
+    .db.prepare(
+      `INSERT INTO agent_session_command_receipts (operation_id, session_id, caller_key, method,
+        fingerprint, status, result_json, rejection_json, accepted_at)
+       VALUES (?, ?, ?, ?, 'x', 'accepted', '{', NULL, 0)`
+    )
+    .run(id, SESSION, CALLER.callerKey, method)
+}
+
+/** What a rewind does to the conversation: a new epoch rebuilt with none of the old rows. */
+async function rewindConversation(): Promise<void> {
+  const journal = rig.host.collaboratorsForTests().sessions.get(SESSION)?.journal
+  if (!journal) {
+    throw new Error('expected an open conversation')
+  }
+  await journal.replaceEpochItems(
+    'handle_forked',
+    rig.store.getRecord(SESSION)!.lease.runtimeFence,
+    []
+  )
+}
+
+/** A stop that leaves a waiting card paused, which only a Resume lifts. */
+async function pausedQueue(): Promise<void> {
+  await rig.workingSend()
+  await queuedDraft('held by stop')
+  await rig.stop()
+  expect(await rig.queuePause()).toEqual({ reason: 'stopped' })
+}
+
+const UNKNOWN = { ok: false, refusal: { code: 'agent_session_operation_unknown' } }
+const CONFLICT = {
+  ok: false,
+  refusal: { code: 'agent_session_operation_conflict', details: { reason: 'operationIdReused' } }
+}
+
+describe('Send now', () => {
+  it.each([
+    ['older than a day', OLD],
+    ['six minutes ahead', FUTURE]
+  ])('accepts an id %s and hands the card off once', async (_case, at) => {
+    await rig.workingSend()
+    const draftId = await queuedDraft('send me now')
+    const id = opId(at)
+    expect(await rig.sendNow(draftId, id)).toMatchObject({
+      ok: true,
+      replayed: false,
+      value: { clientMessageId: id, submission: { queuedMessageId: draftId } }
+    })
+    expect(receipt(id)).toMatchObject({
+      verdict: 'readable',
+      receipt: { method: 'agentSession.queuedMessageSend', result: { kind: 'journal-row' } }
+    })
+    expect(ledgerRow(id)?.outcome).toMatchObject({ status: 'succeeded' })
+  })
+
+  it('answers the same id twice, in turn or at once, from one hand-off', async () => {
+    await rig.workingSend()
+    const draftId = await queuedDraft('send me once')
+    const id = opId()
+    const [first, second] = await Promise.all([rig.sendNow(draftId, id), rig.sendNow(draftId, id)])
+    const third = await rig.sendNow(draftId, id)
+    expect(first).toMatchObject({ ok: true, replayed: false })
+    expect(second).toMatchObject({ ok: true, replayed: true, value: { clientMessageId: id } })
+    expect(third).toMatchObject({ ok: true, replayed: true, value: { clientMessageId: id } })
+    const handoffs = (await rig.host.journalSnapshot(SESSION)).submissions.filter(
+      (entry) => entry.queuedMessageId === draftId
+    )
+    expect(handoffs).toHaveLength(1)
+  })
+
+  it('points another id at the hand-off that already took the card', async () => {
+    await rig.workingSend()
+    const draftId = await queuedDraft('already sent')
+    const first = opId()
+    await rig.sendNow(draftId, first)
+    const alias = opId()
+    expect(await rig.sendNow(draftId, alias)).toMatchObject({
+      ok: true,
+      replayed: false,
+      value: { clientMessageId: first }
+    })
+    expect(receipt(alias)).toMatchObject({ receipt: { result: { kind: 'journal-row' } } })
+    expect(await rig.sendNow(draftId, alias)).toMatchObject({
+      ok: true,
+      replayed: true,
+      value: { clientMessageId: first }
+    })
+  })
+
+  it('replays the hand-off it acknowledged, not a later one that took the card after it came back', async () => {
+    await rig.workingSend()
+    const draftId = await queuedDraft('card')
+    const first = opId()
+    expect(await rig.sendNow(draftId, first)).toMatchObject({ value: { clientMessageId: first } })
+    const ack = opId()
+    expect(await rig.sendNow(draftId, ack)).toMatchObject({
+      ok: true,
+      replayed: false,
+      value: { clientMessageId: first }
+    })
+    // The provider refuses the first hand-off: the card comes back, and another Send takes it.
+    await rig.settleRejected(first, 'nope')
+    const later = opId()
+    expect(await rig.sendNow(draftId, later)).toMatchObject({
+      ok: true,
+      value: { clientMessageId: later }
+    })
+    expect(await rig.sendNow(draftId, ack)).toMatchObject({
+      ok: true,
+      replayed: true,
+      value: { clientMessageId: first }
+    })
+  })
+
+  it('replays its hand-off once the dispatched card is pruned', async () => {
+    await rig.workingSend()
+    const draftId = await queuedDraft('pruned card')
+    const first = opId()
+    await rig.sendNow(draftId, first)
+    const ack = opId()
+    await rig.sendNow(draftId, ack)
+    openTestJournalHostDatabase(rig.root)
+      .db.prepare('DELETE FROM queued_messages WHERE session_id = ? AND message_id = ?')
+      .run(SESSION, draftId)
+    for (const id of [first, ack]) {
+      expect(await rig.sendNow(draftId, id)).toMatchObject({
+        ok: true,
+        replayed: true,
+        value: { clientMessageId: first }
+      })
+    }
+  })
+
+  it('refuses the same id naming another card', async () => {
+    await rig.workingSend()
+    const first = await queuedDraft('first card')
+    const second = await queuedDraft('second card')
+    const id = opId()
+    await rig.sendNow(first, id)
+    expect(await rig.sendNow(second, id)).toMatchObject({
+      ok: false,
+      refusal: {
+        code: 'agent_session_operation_conflict',
+        details: { reason: 'operationIdReused' }
+      }
+    })
+    expect(await rig.drafts()).toContainEqual(expect.objectContaining({ messageId: second }))
+  })
+
+  it('rolls the hand-off back when its receipt cannot be saved, then sends on a retry', async () => {
+    await rig.workingSend()
+    const draftId = await queuedDraft('roll me back')
+    const id = opId()
+    const restore = failReceiptWrites()
+    await expect(rig.sendNow(draftId, id)).rejects.toThrow('receipt unavailable')
+    expect(await rig.handoff(draftId)).toBeUndefined()
+    expect(await rig.drafts()).toContainEqual(expect.objectContaining({ messageId: draftId }))
+    expect(receipt(id)).toEqual({ verdict: 'absent' })
+    expect(ledgerRow(id)).toBeUndefined()
+    restore()
+    expect(await rig.sendNow(draftId, id)).toMatchObject({ ok: true, replayed: false })
+  })
+
+  it('records nothing for a refusal before acceptance', async () => {
+    const id = opId()
+    expect(await rig.sendNow('no-such-card', id)).toMatchObject({
+      ok: false,
+      refusal: { code: 'agent_session_operation_invalid' }
+    })
+    expect(receipt(id)).toEqual({ verdict: 'absent' })
+    expect(ledgerRow(id)).toBeUndefined()
+  })
+
+  it('answers unknown, never success, when acknowledging a hand-off cannot be recorded', async () => {
+    await rig.workingSend()
+    const draftId = await queuedDraft('sent by another')
+    const first = opId()
+    await rig.sendNow(draftId, first)
+    const alias = opId()
+    const restore = failReceiptWrites()
+    expect(await rig.sendNow(draftId, alias)).toMatchObject(UNKNOWN)
+    restore()
+    expect(receipt(alias)).toEqual({ verdict: 'absent' })
+    expect(ledgerRow(alias)).toBeUndefined()
+    const handoffs = (await rig.host.journalSnapshot(SESSION)).submissions.filter(
+      (entry) => entry.queuedMessageId === draftId
+    )
+    expect(handoffs).toHaveLength(1)
+  })
+
+  it('answers an unreadable receipt as unknown and hands nothing off', async () => {
+    await rig.workingSend()
+    const draftId = await queuedDraft('unknown send')
+    const id = opId()
+    writeUnreadableReceipt(id, 'agentSession.queuedMessageSend')
+    expect(await rig.sendNow(draftId, id)).toMatchObject(UNKNOWN)
+    expect(await rig.handoff(draftId)).toBeUndefined()
+  })
+
+  it('answers a retry after a rewind dropped its hand-off as spent, and sends nothing again', async () => {
+    await rig.workingSend()
+    const draftId = await queuedDraft('rewound away')
+    const id = opId()
+    expect(await rig.sendNow(draftId, id)).toMatchObject({ ok: true, replayed: false })
+    await rewindConversation()
+    expect(await rig.sendNow(draftId, id)).toMatchObject({
+      ok: false,
+      refusal: { code: 'agent_session_operation_unknown', details: { reason: 'resultLost' } }
+    })
+    expect(await rig.handoff(draftId)).toBeUndefined()
+  })
+})
+
+describe('Delete', () => {
+  it('withdraws once under an old or future id and replays the withdrawal', async () => {
+    await rig.workingSend()
+    const draftId = await queuedDraft('delete me')
+    const id = opId(OLD)
+    expect(await rig.deleteQueued(draftId, id)).toMatchObject({
+      ok: true,
+      replayed: false,
+      value: { deleted: true, messageId: draftId }
+    })
+    expect(receipt(id)).toMatchObject({
+      receipt: { result: { kind: 'queued-draft', messageId: draftId } }
+    })
+    expect(await rig.deleteQueued(draftId, id)).toMatchObject({
+      ok: true,
+      replayed: true,
+      value: { deleted: true, messageId: draftId }
+    })
+    const future = opId(FUTURE)
+    expect(await rig.deleteQueued(draftId, future)).toMatchObject({
+      ok: true,
+      replayed: false,
+      value: { deleted: false, disposition: 'withdrawn' }
+    })
+  })
+
+  it.each(['missing', 'withdrawn'] as const)(
+    "replays a no-op's first answer when the card was %s",
+    async (disposition) => {
+      await rig.workingSend()
+      const draftId = disposition === 'missing' ? 'no-such-card' : await queuedDraft('gone')
+      if (disposition === 'withdrawn') {
+        await rig.deleteQueued(draftId)
+      }
+      const id = opId()
+      const answer = { deleted: false, messageId: draftId, disposition }
+      expect(await rig.deleteQueued(draftId, id)).toMatchObject({
+        ok: true,
+        replayed: false,
+        value: answer
+      })
+      expect(receipt(id)).toMatchObject({
+        receipt: {
+          result: {
+            kind: 'no-op',
+            outcome: { kind: 'queue-delete', messageId: draftId, disposition }
+          }
+        }
+      })
+      expect(await rig.deleteQueued(draftId, id)).toEqual(
+        expect.objectContaining({ ok: true, replayed: true, value: answer })
+      )
+    }
+  )
+
+  it('answers concurrent copies of one id with one withdrawal', async () => {
+    await rig.workingSend()
+    const draftId = await queuedDraft('delete once')
+    const id = opId()
+    const [first, second] = await Promise.all([
+      rig.deleteQueued(draftId, id),
+      rig.deleteQueued(draftId, id)
+    ])
+    expect(first).toMatchObject({ ok: true, replayed: false, value: { deleted: true } })
+    expect(second).toMatchObject({ ok: true, replayed: true, value: { deleted: true } })
+  })
+
+  it('keeps the card when its receipt cannot be saved', async () => {
+    await rig.workingSend()
+    const draftId = await queuedDraft('keep me')
+    const id = opId()
+    const restore = failReceiptWrites()
+    await expect(rig.deleteQueued(draftId, id)).rejects.toThrow('receipt unavailable')
+    restore()
+    expect(await rig.drafts()).toContainEqual(
+      expect.objectContaining({ messageId: draftId, state: 'waiting' })
+    )
+    expect(ledgerRow(id)).toBeUndefined()
+  })
+
+  it('answers an unreadable receipt as unknown and withdraws nothing', async () => {
+    await rig.workingSend()
+    const draftId = await queuedDraft('unknown delete')
+    const id = opId()
+    writeUnreadableReceipt(id, 'agentSession.queuedMessageDelete')
+    expect(await rig.deleteQueued(draftId, id)).toMatchObject(UNKNOWN)
+    expect(await rig.drafts()).toContainEqual(
+      expect.objectContaining({ messageId: draftId, state: 'waiting' })
+    )
+  })
+
+  it('refuses the same id naming another card and keeps that card', async () => {
+    await rig.workingSend()
+    const first = await queuedDraft('first card')
+    const second = await queuedDraft('second card')
+    const id = opId()
+    await rig.deleteQueued(first, id)
+    expect(await rig.deleteQueued(second, id)).toMatchObject(CONFLICT)
+    expect(await rig.drafts()).toContainEqual(
+      expect.objectContaining({ messageId: second, state: 'waiting' })
+    )
+  })
+
+  it('answers unknown, never success, when a no-op cannot be recorded, and records nothing', async () => {
+    await rig.workingSend()
+    const id = opId()
+    const restore = failReceiptWrites()
+    expect(await rig.deleteQueued('no-such-card', id)).toMatchObject(UNKNOWN)
+    restore()
+    expect(receipt(id)).toEqual({ verdict: 'absent' })
+    expect(ledgerRow(id)).toBeUndefined()
+  })
+
+  it('answers a retry after a rewind from its receipt, withdrawing nothing again', async () => {
+    await rig.workingSend()
+    const draftId = await queuedDraft('deleted then rewound')
+    const id = opId()
+    await rig.deleteQueued(draftId, id)
+    await rewindConversation()
+    expect(await rig.deleteQueued(draftId, id)).toMatchObject({
+      ok: true,
+      replayed: true,
+      value: { deleted: true, messageId: draftId }
+    })
+  })
+})
+
+describe('Resume', () => {
+  it('lifts the pause once under an old id and never lifts a later pause under it', async () => {
+    await rig.workingSend()
+    await queuedDraft('held by stop')
+    await queuedDraft('held by the second stop')
+    await rig.stop()
+    const id = opId(OLD)
+    expect(await rig.resume(id)).toMatchObject({
+      ok: true,
+      replayed: false,
+      value: { resumed: true }
+    })
+    expect(receipt(id)).toMatchObject({ receipt: { result: { kind: 'journal-row' } } })
+    await rig.stop()
+    expect(await rig.queuePause()).toEqual({ reason: 'stopped' })
+    expect(await rig.resume(id)).toMatchObject({
+      ok: true,
+      replayed: true,
+      value: { resumed: false }
+    })
+    expect(await rig.queuePause()).toEqual({ reason: 'stopped' })
+  })
+
+  it('records a no-op Resume and replays it', async () => {
+    const id = opId(FUTURE)
+    expect(await rig.resume(id)).toMatchObject({
+      ok: true,
+      replayed: false,
+      value: { resumed: false }
+    })
+    expect(receipt(id)).toMatchObject({
+      receipt: { result: { kind: 'no-op', outcome: { kind: 'queue-resume', resumed: false } } }
+    })
+    expect(await rig.resume(id)).toMatchObject({
+      ok: true,
+      replayed: true,
+      value: { resumed: false }
+    })
+  })
+
+  it('keeps the pause when its receipt cannot be saved', async () => {
+    await rig.workingSend()
+    await queuedDraft('still held')
+    await rig.stop()
+    const restore = failReceiptWrites()
+    await expect(rig.resume(opId())).rejects.toThrow('receipt unavailable')
+    restore()
+    expect(await rig.queuePause()).toEqual({ reason: 'stopped' })
+  })
+
+  it('answers concurrent copies of one id with one Resume', async () => {
+    await pausedQueue()
+    const id = opId()
+    const [first, second] = await Promise.all([rig.resume(id), rig.resume(id)])
+    expect(first).toMatchObject({ ok: true, replayed: false, value: { resumed: true } })
+    expect(second).toMatchObject({ ok: true, replayed: true, value: { resumed: false } })
+    expect(await rig.queuePause()).toBeNull()
+  })
+
+  it('answers unknown, never success, when a no-op cannot be recorded, and records nothing', async () => {
+    const id = opId()
+    const restore = failReceiptWrites()
+    expect(await rig.resume(id)).toMatchObject(UNKNOWN)
+    restore()
+    expect(receipt(id)).toEqual({ verdict: 'absent' })
+    expect(ledgerRow(id)).toBeUndefined()
+  })
+
+  it("answers a no-op on a newer Orca's records with that refusal, not unknown", async () => {
+    await rig.workingSend()
+    const id = opId()
+    Object.defineProperty(openTestJournalHostDatabase(rig.root), 'readOnly', { value: true })
+    expect(await rig.resume(id)).toMatchObject({
+      ok: false,
+      refusal: { code: 'agent_session_journal_unreadable' }
+    })
+  })
+
+  it('answers an unreadable receipt as unknown and lifts nothing', async () => {
+    await pausedQueue()
+    const id = opId()
+    writeUnreadableReceipt(id, 'agentSession.queuedMessagesResume')
+    expect(await rig.resume(id)).toMatchObject(UNKNOWN)
+    expect(await rig.queuePause()).toEqual({ reason: 'stopped' })
+  })
+
+  it('answers a retry after a rewind from its receipt and never lifts the pause a later Stop set', async () => {
+    await pausedQueue()
+    const id = opId()
+    expect(await rig.resume(id)).toMatchObject({ ok: true, value: { resumed: true } })
+    await rewindConversation()
+    await rig.stop()
+    expect(await rig.queuePause()).toEqual({ reason: 'stopped' })
+    expect(await rig.resume(id)).toMatchObject({
+      ok: true,
+      replayed: true,
+      value: { resumed: false }
+    })
+    expect(await rig.queuePause()).toEqual({ reason: 'stopped' })
+  })
+})
+
+describe.each([
+  ['Delete', (id: string) => rig.deleteQueued('some-card', id)],
+  ['Resume', (id: string) => rig.resume(id)]
+])('%s refused before acceptance', (_action, act) => {
+  it('records nothing', async () => {
+    const id = 'not-an-operation-id'
+    expect(await act(id)).toMatchObject({
+      ok: false,
+      refusal: { code: 'agent_session_operation_invalid' }
+    })
+    expect(receipt(id)).toEqual({ verdict: 'absent' })
+    expect(ledgerRow(id)).toBeUndefined()
+  })
+})
+
+describe('an id reused across commands', () => {
+  const conflict = CONFLICT
+
+  it('refuses a Stop reusing a Delete id, through the compatibility ledger row', async () => {
+    await rig.workingSend()
+    const draftId = await queuedDraft('delete then stop')
+    const id = opId()
+    await rig.deleteQueued(draftId, id)
+    expect(await rig.stop(id)).toMatchObject(conflict)
+  })
+
+  it('refuses a Delete or Resume reusing a Stop id, committing nothing', async () => {
+    await rig.workingSend()
+    const draftId = await queuedDraft('stop then delete')
+    const id = opId()
+    await rig.stop(id)
+    expect(await rig.deleteQueued(draftId, id)).toMatchObject(conflict)
+    expect(await rig.resume(id)).toMatchObject(conflict)
+    expect(await rig.queuePause()).toEqual({ reason: 'stopped' })
+    expect(receipt(id)).toEqual({ verdict: 'absent' })
+  })
+
+  it('refuses a Send now reusing a Delete id, and the reverse', async () => {
+    await rig.workingSend()
+    const first = await queuedDraft('first')
+    const second = await queuedDraft('second')
+    const deleteId = opId()
+    await rig.deleteQueued(first, deleteId)
+    expect(await rig.sendNow(second, deleteId)).toMatchObject(conflict)
+    const sendId = opId()
+    await rig.sendNow(second, sendId)
+    expect(await rig.deleteQueued(second, sendId)).toMatchObject(conflict)
+  })
+})

src/main/native-chat/agent-session-wire/structured-agent-session-queued-mutations.ts

@@ -1,8 +1,7 @@
-// `agentSession.queuedMessageSend` / `agentSession.queuedMessageDelete`, and
-// mutations of the conversation's queued drafts. Settling
-// operations stamp op-scoped tombstone receipts, so a lost acknowledgement
-// replays from the rows themselves — never from the operation ledger, which
-// records only that an operation happened. No mutation returns draft text:
+// `agentSession.queuedMessageSend` / `agentSession.queuedMessageDelete` / Resume: mutations of
+// the conversation's queued drafts. Each action's command receipt commits with the row it
+// changes, so a lost acknowledgement replays from that receipt and the rows it points at, never
+// from the operation ledger. No mutation returns draft text:
 // the published list is the one authority a client renders.
 
 import type { AgentJournalSubmission } from '../../../shared/agent-session-journal-types'
@@ -16,12 +15,14 @@ import type {
 } from '../../../shared/agent-session-wire'
 import type { AgentSessionJournal } from '../agent-session-journal/journal-store'
 import { QueuedMessageNotConsumableError } from '../agent-session-journal/queued-message-consume-error'
-import {
-  isJournalWrittenByNewerOrca,
-  journalOpenRefusal
-} from '../agent-session-journal/journal-open-failure'
+import { refusingNewerOrcaJournal } from '../agent-session-journal/journal-open-failure'
 import type { QueuedMessageRow } from '../agent-session-journal/queued-message-table'
-import type { MutationPlan } from './structured-agent-session-mutation-plans'
+import {
+  journalRowReceiptResult,
+  type MutationPlan
+} from './structured-agent-session-mutation-plans'
+import type { JournalOperationReceipt } from '../agent-session-journal/journal-row-writer'
+import type { CommandReceiptResult } from '../agent-session-journal/command-receipt-schema'
 import { structuredQueueHold } from './structured-agent-session-queued-messages'
 import {
   resumeStructuredQueue,
@@ -52,6 +53,17 @@ function submissionFor(
   return ctx.journal.submissions().find((entry) => entry.clientMessageId === clientMessageId)
 }
 
+/** Where the journal wrote a hand-off's submission, so a replay answers with that one. */
+function submissionRowReceipt(
+  ctx: AgentSessionTurnContext,
+  submission: AgentJournalSubmission | undefined
+): CommandReceiptResult {
+  if (submission?.submittedSequence === undefined) {
+    throw new Error('an acknowledged hand-off requires its submission row')
+  }
+  return { kind: 'journal-row', epoch: ctx.journal.epoch, sequence: submission.submittedSequence }
+}
+
 /** One transaction, stamped with the operation's caller-scoped key so a replay
  *  answers "spent" from the receipts. Withdrawal retires any hold in the same
  *  UPDATE, and the store's commit notification publishes the change. */
@@ -62,12 +74,16 @@ export async function withdrawQueuedMessagesForOperation(
     messageIds: readonly string[]
     callerKey: string
     operationId: string
-  }
+  },
+  receipt?: JournalOperationReceipt
 ): Promise<QueuedMessageRow[]> {
-  return journal.queuedMessages.withdraw({
-    messageIds: input.messageIds,
-    settledByOp: agentSessionOperationKey(input.callerKey, input.operationId)
-  })
+  return journal.queuedMessages.withdraw(
+    {
+      messageIds: input.messageIds,
+      settledByOp: agentSessionOperationKey(input.callerKey, input.operationId)
+    },
+    receipt
+  )
 }
 
 /** Draft actions run like any mutation: admitted on the session's lane, the
@@ -87,20 +103,6 @@ export function mutateQueued<TValue>(
   )
 }
 
-/** A newer Orca's journal refuses a draft write with the words a send gets there. */
-async function refusingNewerOrcaJournal<TValue>(
-  run: Promise<TurnOutcome<TValue>>
-): Promise<TurnOutcome<TValue>> {
-  try {
-    return await run
-  } catch (error) {
-    if (isJournalWrittenByNewerOrca(error)) {
-      return { ok: false, refusal: journalOpenRefusal(error) }
-    }
-    throw error
-  }
-}
-
 /**
  * Send-now. It overrides ONLY queue policy — FIFO order, pause, the busy-turn
  * wait — through the same send block and pending-prompt gates as any send;
@@ -119,13 +121,13 @@ export function sendQueuedStructuredAgentMessage(
     method: 'agentSession.queuedMessageSend',
     fields: { messageId },
     conversationWrite: true,
+    // Its own submission's row, or the row of the hand-off that already took the card.
+    commandReceipt: {
+      result: (row) => journalRowReceiptResult(row, 'submission'),
+      unwritten: (value, ctx) =>
+        submissionRowReceipt(ctx, 'submission' in value ? value.submission : undefined)
+    },
     run: async (ctx): Promise<TurnOutcome<AgentSessionSendResult>> => {
-      // A rerun of this operation after it consumed the card (its answer never
-      // settled): answer with the submission it made, never append it again.
-      const consumedHere = submissionFor(ctx, operationId)
-      if (consumedHere?.queuedMessageId === messageId) {
-        return { ok: true, value: { clientMessageId: operationId, submission: consumedHere } }
-      }
       // The one queue gate; Send-now's override set is exactly `working` (plus
       // FIFO order and the stored hold, which the consume below clears).
       const record = context.deps.store.getRecord(ctx.sessionId)
@@ -172,7 +174,8 @@ export function sendQueuedStructuredAgentMessage(
             expect: row.state,
             settledByOp: agentSessionOperationKey(ctx.resolvedBy, operationId),
             hostInstance: structuredAgentSessionHostInstance()
-          }
+          },
+          ctx.operationReceipt
         )
       } catch (error) {
         if (error instanceof QueuedMessageNotConsumableError) {
@@ -184,19 +187,17 @@ export function sendQueuedStructuredAgentMessage(
       if (!submission) {
         throw new Error('agent_session_submission_lost')
       }
-      context.wakeDelivery(ctx.sessionId)
       return { ok: true, value: { clientMessageId: submissionId, submission } }
     },
-    replay: (ctx) => {
-      const opKey = agentSessionOperationKey(ctx.resolvedBy, operationId)
-      const row = ctx.journal.queuedMessages
-        .receipts(opKey)
-        .find((receipt) => receipt.messageId === messageId)
-      if (!row || row.state !== 'dispatched') {
-        return null
-      }
-      const submission = row.consumedAs === null ? undefined : submissionFor(ctx, row.consumedAs)
-      return submission ? { clientMessageId: submission.clientMessageId, submission } : null
+    // The hand-off the receipt points at, never whichever one the card holds now.
+    replay: (ctx, _outcome, receipt) => {
+      const submission =
+        receipt?.kind === 'journal-row' && receipt.epoch === ctx.journal.epoch
+          ? ctx.journal.submissions().find((entry) => entry.submittedSequence === receipt.sequence)
+          : undefined
+      return submission?.queuedMessageId === messageId
+        ? { clientMessageId: submission.clientMessageId, submission }
+        : null
     }
   }
   return mutateQueued(context, caller, params.envelope, plan)
@@ -216,6 +217,17 @@ export function deleteQueuedStructuredAgentMessage(
     method: 'agentSession.queuedMessageDelete',
     fields: { messageId },
     conversationWrite: true,
+    // The withdrawn card, or a no-op's answer when there was nothing left to withdraw.
+    commandReceipt: {
+      result: () => ({ kind: 'queued-draft', messageId }),
+      unwritten: (value) =>
+        value.deleted
+          ? null
+          : {
+              kind: 'no-op',
+              outcome: { kind: 'queue-delete', messageId, disposition: value.disposition }
+            }
+    },
     run: async (ctx): Promise<TurnOutcome<AgentSessionQueuedMessageDeleteResult>> => {
       const row = ctx.journal.queuedMessages.get(messageId)
       if (!row) {
@@ -230,21 +242,26 @@ export function deleteQueuedStructuredAgentMessage(
       // The withdrawal notifies through the journal's commit listener, which
       // also re-derives the drain — deleting a returned card can unblock the
       // drafts behind it.
-      const withdrawn = await withdrawQueuedMessagesForOperation(ctx.journal, {
-        sessionId: ctx.sessionId,
-        messageIds: [messageId],
-        callerKey: ctx.resolvedBy,
-        operationId
-      })
+      const withdrawn = await withdrawQueuedMessagesForOperation(
+        ctx.journal,
+        {
+          sessionId: ctx.sessionId,
+          messageIds: [messageId],
+          callerKey: ctx.resolvedBy,
+          operationId
+        },
+        ctx.operationReceipt
+      )
       return withdrawn.length > 0
         ? { ok: true, value: { deleted: true, messageId } }
         : { ok: true, value: { deleted: false, messageId, disposition: 'withdrawn' } }
     },
-    replay: (ctx) => {
-      const replayed = ctx.journal.queuedMessages
-        .receipts(agentSessionOperationKey(ctx.resolvedBy, operationId))
-        .some((row) => row.messageId === messageId && row.state === 'withdrawn')
-      return replayed ? { deleted: true, messageId } : null
+    // The receipt alone answers, so a pruned tombstone cannot lose it.
+    replay: (_ctx, _outcome, receipt) => {
+      if (receipt?.kind === 'no-op' && receipt.outcome.kind === 'queue-delete') {
+        return { deleted: false, messageId, disposition: receipt.outcome.disposition }
+      }
+      return receipt?.kind === 'queued-draft' ? { deleted: true, messageId } : null
     }
   }
   return mutateQueued(context, caller, params.envelope, plan)
@@ -262,11 +279,17 @@ export function resumeStructuredAgentQueue(
     method: 'agentSession.queuedMessagesResume',
     fields: {},
     conversationWrite: true,
+    commandReceipt: {
+      result: (row) => journalRowReceiptResult(row, 'tombstone'),
+      unwritten: () => ({ kind: 'no-op', outcome: { kind: 'queue-resume', resumed: false } })
+    },
     // The Resume row notifies through the journal's commit listener, which publishes the
     // lifted pause and wakes the drain.
     run: async (ctx) => ({
       ok: true,
-      value: { resumed: await resumeStructuredQueue(ctx.journal, ctx.fence) }
+      value: {
+        resumed: await resumeStructuredQueue(ctx.journal, ctx.fence, ctx.operationReceipt)
+      }
     }),
     // Like Stop's replay: the Resume already ran, so this one lifts nothing.
     replay: () => ({ resumed: false })

src/main/native-chat/agent-session-wire/structured-agent-session-queued-pause.ts

@@ -5,6 +5,7 @@
 import { randomUUID } from 'node:crypto'
 import type { AgentSessionJournal } from '../agent-session-journal/journal-store'
 import type { DerivedQueuePause } from '../agent-session-journal/queued-message-pause'
+import type { JournalOperationReceipt } from '../agent-session-journal/journal-row-writer'
 import type { StructuredAgentSessionLogger } from './structured-agent-session-logger'
 
 /** A per-process id, minted once per host process like the runtime's own
@@ -56,11 +57,12 @@ export async function markStructuredQueueReopen(
 /** Resume: a journal row that ends every pause. Returns whether the queue was paused. */
 export async function resumeStructuredQueue(
   journal: Pick<AgentSessionJournal, 'queuedMessages' | 'appendQueueResume'>,
-  fence: number
+  fence: number,
+  receipt?: JournalOperationReceipt
 ): Promise<boolean> {
   if (structuredQueuePauses(journal).length === 0) {
     return false
   }
-  await journal.appendQueueResume(fence)
+  await journal.appendQueueResume(fence, receipt)
   return true
 }

src/main/native-chat/agent-session-wire/structured-agent-session-turns-prompt.ts

@@ -23,6 +23,8 @@ import {
   AgentSessionPromptAnswerRejectedError,
   AgentSessionPromptUnavailableError
 } from './structured-agent-session-adapter'
+import type { CommandReceiptResult } from '../agent-session-journal/command-receipt-schema'
+import type { JournalRow } from '../agent-session-journal/journal-row-schema'
 import { settledPrompt, validatePendingPrompt } from './structured-agent-session-prompt-state'
 import type { AgentSessionTurnContext, TurnOutcome } from './structured-agent-session-turns'
 
@@ -132,7 +134,9 @@ export async function performPrompt(
           identity,
           { ...prompt, resolution },
           // A revision: the prompt keeps the turn it was raised in.
-          { fence: ctx.fence, turnScope: ctx.journal.liveTurnScope() }
+          { fence: ctx.fence, turnScope: ctx.journal.liveTurnScope() },
+          // The answer's identity commits here, before the adapter tells the provider.
+          ctx.operationReceipt
         )
       }
     })
@@ -167,3 +171,49 @@ export async function performPrompt(
     value: { itemId: appended.itemId, revision: appended.revision, resolution }
   }
 }
+
+/** The answer a resolved prompt row carries, the one its writer was accepted with. */
+export function promptRowAnswer(row: JournalRow | undefined): AgentSessionPromptResult {
+  const body = row?.kind === 'item' ? row.body : undefined
+  if (row?.kind !== 'item' || (body?.kind !== 'approval' && body?.kind !== 'question')) {
+    throw new Error('an accepted prompt answer requires its prompt row')
+  }
+  return { itemId: row.itemId, revision: row.revision, resolution: body.resolution }
+}
+
+/** What a receipt keeps of an accepted answer: everything its replay says, so the journal's
+ *  later revisions or a rewind never change it. */
+export function promptAnswerReceipt(answer: AgentSessionPromptResult): CommandReceiptResult {
+  const { state, selectedOptionId, answers, resolvedBy, resolvedAt } = answer.resolution
+  if (state !== 'resolved') {
+    throw new Error('an accepted prompt answer is resolved')
+  }
+  return {
+    kind: 'prompt-answer',
+    itemId: answer.itemId,
+    revision: answer.revision,
+    resolution: {
+      state,
+      selectedOptionId,
+      ...(answers
+        ? {
+            answers: answers.map(({ questionId, optionIds, other }) => ({
+              questionId,
+              optionIds,
+              ...(other === undefined ? {} : { other })
+            }))
+          }
+        : {}),
+      resolvedBy,
+      resolvedAt
+    }
+  }
+}
+
+export function acceptedPromptAnswer(
+  receipt: CommandReceiptResult | undefined
+): AgentSessionPromptResult | null {
+  return receipt?.kind === 'prompt-answer'
+    ? { itemId: receipt.itemId, revision: receipt.revision, resolution: receipt.resolution }
+    : null
+}

src/main/native-chat/agent-session-wire/structured-agent-session-turns.ts

@@ -68,7 +68,7 @@ export type AgentSessionTurnContext = {
   publish: () => void
   /** Who a Stop's refusal row names. */
   failureTextContext?: AgentSessionFailureWordsContext
-  /** The operation's success, committed with the row that accepts it (`settlesWithWrite` or `acceptsWithCommandReceipt`). */
+  /** The operation's success, committed with the row that accepts it (`settlesWithWrite` or `commandReceipt`). */
   operationReceipt?: JournalOperationReceipt
   now: () => number
 }

src/main/runtime/agent-session-record-store.ts

@@ -69,6 +69,7 @@ import {
   showAgentSessionTabs
 } from './agent-session-tab-table'
 import type { JournalHostDatabase } from '../native-chat/agent-session-journal/journal-host-database'
+import type { JournalOperationReceipt } from '../native-chat/agent-session-journal/journal-row-writer'
 import { agentSessionOperationOutcomeReceipt } from './agent-session-operation-receipt'
 import { loadAgentSessionStoreRows } from './agent-session-record-rows'
 import { AgentSessionStoreTransactions } from './agent-session-store-transactions'
@@ -311,6 +312,9 @@ export class AgentSessionRecordStore {
   operationOutcomeReceipt = (args: Parameters<typeof agentSessionOperationOutcomeReceipt>[1]) =>
     agentSessionOperationOutcomeReceipt(this.transactions, args)
 
+  commitOperationReceipt = (receipt: JournalOperationReceipt) =>
+    this.transactions.commitReceipt(receipt)
+
   readCommandReceipt = (...args: Parameters<AgentSessionStoreTransactions['readCommandReceipt']>) =>
     this.transactions.readCommandReceipt(...args)
 

src/main/runtime/agent-session-store-transactions.ts

@@ -175,6 +175,16 @@ export class AgentSessionStoreTransactions {
     }
   }
 
+  /** A receipt no journal write carries (a command that changed nothing), in its own transaction. */
+  commitReceipt(receipt: JournalOperationReceipt): Promise<void> {
+    const run = this.queue.then(() => {
+      this.journalDatabase.transaction((db) => receipt.write(db))
+      receipt.committed()
+    })
+    this.queue = run.catch(() => {})
+    return run
+  }
+
   private commit<T>(apply: (draft: AgentSessionStoreState) => T, inMemoryWhenReadOnly: boolean): T {
     const readOnly = this.journalDatabase.readOnly
     if (readOnly && !inMemoryWhenReadOnly) {

src/shared/agent-session-mutation-envelope.ts

@@ -132,7 +132,9 @@ export function agentSessionLedgerRefusal(
 
 /** Why the single admission oracle said no, mapped to what the client can do
  *  about it. The predicate itself is never re-implemented here. */
-function refuseUnlessWriterAdmitted(lease: AgentSessionLease): AgentSessionWireRefusal | null {
+export function refuseUnlessWriterAdmitted(
+  lease: AgentSessionLease
+): AgentSessionWireRefusal | null {
   if (agentSessionLeaseAdmitsWriter(lease)) {
     return null
   }