feat(native-chat): hold mid-turn messages in a host-owned queue

mobile/src/session/mobile-structured-send-delivery.test.ts

@@ -39,6 +39,29 @@ describe('mobileStructuredSendDelivery', () => {
     }
   })
 
+  it('classifies a queued draft answer as accepted and spends its id, replays included', () => {
+    // The host holds the message now; a later identical send is a new message.
+    for (const state of ['waiting', 'dispatched', 'returned', 'withdrawn'] as const) {
+      const queued: StructuredAgentSessionMutationCallResult<AgentSessionSendResult> = {
+        status: 'accepted',
+        value: {
+          clientMessageId: 'client-1',
+          queued: { messageId: 'client-1', position: 1, state }
+        }
+      }
+      expect(mobileStructuredSendDelivery(queued)).toEqual({
+        outcome: 'accepted',
+        operationIdSpent: true,
+        error: null
+      })
+      expect(mobileStructuredSendDelivery(queued, true)).toEqual({
+        outcome: 'accepted',
+        operationIdSpent: true,
+        error: null
+      })
+    }
+  })
+
   it('does not report a retained payload replay as a new accepted send', () => {
     for (const dispatchState of ['accepted', 'pending'] as const) {
       expect(mobileStructuredSendDelivery(accepted(dispatchState), true)).toEqual({

mobile/src/session/mobile-structured-send-delivery.ts

@@ -21,6 +21,7 @@
 //     the retry has to stay a replay. Rotating here is what sent one message to a
 //     model five times.
 
+import type { AgentJournalSubmission } from '../../../src/shared/agent-session-journal-types'
 import type { AgentSessionSendResult } from '../../../src/shared/agent-session-wire'
 import { agentSessionRefusalOperationState } from '../../../src/shared/agent-session-refusal-retry'
 import { structuredAgentSessionRejectionNotice } from '../../../src/shared/structured-agent-session-send-disposition'
@@ -60,7 +61,14 @@ export function mobileStructuredSendDelivery(
       error: result.message
     }
   }
-  const submission = result.value.submission as AgentSessionSendResult['submission'] | undefined
+  if ('queued' in result.value && result.value.queued) {
+    // The host holds (or already settled) the draft: the send is spent — a
+    // later identical message is a new message. A withdrawn replay is spent
+    // too, never unknown: its card was deleted or carried by a /clear.
+    return { outcome: 'accepted', operationIdSpent: true, error: null }
+  }
+  const submission: AgentJournalSubmission | undefined =
+    'submission' in result.value ? result.value.submission : undefined
   if (!submission || submission.dispatchState === 'unknown') {
     return { outcome: 'unknown', operationIdSpent: false, error: null }
   }

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

@@ -7,6 +7,7 @@
 import Database from '../../sqlite/sync-database'
 import { hardenSqliteDatabaseFiles } from '../../sqlite/harden-database-files'
 import { createJournalTablesSql, JOURNAL_DB_SCHEMA_VERSION } from './journal-database-schema'
+import { ensureQueuedMessagesTable } from './queued-message-schema'
 
 export const JOURNAL_BUSY_TIMEOUT_MS = 5000
 
@@ -37,6 +38,10 @@ export function openJournalDatabase(dbPath: string): OpenJournalDatabase {
   try {
     configureJournalPragmas(probe)
     createJournalSchema(probe, stored)
+    // Outside `createJournalSchema` on purpose: its early return skips a db
+    // already at the current version, and this table must exist at EVERY
+    // writable open with no `user_version` bump (see `ensureQueuedMessagesTable`).
+    ensureQueuedMessagesTable(probe)
     hardenSqliteDatabaseFiles(dbPath)
     const opened = { db: probe, readOnly: false }
     transferred = true

src/main/native-chat/agent-session-journal/journal-dispatch-reducer.ts

@@ -6,32 +6,25 @@ import {
   type UnreadAgentSessionFailureFact
 } from '../../../shared/agent-session-failure'
 import { agentJournalSubmissionKey } from '../../../shared/agent-session-journal-item-key'
+import { journalDispatchRowApplies } from './journal-dispatch-settlement'
 import type { JournalReducerState } from './journal-reducer'
-import { placeHandedOverMessage } from './journal-submission-fold'
+import { notePersonTurnAccepted, placeHandedOverMessage } from './journal-submission-fold'
 import type { JournalRow } from './journal-row-schema'
 
 export function applyJournalDispatchRow(
   state: JournalReducerState,
   row: Extract<JournalRow, { kind: 'dispatch' }>
 ): void {
   const submission = state.submissions.get(row.clientMessageId)
-  if (!submission) {
-    return
-  }
-  // `rejected` is terminal; a late `unknown` must not reopen a settled answer.
-  if (submission.dispatchState === 'rejected' || submission.dispatchState === 'accepted') {
+  // Shared with the queued-draft returned hook: a row ignored here must not alter a draft.
+  if (!submission || !journalDispatchRowApplies(submission)) {
     return
   }
   submission.fence = row.fence
   submission.dispatchState = row.state
   submission.providerItemId = row.providerItemId
   submission.reason = row.reason
-  // Read where it can be placed; a kind it cannot place is kept as written, so the classifier
-  // still knows a fact was there without this build claiming what it says.
-  const rejection =
-    row.state === 'rejected'
-      ? (readAgentSessionFailureFact(row.rejection) ?? unreadFailureFact(row.rejection))
-      : undefined
+  const rejection = row.state === 'rejected' ? readStoredRejectionFact(row.rejection) : undefined
   if (rejection) {
     submission.rejection = rejection
   } else {
@@ -47,6 +40,9 @@ export function applyJournalDispatchRow(
   } else {
     delete submission.recovered
   }
+  if (row.state === 'accepted') {
+    notePersonTurnAccepted(state, submission)
+  }
   if (row.state !== 'accepted' || !row.providerItemId) {
     return
   }
@@ -59,6 +55,13 @@ export function applyJournalDispatchRow(
   })
 }
 
+/** A stored rejection fact, read where it can be placed; a kind it cannot place is kept as
+ *  written, so the classifier still knows a fact was there without this build claiming what it
+ *  says. Shared with the queued-draft table, whose returned card mirrors its submission. */
+export function readStoredRejectionFact(value: unknown): UnreadAgentSessionFailureFact | undefined {
+  return readAgentSessionFailureFact(value) ?? unreadFailureFact(value)
+}
+
 function unreadFailureFact(value: unknown): UnreadAgentSessionFailureFact | undefined {
   return typeof value === 'object' &&
     value !== null &&

src/main/native-chat/agent-session-journal/journal-dispatch-settlement.test.ts

@@ -0,0 +1,51 @@
+import { describe, expect, it } from 'vitest'
+import {
+  agentSessionFailureFact,
+  type SubmissionRejectionKind
+} from '../../../shared/agent-session-failure'
+import { agentSessionFailureWords } from '../../../shared/agent-session-failure-words'
+import {
+  DISPATCH_REJECTED_CANCELLED,
+  DISPATCH_REJECTED_HOST_RESTARTED
+} from '../../../shared/structured-agent-session-dispatch-rejection'
+import { rejectedDraftSettlement } from './journal-dispatch-settlement'
+
+function settle(kind: SubmissionRejectionKind) {
+  return rejectedDraftSettlement(
+    agentSessionFailureWords(agentSessionFailureFact(kind), { surface: 'rejection' })
+  )
+}
+
+describe('what a rejection does to the draft it was consumed from', () => {
+  it("a Stop's withdrawal sends it back to waiting, under the queue's pause rather than a hold of its own", () => {
+    expect(settle('cancelled')).toEqual({ state: 'waiting' })
+    expect(rejectedDraftSettlement({ reason: DISPATCH_REJECTED_CANCELLED })).toEqual({
+      state: 'waiting'
+    })
+  })
+
+  it('a restart or close before hand-over sends it back to waiting too', () => {
+    for (const kind of ['hostRestarted', 'chatClosed', 'notDelivered'] as const) {
+      expect(settle(kind)).toEqual({ state: 'waiting' })
+    }
+    expect(rejectedDraftSettlement({ reason: DISPATCH_REJECTED_HOST_RESTARTED })).toEqual({
+      state: 'waiting'
+    })
+  })
+
+  it('a failure returns the card for the user to act on', () => {
+    for (const kind of [
+      'providerRejected',
+      'providerExited',
+      'hostStopped',
+      'startFailed',
+      'writeFailed',
+      'queueFull'
+    ] as const) {
+      expect(settle(kind)).toEqual({ state: 'returned' })
+    }
+    expect(rejectedDraftSettlement({ reason: 'the provider said no' })).toEqual({
+      state: 'returned'
+    })
+  })
+})

src/main/native-chat/agent-session-journal/journal-dispatch-settlement.ts

@@ -0,0 +1,58 @@
+// The one decision for whether a committed dispatch row changes a submission's
+// effective delivery answer, and what a rejection does to the draft it was
+// consumed from. The reducer folds rows through the first and the queued
+// draft's settlement hook fires through it, so the two can never disagree: a
+// row the reducer ignores must not alter a draft.
+
+import type { AgentJournalSubmission } from '../../../shared/agent-session-journal-types'
+import { classifyDispatchRejection } from '../../../shared/structured-agent-session-dispatch-rejection'
+import type { JournalDispatchRow } from './journal-row-schema'
+
+/** `rejected` and `accepted` are terminal; a late row for an absent or settled
+ *  submission must not reopen the answer. */
+export function journalDispatchRowApplies(
+  submission: Pick<AgentJournalSubmission, 'dispatchState'> | undefined
+): boolean {
+  return (
+    submission !== undefined &&
+    submission.dispatchState !== 'rejected' &&
+    submission.dispatchState !== 'accepted'
+  )
+}
+
+/** A consumed draft's submission settled `rejected`: the draft is settled by
+ *  `rejectedDraftSettlement`, since its text has no other holder once it left
+ *  the sender's outbox as a draft. */
+export function consumedSubmissionWasRejected(
+  submission: Pick<AgentJournalSubmission, 'dispatchState'> | undefined
+): boolean {
+  return submission?.dispatchState === 'rejected'
+}
+
+/** What a consumed draft becomes when its submission is rejected. */
+export type RejectedDraftSettlement = { state: 'returned' } | { state: 'waiting' }
+
+/**
+ * Where no one failed the user — a Stop withdrew it, or a restart or close
+ * interrupted it before hand-over — the draft goes back to waiting at its own
+ * position, under whatever pauses the queue: the Stop's own pause, or the
+ * restart's, derived from the host instance. A returned card would block the
+ * drafts behind it on a failure that never happened. A failure returns the
+ * card with its refusal for the user to act on.
+ */
+export function rejectedDraftSettlement(
+  rejection: Pick<AgentJournalSubmission, 'reason'> & { rejection?: unknown }
+): RejectedDraftSettlement {
+  return classifyDispatchRejection(rejection).verdict === null
+    ? { state: 'waiting' }
+    : { state: 'returned' }
+}
+
+/** True when committing this row NEWLY settles the submission to `rejected` —
+ *  the only transition that settles a consumed draft. */
+export function journalDispatchRowNewlyRejects(
+  submission: Pick<AgentJournalSubmission, 'dispatchState'> | undefined,
+  row: Pick<JournalDispatchRow, 'state'>
+): boolean {
+  return row.state === 'rejected' && journalDispatchRowApplies(submission)
+}