diff --git a/src/main/native-chat/agent-session-journal/journal-queued-messages.ts b/src/main/native-chat/agent-session-journal/journal-queued-messages.ts index 3337727689cd..1a66fc156bf7 100644 --- a/src/main/native-chat/agent-session-journal/journal-queued-messages.ts +++ b/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: (run: JournalWriteBody) => Promise - 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> { + 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) } } diff --git a/src/main/native-chat/agent-session-journal/journal-store-contracts.ts b/src/main/native-chat/agent-session-journal/journal-store-contracts.ts index 41b892adf348..39a30289e99e 100644 --- a/src/main/native-chat/agent-session-journal/journal-store-contracts.ts +++ b/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: (run: JournalWriteBody) => Promise + 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 +} diff --git a/src/main/native-chat/agent-session-journal/journal-store.ts b/src/main/native-chat/agent-session-journal/journal-store.ts index dc456d282d1b..3f1aa6ac86b7 100644 --- a/src/main/native-chat/agent-session-journal/journal-store.ts +++ b/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 { + this.queuedMessages.editLeases.dispose() this.queue.markClosed() return this.queue.drain() } diff --git a/src/main/native-chat/agent-session-journal/journal-submission-hook.ts b/src/main/native-chat/agent-session-journal/journal-submission-hook.ts index 84be3ab91c98..36893658f90a 100644 --- a/src/main/native-chat/agent-session-journal/journal-submission-hook.ts +++ b/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, + input: Pick, 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) } diff --git a/src/main/native-chat/agent-session-journal/queued-message-bookkeeping-failure.test.ts b/src/main/native-chat/agent-session-journal/queued-message-bookkeeping-failure.test.ts index 5f9a95a53d20..bc8478602d01 100644 --- a/src/main/native-chat/agent-session-journal/queued-message-bookkeeping-failure.test.ts +++ b/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') diff --git a/src/main/native-chat/agent-session-journal/queued-message-edit-leases.test.ts b/src/main/native-chat/agent-session-journal/queued-message-edit-leases.test.ts new file mode 100644 index 000000000000..7d85e0a58c5b --- /dev/null +++ b/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) + }) +}) diff --git a/src/main/native-chat/agent-session-journal/queued-message-edit-leases.ts b/src/main/native-chat/agent-session-journal/queued-message-edit-leases.ts new file mode 100644 index 000000000000..d698ca1d5eb0 --- /dev/null +++ b/src/main/native-chat/agent-session-journal/queued-message-edit-leases.ts @@ -0,0 +1,187 @@ +// In-memory edit leases for queued cards: while a person edits a card in place, automatic +// delivery stops at it (`nextSendableQueuedCard`). Nothing here is durable. Each lease dies on +// its own deadline, and every read re-derives validity from the card it holds, so a lost +// release, a send, a delete or another client's save never leaves a card held. + +import type { AgentSessionQueuedMessageEditHoldResult } from '../../../shared/agent-session-wire' +import { isUnsettledQueuedMessage, type QueuedMessageRow } from './queued-message-table' + +export const QUEUED_MESSAGE_EDIT_LEASE_MS = 120_000 + +type EditLeaseKey = { callerKey: string; editId: string } + +type EditLease = Pick & { + until: number +} + +/** One per journal handle: closing the conversation drops every lease with it. */ +export class QueuedMessageEditLeases { + private readonly leases = new Map() + private timer: ReturnType | undefined + private timerAt: number | null = null + private wake: (() => void) | undefined + private disposed = false + + constructor( + private readonly rows: () => readonly QueuedMessageRow[], + private readonly now: () => number = () => performance.now() + ) {} + + /** Called once a deadline passes, so the queue re-derives with no other journal activity. */ + onDeadline(wake: () => void): void { + this.wake = wake + } + + /** Cards a live lease holds. `rows` is the caller's own read inside a transaction. */ + heldIds(rows: readonly QueuedMessageRow[] = this.rows()): ReadonlySet { + if (this.leases.size === 0) { + return new Set() + } + const byId = new Map(rows.map((row) => [row.messageId, row])) + const now = this.now() + let lapsed = false + for (const [key, lease] of this.leases) { + const row = byId.get(lease.messageId) + lapsed ||= lease.until <= now + if (lease.until <= now || !row || !leaseMatchesRow(lease, row)) { + this.leases.delete(key) + } + } + this.schedule() + // A deadline pruned here before its timer ran: that timer is gone, so this wakes in its place. + if (lapsed) { + this.wake?.() + } + return new Set([...this.leases.values()].map((lease) => lease.messageId)) + } + + /** Idempotent per caller and edit: a repeat answers the same lease without extending it. */ + acquire( + key: EditLeaseKey, + input: { messageId: string; fingerprint: string } + ): AgentSessionQueuedMessageEditHoldResult { + this.heldIds() + const existing = this.leases.get(leaseKey(key)) + if (existing) { + return existing.messageId === input.messageId && existing.fingerprint === input.fingerprint + ? this.answer(existing) + : { status: 'changed' } + } + const row = this.rows().find((entry) => entry.messageId === input.messageId) + if (this.disposed || !row || !isUnsettledQueuedMessage(row)) { + return { status: 'gone' } + } + if (row.fingerprint !== input.fingerprint) { + return { status: 'changed' } + } + const lease: EditLease = { + messageId: row.messageId, + fingerprint: row.fingerprint, + state: row.state, + consumedAs: row.consumedAs, + until: this.now() + QUEUED_MESSAGE_EDIT_LEASE_MS + } + this.leases.set(leaseKey(key), lease) + this.schedule() + return this.answer(lease) + } + + /** A lease that lapsed or lost its card is not revived: the editor acquires again. */ + renew(key: EditLeaseKey, messageId: string): AgentSessionQueuedMessageEditHoldResult { + this.heldIds() + const lease = this.leases.get(leaseKey(key)) + if (!lease || lease.messageId !== messageId) { + return { status: 'expired' } + } + lease.until = this.now() + QUEUED_MESSAGE_EDIT_LEASE_MS + this.schedule() + return this.answer(lease) + } + + /** Only the exact caller and edit release: never another editor's lease on the same card. */ + release(key: EditLeaseKey, messageId: string): void { + if (this.leases.get(leaseKey(key))?.messageId === messageId) { + this.leases.delete(leaseKey(key)) + this.schedule() + } + } + + /** A save changed the card: every lease on its old text is spent. */ + retire(messageId: string): void { + for (const [key, lease] of this.leases) { + if (lease.messageId === messageId) { + this.leases.delete(key) + } + } + this.schedule() + } + + dispose(): void { + this.disposed = true + this.leases.clear() + this.schedule() + } + + private answer(lease: EditLease): AgentSessionQueuedMessageEditHoldResult { + return { + status: 'held', + fingerprint: lease.fingerprint, + leaseDurationMs: QUEUED_MESSAGE_EDIT_LEASE_MS, + remainingMs: Math.max(0, lease.until - this.now()) + } + } + + /** One timer, at the nearest deadline; unchanged when that deadline is. */ + private schedule(): void { + const next = + this.leases.size === 0 ? null : Math.min(...[...this.leases.values()].map((l) => l.until)) + if (next === this.timerAt) { + return + } + clearTimeout(this.timer) + this.timer = undefined + this.timerAt = next + if (next === null) { + return + } + this.timer = setTimeout( + () => { + this.timer = undefined + this.timerAt = null + const now = this.now() + for (const [key, lease] of this.leases) { + if (lease.until <= now) { + this.leases.delete(key) + } + } + this.schedule() + this.wake?.() + }, + Math.max(1, next - this.now()) + ) + // Read untyped: this module also compiles into the phone's tests, where a timer is a number. + const timer: unknown = this.timer + if ( + typeof timer === 'object' && + timer && + 'unref' in timer && + typeof timer.unref === 'function' + ) { + timer.unref() + } + } +} + +function leaseKey(key: EditLeaseKey): string { + return JSON.stringify([key.callerKey, key.editId]) +} + +/** A hand-off or a return in between re-scopes the card even when its text is unchanged. */ +function leaseMatchesRow(lease: EditLease, row: QueuedMessageRow): boolean { + return ( + isUnsettledQueuedMessage(row) && + row.fingerprint === lease.fingerprint && + row.state === lease.state && + row.consumedAs === lease.consumedAs + ) +} diff --git a/src/main/native-chat/agent-session-journal/queued-message-pause.test.ts b/src/main/native-chat/agent-session-journal/queued-message-pause.test.ts index bedd7bd8351b..8064412b49b7 100644 --- a/src/main/native-chat/agent-session-journal/queued-message-pause.test.ts +++ b/src/main/native-chat/agent-session-journal/queued-message-pause.test.ts @@ -16,7 +16,7 @@ import { liveTestJournalRows, updateTestJournalRowJson } from './journal-host-database-test-support' -import { QueuedMessageNotConsumableError } from './journal-queued-messages' +import { QueuedMessageNotConsumableError } from './queued-message-consume-error' import { applyJournalRow, createJournalReducerState } from './journal-reducer' import { parseJournalRow } from './journal-row-schema' import type { AgentSessionJournal } from './journal-store' diff --git a/src/main/native-chat/agent-session-journal/queued-message-pause.ts b/src/main/native-chat/agent-session-journal/queued-message-pause.ts index 688265f7c278..b0289cbfe33c 100644 --- a/src/main/native-chat/agent-session-journal/queued-message-pause.ts +++ b/src/main/native-chat/agent-session-journal/queued-message-pause.ts @@ -256,19 +256,20 @@ export function queuePauseHolding( } /** The card the queue sends next: the oldest waiting one with no hold of its own, unless a - * returned card or one a pause holds comes first. The queue never reorders, so a newer card never - * overtakes one a pause holds; a card held on its own is passed over. The drain's pick and its - * consume both read this. */ -export function nextSendableQueuedCard( + * returned card, one a pause holds, or one being edited (`editHeld`) comes first. The queue never + * reorders, so a newer card never overtakes either; a card held on its own is passed over. The + * drain's pick and its consume both read this. */ +export function nextSendableQueuedCard( pauses: readonly DerivedQueuePause[], - cards: readonly T[] + cards: readonly T[], + editHeld: ReadonlySet = new Set() ): T | null { for (const card of cards) { if (card.state === 'returned' || queuePauseHolding(pauses, card)) { return null } if (card.state === 'waiting' && card.holdReason === null) { - return card + return editHeld.has(card.messageId) ? null : card } } return null diff --git a/src/main/native-chat/agent-session-journal/queued-message-settlement.ts b/src/main/native-chat/agent-session-journal/queued-message-settlement.ts index dffe87f3af09..212f572fd6cf 100644 --- a/src/main/native-chat/agent-session-journal/queued-message-settlement.ts +++ b/src/main/native-chat/agent-session-journal/queued-message-settlement.ts @@ -37,6 +37,21 @@ export function queuedMessageSettlementOwed( ) } +/** A card waits, or is mid-hand-off and may come back to waiting: a chat that stops running + * marks it (`AgentSessionJournal.markQueueReopen`). */ +export function queuedMessagesAwaitReopenMark( + rows: readonly QueuedMessageRow[], + submissions: Submissions +): boolean { + return rows.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') + }) +} + /** Applies each owed settlement, and withdraws each waiting draft an applied echo proves * delivered (`draftsDeliveredByAppliedEcho`); returns how many drafts changed. */ export function settleOwedQueuedMessages( diff --git a/src/main/native-chat/agent-session-journal/queued-message-store.test.ts b/src/main/native-chat/agent-session-journal/queued-message-store.test.ts index 02e07bc53ee6..991b384b6402 100644 --- a/src/main/native-chat/agent-session-journal/queued-message-store.test.ts +++ b/src/main/native-chat/agent-session-journal/queued-message-store.test.ts @@ -16,11 +16,8 @@ import { agentSessionFailureWords } from '../../../shared/agent-session-failure- import Database from '../../sqlite/sync-database' import { JOURNAL_DB_SCHEMA_VERSION } from './journal-database-schema' import { journalDatabasePath } from './journal-host-database' -import { - JournalQueuedMessages, - QUEUED_MESSAGE_REPLAY_WINDOW_MS, - QueuedMessageNotConsumableError -} from './journal-queued-messages' +import { JournalQueuedMessages, QUEUED_MESSAGE_REPLAY_WINDOW_MS } from './journal-queued-messages' +import { QueuedMessageNotConsumableError } from './queued-message-consume-error' import type { AgentSessionJournal } from './journal-store' import type { AgentMessageSource } from '../../../shared/agent-session-message-source' import { @@ -319,6 +316,31 @@ describe('consume', () => { expect(journal.queuedMessages.get('draft-1')?.state).toBe('withdrawn') }) + it('a consume carrying text the card no longer holds loses: an edit since the read never sends the old text', async () => { + const journal = await open() + await queueDraft(journal, 'draft-1', 'old') + const read = journal.queuedMessages.get('draft-1')! + await journal.queuedMessages.update({ + messageId: 'draft-1', + expectedBodyFingerprint: read.fingerprint, + text: 'new' + }) + await expect( + journal.appendSubmission( + { + clientMessageId: 'sub-draft-1', + payloadFingerprint: read.fingerprint, + body: read.body, + fence: 0, + handoverRecorded: true + }, + { messageId: 'draft-1', expect: 'waiting', settledByOp: null } + ) + ).rejects.toBeInstanceOf(QueuedMessageNotConsumableError) + expect(journal.submissions()).toHaveLength(0) + expect(journal.queuedMessages.get('draft-1')?.state).toBe('waiting') + }) + it('a failed submission insert rolls the draft transition back', async () => { const journal = await open() await queueDraft(journal, 'draft-1') diff --git a/src/main/native-chat/agent-session-journal/queued-message-table.ts b/src/main/native-chat/agent-session-journal/queued-message-table.ts index 303ea060c633..9f2559804486 100644 --- a/src/main/native-chat/agent-session-journal/queued-message-table.ts +++ b/src/main/native-chat/agent-session-journal/queued-message-table.ts @@ -155,6 +155,8 @@ export function consumeQueuedMessageInTransaction( sessionId: string messageId: string expect: 'waiting' | 'returned' + /** The body being sent: a card edited since it was read is not consumed with its old text. */ + fingerprint: string /** The fresh submission id; never the draft's own id. */ consumedAs: string settledByOp: string | null @@ -171,7 +173,7 @@ export function consumeQueuedMessageInTransaction( `UPDATE queued_messages SET state = 'dispatched', hold_reason = NULL, returned_reason = NULL, returned_rejection = NULL, settled_at = ?, settled_by_op = ?, consumed_as = ?, host_instance = COALESCE(?, host_instance) - WHERE session_id = ? AND message_id = ? AND state = ?` + WHERE session_id = ? AND message_id = ? AND state = ? AND fingerprint = ?` ) .run( input.now, @@ -180,7 +182,8 @@ export function consumeQueuedMessageInTransaction( input.hostInstance ?? null, input.sessionId, input.messageId, - input.expect + input.expect, + input.fingerprint ) return Number(changed.changes ?? 0) === 1 } diff --git a/src/main/native-chat/agent-session-journal/queued-message-text-update.test.ts b/src/main/native-chat/agent-session-journal/queued-message-text-update.test.ts new file mode 100644 index 000000000000..d48ce57179db --- /dev/null +++ b/src/main/native-chat/agent-session-journal/queued-message-text-update.test.ts @@ -0,0 +1,181 @@ +import { afterEach, beforeEach, describe, expect, it } from 'vitest' +import Database from '../../sqlite/sync-database' +import { ensureQueuedMessagesTable } from './queued-message-schema' +import { getQueuedMessage, insertQueuedMessage } from './queued-message-table' +import { updateQueuedMessageText } from './queued-message-text-update' +import type { AgentJournalMessageItem } from '../../../shared/agent-session-journal-types' +import { agentSessionSendBodyFingerprint } from '../../../shared/structured-agent-session-send-mutation' +import { MAX_PROMPT_BYTES } from '../../../shared/rpc-contract/structured-agent-session-params' + +let db: Database.Database +const sessionId = 'folder-session' +const initial: AgentJournalMessageItem = { + kind: 'message', + role: 'user', + blocks: [{ type: 'text', text: 'base' }] +} +beforeEach(() => { + db = new Database(':memory:') + ensureQueuedMessagesTable(db) +}) +afterEach(() => db.close()) +function insert(body = initial, messageId = 'card') { + return insertQueuedMessage(db, { + sessionId, + messageId, + body, + fingerprint: agentSessionSendBodyFingerprint(sessionId, body), + hostInstance: 'remote-host', + queuedAt: { epoch: 'epoch', sequence: 4 }, + now: 1, + holdReason: 'send_failed' + }) +} +function update(text: string, fingerprint = agentSessionSendBodyFingerprint(sessionId, initial)) { + return updateQueuedMessageText(db, sessionId, { + messageId: 'card', + expectedBodyFingerprint: fingerprint, + text + }) +} + +describe('atomic queued text update', () => { + it('preserves identity, cursor, order, sender, kind, role, send mode and attachment order', () => { + const body: AgentJournalMessageItem = { + ...initial, + sentAs: 'goal', + from: { kind: 'agent', senders: [], orchestration: null }, + blocks: [ + { type: 'image-ref', path: 'before.png' }, + { type: 'text', text: 'one' }, + { type: 'image-ref', url: 'https://example.com/image.png' }, + { type: 'text', text: 'two' } + ] + } + const before = insert(body) + expect(update('edited', before.fingerprint).status).toBe('updated') + const after = getQueuedMessage(db, sessionId, 'card')! + expect(after).toEqual({ + ...before, + body: { + ...body, + blocks: [body.blocks[0]!, { type: 'text', text: 'edited' }, body.blocks[2]!] + }, + fingerprint: after.fingerprint + }) + expect(after.fingerprint).toBe(agentSessionSendBodyFingerprint(sessionId, after.body)) + }) + it('saving the same text keeps several text blocks as they are', () => { + const body: AgentJournalMessageItem = { + ...initial, + blocks: [ + { type: 'text', text: 'one' }, + { type: 'text', text: 'two' } + ] + } + const before = insert(body) + expect(update('one\ntwo', before.fingerprint).status).toBe('unchanged') + expect(getQueuedMessage(db, sessionId, 'card')).toEqual(before) + }) + it('retains returned-card refusal and failure hold rather than requeueing', () => { + insert() + db.prepare( + "UPDATE queued_messages SET state = 'returned', returned_reason = 'refused', consumed_as = 'submission' WHERE message_id = 'card'" + ).run() + const before = getQueuedMessage(db, sessionId, 'card')! + expect(update('edit').status).toBe('updated') + expect(getQueuedMessage(db, sessionId, 'card')).toMatchObject({ + state: 'returned', + returnedReason: before.returnedReason, + consumedAs: before.consumedAs, + holdReason: before.holdReason, + position: before.position + }) + }) + it('checks the resulting body before the expected base, making lost-answer retry safe', () => { + insert() + expect(update('edit').status).toBe('updated') + expect(update('edit').status).toBe('unchanged') + expect(update('other').status).toBe('changed') + }) + it('accepts semantic A/B/A and preserves the latest host-owned sender', () => { + insert() + const base = agentSessionSendBodyFingerprint(sessionId, initial) + update('B') + const second = getQueuedMessage(db, sessionId, 'card')! + update('base', second.fingerprint) + const from: NonNullable = { + kind: 'agent', + senders: [], + orchestration: null + } + const latestBody = { ...initial, from } + db.prepare("UPDATE queued_messages SET body_json = ? WHERE message_id = 'card'").run( + JSON.stringify(latestBody) + ) + expect(agentSessionSendBodyFingerprint(sessionId, latestBody)).toBe(base) + expect(update('C', base).status).toBe('updated') + expect(getQueuedMessage(db, sessionId, 'card')?.body.from).toEqual(from) + }) + it('refuses empty, oversized, command and unsupported bodies without modifying them', () => { + insert() + expect(update(' ').status).toBe('not-editable') + expect(update('é'.repeat(MAX_PROMPT_BYTES)).status).toBe('not-editable') + db.prepare('DELETE FROM queued_messages').run() + const command = insert({ ...initial, command: { name: 'compact' } }) + expect(update('edit', command.fingerprint).status).toBe('not-editable') + expect(getQueuedMessage(db, sessionId, 'card')).toEqual(command) + db.prepare('DELETE FROM queued_messages').run() + const unsupported = insert({ + ...initial, + blocks: [{ type: 'tool-result', callId: 'tool', output: 'retained' }] + }) + expect(update('edit', unsupported.fingerprint).status).toBe('not-editable') + expect(getQueuedMessage(db, sessionId, 'card')).toEqual(unsupported) + }) + it('allows empty text with a retained attachment, and inserts text before one with none', () => { + const before = insert({ + ...initial, + blocks: [ + { type: 'text', text: 'caption' }, + { type: 'image-ref', path: 'file.png' } + ] + }) + expect(update('', before.fingerprint).status).toBe('updated') + expect(getQueuedMessage(db, sessionId, 'card')?.body.blocks).toEqual([ + { type: 'text', text: '' }, + { type: 'image-ref', path: 'file.png' } + ]) + const imageOnly = insert( + { ...initial, blocks: [{ type: 'image-ref', path: 'file.png' }] }, + 'image' + ) + expect( + updateQueuedMessageText(db, sessionId, { + messageId: 'image', + expectedBodyFingerprint: imageOnly.fingerprint, + text: 'added' + }).status + ).toBe('updated') + expect(getQueuedMessage(db, sessionId, 'image')?.body.blocks).toEqual([ + { type: 'text', text: 'added' }, + { type: 'image-ref', path: 'file.png' } + ]) + }) + it('a failed transaction rolls back body and fingerprint together', () => { + const before = insert() + db.exec('BEGIN IMMEDIATE') + update('edit') + db.exec('ROLLBACK') + expect(getQueuedMessage(db, sessionId, 'card')).toEqual(before) + }) + it('dispatch, withdrawal and a pruned tombstone all answer gone', () => { + insert() + db.prepare("UPDATE queued_messages SET state = 'dispatched' WHERE message_id = 'card'").run() + expect(update('edit')).toMatchObject({ status: 'gone', disposition: 'dispatched' }) + db.prepare("UPDATE queued_messages SET state = 'withdrawn' WHERE message_id = 'card'").run() + expect(update('edit')).toMatchObject({ status: 'gone', disposition: 'withdrawn' }) + db.prepare('DELETE FROM queued_messages').run() + expect(update('edit')).toMatchObject({ status: 'gone', disposition: 'missing' }) + }) +}) diff --git a/src/main/native-chat/agent-session-journal/queued-message-text-update.ts b/src/main/native-chat/agent-session-journal/queued-message-text-update.ts new file mode 100644 index 000000000000..1c631b37c82f --- /dev/null +++ b/src/main/native-chat/agent-session-journal/queued-message-text-update.ts @@ -0,0 +1,66 @@ +import type Database from '../../sqlite/sync-database' +import type { AgentJournalMessageItem } from '../../../shared/agent-session-journal-types' +import type { AgentSessionQueuedMessageUpdateResult } from '../../../shared/agent-session-wire' +import { agentSessionSendBodyFingerprint } from '../../../shared/structured-agent-session-send-mutation' +import { queuedMessageWithEditedText } from '../../../shared/queued-message-text-edit' +import { + getQueuedMessage, + isUnsettledQueuedMessage, + type QueuedMessageRow +} from './queued-message-table' + +export type QueuedMessageTextUpdate = { + messageId: string + expectedBodyFingerprint: string + text: string +} + +/** Compare-and-set on the body fingerprint. The resulting body is checked first, so a retried + * Save whose first answer was lost reads `unchanged`, and one a newer edit superseded reads + * `changed`: a retry never overwrites newer text. */ +export function queuedMessageTextUpdateDecision( + row: QueuedMessageRow | null, + sessionId: string, + input: QueuedMessageTextUpdate +): + | AgentSessionQueuedMessageUpdateResult + | { status: 'ready'; body: AgentJournalMessageItem; fingerprint: string } { + const { messageId } = input + if (!row || !isUnsettledQueuedMessage(row)) { + const disposition = !row ? 'missing' : row.state === 'dispatched' ? 'dispatched' : 'withdrawn' + return { status: 'gone', messageId, disposition } + } + const body = queuedMessageWithEditedText(row.body, input.text) + if (!body) { + return { status: 'not-editable', messageId } + } + const fingerprint = agentSessionSendBodyFingerprint(sessionId, body) + if (row.fingerprint === fingerprint) { + return { status: 'unchanged', messageId, fingerprint } + } + if (row.fingerprint !== input.expectedBodyFingerprint) { + return { status: 'changed', messageId } + } + return { status: 'ready', body, fingerprint } +} + +/** Inside the journal's draft transaction; only body and fingerprint change, so the card keeps + * its id, position, state, holds and hand-off link. */ +export function updateQueuedMessageText( + db: Database.Database, + sessionId: string, + input: QueuedMessageTextUpdate +): AgentSessionQueuedMessageUpdateResult { + const decision = queuedMessageTextUpdateDecision( + getQueuedMessage(db, sessionId, input.messageId), + sessionId, + input + ) + if (decision.status !== 'ready') { + return decision + } + db.prepare( + 'UPDATE queued_messages SET body_json = ?, fingerprint = ? WHERE session_id = ? AND message_id = ?' + ).run(JSON.stringify(decision.body), decision.fingerprint, sessionId, input.messageId) + return { status: 'updated', messageId: input.messageId, fingerprint: decision.fingerprint } +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts index 251ccc88e32b..78dcd0b1acd6 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts @@ -65,7 +65,7 @@ export class StructuredAgentSessionHost { }, deliverSettleEdge: (id, journal) => this.conversationDelivery.afterSettleEdge(id, journal), logger: sessionLogger.deferredStructuredAgentSessionLogger(() => this.deps.logger), - onOpened: (sessionId) => this.queued.drain.schedule(sessionId), + onOpened: (sessionId) => this.queued.onOpened(sessionId), now: () => this.now() }) private readonly queued = wireStructuredAgentSessionQueuedMessages(this.sessions, () => @@ -301,6 +301,8 @@ export class StructuredAgentSessionHost { queuedMessageSend = this.queued.queuedMessageSend queuedMessageDelete = this.queued.queuedMessageDelete queuedMessagesResume = this.queued.queuedMessagesResume + queuedMessageUpdate = this.queued.queuedMessageUpdate + queuedMessageEditHold = this.queued.queuedMessageEditHold waitForSendSettlement = this.clientDelivery.waitForSendSettlement diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-edit.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-edit.test.ts new file mode 100644 index 000000000000..9e4280fb54f7 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-edit.test.ts @@ -0,0 +1,335 @@ +// Editing a queued card in place, through the real host, journal and drain: Save keeps the card's +// identity, an edit lease stops automatic delivery at the card (and so at everything behind it), +// and no lease outlives its deadline, its card, or its conversation. + +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { + createQueuedMessageTestRig, + eventually, + QUEUED_RIG_CALLER, + type QueuedMessageTestRig +} from './structured-agent-session-queued-message-rig.test-fixture' +import { + HOST_TEST_SESSION, + hostTestMessage, + hostTestOperationId +} from './structured-agent-session-host-test-data' +import { agentSessionSendBodyFingerprint } from '../../../shared/structured-agent-session-send-mutation' +import { + readQueuePublication, + structuredQueueSendGate +} from './structured-agent-session-queued-publication' +import { shouldQueueStructuredAgentSessionSend } from './structured-agent-session-queued-messages' +import { QUEUED_MESSAGES_PUBLISHED_MAX_BYTES } from './structured-agent-session-queued-published-bytes' +import { QUEUED_MESSAGE_EDIT_LEASE_MS } from '../agent-session-journal/queued-message-edit-leases' + +let rig: QueuedMessageTestRig +beforeEach(async () => { + rig = await createQueuedMessageTestRig() +}) +afterEach(() => rig.dispose()) + +function journal() { + const value = rig.host.collaboratorsForTests().sessions.get(HOST_TEST_SESSION)?.journal + if (!value) { + throw new Error('missing test conversation') + } + return value +} +async function queue(text: string, options?: { internal: true }) { + const result = await rig.send(text, 'queue-if-active', options).result + if (!result.ok || !('queued' in result.value)) { + throw new Error('not queued') + } + return result.value.queued.messageId +} +function fingerprint(id: string) { + const card = journal().queuedMessages.get(id) + if (!card) { + throw new Error('missing card') + } + return card.fingerprint +} +function update(messageId: string, expected: string, text: string, operationId?: string) { + const fields = { messageId, expectedBodyFingerprint: expected, text } + return rig.host.queuedMessageUpdate(QUEUED_RIG_CALLER, { + envelope: rig.envelope( + fields, + 'agentSession.queuedMessageUpdate', + operationId ?? hostTestOperationId() + ), + ...fields + }) +} +function hold( + messageId: string, + action: 'acquire' | 'renew' | 'release' = 'acquire', + editId = 'edit-1', + caller = QUEUED_RIG_CALLER +) { + const common = { sessionId: HOST_TEST_SESSION, messageId, editId } + return rig.host.queuedMessageEditHold( + caller, + action === 'acquire' + ? { ...common, action, expectedBodyFingerprint: fingerprint(messageId) } + : { ...common, action } + ) +} +function publication() { + return readQueuePublication(journal(), structuredQueueSendGate(rig.store, HOST_TEST_SESSION)) +} +function published(id: string) { + return publication().queuedMessages.find((message) => message.messageId === id) +} +function held() { + return [...journal().queuedMessages.editLeases.heldIds()] +} + +describe('saving a queued card in place', () => { + it('keeps id, position and sender; text that reads like a command stays text', async () => { + await rig.workingSend() + const from = { kind: 'agent' as const, senders: [], orchestration: null } + const queued = await rig.send('original', 'queue-if-active', { internal: true, from }).result + if (!queued.ok || !('queued' in queued.value)) { + throw new Error('not queued') + } + const id = queued.value.queued.messageId + const before = journal().queuedMessages.get(id)! + expect(await update(id, before.fingerprint, '/compact')).toMatchObject({ + ok: true, + value: { status: 'updated' } + }) + const body = { ...before.body, blocks: [{ type: 'text' as const, text: '/compact' }] } + expect(journal().queuedMessages.get(id)).toEqual({ + ...before, + body, + fingerprint: agentSessionSendBodyFingerprint(HOST_TEST_SESSION, body) + }) + expect(published(id)?.body).toEqual(body) + }) + + it('a resent Save reads the card as it stands and never overwrites newer text', async () => { + await rig.workingSend() + const id = await queue('base') + const base = fingerprint(id) + const operation = hostTestOperationId() + expect(await update(id, base, 'mine', operation)).toMatchObject({ + value: { status: 'updated' } + }) + expect(await update(id, base, 'mine', operation)).toMatchObject({ + value: { status: 'unchanged' } + }) + expect(await update(id, fingerprint(id), 'newer')).toMatchObject({ + value: { status: 'updated' } + }) + expect(await update(id, base, 'mine', operation)).toMatchObject({ + value: { status: 'changed' } + }) + expect(await update(id, base, 'another draft')).toMatchObject({ + value: { status: 'changed' } + }) + expect(journal().queuedMessages.get(id)?.body.blocks).toEqual([{ type: 'text', text: 'newer' }]) + }) + + it('after Send or Delete elsewhere, Save answers gone and recreates nothing', async () => { + await rig.workingSend() + const deleted = await queue('deleted') + const sent = await queue('sent') + const bases = [fingerprint(deleted), fingerprint(sent)] as const + await rig.deleteQueued(deleted) + await rig.sendNow(sent) + expect(await update(deleted, bases[0], 'edit')).toMatchObject({ + value: { status: 'gone', disposition: 'withdrawn' } + }) + expect(await update(sent, bases[1], 'edit')).toMatchObject({ + value: { status: 'gone', disposition: 'dispatched' } + }) + expect(await rig.drafts()).toEqual([]) + }) + + it('counts the edited card once against the queue size bound', async () => { + await rig.workingSend() + const big = await queue('x'.repeat(QUEUED_MESSAGES_PUBLISHED_MAX_BYTES - 120 * 1024)) + const small = await queue('small') + expect(await update(small, fingerprint(small), 'y'.repeat(200 * 1024))).toMatchObject({ + ok: false, + refusal: { details: { reason: 'queueTooLarge' } } + }) + expect(journal().queuedMessages.get(small)?.body.blocks).toEqual([ + { type: 'text', text: 'small' } + ]) + expect(await update(big, fingerprint(big), 'z'.repeat(250 * 1024))).toMatchObject({ + ok: true, + value: { status: 'updated' } + }) + }) + + it('a command card is not editable', async () => { + await rig.workingSend() + const body = { ...hostTestMessage('/compact'), command: { name: 'compact' } } + const command = await journal().queuedMessages.insert({ + messageId: 'command-card', + body, + fingerprint: agentSessionSendBodyFingerprint(HOST_TEST_SESSION, body), + hostInstance: 'host' + }) + expect(await hold(command.messageId)).toEqual({ status: 'not-editable' }) + expect(await update(command.messageId, command.fingerprint, 'text')).toMatchObject({ + value: { status: 'not-editable' } + }) + }) +}) + +describe('edit leases and automatic delivery', () => { + it('earlier cards deliver; the edited card and those behind it wait; a new send still queues', async () => { + const work = await rig.workingSend() + const first = await queue('first') + const edited = await queue('edited') + const last = await queue('last') + expect(await hold(edited)).toMatchObject({ status: 'held', leaseDurationMs: 120_000 }) + expect(held()).toEqual([edited]) + expect(published(edited)?.paused).toBeUndefined() + await rig.settleAccepted(work, 'work-done') + await eventually(async () => expect(await rig.handoff(first)).toBeDefined()) + await rig.settleAccepted(await rig.handoffId(first), 'first-done') + expect(publication().nextQueuedMessageId).toBeNull() + expect(journal().queuedMessages.get(edited)?.state).toBe('waiting') + expect(journal().queuedMessages.get(last)?.state).toBe('waiting') + expect( + shouldQueueStructuredAgentSessionSend({ + journal: journal(), + ...structuredQueueSendGate(rig.store, HOST_TEST_SESSION)() + }) + ).toBe(true) + // An explicit Send still overrides, and spends the lease with the card. + expect(await rig.sendNow(edited)).toMatchObject({ ok: true }) + expect(journal().queuedMessages.editLeases.heldIds().size).toBe(0) + }) + + it('Save then release lets the edited card drain with its new text', async () => { + const work = await rig.workingSend() + const id = await queue('before') + await hold(id) + await rig.settleAccepted(work, 'work-done') + expect(await update(id, fingerprint(id), 'after')).toMatchObject({ + value: { status: 'updated' } + }) + expect(held()).toEqual([]) + await hold(id, 'release') + await eventually(async () => expect(await rig.handoff(id)).toBeDefined()) + const submissions = (await rig.host.journalSnapshot(HOST_TEST_SESSION)).submissions + expect(submissions.at(-1)?.payloadFingerprint).toBe( + agentSessionSendBodyFingerprint(HOST_TEST_SESSION, { + kind: 'message', + role: 'user', + blocks: [{ type: 'text', text: 'after' }] + }) + ) + }) + + it("one editor's release keeps another's hold; a renewal re-sends nothing and is not activity", async () => { + await rig.workingSend() + const id = await queue('base') + await hold(id) + await hold(id, 'acquire', 'edit-1', { callerKey: 'phone' }) + const before = publication() + const touch = vi.spyOn(rig.host.collaboratorsForTests().sessions, 'touch') + expect(await hold(id, 'renew')).toMatchObject({ status: 'held' }) + // An open editor does not keep the agent's process from its idle stop. + expect(touch).not.toHaveBeenCalled() + expect(publication()).toBe(before) + expect(await hold(id, 'release')).toEqual({ status: 'released' }) + expect(held()).toEqual([id]) + await hold(id, 'release', 'edit-1', { callerKey: 'phone' }) + expect(held()).toEqual([]) + }) + + it('acquiring after delivery won answers gone; Save without a lease still works', async () => { + await rig.workingSend() + const id = await queue('base') + expect(await update(id, fingerprint(id), 'saved')).toMatchObject({ + value: { status: 'updated' } + }) + await rig.sendNow(id) + expect( + await rig.host.queuedMessageEditHold(QUEUED_RIG_CALLER, { + sessionId: HOST_TEST_SESSION, + messageId: id, + editId: 'late', + action: 'acquire', + expectedBodyFingerprint: 'f'.repeat(64) + }) + ).toEqual({ status: 'gone' }) + }) + + it('a lapsed lease wakes delivery with no journal activity, even when publishing fails', async () => { + const work = await rig.workingSend() + const id = await queue('edited') + let clock = 0 + const time = vi.spyOn(performance, 'now').mockImplementation(() => clock) + vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout'] }) + try { + await hold(id) + await rig.settleAccepted(work, 'work-done') + await vi.advanceTimersByTimeAsync(0) + expect(publication().nextQueuedMessageId).toBeNull() + const publish = vi + .spyOn(rig.host.collaboratorsForTests().subscribers, 'publish') + .mockImplementationOnce(() => { + throw new Error('publication failed') + }) + clock = QUEUED_MESSAGE_EDIT_LEASE_MS + await vi.advanceTimersByTimeAsync(QUEUED_MESSAGE_EDIT_LEASE_MS) + publish.mockRestore() + vi.useRealTimers() + await eventually(async () => expect(await rig.handoff(id)).toBeDefined()) + } finally { + vi.useRealTimers() + time.mockRestore() + } + }) + + it('Stop and Resume leave the edit hold alone and never publish it as a pause', async () => { + await rig.workingSend() + const id = await queue('base') + await hold(id) + await rig.stop() + await rig.resume() + expect(held()).toEqual([id]) + expect(published(id)?.paused).toBeUndefined() + expect(publication().nextQueuedMessageId).toBeNull() + }) + + it('an edit open across /clear keeps its hold and saves into the same card', async () => { + const working = await rig.workingSend() + const id = await queue('before clear') + await hold(id) + await rig.stop() + await rig.settleAccepted(working, 'answer') + const fields = { command: 'clear' as const } + expect( + await rig.host.conversationCommand(QUEUED_RIG_CALLER, { + envelope: rig.envelope(fields, 'agentSession.conversationCommand', hostTestOperationId()), + ...fields + }) + ).toMatchObject({ ok: true }) + expect(held()).toEqual([id]) + expect(await update(id, fingerprint(id), 'after clear')).toMatchObject({ + value: { status: 'updated' } + }) + expect(published(id)?.body.blocks).toEqual([{ type: 'text', text: 'after clear' }]) + expect(publication().queuePause).toBeNull() + expect(journal().queuedMessages.get(id)?.state).toBe('waiting') + }) + + it('release never opens a closed conversation; closing drops every lease', async () => { + await rig.workingSend() + const id = await queue('base') + await hold(id) + const leases = journal().queuedMessages.editLeases + await rig.host.close(HOST_TEST_SESSION, 'evict') + expect(leases.heldIds([]).size).toBe(0) + expect(await hold(id, 'release')).toEqual({ status: 'released' }) + expect(rig.host.collaboratorsForTests().sessions.get(HOST_TEST_SESSION)).toBeUndefined() + }) +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-edit.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-edit.ts new file mode 100644 index 000000000000..d7cc959662e6 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-edit.ts @@ -0,0 +1,117 @@ +// Editing a queued card in place: Save is a compare-and-set on the card's body fingerprint, and +// an edit lease keeps automatic delivery from sending the card while someone types. The card keeps +// its id, position, sender and attachments; only its text changes. + +import type { + AgentSessionMutationEnvelope, + AgentSessionMutationResult, + AgentSessionQueuedMessageEditHoldParams, + AgentSessionQueuedMessageEditHoldResult, + AgentSessionQueuedMessageUpdateResult +} from '../../../shared/agent-session-wire' +import { queuedMessageEditableText } from '../../../shared/queued-message-text-edit' +import { queuedMessageTextUpdateDecision } from '../agent-session-journal/queued-message-text-update' +import type { StructuredAgentSessionMutationContext } from './structured-agent-session-mutation-context' +import type { StructuredAgentSessionCaller } from './structured-agent-session-host-types' +import type { MutationPlan } from './structured-agent-session-mutation-plans' +import { mutateQueued } from './structured-agent-session-queued-mutations' +import { queuedMessagesPublishedBytesRefusal } from './structured-agent-session-queued-published-bytes' + +/** Save. Editing sends nothing, so it pays no send admission, like Delete. A resend under the + * same id simply runs again: the compare-and-set answers it from the card as it stands. */ +export function updateQueuedStructuredAgentMessage( + context: StructuredAgentSessionMutationContext, + caller: StructuredAgentSessionCaller, + params: { + envelope: AgentSessionMutationEnvelope + messageId: string + expectedBodyFingerprint: string + text: string + } +): Promise> { + const { envelope, ...fields } = params + const plan: MutationPlan = { + method: 'agentSession.queuedMessageUpdate', + fields, + conversationWrite: true, + run: async (ctx) => { + const queued = ctx.journal.queuedMessages + const decision = queuedMessageTextUpdateDecision( + queued.get(fields.messageId), + ctx.sessionId, + fields + ) + if (decision.status !== 'ready') { + return { ok: true, value: decision } + } + // Editing an agent's card is still the person's action, so it may use their reserve. + const refusal = queuedMessagesPublishedBytesRefusal( + ctx.journal, + decision.body, + true, + fields.messageId + ) + return refusal ? { ok: false, refusal } : { ok: true, value: await queued.update(fields) } + }, + replay: () => null, + rerunWhenReplayMissing: () => true + } + return mutateQueued(context, caller, envelope, plan) +} + +/** Acquire, renew and release an edit lease. Ephemeral: no ledger row, no agent started, and a + * release never opens a conversation (a closed one already dropped its leases). */ +export function holdQueuedStructuredAgentMessageEdit( + context: StructuredAgentSessionMutationContext, + caller: StructuredAgentSessionCaller, + params: AgentSessionQueuedMessageEditHoldParams +): Promise { + const { sessionId, messageId } = params + const key = { callerKey: caller.callerKey, editId: params.editId } + return context.serialize(sessionId, async () => { + if (params.action === 'release') { + const journal = context.sessions.get(sessionId)?.journal + journal?.queuedMessages.editLeases.release(key, messageId) + if (journal) { + republish(context, sessionId, journal) + } + return { status: 'released' } + } + const session = await context.openConversation(sessionId) + if (!session) { + return { status: 'gone' } + } + const queued = session.journal.queuedMessages + if (params.action !== 'acquire') { + // Holds nothing new, so nothing to publish (a lapsed lease it prunes wakes the queue itself); + // and a publish is idle-sweep activity, which an editor left open must not renew forever. + return queued.editLeases.renew(key, messageId) + } + const row = queued.get(messageId) + const result: AgentSessionQueuedMessageEditHoldResult = + row && queuedMessageEditableText(row.body) === null + ? { status: 'not-editable' } + : queued.editLeases.acquire(key, { messageId, fingerprint: params.expectedBodyFingerprint }) + republish(context, sessionId, session.journal) + return result + }) +} + +/** Leases live outside the journal, so no commit announces them: publish the next card to send + * and wake the drain here. A failed publish is logged; the lease still stands. */ +export function republish( + context: StructuredAgentSessionMutationContext, + sessionId: string, + journal: Parameters[1] +): void { + try { + context.publish(sessionId, journal) + } catch (error) { + context.deps.logger.warn('publishing a queued edit hold failed', { + scope: 'queued-edit-hold', + sessionId, + error + }) + } + context.wakeQueuedDrain?.(sessionId) +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-messages.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-messages.ts index 249396d38b5d..ac85a5ed8bfb 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-messages.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-messages.ts @@ -22,7 +22,7 @@ import type { AgentSessionRecord } from '../../../shared/agent-session-record' import { isStructuredAgentSessionMainAgentWorking } from '../../../shared/structured-agent-session-main-agent-working' import type { AgentSessionJournal } from '../agent-session-journal/journal-store' import type { AgentSessionTurnContext } from './structured-agent-session-turns' -import { QueuedMessageNotConsumableError } from '../agent-session-journal/journal-queued-messages' +import { QueuedMessageNotConsumableError } from '../agent-session-journal/queued-message-consume-error' import type { QueuedMessageRow } from '../agent-session-journal/queued-message-table' import type { StructuredAgentSessionHostSession } from './structured-agent-session-host-types' import { @@ -57,16 +57,19 @@ export function pendingPromptExists(journal: Pick + journal: Pick, + automatic: boolean ): QueuedMessageRow | null { const rows = journal.queuedMessages.list() // Nothing waiting costs no pause derivation: this runs on every journal publish. if (!rows.some((row) => row.state === 'waiting')) { return null } - return nextSendableQueuedCard(structuredQueuePauses(journal), rows) + const edited = automatic ? journal.queuedMessages.editLeases.heldIds(rows) : undefined + return nextSendableQueuedCard(structuredQueuePauses(journal), rows, edited) } /** @@ -125,7 +128,7 @@ export function nextStructuredQueuedMessage(input: { record: AgentSessionRecord | null fence: number }): QueuedMessageRow | null { - const next = oldestActionableQueuedMessage(input.journal) + const next = oldestActionableQueuedMessage(input.journal, true) const { journal, fence } = input // The gate's cheap `working` first: publication asks on every streamed frame, and the gate's // prompt check walks the whole fold. @@ -160,7 +163,7 @@ export function shouldQueueStructuredAgentSessionSend(input: { if (hold !== null) { return true } - return oldestActionableQueuedMessage(input.journal) !== null + return oldestActionableQueuedMessage(input.journal, false) !== null } /** @@ -289,7 +292,7 @@ export class StructuredAgentSessionQueuedMessageDrain { try { if ( !journal.queuedMessages.settlementOwed() && - (oldestActionableQueuedMessage(journal) === null || + (oldestActionableQueuedMessage(journal, true) === null || isStructuredAgentSessionMainAgentWorking( journal.activeTurnId(), journal.submissions(), diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-mutations.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-mutations.ts index 28be32c15759..ab8a9384f091 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-mutations.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-mutations.ts @@ -15,7 +15,7 @@ import type { AgentSessionSendResult } from '../../../shared/agent-session-wire' import type { AgentSessionJournal } from '../agent-session-journal/journal-store' -import { QueuedMessageNotConsumableError } from '../agent-session-journal/journal-queued-messages' +import { QueuedMessageNotConsumableError } from '../agent-session-journal/queued-message-consume-error' import { isJournalWrittenByNewerOrca, journalOpenRefusal @@ -70,10 +70,9 @@ export async function withdrawQueuedMessagesForOperation( }) } - /** Draft actions run like any mutation: admitted on the session's lane, the * conversation opened for the write. */ -function mutateQueued( +export function mutateQueued( context: StructuredAgentSessionMutationContext, caller: StructuredAgentSessionCaller, envelope: AgentSessionMutationEnvelope, diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-published-bytes.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-published-bytes.ts index 4e1222b83429..e55450018906 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-published-bytes.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-published-bytes.ts @@ -20,14 +20,16 @@ export const QUEUED_MESSAGES_PUBLISHED_MAX_BYTES = export const QUEUED_MESSAGES_PERSON_RESERVE_BYTES = MAX_PROMPT_BYTES + 1024 /** Refused readably rather than trimmed, when the published cards would outgrow their frame room. - * Another agent's card must also leave the person's reserve free. */ + * Another agent's card must also leave the person's reserve free. `replacing` is an edit's own + * card, whose old body the new one takes the place of. */ export function queuedMessagesPublishedBytesRefusal( journal: AgentSessionJournal, body: AgentJournalMessageItem, - fromPerson: boolean + fromPerson: boolean, + replacing?: string ): AgentSessionWireRefusal | null { const bytes = unsettledQueuedMessages(journal).reduce( - (sum, row) => sum + publishedBodyBytes(row.body), + (sum, row) => (row.messageId === replacing ? sum : sum + publishedBodyBytes(row.body)), publishedBodyBytes(body) ) const limit = fromPerson diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-wiring.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-wiring.ts index a5ddfeb64748..fc8078ad649f 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-wiring.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-wiring.ts @@ -1,5 +1,5 @@ // Host wiring for mid-turn queueing: builds the serialized drain from the -// host's mutation context and exposes the draft actions (Send, Delete, Resume), so the host +// host's mutation context and exposes the draft actions (Send, Delete, Resume, Edit), so the host // class stays a description of its surface. import type { StructuredAgentSessionMutationContext } from './structured-agent-session-host-mutations' @@ -15,6 +15,11 @@ import { sendQueuedStructuredAgentMessage } from './structured-agent-session-queued-mutations' import { deferredStructuredAgentSessionLogger } from './structured-agent-session-logger' +import { + holdQueuedStructuredAgentMessageEdit, + republish, + updateQueuedStructuredAgentMessage +} from './structured-agent-session-queued-edit' /** `sessions` are the live conversations (their `touch` is the idle sweep's activity renewal, * which the drain's schedule rides); everything else comes from the host's mutation context, @@ -37,6 +42,34 @@ export function wireStructuredAgentSessionQueuedMessages( }) return { drain, + /** A conversation (re)opened: its edit leases wake the queue when a deadline passes. */ + onOpened: (sessionId: string) => { + const journal = sessions.get(sessionId)?.journal + journal?.queuedMessages.editLeases.onDeadline(() => { + void context() + .serialize(sessionId, async () => { + if (sessions.get(sessionId)?.journal === journal) { + republish(context(), sessionId, journal) + } + }) + .catch((error: unknown) => + context().deps.logger.warn('expiring a queued edit hold failed', { + scope: 'queued-edit-hold', + sessionId, + error + }) + ) + }) + drain.schedule(sessionId) + }, + queuedMessageUpdate: ( + caller: StructuredAgentSessionCaller, + params: Parameters[2] + ) => updateQueuedStructuredAgentMessage(context(), caller, params), + queuedMessageEditHold: ( + caller: StructuredAgentSessionCaller, + params: Parameters[2] + ) => holdQueuedStructuredAgentMessageEdit(context(), caller, params), /** Every journal publish: turn, submission, prompt, command and Stop * settlements are all commits, and each re-derives the drain's gates. */ onJournalActivity: (sessionId: string) => { diff --git a/src/main/runtime/mobile-rpc-allowlist.test.ts b/src/main/runtime/mobile-rpc-allowlist.test.ts index e761fd9e04c4..9ce687d486de 100644 --- a/src/main/runtime/mobile-rpc-allowlist.test.ts +++ b/src/main/runtime/mobile-rpc-allowlist.test.ts @@ -156,6 +156,8 @@ describe('mobile RPC allowlist', () => { 'agentSession.cancel', 'agentSession.queuedMessageSend', 'agentSession.queuedMessageDelete', + 'agentSession.queuedMessageUpdate', + 'agentSession.queuedMessageEditHold', 'agentSession.queuedMessagesResume', 'agentSession.close', 'agentSession.respondToApproval', diff --git a/src/main/runtime/rpc/methods/structured-agent-session-queued-edit.test.ts b/src/main/runtime/rpc/methods/structured-agent-session-queued-edit.test.ts new file mode 100644 index 000000000000..a2bb215e160e --- /dev/null +++ b/src/main/runtime/rpc/methods/structured-agent-session-queued-edit.test.ts @@ -0,0 +1,88 @@ +// The two in-place edit methods on the wire: registered and advertised apart from the dark +// mid-turn queue, the caller taken from the authenticated transport, strict params, and a lease +// release that cleanup admission always lets through. + +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { + AGENT_SESSION_QUEUED_MESSAGE_EDIT_RUNTIME_CAPABILITY, + AGENT_SESSION_QUEUED_MESSAGES_RUNTIME_CAPABILITY, + RUNTIME_CAPABILITIES +} from '../../../../shared/protocol-version' +import { + call, + clearStructuredHostStub, + envelope, + hostCalls, + installStructuredHostStub, + SESSION, + STRUCTURED_CLIENT +} from './structured-agent-session-rpc.test-fixture' +import { + QueuedMessageActionParams, + QueuedMessageEditHoldParams, + QueuedMessageUpdateParams +} from './structured-agent-session-schemas' + +const update = { + envelope: envelope(), + messageId: 'card', + expectedBodyFingerprint: 'a'.repeat(64), + text: 'updated' +} +const acquire = { + sessionId: SESSION, + messageId: 'card', + editId: 'edit', + action: 'acquire', + expectedBodyFingerprint: 'a'.repeat(64) +} +beforeEach(() => { + installStructuredHostStub() + hostCalls.queuedMessageUpdate = vi.fn(async () => ({ ok: true, value: { status: 'updated' } })) + hostCalls.queuedMessageEditHold = vi.fn(async () => ({ status: 'held' })) +}) +afterEach(clearStructuredHostStub) + +describe('queued edit methods', () => { + it('the host advertises editing on its own, with mid-turn queueing still dark', () => { + expect(RUNTIME_CAPABILITIES).toContain(AGENT_SESSION_QUEUED_MESSAGE_EDIT_RUNTIME_CAPABILITY) + expect(RUNTIME_CAPABILITIES).not.toContain(AGENT_SESSION_QUEUED_MESSAGES_RUNTIME_CAPABILITY) + }) + + it.each([ + ['agentSession.queuedMessageUpdate', update, 'queuedMessageUpdate'], + ['agentSession.queuedMessageEditHold', acquire, 'queuedMessageEditHold'] + ] as const)('%s reaches the host as the authenticated caller', async (method, params, name) => { + expect(await call(method, params, STRUCTURED_CLIENT)).toMatchObject({ ok: true }) + expect(hostCalls[name]).toHaveBeenCalledWith({ callerKey: 'trusted-local:runtime' }, params) + }) + + it('a release passes cleanup admission even for an audience that cannot start this agent', async () => { + hostCalls.sessionAgent = vi.fn(() => 'pi') + const release = { sessionId: SESSION, messageId: 'card', editId: 'edit', action: 'release' } + expect( + await call('agentSession.queuedMessageEditHold', acquire, STRUCTURED_CLIENT) + ).toMatchObject({ ok: false }) + expect( + await call('agentSession.queuedMessageEditHold', release, STRUCTURED_CLIENT) + ).toMatchObject({ ok: true }) + }) + + it('strict params refuse a forged caller, oversized text and unknown fields', () => { + expect(QueuedMessageUpdateParams.safeParse(update).success).toBe(true) + expect( + QueuedMessageUpdateParams.safeParse({ ...update, text: 'é'.repeat(256 * 1024) }).success + ).toBe(false) + expect(QueuedMessageUpdateParams.safeParse({ ...update, sentAs: 'goal' }).success).toBe(false) + expect(QueuedMessageEditHoldParams.safeParse(acquire).success).toBe(true) + expect( + QueuedMessageEditHoldParams.safeParse({ ...acquire, callerKey: 'another-client' }).success + ).toBe(false) + expect(QueuedMessageEditHoldParams.safeParse({ ...acquire, action: 'renew' }).success).toBe( + false + ) + expect( + QueuedMessageActionParams.safeParse({ envelope: envelope(), messageId: 'card' }).success + ).toBe(true) + }) +}) diff --git a/src/main/runtime/rpc/methods/structured-agent-session-queued-methods.ts b/src/main/runtime/rpc/methods/structured-agent-session-queued-methods.ts index 5689330e3c3b..821bfd6f1717 100644 --- a/src/main/runtime/rpc/methods/structured-agent-session-queued-methods.ts +++ b/src/main/runtime/rpc/methods/structured-agent-session-queued-methods.ts @@ -1,18 +1,41 @@ -// The queued-message actions: Send-now and Delete on one card, and Resume on a -// paused queue. Not gated on agent-session.queued-messages.v1: a host without it -// still publishes the cards it kept unsent. A host older than the queue lacks them. +// The queued-message actions: Send-now, Delete and in-place Edit on one card, and Resume on a +// paused queue. Not gated on agent-session.queued-messages.v1: a host without it still publishes +// the cards it kept unsent. A host older than the queue, or than edit-v1, lacks them. import { defineMethod } from '../core' import { requireStructuredSessionHost as requireSessionHost, + requireStructuredCleanupHost, structuredCallerFor as callerFor } from './structured-agent-session-gate' import { QueuedMessageActionParams, + QueuedMessageUpdateParams, + QueuedMessageEditHoldParams, QueuedMessagesResumeParams } from './structured-agent-session-schemas' export const STRUCTURED_AGENT_SESSION_QUEUED_METHODS = [ + defineMethod({ + name: 'agentSession.queuedMessageUpdate', + permission: 'workspace', + params: QueuedMessageUpdateParams, + handler: async (params, ctx) => + requireSessionHost(ctx, params.envelope.sessionId).queuedMessageUpdate(callerFor(ctx), params) + }), + defineMethod({ + name: 'agentSession.queuedMessageEditHold', + permission: 'workspace', + params: QueuedMessageEditHoldParams, + handler: async (params, ctx) => { + // Releasing a lease is cleanup: no admission condition may strand it until its deadline. + const host = + params.action === 'release' + ? requireStructuredCleanupHost(ctx) + : requireSessionHost(ctx, params.sessionId) + return host.queuedMessageEditHold(callerFor(ctx), params) + } + }), defineMethod({ name: 'agentSession.queuedMessageSend', permission: 'workspace', diff --git a/src/main/runtime/rpc/methods/structured-agent-session-schemas.ts b/src/main/runtime/rpc/methods/structured-agent-session-schemas.ts index 1f749d50490d..76ce9c1b8249 100644 --- a/src/main/runtime/rpc/methods/structured-agent-session-schemas.ts +++ b/src/main/runtime/rpc/methods/structured-agent-session-schemas.ts @@ -34,3 +34,7 @@ export { UnsubscribeParams } from '../../../../shared/rpc-contract/structured-agent-session-params' export { ContinueInterruptedParams } from '../../../../shared/rpc-contract/structured-agent-session-continue-params' +export { + QueuedMessageUpdateParams, + QueuedMessageEditHoldParams +} from '../../../../shared/rpc-contract/structured-agent-session-queued-edit-params' diff --git a/src/main/runtime/rpc/methods/structured-agent-session.test.ts b/src/main/runtime/rpc/methods/structured-agent-session.test.ts index e287472bb6ce..f67dd8ed74fb 100644 --- a/src/main/runtime/rpc/methods/structured-agent-session.test.ts +++ b/src/main/runtime/rpc/methods/structured-agent-session.test.ts @@ -175,7 +175,7 @@ describe('capability gating', () => { } // Bump deliberately: the whole agentSession.* surface is behind the structured capability, // so an additive method is invisible to old clients and needs no protocol bump. - expect(STRUCTURED_AGENT_SESSION_METHODS).toHaveLength(34) + expect(STRUCTURED_AGENT_SESSION_METHODS).toHaveLength(36) }) it('hides the surface from a declared client that did not advertise it', async () => { diff --git a/src/main/runtime/runtime-rpc/runtime-rpc-mobile-agent-session-methods.ts b/src/main/runtime/runtime-rpc/runtime-rpc-mobile-agent-session-methods.ts index 56b30d837a5d..6ceb45937d8f 100644 --- a/src/main/runtime/runtime-rpc/runtime-rpc-mobile-agent-session-methods.ts +++ b/src/main/runtime/runtime-rpc/runtime-rpc-mobile-agent-session-methods.ts @@ -8,6 +8,8 @@ export const MOBILE_AGENT_SESSION_RPC_METHODS = [ 'agentSession.cancel', 'agentSession.queuedMessageSend', 'agentSession.queuedMessageDelete', + 'agentSession.queuedMessageUpdate', + 'agentSession.queuedMessageEditHold', 'agentSession.queuedMessagesResume', 'agentSession.close', 'agentSession.respondToApproval', diff --git a/src/renderer/src/components/ButtonKeyHint.test.tsx b/src/renderer/src/components/ButtonKeyHint.test.tsx new file mode 100644 index 000000000000..669951688e41 --- /dev/null +++ b/src/renderer/src/components/ButtonKeyHint.test.tsx @@ -0,0 +1,27 @@ +// @vitest-environment happy-dom +import { cleanup, render, screen } from '@testing-library/react' +import { afterEach, describe, expect, it } from 'vitest' +import { Button } from '@/components/ui/button' +import { ButtonKeyHint } from './ButtonKeyHint' + +afterEach(cleanup) + +describe('ButtonKeyHint', () => { + it('shows the modifier before the Enter glyph without joining the button name', () => { + render( + + ) + const button = screen.getByRole('button', { name: 'Create worktree' }) + const badge = button.querySelector('[aria-hidden="true"]') + expect(badge?.textContent).toBe('⌘') + expect(badge?.querySelector('svg')).toBeTruthy() + }) + it('shows only the Enter glyph when plain Enter submits', () => { + const { container } = render() + expect(container.firstElementChild?.textContent).toBe('') + expect(container.querySelector('svg')).toBeTruthy() + }) +}) diff --git a/src/renderer/src/components/ButtonKeyHint.tsx b/src/renderer/src/components/ButtonKeyHint.tsx new file mode 100644 index 000000000000..23dc2ac44c2a --- /dev/null +++ b/src/renderer/src/components/ButtonKeyHint.tsx @@ -0,0 +1,21 @@ +import React from 'react' +import { CornerDownLeft } from 'lucide-react' + +/** The Enter-key badge that trails a filled primary button's label ("Create worktree ⌘↵"). + * Decorative: the button's name is its label, not the glyphs. */ +export function ButtonKeyHint({ + modifierLabel +}: { + /** The platform modifier shown before ↵ (e.g. "⌘" or "Ctrl"); omit when plain Enter submits. */ + modifierLabel?: string +}): React.JSX.Element { + return ( + + {modifierLabel ? {modifierLabel} : null} + + + ) +} diff --git a/src/renderer/src/components/native-chat/NativeChatQueuedMessageCard.tsx b/src/renderer/src/components/native-chat/NativeChatQueuedMessageCard.tsx index 9201e87ef459..c372efbe93dc 100644 --- a/src/renderer/src/components/native-chat/NativeChatQueuedMessageCard.tsx +++ b/src/renderer/src/components/native-chat/NativeChatQueuedMessageCard.tsx @@ -35,6 +35,8 @@ import { useNativeChatClippedLine } from './use-native-chat-clipped-line' import { agentSessionFailureStatedByStartRow } from './structured-agent-session-delivery-notices' import type { AgentSessionFailureFact } from '../../../../shared/agent-session-failure' import { agentSessionWriteNotDoneParts } from '../../../../shared/agent-session-refusal-notice' +import { NativeChatQueuedMessageEditor } from './NativeChatQueuedMessageEditor' +import type { QueuedMessageInlineEditor } from './use-structured-agent-session-queued-edit' /** The visible caption under the text; the default waiting hold needs none. */ export function queuedMessageCardCaption( @@ -138,6 +140,8 @@ export function NativeChatQueuedMessageCard({ onSteer, onDelete, onEdit, + editor, + editDisabled = false, onTurnOffQueueing }: { card: QueuedMessageCard @@ -151,7 +155,12 @@ export function NativeChatQueuedMessageCard({ steerHeld?: boolean onSteer: () => void onDelete: () => void - onEdit: () => void + /** Absent when the host cannot edit in place: the card then offers no Edit at all. */ + onEdit?: () => void + /** This card is being edited here: the editor replaces its text and its actions. */ + editor?: QueuedMessageInlineEditor + /** Another card in this pane is being edited. */ + editDisabled?: boolean /** Absent when the host does not queue sends, so there is nothing to turn off. */ onTurnOffQueueing?: () => void }): React.JSX.Element { @@ -164,6 +173,21 @@ export function NativeChatQueuedMessageCard({ const [expanded, setExpanded] = useState(false) const [clipped, measureLine] = useNativeChatClippedLine(false) const textId = useId() + const sender = card.from ? ( + + ) : null + if (editor) { + // The editor's own header stands in for the icon, the actions and the caption. + return ( +
  • + +
    • + ) + } return (
  • )}
    - {card.from ? ( - - ) : null} + {sender} {expanded ? null : (

    {card.text} @@ -256,38 +274,43 @@ export function NativeChatQueuedMessageCard({ {translate('components.native-chat.queuedMessages.delete', 'Delete')} - - - - - - {/* A command's text is not a draft: edited, it would become a message. */} - {card.command ? null : ( - - - {translate('components.native-chat.queuedMessages.editMessage', 'Edit message')} - - )} - {onTurnOffQueueing ? ( - - {translate( - 'components.native-chat.queuedMessages.turnOffQueueing', - 'Turn off queueing' + {(!card.command && onEdit) || onTurnOffQueueing ? ( + + + + + + {/* A command's text is not a draft: edited, it would become a message. */} + {card.command || !onEdit ? null : ( + + + {translate( + 'components.native-chat.queuedMessages.editMessage', + 'Edit message' + )} + + )} + {onTurnOffQueueing ? ( + + {translate( + 'components.native-chat.queuedMessages.turnOffQueueing', + 'Turn off queueing' + )} + + ) : null} + + + ) : null} )}

    diff --git a/src/renderer/src/components/native-chat/NativeChatQueuedMessageEditor.test.tsx b/src/renderer/src/components/native-chat/NativeChatQueuedMessageEditor.test.tsx new file mode 100644 index 000000000000..5afab04acf62 --- /dev/null +++ b/src/renderer/src/components/native-chat/NativeChatQueuedMessageEditor.test.tsx @@ -0,0 +1,192 @@ +// @vitest-environment happy-dom +import { cleanup, fireEvent, render, screen } from '@testing-library/react' +import { afterEach, describe, expect, it, vi } from 'vitest' +import { NativeChatQueuedMessageEditor } from './NativeChatQueuedMessageEditor' +import { NativeChatApprovalCard } from './NativeChatApprovalCard' +import { NativeChatQuestionCard } from './NativeChatQuestionCard' +import type { QueuedMessageInlineEditor } from './use-structured-agent-session-queued-edit' + +vi.mock('@/i18n/i18n', () => ({ + translate: (_key: string, fallback: string, values?: Record) => + fallback.replace(/\{\{(\w+)\}\}/g, (_match, name: string) => values?.[name] ?? '') +})) +afterEach(cleanup) +function editor(): QueuedMessageInlineEditor { + return { + messageId: 'mail', + text: 'full message', + acquiring: false, + saving: false, + canSave: true, + change: vi.fn(), + save: vi.fn(), + cancel: vi.fn() + } +} +function view( + edit: QueuedMessageInlineEditor, + prompt: 'question' | 'approval' | null, + answer = vi.fn(), + promptCancel = vi.fn() +) { + return ( +
    +
      +
    • + +
      • +
    + {prompt === 'approval' ? ( + + ) : null} + {prompt === 'question' ? ( + + ) : null} +
    + ) +} +describe('queued card textarea', () => { + it.each(['question', 'approval'] as const)( + 'stays mounted and focused when a %s opens, its keys apart from the prompt', + (prompt) => { + const edit = editor() + const answer = vi.fn() + const promptCancel = vi.fn() + const { rerender } = render(view(edit, null, answer, promptCancel)) + const field = screen.getByRole('textbox', { name: 'Edit message' }) + expect(document.activeElement).toBe(field) + rerender(view(edit, prompt, answer, promptCancel)) + expect(document.activeElement).toBe(field) + fireEvent.keyDown(field, { key: 'Enter' }) + expect(edit.save).toHaveBeenCalledOnce() + expect(answer).not.toHaveBeenCalled() + fireEvent.keyDown(field, { key: 'Escape' }) + expect(edit.cancel).toHaveBeenCalledOnce() + expect(promptCancel).not.toHaveBeenCalled() + } + ) + it.each(['question', 'approval'] as const)( + 'focuses and saves with a %s already open', + (prompt) => { + const edit = editor() + render(view(edit, prompt)) + const field = screen.getByRole('textbox', { name: 'Edit message' }) + expect(document.activeElement).toBe(field) + fireEvent.click(screen.getByRole('button', { name: 'Save' })) + expect(edit.save).toHaveBeenCalledOnce() + } + ) + it('Enter and the platform modifier save, Shift+Enter is a newline and handled keys do not bubble', () => { + const edit = editor() + const parent = vi.fn() + render( +
    + +
    + ) + const field = screen.getByRole('textbox') + fireEvent.keyDown(field, { key: 'Enter' }) + expect(edit.save).toHaveBeenCalledOnce() + expect(parent).not.toHaveBeenCalled() + fireEvent.keyDown(field, { key: 'Enter', shiftKey: true }) + expect(edit.save).toHaveBeenCalledOnce() + for (const [platform, modifier] of [ + ['Macintosh', { metaKey: true }], + ['Linux', { ctrlKey: true }], + ['Windows', { ctrlKey: true }] + ] as const) { + vi.spyOn(navigator, 'userAgent', 'get').mockReturnValue(platform) + fireEvent.keyDown(field, { key: 'Enter', ...modifier }) + } + expect(edit.save).toHaveBeenCalledTimes(4) + vi.restoreAllMocks() + }) + it('IME confirmation and Escape never commit/cancel or reach the prompt', () => { + const edit = editor() + const parent = vi.fn() + render( +
    + +
    + ) + const field = screen.getByRole('textbox') + fireEvent.compositionStart(field) + fireEvent.keyDown(field, { key: 'Enter', isComposing: true }) + fireEvent.keyDown(field, { key: 'Escape', isComposing: true }) + fireEvent.compositionEnd(field) + fireEvent.keyDown(field, { key: 'Enter', keyCode: 229 }) + expect(edit.save).not.toHaveBeenCalled() + expect(edit.cancel).not.toHaveBeenCalled() + expect(parent).not.toHaveBeenCalled() + }) + it('acquisition seeds a read-only field with the X always usable', () => { + const edit = { ...editor(), acquiring: true } + render() + expect(screen.getByRole('textbox').getAttribute('readonly')).not.toBeNull() + expect(screen.getByRole('button', { name: 'Save' }).hasAttribute('disabled')).toBe(true) + fireEvent.click(screen.getByRole('button', { name: 'Cancel editing' })) + expect(edit.cancel).toHaveBeenCalledOnce() + }) + it('the header X cancels and saves nothing; Save is the only footer action', () => { + const edit = editor() + render() + expect(screen.getByText('Editing message')).toBeTruthy() + fireEvent.click(screen.getByRole('button', { name: 'Cancel editing' })) + expect(edit.cancel).toHaveBeenCalledOnce() + expect(edit.save).not.toHaveBeenCalled() + expect(screen.queryByRole('button', { name: 'Cancel' })).toBeNull() + expect( + screen + .getAllByRole('button') + .map((button) => button.getAttribute('aria-label') ?? button.textContent) + ).toEqual(['Cancel editing', 'Save']) + }) + it.each([ + { state: 'saving', edit: { saving: true } }, + { state: 'unchanged', edit: { canSave: false } } + ])('Save is disabled while $state and enabled otherwise', ({ edit: overrides }) => { + const edit = { ...editor(), ...overrides } + const { rerender } = render() + const save = screen.getByRole('button', { name: 'Save' }) + expect(save.hasAttribute('disabled')).toBe(true) + fireEvent.click(save) + expect(edit.save).not.toHaveBeenCalled() + rerender() + fireEvent.click(screen.getByRole('button', { name: 'Save' })) + expect(edit.save).toHaveBeenCalledOnce() + }) + it('Save carries a decorative Enter badge and no key hint text', () => { + for (const platform of ['Macintosh', 'Windows']) { + vi.spyOn(navigator, 'userAgent', 'get').mockReturnValue(platform) + const { container, unmount } = render() + const save = screen.getByRole('button', { name: 'Save' }) + const badge = save.querySelector('[aria-hidden="true"]') + expect(badge?.querySelector('svg')).toBeTruthy() + expect(badge?.textContent).toBe('') + expect(container.textContent).not.toMatch(/Shift|⇧|new line/) + unmount() + } + vi.restoreAllMocks() + }) +}) diff --git a/src/renderer/src/components/native-chat/NativeChatQueuedMessageEditor.tsx b/src/renderer/src/components/native-chat/NativeChatQueuedMessageEditor.tsx new file mode 100644 index 000000000000..71754511ed02 --- /dev/null +++ b/src/renderer/src/components/native-chat/NativeChatQueuedMessageEditor.tsx @@ -0,0 +1,109 @@ +import { useLayoutEffect, useRef, type ReactNode } from 'react' +import { Pencil, X } from 'lucide-react' +import { Textarea } from '@/components/ui/textarea' +import { Button } from '@/components/ui/button' +import { translate } from '@/i18n/i18n' +import { ButtonKeyHint } from '@/components/ButtonKeyHint' +import { isMacPlatform } from './native-chat-shortcut' +import type { QueuedMessageInlineEditor } from './use-structured-agent-session-queued-edit' + +/** The inline editor that replaces a queued card's row while it is edited. It lives in the card, + * not the composer, so it stays usable while a question or approval holds the composer's slot. */ +export function NativeChatQueuedMessageEditor({ + editor, + sender +}: { + editor: QueuedMessageInlineEditor + /** The card's "From " line, kept beside the label. */ + sender?: ReactNode +}): React.JSX.Element { + const field = useRef(null) + useLayoutEffect(() => { + if (editor.acquiring) { + return + } + const textarea = field.current + textarea?.focus({ preventScroll: true }) + const list = textarea?.closest('ul') + const row = textarea?.closest('li') + if (list && row) { + const rowBounds = row.getBoundingClientRect() + const listBounds = list.getBoundingClientRect() + if (rowBounds.top < listBounds.top) { + list.scrollTop += rowBounds.top - listBounds.top + } else if (rowBounds.bottom > listBounds.bottom) { + list.scrollTop += rowBounds.bottom - listBounds.bottom + } + } + }, [editor.acquiring]) + return ( +
    +
    + + + {translate('components.native-chat.queuedMessages.editingMessage', 'Editing message')} + + {sender ? ( + <> + · +
    {sender}
    + + ) : null} + +
    +