native chat: queued-message actions and question answers accept through their own receipt

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,17 +18,20 @@ 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 {
   agentSessionOperationOutcomeUnknown,
   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
   })