fix(native-chat): edit a queued message in place, even while a question is open

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

@@ -5,19 +5,14 @@
 // working status, teardown, or the idle sweep.
 
 import type Database from '../../sqlite/sync-database'
-import type {
-  AgentJournalCursor,
-  AgentJournalMessageItem
-} from '../../../shared/agent-session-journal-types'
+import type { AgentJournalMessageItem } from '../../../shared/agent-session-journal-types'
 import {
   AGENT_SESSION_MAX_NEW_OPERATION_AGE_MS,
   AGENT_SESSION_OPERATION_FUTURE_SKEW_MS
 } from '../../../shared/agent-session-host-authority'
-import type { JournalHostDatabase } from './journal-host-database'
-import type { JournalReducerState } from './journal-reducer'
 import type { JournalRow } from './journal-row-schema'
 import type { JournalOperationReceipt } from './journal-row-writer'
-import type { JournalSubmissionConsume } from './journal-store-contracts'
+import type { JournalQueuedMessagesDeps, JournalSubmissionConsume } from './journal-store-contracts'
 import { holdQueuedMessages } from './queued-message-holds'
 import {
   deriveQueuePauses,
@@ -41,38 +36,22 @@ import { moveQueuedMessages, type QueuedMessagePositionMove } from './queued-mes
 import { pruneQueuedMessages, retainedSubmissionVerdict } from './queued-message-retention'
 import {
   queuedMessageSettlementOwed,
+  queuedMessagesAwaitReopenMark,
   settleOwedQueuedMessages,
   settleQueuedMessagesForRow
 } from './queued-message-settlement'
 import { AgentSessionJournalError, assertJournalWritable } from './journal-write-guards'
-import type { JournalAttachmentClaim } from './journal-submission-hook'
-import type { JournalWriteBody, JournalWriteResult } from './journal-write-queue'
+import type { JournalWriteResult } from './journal-write-queue'
 import { QueuedMessageNotConsumableError } from './queued-message-consume-error'
-export { QueuedMessageNotConsumableError } from './queued-message-consume-error'
+import { QueuedMessageEditLeases } from './queued-message-edit-leases'
+import { updateQueuedMessageText, type QueuedMessageTextUpdate } from './queued-message-text-update'
 
 /** Tombstones must outlive the window in which their operation id could still be admitted as new. */
 export const QUEUED_MESSAGE_REPLAY_WINDOW_MS =
   AGENT_SESSION_MAX_NEW_OPERATION_AGE_MS + AGENT_SESSION_OPERATION_FUTURE_SKEW_MS
 
-export type JournalQueuedMessagesDeps = {
-  sessionId: string
-  now: () => number
-  serialize: <T>(run: JournalWriteBody<T>) => Promise<T>
-  database: () => JournalHostDatabase
-  readOnly: () => boolean
-  state: () => JournalReducerState
-  /** Where the reopen's pause begins when this handle could not mark it (`reopenFloor`). */
-  reopenFloor: () => AgentJournalCursor | null
-  /** The journal's own commit notification. Every standalone draft-table
-   *  transaction that changed rows fires it after COMMIT, so a draft or hold
-   *  change publishes and wakes the drain through the same path a journal row
-   *  does — no call site can forget. In-transaction consume and the returned
-   *  transition already ride their row's own commit. */
-  committed: () => void
-  claimAttachments: JournalAttachmentClaim
-}
-
 export class JournalQueuedMessages {
+  readonly editLeases = new QueuedMessageEditLeases(() => this.list())
   /** Bumped on every draft-table write, so publication memos recompute only when they must. */
   private changeRevision = 0
   private listed: { revision: number; rows: readonly QueuedMessageRow[] } | null = null
@@ -108,6 +87,15 @@ export class JournalQueuedMessages {
     return getQueuedMessage(this.deps.database().db, this.deps.sessionId, messageId)
   }
 
+  /** An in-place text edit; a changed card spends every edit lease on its old text. */
+  update(input: QueuedMessageTextUpdate): Promise<ReturnType<typeof updateQueuedMessageText>> {
+    return this.transact(
+      (db) => updateQueuedMessageText(db, this.deps.sessionId, input),
+      (result) => result.status === 'updated',
+      () => this.editLeases.retire(input.messageId)
+    )
+  }
+
   /** Replay receipts for one caller-scoped operation key. */
   receipts(settledByOp: string): QueuedMessageRow[] {
     return queuedMessagesSettledByOp(this.deps.database().db, this.deps.sessionId, settledByOp)
@@ -168,17 +156,9 @@ export class JournalQueuedMessages {
     return this.derivePauses(this.list())
   }
 
-  /** A card waits, or is mid-hand-off and may come back to waiting: a chat that stops running
-   *  marks it (`AgentSessionJournal.markQueueReopen`). */
+  /** `queuedMessagesAwaitReopenMark`. */
   awaitReopenMark(): boolean {
-    const { submissions } = this.deps.state()
-    return this.list().some((row) => {
-      if (row.state === 'waiting') {
-        return true
-      }
-      const handOff = row.consumedAs ? submissions.get(row.consumedAs)?.dispatchState : undefined
-      return row.state === 'dispatched' && (handOff === 'pending' || handOff === 'unknown')
-    })
+    return queuedMessagesAwaitReopenMark(this.list(), this.deps.state().submissions)
   }
 
   /** The person's Stop still pausing the queue, if any (`journalUserStopInForce`). */
@@ -289,7 +269,7 @@ export class JournalQueuedMessages {
    *  throws so the whole append — draft transition AND submission row — rolls back. */
   consumeInTransaction(
     db: Database.Database,
-    input: JournalSubmissionConsume & { consumedAs: string }
+    input: JournalSubmissionConsume & { consumedAs: string; fingerprint: string }
   ): void {
     const { db: own } = this.deps.database()
     if (own !== db) {
@@ -302,7 +282,8 @@ export class JournalQueuedMessages {
       // this guards any pause-relevant row written off that lane from overtaking a held card.
       const cards = listQueuedMessages(db, this.deps.sessionId)
       const pauses = this.derivePauses(cards)
-      if (nextSendableQueuedCard(pauses, cards)?.messageId !== input.messageId) {
+      const held = this.editLeases.heldIds(cards)
+      if (nextSendableQueuedCard(pauses, cards, held)?.messageId !== input.messageId) {
         throw new QueuedMessageNotConsumableError(input.messageId, input.expect)
       }
     }

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

@@ -15,6 +15,9 @@ import type { AgentSessionMessageSource } from '../../../shared/agent-session-me
 import type { JournalHostDatabase } from './journal-host-database'
 import type { JournalLifecycleMutationInput } from './journal-row-builders'
 import type { JournalRow } from './journal-row-schema'
+import type { JournalReducerState } from './journal-reducer'
+import type { JournalAttachmentClaim } from './journal-submission-hook'
+import type { JournalWriteBody } from './journal-write-queue'
 
 export type AgentSessionJournalOptions = {
   identity: AgentSessionJournalIdentity
@@ -122,3 +125,22 @@ export type JournalItemAppendInput = {
   body: AgentJournalItemBody
   options: JournalItemAppendOptions
 }
+
+/** What `JournalQueuedMessages` reads from the journal that owns it. */
+export type JournalQueuedMessagesDeps = {
+  sessionId: string
+  now: () => number
+  serialize: <T>(run: JournalWriteBody<T>) => Promise<T>
+  database: () => JournalHostDatabase
+  readOnly: () => boolean
+  state: () => JournalReducerState
+  /** Where the reopen's pause begins when this handle could not mark it (`reopenFloor`). */
+  reopenFloor: () => AgentJournalCursor | null
+  /** The journal's own commit notification. Every standalone draft-table
+   *  transaction that changed rows fires it after COMMIT, so a draft or hold
+   *  change publishes and wakes the drain through the same path a journal row
+   *  does — no call site can forget. In-transaction consume and the returned
+   *  transition already ride their row's own commit. */
+  committed: () => void
+  claimAttachments: JournalAttachmentClaim
+}

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

@@ -179,6 +179,7 @@ export class AgentSessionJournal {
   /** Refuses every later write and resolves once the admitted ones have landed. Holds no
    *  connection, so there is nothing to release and nothing that can fail. */
   close(): Promise<void> {
+    this.queuedMessages.editLeases.dispose()
     this.queue.markClosed()
     return this.queue.drain()
   }

src/main/native-chat/agent-session-journal/journal-submission-hook.ts

@@ -18,12 +18,16 @@ export type JournalAttachmentClaim = (
 export function journalSubmissionHook(
   queuedMessages: JournalQueuedMessages,
   claimAttachments: JournalAttachmentClaim,
-  input: Pick<JournalSubmissionInput, 'clientMessageId' | 'body' | 'origin'>,
+  input: Pick<JournalSubmissionInput, 'clientMessageId' | 'body' | 'origin' | 'payloadFingerprint'>,
   consume: JournalSubmissionConsume | undefined
 ): JournalRowTransactionHook {
   return (db) => {
     if (consume) {
-      queuedMessages.consumeInTransaction(db, { ...consume, consumedAs: input.clientMessageId })
+      queuedMessages.consumeInTransaction(db, {
+        ...consume,
+        consumedAs: input.clientMessageId,
+        fingerprint: input.payloadFingerprint
+      })
     }
     claimAttachments(db, input.body, input.origin === 'client' && !consume)
   }

src/main/native-chat/agent-session-journal/queued-message-bookkeeping-failure.test.ts

@@ -179,7 +179,12 @@ describe('draft bookkeeping inside a journal append', () => {
       try {
         await expect(
           journal.appendSubmission(
-            { clientMessageId: 'sub-draft-1', payloadFingerprint: 'fp', body: BODY, fence: 0 },
+            {
+              clientMessageId: 'sub-draft-1',
+              payloadFingerprint: 'fp-draft-1',
+              body: BODY,
+              fence: 0
+            },
             { messageId: 'draft-1', expect: 'waiting', settledByOp: null }
           )
         ).rejects.toThrow('SQLITE_FULL')

src/main/native-chat/agent-session-journal/queued-message-edit-leases.test.ts

@@ -0,0 +1,159 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+import { QueuedMessageEditLeases, QUEUED_MESSAGE_EDIT_LEASE_MS } from './queued-message-edit-leases'
+import type { QueuedMessageRow } from './queued-message-table'
+
+function card(messageId = 'card'): QueuedMessageRow {
+  return {
+    sessionId: 'folder-session',
+    messageId,
+    position: 1,
+    body: { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'hello' }] },
+    fingerprint: 'base',
+    createdAt: 1,
+    hostInstance: 'host-a',
+    state: 'waiting',
+    holdReason: null,
+    returnedReason: null,
+    returnedRejection: null,
+    settledAt: null,
+    settledByOp: null,
+    consumedAs: null,
+    carriedFrom: null,
+    queuedAt: null
+  }
+}
+const desktop = { callerKey: 'desktop', editId: 'edit-1' }
+const base = { messageId: 'card', fingerprint: 'base' }
+let rows: QueuedMessageRow[]
+let leases: QueuedMessageEditLeases
+let clock: number
+const wake = vi.fn()
+
+beforeEach(() => {
+  vi.useFakeTimers()
+  clock = 0
+  rows = [card()]
+  wake.mockReset()
+  leases = new QueuedMessageEditLeases(
+    () => rows,
+    () => clock
+  )
+  leases.onDeadline(wake)
+})
+afterEach(() => {
+  leases.dispose()
+  vi.useRealTimers()
+})
+
+function advance(ms: number): void {
+  clock += ms
+  vi.advanceTimersByTime(ms)
+}
+
+describe('queued edit leases', () => {
+  it('a repeated acquire answers the same lease without extending it', () => {
+    expect(leases.acquire(desktop, base)).toMatchObject({ status: 'held', remainingMs: 120_000 })
+    advance(30_000)
+    expect(leases.acquire(desktop, base)).toMatchObject({ status: 'held', remainingMs: 90_000 })
+  })
+
+  it('acquire refuses a card that is gone or whose text moved on', () => {
+    expect(leases.acquire(desktop, { ...base, fingerprint: 'stale' })).toEqual({
+      status: 'changed'
+    })
+    expect(leases.acquire(desktop, { ...base, messageId: 'other' })).toEqual({ status: 'gone' })
+    rows[0]!.state = 'dispatched'
+    expect(leases.acquire(desktop, base)).toEqual({ status: 'gone' })
+    expect(leases.heldIds().size).toBe(0)
+  })
+
+  it('expires on its own and wakes the queue with no other activity', () => {
+    leases.acquire(desktop, base)
+    advance(QUEUED_MESSAGE_EDIT_LEASE_MS - 1)
+    expect(wake).not.toHaveBeenCalled()
+    expect(leases.heldIds().has('card')).toBe(true)
+    advance(1)
+    expect(wake).toHaveBeenCalledOnce()
+    expect(leases.heldIds().size).toBe(0)
+  })
+
+  it('renewal extends the deadline; a lapsed lease is not revived', () => {
+    leases.acquire(desktop, base)
+    advance(100_000)
+    expect(leases.renew(desktop, 'card')).toMatchObject({ status: 'held', remainingMs: 120_000 })
+    advance(119_999)
+    expect(wake).not.toHaveBeenCalled()
+    advance(1)
+    expect(wake).toHaveBeenCalledOnce()
+    expect(leases.renew(desktop, 'card')).toEqual({ status: 'expired' })
+  })
+
+  it("a renew that prunes another editor's lapsed lease before its timer runs still wakes the queue", () => {
+    rows.push({ ...card('second'), position: 2, fingerprint: 'second-base' })
+    const phone = { callerKey: 'phone', editId: 'edit-9' }
+    leases.acquire(desktop, base)
+    clock = 10_000
+    leases.acquire(phone, { messageId: 'second', fingerprint: 'second-base' })
+    // The first deadline has passed on the host clock; its timer is late and has not run.
+    clock = QUEUED_MESSAGE_EDIT_LEASE_MS + 1
+    expect(leases.renew(phone, 'second')).toMatchObject({ status: 'held' })
+    expect([...leases.heldIds()]).toEqual(['second'])
+    vi.advanceTimersByTime(QUEUED_MESSAGE_EDIT_LEASE_MS - 1)
+    expect(wake).toHaveBeenCalledOnce()
+  })
+
+  it('release frees only the exact caller and edit, never another editor on the same card', () => {
+    leases.acquire(desktop, base)
+    leases.acquire({ callerKey: 'phone', editId: 'edit-1' }, base)
+    leases.acquire({ ...desktop, editId: 'edit-2' }, base)
+    leases.release(desktop, 'card')
+    leases.release(desktop, 'card')
+    expect(leases.heldIds().has('card')).toBe(true)
+    leases.release({ callerKey: 'phone', editId: 'edit-1' }, 'card')
+    expect(leases.heldIds().has('card')).toBe(true)
+    leases.release({ ...desktop, editId: 'edit-2' }, 'card')
+    expect(leases.heldIds().size).toBe(0)
+  })
+
+  it.each(['text changed', 'sent', 'deleted', 'handed off and returned'] as const)(
+    'every read re-derives: a card whose %s holds nothing',
+    (change) => {
+      leases.acquire(desktop, base)
+      const row = rows[0]!
+      if (change === 'text changed') {
+        row.fingerprint = 'newer'
+      } else if (change === 'sent') {
+        row.state = 'dispatched'
+      } else if (change === 'deleted') {
+        rows = []
+      } else {
+        Object.assign(row, { state: 'returned', consumedAs: 'submission-2' })
+      }
+      expect(leases.heldIds().size).toBe(0)
+      expect(leases.renew(desktop, 'card')).toEqual({ status: 'expired' })
+    }
+  )
+
+  it('a save retires every lease on the card; others stay', () => {
+    rows.push(card('second'))
+    leases.acquire(desktop, base)
+    leases.acquire({ callerKey: 'phone', editId: 'edit-9' }, base)
+    leases.acquire({ ...desktop, editId: 'edit-2' }, { ...base, messageId: 'second' })
+    leases.retire('card')
+    expect([...leases.heldIds()]).toEqual(['second'])
+  })
+
+  it('reads the rows it is given inside a transaction', () => {
+    leases.acquire(desktop, base)
+    expect(leases.heldIds([{ ...card(), state: 'withdrawn' }]).size).toBe(0)
+  })
+
+  it('dispose drops every lease and deadline; a closed handle grants none', () => {
+    leases.acquire(desktop, base)
+    leases.dispose()
+    advance(QUEUED_MESSAGE_EDIT_LEASE_MS)
+    expect(wake).not.toHaveBeenCalled()
+    expect(leases.acquire(desktop, base)).toEqual({ status: 'gone' })
+    expect(vi.getTimerCount()).toBe(0)
+  })
+})