From da06bcfb780beb0e9edb49d769210505b9e22cc7 Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Thu, 8 Oct 2026 19:59:05 -0700 Subject: [PATCH 01/12] feat(native-chat): edit a queued message in place on the host A queued card's Edit used to copy its text into the composer and delete the card, so the message lost its place, an agent-mail card lost its sender, and while a question or approval replaced the composer the text vanished. The execution host now edits the card itself: - agentSession.queuedMessageUpdate replaces only the card's text, as a compare-and-set on the body fingerprint the client last saw. It keeps the id, position, sender, kind, send mode and attachments; a command card is not editable. A resent Save answers from the card as it stands (unchanged or changed), so a lost answer never overwrites newer text. - agentSession.queuedMessageEditHold acquires, renews and releases an in-memory edit lease per caller and edit. While one is live, automatic delivery stops at the card (earlier cards still send, later ones wait); an explicit Send still overrides. Leases expire after two minutes on their own, re-derive validity from the card on every read, and die with the conversation. Release uses cleanup admission. - The published card carries an optional editHeld marker, never `paused`. - Both methods sit behind the new agent-session.queued-message-edit.v1 capability, independent of the dark mid-turn queue capability, and are on the mobile allowlist for the phone follow-up. --- .../journal-queued-messages.ts | 26 +- .../agent-session-journal/journal-store.ts | 1 + .../queued-message-edit-leases.test.ts | 145 +++++++++ .../queued-message-edit-leases.ts | 172 ++++++++++ .../queued-message-not-consumable.ts | 11 + .../queued-message-pause.test.ts | 2 +- .../queued-message-pause.ts | 13 +- .../queued-message-store.test.ts | 7 +- .../queued-message-text-update.test.ts | 149 +++++++++ .../queued-message-text-update.ts | 66 ++++ .../structured-agent-session-host.ts | 4 +- ...ructured-agent-session-queued-edit.test.ts | 307 ++++++++++++++++++ .../structured-agent-session-queued-edit.ts | 123 +++++++ ...tructured-agent-session-queued-messages.ts | 17 +- ...ructured-agent-session-queued-mutations.ts | 4 +- ...ctured-agent-session-queued-publication.ts | 15 +- ...ed-agent-session-queued-published-bytes.ts | 8 +- .../structured-agent-session-queued-wiring.ts | 36 +- src/main/runtime/mobile-rpc-allowlist.test.ts | 2 + ...ructured-agent-session-queued-edit.test.ts | 88 +++++ ...structured-agent-session-queued-methods.ts | 29 +- .../structured-agent-session-schemas.ts | 4 + ...untime-rpc-mobile-agent-session-methods.ts | 2 + .../agent-session-queued-message-wire.ts | 23 ++ src/shared/protocol-version.ts | 7 + src/shared/queued-message-text-edit.ts | 40 +++ .../rpc-params-catalog.generated.ts | 6 + ...ctured-agent-session-queued-edit-params.ts | 27 ++ 28 files changed, 1290 insertions(+), 44 deletions(-) create mode 100644 src/main/native-chat/agent-session-journal/queued-message-edit-leases.test.ts create mode 100644 src/main/native-chat/agent-session-journal/queued-message-edit-leases.ts create mode 100644 src/main/native-chat/agent-session-journal/queued-message-not-consumable.ts create mode 100644 src/main/native-chat/agent-session-journal/queued-message-text-update.test.ts create mode 100644 src/main/native-chat/agent-session-journal/queued-message-text-update.ts create mode 100644 src/main/native-chat/agent-session-wire/structured-agent-session-queued-edit.test.ts create mode 100644 src/main/native-chat/agent-session-wire/structured-agent-session-queued-edit.ts create mode 100644 src/main/runtime/rpc/methods/structured-agent-session-queued-edit.test.ts create mode 100644 src/shared/queued-message-text-edit.ts create mode 100644 src/shared/rpc-contract/structured-agent-session-queued-edit-params.ts 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 5b0e2580c45e..dacb186bac11 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 @@ -47,6 +47,9 @@ import { import { AgentSessionJournalError, assertJournalWritable } from './journal-write-guards' import type { JournalAttachmentClaim } from './journal-submission-hook' import type { JournalWriteBody, JournalWriteResult } from './journal-write-queue' +import { QueuedMessageEditLeases } from './queued-message-edit-leases' +import { QueuedMessageNotConsumableError } from './queued-message-not-consumable' +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 = @@ -71,6 +74,7 @@ export type JournalQueuedMessagesDeps = { } 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 @@ -106,6 +110,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) @@ -290,7 +303,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) } } @@ -369,13 +383,3 @@ export class JournalQueuedMessages { ).then(() => undefined) } } - -export class QueuedMessageNotConsumableError extends Error { - constructor( - readonly messageId: string, - readonly expected: 'waiting' | 'returned' - ) { - super(`queued message ${messageId} is no longer ${expected}`) - this.name = 'QueuedMessageNotConsumableError' - } -} 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 06bdfba25fea..a65c382b1135 100644 --- a/src/main/native-chat/agent-session-journal/journal-store.ts +++ b/src/main/native-chat/agent-session-journal/journal-store.ts @@ -172,6 +172,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/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..e712ef18a35e --- /dev/null +++ b/src/main/native-chat/agent-session-journal/queued-message-edit-leases.test.ts @@ -0,0 +1,145 @@ +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('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..d25b749fdc3d --- /dev/null +++ b/src/main/native-chat/agent-session-journal/queued-message-edit-leases.ts @@ -0,0 +1,172 @@ +// 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() + for (const [key, lease] of this.leases) { + const row = byId.get(lease.messageId) + if (lease.until <= now || !row || !leaseMatchesRow(lease, row)) { + this.leases.delete(key) + } + } + this.schedule() + 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()) + ) + this.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-not-consumable.ts b/src/main/native-chat/agent-session-journal/queued-message-not-consumable.ts new file mode 100644 index 000000000000..86ae3dbc41a6 --- /dev/null +++ b/src/main/native-chat/agent-session-journal/queued-message-not-consumable.ts @@ -0,0 +1,11 @@ +/** A consume's compare-and-set lost: the card is no longer in the state the caller read, so the + * whole append it rode in rolls back. */ +export class QueuedMessageNotConsumableError extends Error { + constructor( + readonly messageId: string, + readonly expected: 'waiting' | 'returned' + ) { + super(`queued message ${messageId} is no longer ${expected}`) + this.name = 'QueuedMessageNotConsumableError' + } +} 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 dabc1d64335d..ea649c17809f 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 @@ -17,7 +17,7 @@ import { liveTestJournalRows, updateTestJournalRowJson } from './journal-host-database-test-support' -import { QueuedMessageNotConsumableError } from './journal-queued-messages' +import { QueuedMessageNotConsumableError } from './queued-message-not-consumable' 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 a7a652d81b7c..a561782cd8b0 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 @@ -217,19 +217,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-store.test.ts b/src/main/native-chat/agent-session-journal/queued-message-store.test.ts index 02e07bc53ee6..b63cac121fee 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-not-consumable' import type { AgentSessionJournal } from './journal-store' import type { AgentMessageSource } from '../../../shared/agent-session-message-source' import { 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..e9b57abf158d --- /dev/null +++ b/src/main/native-chat/agent-session-journal/queued-message-text-update.test.ts @@ -0,0 +1,149 @@ +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', + carriedFrom: 'old-folder-session', + 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('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, inserting text before it', () => { + const before = insert({ ...initial, blocks: [{ 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' } + ]) + }) + 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 1a08095a82ec..f9ab7df5c20d 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..82e7bca7bd9a --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-edit.test.ts @@ -0,0 +1,307 @@ +// 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) +} + +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(published(edited)?.editHeld).toBe(true) + 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(published(id)?.editHeld).toBeUndefined() + 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", async () => { + await rig.workingSend() + const id = await queue('base') + await hold(id) + await hold(id, 'acquire', 'edit-1', { callerKey: 'phone' }) + const before = publication() + expect(await hold(id, 'renew')).toMatchObject({ status: 'held' }) + expect(publication()).toBe(before) + expect(await hold(id, 'release')).toEqual({ status: 'released' }) + expect(published(id)?.editHeld).toBe(true) + await hold(id, 'release', 'edit-1', { callerKey: 'phone' }) + expect(published(id)?.editHeld).toBeUndefined() + }) + + 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(published(id)).toMatchObject({ editHeld: true }) + expect(published(id)?.paused).toBeUndefined() + expect(publication().nextQueuedMessageId).toBeNull() + }) + + 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..6a17d8851f7f --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-edit.ts @@ -0,0 +1,123 @@ +// 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, + touch: (sessionId: string) => void +): 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' } + } + // An open editor is activity: the idle sweep must not close the conversation under it. + touch(sessionId) + const queued = session.journal.queuedMessages + let result: AgentSessionQueuedMessageEditHoldResult + if (params.action === 'acquire') { + const row = queued.get(messageId) + result = + row && queuedMessageEditableText(row.body) === null + ? { status: 'not-editable' } + : queued.editLeases.acquire(key, { + messageId, + fingerprint: params.expectedBodyFingerprint + }) + } else { + result = queued.editLeases.renew(key, messageId) + } + republish(context, sessionId, session.journal) + return result + }) +} + +/** Leases live outside the journal, so no commit announces them: publish the held marker 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..212924851a38 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-not-consumable' 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 f57723ed61c7..136a7dcc15d7 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-not-consumable' import { isJournalWrittenByNewerOrca, journalOpenRefusal @@ -141,7 +141,7 @@ export async function carryQueuedMessagesToClearReplacement( /** 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-publication.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-publication.ts index d48d9bf27c39..fd8c2d553fcd 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-publication.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-publication.ts @@ -42,8 +42,12 @@ export function structuredQueueSendGate( } /** Waiting and returned rows only. `paused` is a per-card hold (a failed conversion); a person's - * Stop pauses the queue, published once beside it. */ -function computePublishedQueuedMessages(journal: AgentSessionJournal): AgentSessionQueuedMessage[] { + * Stop pauses the queue, published once beside it. `editHeld` is never `paused`: an older client + * would caption it with a hold whose release it does not own. */ +function computePublishedQueuedMessages( + journal: AgentSessionJournal, + edited: ReadonlySet +): AgentSessionQueuedMessage[] { const published: AgentSessionQueuedMessage[] = [] for (const row of journal.queuedMessages.list()) { if (row.state !== 'waiting' && row.state !== 'returned') { @@ -55,6 +59,7 @@ function computePublishedQueuedMessages(journal: AgentSessionJournal): AgentSess position: row.position, body: row.body, state: row.state, + ...(edited.has(row.messageId) ? { editHeld: true as const } : {}), ...(held ? { paused: true as const } : {}), // The stored reason is a typed marker; an unknown one reads as a plain hold. ...(held && isPublishedPausedReason(row.holdReason) ? { pausedReason: row.holdReason } : {}), @@ -82,12 +87,14 @@ const listMemos = new WeakMap() const publications = new WeakMap() function readPublishedQueuedMessages(journal: AgentSessionJournal): AgentSessionQueuedMessage[] { - const key = String(journal.queuedMessages.revision()) + // A renewal moves no held set, so it re-sends nothing. + const edited = journal.queuedMessages.editLeases.heldIds() + const key = JSON.stringify([journal.queuedMessages.revision(), ...edited]) const memo = listMemos.get(journal) if (memo && memo.key === key) { return memo.list } - const list = computePublishedQueuedMessages(journal) + const list = computePublishedQueuedMessages(journal, edited) // Belt for the identity dedup: equal recomputed content keeps the previous reference. const serialized = JSON.stringify(list) if (memo && memo.serialized === serialized) { 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..bf7a61823268 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,35 @@ 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, (id) => sessions.touch(id)), /** 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/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/shared/agent-session-queued-message-wire.ts b/src/shared/agent-session-queued-message-wire.ts index d10d263aa024..ac80de57c6ce 100644 --- a/src/shared/agent-session-queued-message-wire.ts +++ b/src/shared/agent-session-queued-message-wire.ts @@ -40,6 +40,9 @@ export type AgentSessionQueuedMessage = { state: 'waiting' | 'returned' /** This one card is held, whatever the queue's pause: its conversion failed. */ paused?: true + /** Someone is editing this card in place: automatic delivery waits at it until the edit ends + * or its lease expires. Absent on older hosts, and never paired with `paused`. */ + editHeld?: true /** Why it is held, as a marker the client localizes: 'send_failed' ("couldn't send"); only an * explicit Send releases it. A client must treat an unknown marker as a plain hold, so a newer * host can add one. The queue-level pause is `queuePause`, published beside the list. */ @@ -59,3 +62,23 @@ export type AgentSessionQueuedMessageDeleteResult = /** `dispatched` means it already became a submission; `missing` covers a * pruned tombstone. Replays answer from tombstone receipts. */ | { deleted: false; messageId: string; disposition: 'dispatched' | 'withdrawn' | 'missing' } + +/** `agentSession.queuedMessageUpdate` (`agent-session.queued-message-edit.v1`). `unchanged`: the + * card already reads as asked, which is also how a resent Save whose answer was lost succeeds. + * `changed`: someone else's edit landed first; nothing was overwritten. */ +export type AgentSessionQueuedMessageUpdateResult = + | { status: 'updated' | 'unchanged'; messageId: string; fingerprint: string } + | { status: 'changed' | 'not-editable'; messageId: string } + | { status: 'gone'; messageId: string; disposition: 'dispatched' | 'withdrawn' | 'missing' } + +/** `agentSession.queuedMessageEditHold`: one editor's lease, keyed by the authenticated caller and + * `editId`. It expires on its own; renewing extends it, and release frees only this editor's. */ +export type AgentSessionQueuedMessageEditHoldParams = { + sessionId: string + messageId: string + editId: string +} & ({ action: 'acquire'; expectedBodyFingerprint: string } | { action: 'renew' | 'release' }) + +export type AgentSessionQueuedMessageEditHoldResult = + | { status: 'held'; fingerprint: string; leaseDurationMs: number; remainingMs: number } + | { status: 'released' | 'expired' | 'changed' | 'gone' | 'not-editable' } diff --git a/src/shared/protocol-version.ts b/src/shared/protocol-version.ts index 95da4aba339d..c78241c8c773 100644 --- a/src/shared/protocol-version.ts +++ b/src/shared/protocol-version.ts @@ -250,6 +250,12 @@ export const AGENT_SESSION_SEND_ANSWERS_PROOF_RUNTIME_CAPABILITY = // gates the rollout. export const AGENT_SESSION_QUEUED_MESSAGES_RUNTIME_CAPABILITY = 'agent-session.queued-messages.v1' as const +// Why: a host advertising this edits a queued card in place (`agentSession.queuedMessageUpdate`) +// and holds it from automatic delivery while edited (`agentSession.queuedMessageEditHold`). Apart +// from queued-messages.v1: kept and agent-mail cards are editable with mid-turn queueing still off. +// A client offers Edit only when the host advertises it; there is no copy-and-delete fallback. +export const AGENT_SESSION_QUEUED_MESSAGE_EDIT_RUNTIME_CAPABILITY = + 'agent-session.queued-message-edit.v1' as const // Why: `agentSession.conversationCommand`'s params are strict, so an older host rejects // `delivery`. A host advertising this holds a /compact sent while the agent works as a queued // card instead of refusing it. Clients ask only when queued-messages.v1 is advertised too: @@ -394,6 +400,7 @@ export const RUNTIME_CAPABILITIES = [ AGENT_SESSION_ACCEPTED_SEND_RUNTIME_CAPABILITY, AGENT_SESSION_SEND_ANSWERS_PROOF_RUNTIME_CAPABILITY, AGENT_SESSION_QUEUED_COMMANDS_RUNTIME_CAPABILITY, + AGENT_SESSION_QUEUED_MESSAGE_EDIT_RUNTIME_CAPABILITY, AGENT_SESSION_CONTINUE_INTERRUPTED_RUNTIME_CAPABILITY, ...STRUCTURED_AGENT_SESSION_SURFACE_RUNTIME_CAPABILITIES, ...AGENT_SESSION_ATTENTION_RUNTIME_CAPABILITIES, diff --git a/src/shared/queued-message-text-edit.ts b/src/shared/queued-message-text-edit.ts new file mode 100644 index 000000000000..219232d24909 --- /dev/null +++ b/src/shared/queued-message-text-edit.ts @@ -0,0 +1,40 @@ +import type { AgentJournalMessageItem } from './agent-session-journal-types' +import { MAX_PROMPT_BYTES } from './rpc-contract/structured-agent-session-params' + +/** The text an in-place edit starts from, or null when the card is not editable: a command is + * not a draft, and a block this edit cannot carry over unchanged refuses rather than drops. */ +export function queuedMessageEditableText(body: AgentJournalMessageItem): string | null { + if (body.command || body.blocks.some((b) => b.type !== 'text' && b.type !== 'image-ref')) { + return null + } + return body.blocks.flatMap((block) => (block.type === 'text' ? [block.text] : [])).join('\n') +} + +/** The body with its text replaced by one block at the first text block's place; attachments, + * sender, kind and send mode stay. Null when the result would not be a sendable draft. */ +export function queuedMessageWithEditedText( + body: AgentJournalMessageItem, + text: string +): AgentJournalMessageItem | null { + if ( + queuedMessageEditableText(body) === null || + new TextEncoder().encode(text).length > MAX_PROMPT_BYTES || + (!text.trim() && !body.blocks.some((block) => block.type === 'image-ref')) + ) { + return null + } + const blocks: AgentJournalMessageItem['blocks'] = [] + let placed = false + for (const block of body.blocks) { + if (block.type !== 'text') { + blocks.push(block) + } else if (!placed) { + blocks.push({ type: 'text', text }) + placed = true + } + } + if (!placed) { + blocks.unshift({ type: 'text', text }) + } + return { ...body, blocks } +} diff --git a/src/shared/rpc-contract/rpc-params-catalog.generated.ts b/src/shared/rpc-contract/rpc-params-catalog.generated.ts index ce0481c1f171..10a2a802ae02 100644 --- a/src/shared/rpc-contract/rpc-params-catalog.generated.ts +++ b/src/shared/rpc-contract/rpc-params-catalog.generated.ts @@ -515,6 +515,10 @@ import { ThreadGoalParams, UnsubscribeParams } from './structured-agent-session-params' +import { + QueuedMessageEditHoldParams, + QueuedMessageUpdateParams +} from './structured-agent-session-queued-edit-params' import { TerminalAdoptOrphans } from './terminal-orphan-params' import { TerminalQuickCommandsUpdate } from './terminal-quick-command-params' import { @@ -622,7 +626,9 @@ export const RPC_PARAMS_BY_METHOD = { 'agentSession.modelCatalog': ModelCatalogParams, 'agentSession.options': OptionsParams, 'agentSession.queuedMessageDelete': QueuedMessageActionParams, + 'agentSession.queuedMessageEditHold': QueuedMessageEditHoldParams, 'agentSession.queuedMessageSend': QueuedMessageActionParams, + 'agentSession.queuedMessageUpdate': QueuedMessageUpdateParams, 'agentSession.queuedMessagesResume': QueuedMessagesResumeParams, 'agentSession.readVisual': ReadVisualParams, 'agentSession.release': HoldParams, diff --git a/src/shared/rpc-contract/structured-agent-session-queued-edit-params.ts b/src/shared/rpc-contract/structured-agent-session-queued-edit-params.ts new file mode 100644 index 000000000000..2bbf1f8abd4b --- /dev/null +++ b/src/shared/rpc-contract/structured-agent-session-queued-edit-params.ts @@ -0,0 +1,27 @@ +import { z } from 'zod' +import { Identifier, SessionId } from './structured-agent-session-identifiers' +import { MutationEnvelope, MAX_PROMPT_BYTES } from './structured-agent-session-params' + +export const QueuedMessageUpdateParams = z + .object({ + envelope: MutationEnvelope, + messageId: Identifier('Invalid queued message id'), + expectedBodyFingerprint: z.string().regex(/^[0-9a-f]{64}$/), + text: z.string().refine((text) => new TextEncoder().encode(text).length <= MAX_PROMPT_BYTES) + }) + .strict() + +const QueuedMessageEditIdentity = z.object({ + sessionId: SessionId, + messageId: Identifier('Invalid queued message id'), + editId: Identifier('Invalid edit id') +}) + +export const QueuedMessageEditHoldParams = z.discriminatedUnion('action', [ + QueuedMessageEditIdentity.extend({ + action: z.literal('acquire'), + expectedBodyFingerprint: z.string().regex(/^[0-9a-f]{64}$/) + }).strict(), + QueuedMessageEditIdentity.extend({ action: z.literal('renew') }).strict(), + QueuedMessageEditIdentity.extend({ action: z.literal('release') }).strict() +]) From 7a745136d904cf2e797f412d349919ac216e0d0c Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Thu, 8 Oct 2026 19:59:27 -0700 Subject: [PATCH 02/12] feat(native-chat): edit a queued card inline on desktop Edit message now opens an editor inside the queued card instead of moving the text into the composer. The card keeps its place and sender, and the editor stays usable while a question or approval holds the composer's slot (it does not lose focus to the prompt card). - Enter or Cmd/Ctrl+Enter saves, Shift+Enter is a newline, Escape cancels; IME composition keys never save or cancel. Save and Cancel replace the card's own actions while it is edited; one editor per pane. - The editor holds the host's edit lease, renewing every 30 seconds and taking it again after a lapse. Lease failures are only logged: they never block typing, Save or Cancel. - If the card is sent or removed elsewhere mid-edit, a clean edit just closes; changed text goes into the chat box after any existing draft, with one toast. A Save that lands and then sends counts as success. - Against a host without agent-session.queued-message-edit.v1 the card offers no Edit (no copy-and-delete fallback); an empty overflow menu is hidden. --- .../NativeChatQueuedMessageCard.tsx | 94 +++-- .../NativeChatQueuedMessageEditor.test.tsx | 139 +++++++ .../NativeChatQueuedMessageEditor.tsx | 81 ++++ .../NativeChatQueuedMessageList.test.tsx | 52 ++- .../NativeChatQueuedMessageList.tsx | 21 +- ...native-chat-submission-navigation.test.tsx | 2 + .../structured-agent-session-queued-cards.ts | 3 + ...uctured-agent-session-queued-edit-lease.ts | 95 +++++ .../use-native-chat-prompt-card-focus.ts | 7 +- ...uctured-agent-session-queued-edit.test.tsx | 359 ++++++++++++++++++ ...se-structured-agent-session-queued-edit.ts | 281 ++++++++++++++ ...ent-session-queued-messages.clear.test.tsx | 11 +- ...nt-session-queued-messages.resume.test.tsx | 14 +- ...red-agent-session-queued-messages.test.tsx | 89 +---- ...tructured-agent-session-queued-messages.ts | 71 ++-- .../use-structured-agent-session.ts | 9 +- ...ructured-native-chat-submit-reveal.test.ts | 2 + src/renderer/src/i18n/locales/en.json | 8 +- src/renderer/src/i18n/locales/es.json | 8 +- src/renderer/src/i18n/locales/fr.json | 8 +- src/renderer/src/i18n/locales/ja.json | 8 +- src/renderer/src/i18n/locales/ko.json | 8 +- src/renderer/src/i18n/locales/zh.json | 8 +- ...tructured-agent-session-host-capability.ts | 12 + 24 files changed, 1217 insertions(+), 173 deletions(-) create mode 100644 src/renderer/src/components/native-chat/NativeChatQueuedMessageEditor.test.tsx create mode 100644 src/renderer/src/components/native-chat/NativeChatQueuedMessageEditor.tsx create mode 100644 src/renderer/src/components/native-chat/structured-agent-session-queued-edit-lease.ts create mode 100644 src/renderer/src/components/native-chat/use-structured-agent-session-queued-edit.test.tsx create mode 100644 src/renderer/src/components/native-chat/use-structured-agent-session-queued-edit.ts diff --git a/src/renderer/src/components/native-chat/NativeChatQueuedMessageCard.tsx b/src/renderer/src/components/native-chat/NativeChatQueuedMessageCard.tsx index 0b83b794b055..953697efc0e6 100644 --- a/src/renderer/src/components/native-chat/NativeChatQueuedMessageCard.tsx +++ b/src/renderer/src/components/native-chat/NativeChatQueuedMessageCard.tsx @@ -31,6 +31,8 @@ import { queuedCardSenderLine } from './native-chat-agent-message-sender-label' 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( @@ -133,6 +135,8 @@ export function NativeChatQueuedMessageCard({ onSteer, onDelete, onEdit, + editor, + editDisabled = false, onTurnOffQueueing }: { card: QueuedMessageCard @@ -144,7 +148,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 { @@ -156,7 +165,9 @@ export function NativeChatQueuedMessageCard({
  • {returned || card.pausedReason === QUEUED_MESSAGE_PAUSED_SEND_FAILED ? ( @@ -169,9 +180,13 @@ export function NativeChatQueuedMessageCard({ {queuedCardSenderLine(card.from)}

    ) : null} -

    - {card.text} -

    + {editor ? ( + + ) : ( +

    + {card.text} +

    + )} {caption ? (

    {/* Nothing acts on a send still on its way: the host holds no card for it yet. */} - {card.hold === 'sending' ? null : ( + {card.hold === 'sending' || editor ? null : ( <> {/* A command never steers: its Send shows only while the agent is idle. */} {card.waitsForAgent ? null : ( @@ -196,7 +211,8 @@ export function NativeChatQueuedMessageCard({ variant="ghost" size="xs" onClick={onSteer} - disabled={steerHeld} + // Someone is editing it on another device; their Save, Cancel or lapse frees it. + disabled={steerHeld || card.editHeld} > {sendNow.steers ? ( @@ -232,38 +248,40 @@ 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..a2f37995f299 --- /dev/null +++ b/src/renderer/src/components/native-chat/NativeChatQueuedMessageEditor.test.tsx @@ -0,0 +1,139 @@ +// @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) => fallback })) +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, with separate Save/Cancel', + (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 Cancel 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' })) + expect(edit.cancel).toHaveBeenCalledOnce() + }) +}) 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..083864daee7a --- /dev/null +++ b/src/renderer/src/components/native-chat/NativeChatQueuedMessageEditor.tsx @@ -0,0 +1,81 @@ +import { useLayoutEffect, useRef } from 'react' +import { Textarea } from '@/components/ui/textarea' +import { Button } from '@/components/ui/button' +import { translate } from '@/i18n/i18n' +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 text 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 +}: { + editor: QueuedMessageInlineEditor +}): 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 ( +
    +