diff --git a/mobile/src/session/mobile-native-chat-merged-snapshot-parity.test.ts b/mobile/src/session/mobile-native-chat-merged-snapshot-parity.test.ts index 9d75bb02d7d6..5c38fa7be907 100644 --- a/mobile/src/session/mobile-native-chat-merged-snapshot-parity.test.ts +++ b/mobile/src/session/mobile-native-chat-merged-snapshot-parity.test.ts @@ -148,11 +148,13 @@ function comparable(merged: Drawn, fresh: Drawn): { merged: string[]; fresh: str } /** Every frame from each subscribe point in `starts`, `rowsPerFrame` rows at a time; with how many - * of those frames the merged state held only a window of the chat. */ + * of those frames the merged state held only a window of the chat. `kept`: the never-opened send + * became a card, so from its rejection on neither list draws it nor its stop row. */ function compareFrames( journal: StopJournal, starts: readonly number[], - rowsPerFrame: number + rowsPerFrame: number, + kept = false ): { differing: string[]; windowed: number } { const last = journal.rows.length const fresh = new Map() @@ -165,15 +167,19 @@ function compareFrames( fresh.set(upTo, drawnThen) return drawnThen } + // Whether a list draws the never-opened send as the Stop left it: its stop row, or for a kept + // send, nothing of it at all. + const asStopped = (ids: readonly string[]): boolean => + kept ? !ids.includes(STOP_ROW) && !ids.includes(`orca:${NEVER_OPENED}`) : ids.includes(STOP_ROW) const freshListing = new Map() const freshListed = (upTo: number): boolean => { const cached = freshListing.get(upTo) if (cached !== undefined) { return cached } - const shown = listed( - apply(EMPTY_STRUCTURED_AGENT_SESSION, [journal.snapshotAt(upTo)]) - ).includes(STOP_ROW) + const shown = asStopped( + listed(apply(EMPTY_STRUCTURED_AGENT_SESSION, [journal.snapshotAt(upTo)])) + ) freshListing.set(upTo, shown) return shown } @@ -191,7 +197,7 @@ function compareFrames( if (JSON.stringify(compared.merged) !== JSON.stringify(compared.fresh)) { differing.push(`subscribed at ${start}, frame through ${upTo}`) } - if (upTo >= journal.takenBack && !listed(merged).includes(STOP_ROW)) { + if (upTo >= journal.takenBack && !asStopped(listed(merged))) { differing.push(`subscribed at ${start}, no stop row in the list through ${upTo}`) } if (upTo >= journal.takenBack && !freshListed(upTo)) { @@ -231,6 +237,29 @@ describe.each(['after-end', 'during-turn'] as const)( } ) +// The same chat, but the never-opened send was still held by the host when the Stop came: the +// card holds its words, so the phone draws no bubble and no stop row for it, from any frame, and the +// desktop's transcript (which draws other refused sends in place) draws none either. +describe('a short chat whose last send a Stop kept as a card: merged frames draw as a fresh snapshot does', () => { + const journal = stopJournal(8, 'after-end', 'kept') + + it('draws nothing of the kept send on the phone or the desktop', () => { + const state = apply(EMPTY_STRUCTURED_AGENT_SESSION, [journal.snapshotAt(journal.rows.length)]) + expect(drawn(state).rows.map((row) => row.id)).not.toContain(`orca:${NEVER_OPENED}`) + const desktop = projectStructuredAgentSessionMessages(state.items, [], state.submissions, { + rejectedInPlace: true + }) + expect(desktop.map((message) => message.id)).not.toContain(`orca:${NEVER_OPENED}`) + expect(desktop.some((message) => message.stoppedBeforeStart)).toBe(false) + }) + + it.each([1, 2, 3])('from every subscribe point, %i row(s) per frame', (rowsPerFrame) => { + expect( + compareFrames(journal, range(2, journal.rows.length), rowsPerFrame, true).differing + ).toEqual([]) + }) +}) + describe('a chat longer than the phone first loads', () => { const journal = stopJournal(70) const last = journal.rows.length diff --git a/mobile/src/session/mobile-native-chat-stop-journal.test-fixture.ts b/mobile/src/session/mobile-native-chat-stop-journal.test-fixture.ts index bdf77d41296d..91a44c6f7406 100644 --- a/mobile/src/session/mobile-native-chat-stop-journal.test-fixture.ts +++ b/mobile/src/session/mobile-native-chat-stop-journal.test-fixture.ts @@ -42,9 +42,14 @@ export type StopJournal = HostJournalFrames & { * that holds sends while Stopping, as the QA journal shows), or while the turn still runs. */ export type SendWhileStoppingWritten = 'after-end' | 'during-turn' +/** The last send, never opened: handed over and taken back at the child's exit, or still held by + * the host when the Stop came, which kept it as a card. */ +export type NeverOpenedSend = 'taken-back' | 'kept' + export function stopJournal( historyTurns: number, - sendWhileStopping: SendWhileStoppingWritten = 'after-end' + sendWhileStopping: SendWhileStoppingWritten = 'after-end', + neverOpened: NeverOpenedSend = 'taken-back' ): StopJournal { const rows: JournalRow[] = [] const stopping: StopJournalStopping[] = [] @@ -228,9 +233,13 @@ export function stopJournal( openTurn('sent-while-stopping', 't-sent-while-stopping') endTurn('sent-while-stopping', 't-sent-while-stopping', 'completed') // A send the provider never opened; a turn-less Stop ends the child, whose exit takes it back. + // One the host still held is kept as a card ahead of the Stop's event. const neverOpenedSent = submit(NEVER_OPENED) - handOver(NEVER_OPENED) - const stopNeverOpened = stopEvent() + const kept = neverOpened === 'kept' + if (!kept) { + handOver(NEVER_OPENED) + } + const stopNeverOpened = kept ? 0 : stopEvent() const takenBack = add({ kind: 'dispatch', clientMessageId: NEVER_OPENED, @@ -238,9 +247,14 @@ export function stopJournal( providerItemId: null, reason: DISPATCH_REJECTED_CANCELLED, rejection: { kind: 'cancelled' }, - recovered: true + recovered: true, + ...(kept ? { keptAsQueuedMessageId: NEVER_OPENED } : {}) }) - stopping.push({ from: stopNeverOpened, until: takenBack - 1 }) + if (kept) { + stopEvent() + } else { + stopping.push({ from: stopNeverOpened, until: takenBack - 1 }) + } // The next send starts a new child under a new fence. fence = 2 submit('recovery') diff --git a/src/main/native-chat/agent-session-journal/journal-pending-submission-recovery.ts b/src/main/native-chat/agent-session-journal/journal-pending-submission-recovery.ts index 3aacf477d790..80c3783c3d4e 100644 --- a/src/main/native-chat/agent-session-journal/journal-pending-submission-recovery.ts +++ b/src/main/native-chat/agent-session-journal/journal-pending-submission-recovery.ts @@ -63,47 +63,6 @@ export async function markJournalPendingSubmissionsUnknown( return unresolved.map((entry) => entry.clientMessageId) } -/** Settles as `rejected` every submission a child handed over and never echoed, when that child - * ended in its start: one that never answered initialize ran nothing, so each is safe to send - * again. A queued submission was never handed to that child; the delivery loop settles it. */ -export async function rejectJournalPendingSubmissions( - journal: AgentSessionJournal, - fence: number, - rejection: AgentJournalDispatchRejection -): Promise { - const unwritten = journalPendingSubmissionResolutions(journal.submissions(), fence, { rejection }) - for (const resolution of unwritten) { - await journal.resolveDispatch(resolution) - } - return unwritten.map((entry) => entry.clientMessageId) -} - -/** Rejects queued submissions — accepted, never handed over, so provably unwritten. */ -export async function rejectJournalQueuedSubmissions( - journal: AgentSessionJournal, - fence: number, - rejection: AgentJournalDispatchRejection, - which: (submission: AgentJournalSubmission) => boolean = () => true -): Promise { - const queued = journal - .submissions() - .filter((entry) => isQueuedAgentJournalSubmission(entry) && which(entry)) - // Issued together, so the fold shows none of them queued once this call returns: a Stop decides - // whether anything is working from it without awaiting the withdrawal. - await Promise.all( - queued.map((entry) => - journal.resolveDispatch({ - clientMessageId: entry.clientMessageId, - state: 'rejected', - ...rejection, - fence, - recovered: true - }) - ) - ) - return queued.map((entry) => entry.clientMessageId) -} - /** Rows rejecting every submission still queued, read from `state` when called: for an append * that must carry them with what follows, in one transaction. */ export function journalQueuedRejectionRowBuilders( diff --git a/src/main/native-chat/agent-session-journal/journal-queued-rejection-batch.test.ts b/src/main/native-chat/agent-session-journal/journal-queued-rejection-batch.test.ts index cd14ce2fe943..4050926fc85c 100644 --- a/src/main/native-chat/agent-session-journal/journal-queued-rejection-batch.test.ts +++ b/src/main/native-chat/agent-session-journal/journal-queued-rejection-batch.test.ts @@ -14,6 +14,7 @@ import { type AgentSessionJournalIdentity } from '../../../shared/agent-session-journal-types' import type { AgentSessionJournal } from './journal-store' +import { holdUnsentSends } from './journal-unsent-send-hold' import { closeTestJournalHostDatabases, createTrackedJournalOpener @@ -114,12 +115,8 @@ it('writes neither when the row cannot be written', async () => { // A Stop that reaches the lane first takes the message back; the failed start then failed no one. it('writes nothing when a Stop withdrew every queued message first', async () => { const journal = await openWithQueued('first') - const withdrawal = agentSessionFailureWords(agentSessionFailureFact('cancelled'), { - surface: 'rejection' - }) - await Promise.all([ - journal.rejectQueuedSubmissions(0, withdrawal), + holdUnsentSends(journal, { fence: 0, hostInstance: 'proc-1', hold: { cause: 'cancelled' } }), journal.appendLifecycleBatch(startFailureBatch()) ]) 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 25e3a7e8d865..05c76aa846ad 100644 --- a/src/main/native-chat/agent-session-journal/journal-store.ts +++ b/src/main/native-chat/agent-session-journal/journal-store.ts @@ -1,7 +1,6 @@ // Append-only journal store for one agent session. It owns the chat's fold and write queue, and no // connection: every statement goes through the host's one journal database. -import type { AgentJournalDispatchRejection } from '../../../shared/agent-session-failure-words' import { randomUUID } from 'node:crypto' import type { AgentJournalAcceptanceReceipt, @@ -29,11 +28,7 @@ import type { JournalReplacementItem } from './journal-epoch-replacement' import { readJournalSince } from './journal-cursor' import type { JournalHostDatabase } from './journal-host-database' import { journalRowsAfterReader, type JournalLoad } from './journal-open' -import { - markJournalPendingSubmissionsUnknown, - rejectJournalPendingSubmissions, - rejectJournalQueuedSubmissions -} from './journal-pending-submission-recovery' +import { markJournalPendingSubmissionsUnknown } from './journal-pending-submission-recovery' import { applyJournalRow, createJournalReducerState, @@ -355,23 +350,6 @@ export class AgentSessionJournal { return markJournalPendingSubmissionsUnknown(this, fence, reason) } - /** Reject sends a child that ended in its start was handed and never echoed: none ran. */ - async rejectPendingSubmissions( - fence: number, - rejection: AgentJournalDispatchRejection - ): Promise { - return rejectJournalPendingSubmissions(this, fence, rejection) - } - - /** Reject sends accepted but never handed over, optionally only those `which` names. */ - async rejectQueuedSubmissions( - fence: number, - rejection: AgentJournalDispatchRejection, - which?: (submission: AgentJournalSubmission) => boolean - ): Promise { - return rejectJournalQueuedSubmissions(this, fence, rejection, which) - } - /** The escape hatch for a forked handle and an unreadable schema. It invalidates every cursor; * clients reload. */ async rollEpoch(reason: AgentJournalEpochReason, fence: number): Promise { diff --git a/src/main/native-chat/agent-session-journal/journal-unsent-send-hold.test.ts b/src/main/native-chat/agent-session-journal/journal-unsent-send-hold.test.ts index 29fb7eb35fa2..43ca9559a3ef 100644 --- a/src/main/native-chat/agent-session-journal/journal-unsent-send-hold.test.ts +++ b/src/main/native-chat/agent-session-journal/journal-unsent-send-hold.test.ts @@ -445,6 +445,68 @@ describe('a close of the chat', () => { }) }) +describe("a person's Stop", () => { + const CANCELLED = agentSessionFailureWords(agentSessionFailureFact('cancelled'), { + surface: 'rejection' + }) + + it('keeps every queued text of a person by the same rule, ahead of the cards already queued', async () => { + const image: AgentJournalMessageItem = { + kind: 'message', + role: 'user', + blocks: [ + { type: 'text', text: 'look' }, + { type: 'image-ref', path: '/tmp/attachment.png' } + ] + } + const journal = await open() + await journal.queuedMessages.insert({ + messageId: 'queued-card', + body: message('queued card'), + fingerprint: fingerprint(message('queued card')), + hostInstance: HOST + }) + await accept(journal, 'first', { origin: 'client', source: USER_MESSAGE_SOURCE }) + await accept(journal, 'mail', { origin: 'host', source: MAIL_SOURCE }) + await accept(journal, 'compact', { + origin: 'client', + source: USER_MESSAGE_SOURCE, + body: structuredAgentSessionCompactBody() + }) + await accept(journal, 'image', { origin: 'client', source: USER_MESSAGE_SOURCE, body: image }) + await accept(journal, 'second', { origin: 'client', source: USER_MESSAGE_SOURCE }) + await hold(journal, { cause: 'cancelled' }) + + expect(cardOrder(journal)).toEqual(['first', 'second', 'queued-card']) + for (const id of ['first', 'second']) { + expect(journal.submission(id)).toMatchObject({ + dispatchState: 'rejected', + ...CANCELLED, + keptAsQueuedMessageId: id + }) + } + for (const id of ['mail', 'compact', 'image']) { + expect(journal.submission(id)).toMatchObject({ dispatchState: 'rejected', ...CANCELLED }) + expect(journal.submission(id)).not.toHaveProperty('keptAsQueuedMessageId') + } + }) + + // The Stop decides whether the agent still works from the fold, without awaiting the hold. + it('leaves none of them queued in the fold by the time the call returns', async () => { + const journal = await open() + await accept(journal, 'first', { origin: 'client', source: USER_MESSAGE_SOURCE }) + await accept(journal, 'second', { origin: 'client', source: USER_MESSAGE_SOURCE }) + const settled = hold(journal, { cause: 'cancelled' }) + + expect(journal.submissions().map((entry) => entry.dispatchState)).toEqual([ + 'rejected', + 'rejected' + ]) + await settled + expect(cardOrder(journal)).toEqual(['first', 'second']) + }) +}) + describe('the order kept cards stand in', () => { // Sequences restart with each epoch, so a card kept before a rewind has no sequence to compare. it('a card kept before the epoch rolled stays ahead of one kept after it', async () => { diff --git a/src/main/native-chat/agent-session-journal/journal-unsent-send-hold.ts b/src/main/native-chat/agent-session-journal/journal-unsent-send-hold.ts index fd5a774c7bf2..1d101c9cb18f 100644 --- a/src/main/native-chat/agent-session-journal/journal-unsent-send-hold.ts +++ b/src/main/native-chat/agent-session-journal/journal-unsent-send-hold.ts @@ -1,10 +1,10 @@ // What becomes of a send the host accepted and can no longer hand over: an earlier host process -// quit or crashed first (settled at the next open), or the chat is closing. Either way the agent -// provably never got it. A person's message (typed, or a launch's first prompt) is kept as a card -// at the head of the queue, an ordinary one: like every card the chat closed with, it waits for the -// chat's next turn (`queued-message-pause.ts`). Everything else is rejected as before, because -// something else re-derives it or the person re-runs it. The submission itself is always rejected, -// so it is never handed over twice. +// quit or crashed first (settled at the next open), the chat is closing, or a person's Stop took it +// back. Either way the agent provably never got it. A person's message (typed, or a launch's first +// prompt) is kept as a card at the head of the queue, an ordinary one: like every card, it waits +// out the pause in force, the Stop's or the reopen's (`queued-message-pause.ts`). Everything else is +// rejected as before, because something else re-derives it or the person re-runs it. The submission +// itself is always rejected, so it is never handed over twice. import type { AgentJournalItemBody, @@ -27,6 +27,8 @@ export type UnsentSendHold = | { cause: 'hostRestarted' } /** A close of the chat; `which` narrows it to what a close that did not complete closed. */ | { cause: 'chatClosed'; which?: (submission: AgentJournalSubmission) => boolean } + /** A person's Stop: every send still queued. */ + | { cause: 'cancelled' } /** The body an unsent send is kept with, or null when it is rejected instead: * - a card's own hand-off: its rejection already returns the card (`rejectedDraftSettlement`); @@ -98,7 +100,7 @@ export async function holdUnsentSends( : isQueuedAgentJournalSubmission(entry) && (hold.cause === 'hostRestarted' ? journal.wroteBeforeOpen(entry.acceptedSequence) - : (hold.which?.(entry) ?? true)) + : hold.cause === 'cancelled' || (hold.which?.(entry) ?? true)) ) .sort((a, b) => (a.acceptedSequence ?? 0) - (b.acceptedSequence ?? 0)) if (unsent.length === 0) { @@ -112,12 +114,13 @@ export async function holdUnsentSends( journal.itemBody(agentJournalSubmissionKey(submission.clientMessageId)) ) })) - const positions = headOfQueuePositions(journal, kept) + const positions = headOfQueuePositions(journal, kept, hold) const rejection = agentSessionFailureWords(agentSessionFailureFact(hold.cause), { surface: 'rejection' }) - const failures: unknown[] = [] - for (const [index, { submission, body }] of kept.entries()) { + // Issued together, so the fold shows none of them queued once this call returns: a Stop decides + // whether anything is working from it without awaiting the hold. + const settled = kept.map(async ({ submission, body }, index): Promise => { const { clientMessageId } = submission const moves = [ // The cards an earlier settlement kept move with the first row. @@ -166,10 +169,10 @@ export async function holdUnsentSends( card ? { ...reject, keptAsQueuedMessageId: clientMessageId } : reject, keep ) + return undefined } catch (error) { if (!keep) { - failures.push(error) - continue + return error } // Keeping it failed: rejected as before. That loses the message, as every build before this // one did; it is never left queued for a handover nothing will make. @@ -179,9 +182,13 @@ export async function holdUnsentSends( cause: hold.cause, error: error instanceof Error ? error.message : String(error) }) - await journal.resolveDispatch(reject).catch((fallback: unknown) => failures.push(fallback)) + return journal.resolveDispatch(reject).then( + () => undefined, + (fallback: unknown) => fallback + ) } - } + }) + const failures = (await Promise.all(settled)).filter((failure) => failure !== undefined) if (failures.length > 0) { throw new AggregateError(failures, 'settling unsent sends failed') } @@ -194,7 +201,8 @@ export async function holdUnsentSends( * them: sequences restart with each epoch, and a returned hand-off's names its draft's time. */ function headOfQueuePositions( journal: AgentSessionJournal, - batch: readonly { submission: AgentJournalSubmission; body: AgentJournalMessageItem | null }[] + batch: readonly { submission: AgentJournalSubmission; body: AgentJournalMessageItem | null }[], + hold: UnsentSendHold ): { placed: QueuedMessagePositionMove[]; earlier: QueuedMessagePositionMove[] } { const cards = journal.queuedMessages.list() const keptIds = new Set( @@ -216,7 +224,9 @@ function headOfQueuePositions( if (body && !earlier.includes(clientMessageId)) { placed.push({ messageId: clientMessageId, position: 0 }) } else if ( - // Kept by its settlement (`rejectedDraftSettlement`): a Send the person asked for. + // Kept by its settlement (`rejectedDraftSettlement`): a Send the person asked for. A Stop + // returns it where it stood. + hold.cause !== 'cancelled' && submission.origin === 'client' && cards.some((card) => card.consumedAs === clientMessageId) ) { diff --git a/src/main/native-chat/agent-session-journal/queued-message-delivered-echo.test.ts b/src/main/native-chat/agent-session-journal/queued-message-delivered-echo.test.ts index e6905e848460..1fc89aa9363e 100644 --- a/src/main/native-chat/agent-session-journal/queued-message-delivered-echo.test.ts +++ b/src/main/native-chat/agent-session-journal/queued-message-delivered-echo.test.ts @@ -16,6 +16,7 @@ import { agentSessionFailureWords } from '../../../shared/agent-session-failure- import { agentSessionSendBodyFingerprint } from '../../../shared/structured-agent-session-send-mutation' import { JournalQueuedMessages } from './journal-queued-messages' import type { AgentSessionJournal } from './journal-store' +import { holdUnsentSends } from './journal-unsent-send-hold' import { createTrackedJournalOpener } from './journal-host-database-test-support' import { claudeProviderHandle } from '../../../shared/agent-session-provider-handle-encoding' @@ -103,7 +104,11 @@ async function handOffAndReject( fence: 0 }) } else { - await journal.rejectQueuedSubmissions(0, STOP_WITHDRAWAL) + await holdUnsentSends(journal, { + fence: 0, + hostInstance: 'proc-1', + hold: { cause: 'cancelled' } + }) } expect(journal.queuedMessages.get('draft-1')).toMatchObject({ state: 'waiting', 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..77d54ad073c2 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 @@ -22,6 +22,7 @@ import { QueuedMessageNotConsumableError } from './journal-queued-messages' import type { AgentSessionJournal } from './journal-store' +import { holdUnsentSends } from './journal-unsent-send-hold' import type { AgentMessageSource } from '../../../shared/agent-session-message-source' import { closeTestJournalHostDatabases, @@ -50,12 +51,6 @@ function message(text: string): AgentJournalMessageItem { } const journals = createTrackedJournalOpener() -const STOP_WITHDRAWAL = agentSessionFailureWords(agentSessionFailureFact('cancelled'), { - surface: 'rejection' -}) -const HOST_RESTARTED = agentSessionFailureWords(agentSessionFailureFact('hostRestarted'), { - surface: 'rejection' -}) const PROVIDER_REFUSAL = agentSessionFailureFact('providerRejected', { detail: { text: 'Claude refused this payload', audience: 'person' } }) @@ -77,6 +72,24 @@ async function open(): Promise { return journal } +/** A Stop's withdrawal of every queued send, as the chat's Stop issues it. */ +function stopWithdrawal(journal: AgentSessionJournal): Promise { + return holdUnsentSends(journal, { + fence: 0, + hostInstance: 'proc-1', + hold: { cause: 'cancelled' } + }) +} + +/** The open's settlement of what an earlier host process accepted and never handed over. */ +function hostRestartedHold(journal: AgentSessionJournal): Promise { + return holdUnsentSends(journal, { + fence: 0, + hostInstance: 'proc-1', + hold: { cause: 'hostRestarted' } + }) +} + async function queueDraft(journal: AgentSessionJournal, messageId: string, text = 'queued text') { return journal.queuedMessages.insert({ messageId, @@ -377,7 +390,8 @@ describe('returned transition (D1/N4)', () => { await queueDraft(journal, 'draft-1') await consumeDraft(journal, 'draft-1') // The Stop's own withdrawal path: the queued (not handed over) submission. - expect(await journal.rejectQueuedSubmissions(0, STOP_WITHDRAWAL)).toEqual(['sub-draft-1']) + expect(await stopWithdrawal(journal)).not.toBeNull() + expect(journal.submission('sub-draft-1')?.dispatchState).toBe('rejected') // Nothing failed: no refusal to show, its position kept, its spent id recorded. const requeued = { state: 'waiting', @@ -399,7 +413,7 @@ describe('returned transition (D1/N4)', () => { const journal = await open() await queueDraft(journal, 'draft-1') await consumeDraft(journal, 'draft-1') - await journal.rejectQueuedSubmissions(0, STOP_WITHDRAWAL) + await stopWithdrawal(journal) // The spent id already names a rejected submission: one id, one delivery. await expect(consumeDraft(journal, 'draft-1')).rejects.toMatchObject({ code: 'journal_submission_exists' @@ -430,9 +444,7 @@ describe('returned transition (D1/N4)', () => { await journal.close() journal = await open() expect(journal.queuedMessages.get('draft-1')?.state).toBe('dispatched') - await journal.rejectQueuedSubmissions(0, HOST_RESTARTED, (submission) => - journal.wroteBeforeOpen(submission.acceptedSequence) - ) + await hostRestartedHold(journal) // Whoever sent it, it waits with no hold of its own, under the reopen's pause. expect(journal.queuedMessages.get('draft-1')).toMatchObject({ state: 'waiting', @@ -605,10 +617,13 @@ describe('open-time repair and retention', () => { let journal = await open() await queueDraft(journal, 'draft-1') await consumeDraft(journal, 'draft-1', { origin }) - await journal.rejectQueuedSubmissions( - 0, - agentSessionFailureWords(agentSessionFailureFact(cause), { surface: 'rejection' }) - ) + await journal.resolveDispatch({ + clientMessageId: 'sub-draft-1', + state: 'rejected', + ...agentSessionFailureWords(agentSessionFailureFact(cause), { surface: 'rejection' }), + fence: 0, + recovered: true + }) await journal.close() // The hook "was skipped": the draft is back to dispatched behind the stored rejection. const db = new Database(journalDatabasePath(root)) @@ -655,7 +670,7 @@ describe('open-time repair and retention', () => { let journal = await open() await queueDraft(journal, 'draft-1') await consumeDraft(journal, 'draft-1') - await journal.rejectQueuedSubmissions(0, STOP_WITHDRAWAL) + await stopWithdrawal(journal) await journal.close() const db = new Database(journalDatabasePath(root)) db.prepare( @@ -683,7 +698,7 @@ describe('open-time repair and retention', () => { journal = await open() expect(journal.queuedMessages.get('draft-1')?.state).toBe('dispatched') // The delivery loop's leftover rejection now sends it back to waiting. - await journal.rejectQueuedSubmissions(0, HOST_RESTARTED) + await hostRestartedHold(journal) expect(journal.queuedMessages.get('draft-1')).toMatchObject({ state: 'waiting', consumedAs: null diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-chat-stop.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-chat-stop.ts index b6cd948fa4dc..2b4cb388d3e8 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-chat-stop.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-chat-stop.ts @@ -5,8 +5,6 @@ // reachable only through `mutateWithChatStop`, which queues that step in the same synchronous call // as the mutation. -import { agentSessionFailureWords } from '../../../shared/agent-session-failure-words' -import { agentSessionFailureFact } from '../../../shared/agent-session-failure' import { isQueuedAgentJournalSubmission } from '../../../shared/agent-session-queued-submission' import type { AgentSessionCancelResult, @@ -18,6 +16,8 @@ import { type StructuredAgentSessionMutationContext } from './structured-agent-session-mutation-context' import type { StructuredAgentSessionCaller } from './structured-agent-session-host-types' +import { holdUnsentSends } from '../agent-session-journal/journal-unsent-send-hold' +import { structuredAgentSessionHostInstance } from './structured-agent-session-queued-pause' import type { MutationPlan } from './structured-agent-session-mutation-plans' import { runRecordedStop, @@ -77,13 +77,15 @@ export function mutateWithChatStop( async (tookEffect) => { // Read before the withdrawal it decides on is issued. const hadQueued = ctx.journal.submissions().some(isQueuedAgentJournalSubmission) - // Stop withdraws every queued SUBMISSION first, whatever the start or the child is doing. - // Issued, not awaited: the interrupt never waits on bookkeeping. + // Stop withdraws every queued SUBMISSION first, whatever the start or the child is doing, + // keeping a person's words as a card, as a close does. Issued, not awaited: the interrupt + // never waits on bookkeeping. const withdrew = withdrawQueuedForStop(ctx, () => - ctx.journal.rejectQueuedSubmissions( - ctx.fence, - agentSessionFailureWords(agentSessionFailureFact('cancelled'), { surface: 'rejection' }) - ) + holdUnsentSends(ctx.journal, { + fence: ctx.fence, + hostInstance: structuredAgentSessionHostInstance(), + hold: { cause: 'cancelled' } + }).then((newest) => newest !== null) ) const child = context.sessions.get(ctx.sessionId)?.child if (child?.close) { @@ -103,7 +105,7 @@ export function mutateWithChatStop( } if (child?.phase === 'starting') { // A start that may never land takes no interrupt, so Stop ends it; the chat stays. Its end - // settles what the child was handed as stopped (`unrunRejection`). The event is + // settles what the child was handed as a Stop's (`unrunHold`). The event is // issued first and lands behind the withdrawal, in the journal's queue order. const effect = tookEffect() try { @@ -119,11 +121,11 @@ export function mutateWithChatStop( const inFlight = turnId !== undefined || isMainAgentWorking(ctx) const record = context.deps.store.getRecord(ctx.sessionId) if (!child || !inFlight) { - // Nothing to interrupt, so the answer may wait for the withdrawal. + // Nothing to interrupt, so the answer may wait for the withdrawal. The event is issued in + // the hold's tick, so no frame publishes a kept card without its pause. + const effect = hadQueued ? tookEffect() : Promise.resolve() const withdrewAny = await withdrew - if (withdrewAny) { - await tookEffect() - } + await effect return { ok: true, value: { ...named, cancelled: withdrewAny } } } const reach = hadQueued ? 'unrecorded' : stopReachesUnrecordedWork(ctx, turnId) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-child-exit.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-child-exit.ts index 694268a1f8bf..55117104acb4 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-child-exit.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-child-exit.ts @@ -69,7 +69,7 @@ export type StructuredAgentSessionChildExitContext< holdUnrunSends?: ( sessionId: string, fence: number, - cause: 'chatClosed' | 'hostRestarted' + cause: 'chatClosed' | 'hostRestarted' | 'cancelled' ) => Promise /** Lets the child's sink and the adapter's route for it go; absent leaves both to the next attach. */ route?: { @@ -129,21 +129,20 @@ export async function endExitedStructuredAgentSessionChildUnderSerialize< ? agentSessionFailureFact('hostStopped') : undefined : exit.failure - // A send the child was handed and never echoed cannot have run when a person's Stop ended its - // start, or any close ended it before it answered its start: settled as the chat settles a queued - // send for the same end. A Stop withdraws it as cancelled; a quit or close keeps a person's - // message as a held card. A host stop fails the start (`startFailure`). - const unrunRejection = - startClose === 'user-stop' ? agentSessionFailureFact('cancelled') : undefined + // A send the child was handed and never echoed cannot have run when a close ended it before it + // answered its start: settled as the chat settles a queued send for the same end. A Stop, a quit + // or a close keeps a person's message as a held card. A host stop fails the start + // (`startFailure`). const unrunHold = expected && exit.startupUnanswered && close?.cause !== 'host-stop' && - close?.cause !== 'user-stop' && close?.cause !== 'context-clear' - ? close?.quit - ? ('hostRestarted' as const) - : ('chatClosed' as const) + ? close?.cause === 'user-stop' + ? ('cancelled' as const) + : close?.quit + ? ('hostRestarted' as const) + : ('chatClosed' as const) : undefined const endChild = (): void => { context.startupAttempts?.childEnded(sessionId, child) @@ -220,8 +219,7 @@ export async function endExitedStructuredAgentSessionChildUnderSerialize< ...(startFailed && child.generation ? { exitedDuringStartup: { generation: child.generation } } : {}), - exit: watched, - ...(unrunRejection ? { unrunRejection } : {}) + exit: watched }) if (!settled.ok) { logExitFailure(context, sessionId, 'exit-settlement', settled.error) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-claude-hung-start-stop.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-claude-hung-start-stop.test.ts index 347de6cf6643..191d61d0092a 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-claude-hung-start-stop.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-claude-hung-start-stop.test.ts @@ -211,6 +211,28 @@ it.each(['user-close', 'evict'] as const)( } ) +// A person's Stop settles it as a close does: the words come back as a card the Stop holds. +it("keeps a person's message held behind a start that never answered as a card when the Stop ends it", async () => { + const id = await sendHello(false) + + expect( + await host.cancel(CALLER, { envelope: envelope('agentSession.cancel', {}) }) + ).toMatchObject({ ok: true, value: { cancelled: true } }) + + await vi.waitFor(async () => + expect(await submission(id)).toMatchObject({ + dispatchState: 'rejected', + reason: DISPATCH_REJECTED_CANCELLED, + keptAsQueuedMessageId: id + }) + ) + expect(host['sessions'].get(SESSION)?.journal.queuedMessages.get(id)).toMatchObject({ + state: 'waiting', + body: hostTestMessage('hello') + }) + expect(await helloWritten()).toBe(false) +}) + // Once it answered, the CLI may have taken the message without echoing it yet: a close leaves it in // doubt, as for any running child. it('leaves a message in doubt when the start answered and the chat closes before the echo', async () => { diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-create-test-fixture.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-create-test-fixture.ts index ab375470e0ef..470bac2c0e98 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-create-test-fixture.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-create-test-fixture.ts @@ -52,3 +52,18 @@ export function stopCreatedChat( } return host.cancel(CREATE_TEST_CALLER, { envelope }) } + +export function resumeCreatedChat(host: StructuredAgentSessionHost, sessionId = HOST_TEST_SESSION) { + return host.queuedMessagesResume(CREATE_TEST_CALLER, { + envelope: { + sessionId, + clientOperationId: hostTestOperationId(), + expectedRuntimeFence: null, + payloadFingerprint: computeAgentSessionPayloadFingerprint({ + method: 'agentSession.queuedMessagesResume', + sessionId, + fields: {} + }) + } + }) +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-create.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-create.test.ts index c3f2a4855986..c607ba79a2c9 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-create.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-create.test.ts @@ -7,6 +7,7 @@ import { import { CREATE_TEST_CALLER as CALLER, createTestParams, + resumeCreatedChat, stopCreatedChat } from './structured-agent-session-create-test-fixture' import { @@ -25,7 +26,6 @@ import { AgentSessionJournal } from '../agent-session-journal/journal-store' import { AGENT_JOURNAL_THREAD_SCOPE } from '../../../shared/agent-session-journal-types' import { AGENT_SESSION_HISTORY_MAX_LIMIT } from '../../../shared/agent-session-wire' import { projectStructuredAgentSessionMessages } from '../../../shared/structured-agent-session-message-projection' -import { NATIVE_CHAT_STOPPED_BEFORE_START_TEXT } from '../../../shared/native-chat-stopped-before-start' let rig: RestTestRig beforeEach(async () => { @@ -40,6 +40,42 @@ function firstMessage() { return { clientMessageId: hostTestOperationId(), body: hostTestMessage('opening message') } } +function dispatchedTexts(): string[] { + return rig.adapter.dispatch.mock.calls.map(([input]) => + input.body.blocks.map((block) => (block.type === 'text' ? block.text : '')).join('') + ) +} + +/** A Stop kept the first message, unsent, as a card at the head of the queue under the Stop's + * pause, as it keeps any person's held send; neither client draws it in the conversation. */ +async function expectKeptAsCard(clientMessageId: string): Promise { + expect( + (await rig.host.journalSnapshot(SESSION)).submissions.find( + (entry) => entry.clientMessageId === clientMessageId + ) + ).toMatchObject({ + dispatchState: 'rejected', + rejection: { kind: 'cancelled' }, + keptAsQueuedMessageId: clientMessageId + }) + const page = await rig.host.history({ sessionId: SESSION, direction: 'tail' }) + if (!page.ok) { + throw new Error('history refused') + } + expect(page.page.queuedMessages?.map(({ messageId, state }) => ({ messageId, state }))).toEqual([ + { messageId: clientMessageId, state: 'waiting' } + ]) + expect(page.page.queuePause).toEqual({ reason: 'stopped' }) + for (const rejectedInPlace of [true, false]) { + expect( + projectStructuredAgentSessionMessages(page.page.items, [], page.page.submissions, { + rejectedInPlace + }) + ).toEqual([]) + } + expect(rig.adapter.dispatch).not.toHaveBeenCalled() +} + function abortableAcquisition() { const entered = Promise.withResolvers() rig.adapter.acquire.mockImplementationOnce( @@ -161,7 +197,7 @@ it('keeps delivery behind runtime tab publication even after the message is comm await stopCreatedChat(rig.host) }) -it('an immediate Stop cancels the queued first message before acquire and replay never redelivers it', async () => { +it('an immediate Stop keeps the queued first message as a card before acquire and replay never redelivers it', async () => { const first = firstMessage() const params = createTestParams(first) let stopping: ReturnType | undefined @@ -180,14 +216,20 @@ it('an immediate Stop cancels the queued first message before acquire and replay replayed: true, value: { page: { - submissions: [{ clientMessageId: first.clientMessageId, rejection: { kind: 'cancelled' } }] + submissions: [ + { + clientMessageId: first.clientMessageId, + rejection: { kind: 'cancelled' }, + keptAsQueuedMessageId: first.clientMessageId + } + ] } } }) expect( rig.host.collaboratorsForTests().sessions.get(SESSION)?.journal.stopMarks.latest() ).toMatchObject({ event: { reason: 'user-stop' } }) - expect(rig.adapter.dispatch).not.toHaveBeenCalled() + await expectKeptAsCard(first.clientMessageId) }) it('holds the created first message until the child proves its start, then hands it over once', async () => { @@ -226,7 +268,7 @@ it('holds the created first message until the child proves its start, then hands expect(rig.adapter.dispatch).toHaveBeenCalledOnce() }) -it('Stop while the child proves its start withdraws the created first message once, unsent', async () => { +it('Stop while the child proves its start keeps the created first message as a card, and Resume sends it once', async () => { const first = firstMessage() const spawn = rig.adapter.acquire.getMockImplementation()! rig.adapter.acquire.mockImplementationOnce(async (input) => ({ @@ -241,18 +283,17 @@ it('Stop while the child proves its start withdraws the created first message on expect(await stopCreatedChat(rig.host)).toMatchObject({ ok: true, value: { cancelled: true } }) const submissions = (await rig.host.journalSnapshot(SESSION)).submissions - expect(submissions).toEqual([ - expect.objectContaining({ - clientMessageId: first.clientMessageId, - dispatchState: 'rejected', - rejection: expect.objectContaining({ kind: 'cancelled' }) - }) - ]) + expect(submissions).toHaveLength(1) expect(submissions[0].handedOverAt).toBeUndefined() - expect(rig.adapter.dispatch).not.toHaveBeenCalled() + await expectKeptAsCard(first.clientMessageId) + + expect(await resumeCreatedChat(rig.host)).toMatchObject({ ok: true, value: { resumed: true } }) + await vi.waitFor(() => expect(dispatchedTexts()).toEqual(['opening message'])) + await new Promise((resolve) => setTimeout(resolve, 50)) + expect(dispatchedTexts()).toEqual(['opening message']) }) -it('Stop aborts a handshake outside the lane and the next send sends only its own text', async () => { +it('Stop aborts a handshake outside the lane; the next send goes first, then the kept first message once', async () => { const first = firstMessage() const params = createTestParams(first) const entered = abortableAcquisition() @@ -263,16 +304,21 @@ it('Stop aborts a handshake outside the lane and the next send sends only its ow ok: true, replayed: true }) + await expectKeptAsCard(first.clientMessageId) const next = restTestSend('next message', rig.store.getRecord(SESSION)?.lease.runtimeFence) expect(await rig.host.send(CALLER, next)).toMatchObject({ ok: true }) - await vi.waitFor(() => expect(rig.adapter.dispatch).toHaveBeenCalledOnce()) - expect(rig.adapter.dispatch).toHaveBeenCalledWith(expect.objectContaining({ body: next.body })) - const snapshot = await rig.host.journalSnapshot(SESSION) - expect(snapshot.submissions).toHaveLength(2) + // The next message's accepted turn lifts the Stop's pause; the card follows it, once. + await vi.waitFor(() => expect(dispatchedTexts()).toEqual(['next message', 'opening message'])) + await new Promise((resolve) => setTimeout(resolve, 50)) + expect(dispatchedTexts()).toEqual(['next message', 'opening message']) expect( - snapshot.submissions.find(({ clientMessageId }) => clientMessageId === first.clientMessageId) - ?.rejection - ).toMatchObject({ kind: 'cancelled' }) + (await rig.host.journalSnapshot(SESSION)).submissions.find( + ({ clientMessageId }) => clientMessageId === first.clientMessageId + ) + ).toMatchObject({ + rejection: { kind: 'cancelled' }, + keptAsQueuedMessageId: first.clientMessageId + }) }) it('replays the stopped first message even after its row falls outside the hydration page', async () => { @@ -313,14 +359,15 @@ it('replays the stopped first message even after its row falls outside the hydra expect(replay.value.firstMessage).toMatchObject({ clientMessageId: first.clientMessageId, dispatchState: 'rejected', - rejection: { kind: 'cancelled' } + rejection: { kind: 'cancelled' }, + keptAsQueuedMessageId: first.clientMessageId }) expect(append).not.toHaveBeenCalled() expect(rig.adapter.acquire).not.toHaveBeenCalled() expect(rig.adapter.dispatch).not.toHaveBeenCalled() }) -it('admits Stop with the create-returned fence after acquisition advances it and keeps the stopped row', async () => { +it('admits Stop with the create-returned fence after acquisition advances it and keeps the first message as a card', async () => { const first = firstMessage() const entered = abortableAcquisition() const created = await rig.host.create(CALLER, createTestParams(first), { firstMessage: first }) @@ -337,22 +384,7 @@ it('admits Stop with the create-returned fence after acquisition advances it and }) try { await vi.waitFor(() => expect(stopped).toMatchObject({ ok: true, value: { cancelled: true } })) - const snapshot = await rig.host.journalSnapshot(SESSION) - expect(snapshot.submissions).toMatchObject([ - { - clientMessageId: first.clientMessageId, - dispatchState: 'rejected', - rejection: { kind: 'cancelled' } - } - ]) - expect( - projectStructuredAgentSessionMessages(snapshot.items, [], snapshot.submissions, { - rejectedInPlace: true - }) - ).toMatchObject([ - { role: 'user', blocks: first.body.blocks }, - { blocks: [{ type: 'text', text: NATIVE_CHAT_STOPPED_BEFORE_START_TEXT }] } - ]) + await expectKeptAsCard(first.clientMessageId) expect( rig.host.collaboratorsForTests().sessions.get(SESSION)?.journal.stopMarks.latest() ).toMatchObject({ event: { reason: 'user-stop' } }) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-dead-generation-settlement.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-dead-generation-settlement.test.ts index 07510c754e1c..c7e4643a6670 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-dead-generation-settlement.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-dead-generation-settlement.test.ts @@ -279,7 +279,7 @@ describe('dead structured-session generation settlement', () => { const settledSnapshot = journal.snapshot() const closedJournal: Pick< AgentSessionJournal, - 'snapshot' | 'submissions' | 'rejectQueuedSubmissions' | 'appendLifecycleBatch' | 'itemFence' + 'snapshot' | 'submissions' | 'appendLifecycleBatch' | 'itemFence' > = { itemFence: () => undefined, snapshot: () => ({ @@ -287,9 +287,6 @@ describe('dead structured-session generation settlement', () => { items: [settledItem] }), submissions: () => [], - rejectQueuedSubmissions: async () => { - throw new Error('journal_closed') - }, appendLifecycleBatch: async () => { throw new Error('journal_closed') } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-dead-generation-settlement.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-dead-generation-settlement.ts index c9742e18881d..599b46e661c3 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-dead-generation-settlement.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-dead-generation-settlement.ts @@ -81,8 +81,6 @@ export async function settleStructuredAgentSessionDeadGeneration(input: { /** The exit, watched: what that child's own translator could only end `unverifiable` (its stream * closed before the exit was proven) is revised in this batch. */ exit?: StructuredAgentSessionWatchedExit - /** A person's Stop ended a starting child before what it was handed could run: each is rejected so. */ - unrunRejection?: SubmissionRejectionFact }): Promise { try { const hasUnfinishedWork = hasUnfinishedStructuredAgentSessionWork(input.journal, input.exit) @@ -92,15 +90,11 @@ export async function settleStructuredAgentSessionDeadGeneration(input: { } // A queued message is the delivery loop's to settle: it was never handed to this child. A send // a child still starting was handed and never echoed did not run, and its root is gone: it is - // rejected, with the child's own diagnostic or as the Stop that ended it. A proven child's - // handed-over sends stay in doubt. + // rejected with the child's own diagnostic. A proven child's handed-over sends stay in doubt. const startupFailure = input.exitedDuringStartup ? structuredAgentSessionStartFailure({ exit: input.exitFailure }, input.failureTextContext) : null - const closed = input.unrunRejection - const unrun = - startupFailure ?? (closed && agentSessionFailureWords(closed, { surface: 'rejection' })) - const withdrawn = unrun ? [] : codexUnopenedSendResolutions(input.journal, input.fence) + const withdrawn = startupFailure ? [] : codexUnopenedSendResolutions(input.journal, input.fence) const withdrawnIds = new Set(withdrawn.map((entry) => entry.clientMessageId)) const dispatches = [ ...withdrawn, @@ -109,7 +103,7 @@ export async function settleStructuredAgentSessionDeadGeneration(input: { (entry) => !withdrawnIds.has(entry.clientMessageId) ), input.fence, - unrun ? { rejection: unrun } : { reason: input.pendingSubmissionReason } + startupFailure ? { rejection: startupFailure } : { reason: input.pendingSubmissionReason } ) ] const items = input.journal.snapshot().items diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-pause-lift.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-pause-lift.test.ts index 418c51f5cc74..0ab4b509b369 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-pause-lift.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-pause-lift.test.ts @@ -21,6 +21,12 @@ import { import { openRigTurnFor } from './structured-agent-session-queued-rig-turn.test-fixture' import { sameQueuePause } from './structured-agent-session-queued-publication' import { structuredQueuePauses } from './structured-agent-session-queued-pause' +import { stopWithdrawal } from './structured-agent-session-stop-withdrawal.test-fixture' + +vi.mock( + '../agent-session-journal/journal-unsent-send-hold', + () => import('./structured-agent-session-stop-withdrawal.test-fixture') +) let rig: QueuedMessageTestRig @@ -468,9 +474,7 @@ describe('a failed Stop', () => { it('still takes effect when its withdrawal fails, and pauses the queue', async () => { await rig.workingSend() const draftId = await queuedDraft('queued before the stop') - const reject = vi - .spyOn(AgentSessionJournal.prototype, 'rejectQueuedSubmissions') - .mockRejectedValueOnce(new Error('disk full')) + stopWithdrawal.run = () => Promise.reject(new Error('disk full')) const warned = vi.spyOn(console, 'warn').mockImplementation(() => undefined) try { expect(await rig.stop()).toMatchObject({ ok: true }) @@ -479,7 +483,7 @@ describe('a failed Stop', () => { expect.objectContaining({ step: 'withdrawal', error: new Error('disk full') }) ) } finally { - reject.mockRestore() + stopWithdrawal.run = undefined warned.mockRestore() } await expectPaused(draftId) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-stop-row.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-stop-row.test.ts index e6ecad009d4c..1e6a43282911 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-stop-row.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-stop-row.test.ts @@ -134,6 +134,10 @@ describe("Stop's event", () => { await expectHeld('stopped', first) const person = rig.send('the person asks for a turn') await person.result + // Handed over, so the later Stop takes it back from the agent rather than keeping it as a card. + await eventually(async () => + expect((await rig.submission(person.id))?.handedOverAt).toBeDefined() + ) // A later Stop supersedes: the send made before it no longer lifts anything. await rig.stop() await rig.settleAccepted(person.id, 'person') diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-stop.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-stop.ts index fea528147494..5be945ff3090 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-stop.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-stop.ts @@ -104,17 +104,18 @@ export function sentSinceStop( } /** The Stop's withdrawal of every queued send, issued at once and never awaited ahead of the - * interrupt. Bookkeeping: one that fails is reported and withdrew nothing. */ + * interrupt; `withdraw` resolves whether it settled any. Bookkeeping: one that fails is reported + * and withdrew nothing. */ export function withdrawQueuedForStop( ctx: AgentSessionTurnContext, - withdraw: () => Promise + withdraw: () => Promise ): Promise { const failed = (error: unknown): boolean => { report(ctx, 'withdrawal', error) return false } try { - return withdraw().then((withdrawn) => withdrawn.length > 0, failed) + return withdraw().catch(failed) } catch (error) { return Promise.resolve(failed(error)) } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-withdrawn-draft.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-withdrawn-draft.test.ts index 6cf71ea894c1..78490b509dac 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-withdrawn-draft.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-withdrawn-draft.test.ts @@ -6,7 +6,6 @@ import { afterEach, beforeEach, expect, it, vi } from 'vitest' import { JournalQueuedMessages } from '../agent-session-journal/journal-queued-messages' -import { AgentSessionJournal } from '../agent-session-journal/journal-store' import { HOST_TEST_SESSION as SESSION } from './structured-agent-session-host-test-data' import { holdDelivery } from './structured-agent-session-delivery-hold.test-fixture' import { @@ -14,6 +13,12 @@ import { eventually, type QueuedMessageTestRig } from './structured-agent-session-queued-message-rig.test-fixture' +import { stopWithdrawal } from './structured-agent-session-stop-withdrawal.test-fixture' + +vi.mock( + '../agent-session-journal/journal-unsent-send-hold', + () => import('./structured-agent-session-stop-withdrawal.test-fixture') +) let rig: QueuedMessageTestRig @@ -99,26 +104,20 @@ it('Stop, then a user send: the withdrawn draft and the paused cards behind it d expect(await rig.drafts()).toEqual([]) }) -// Bookkeeping never fails a Stop: a withdrawal that throws is reported and counts as nothing -// withdrawn, so no pause holds the card it did send back, and that card sends again. -it('a Stop whose withdrawal throws after landing still answers; the draft it released sends again under a fresh id', async () => { +// Bookkeeping never fails a Stop: a withdrawal that throws is reported. The Stop's event was +// issued with it, so the card it did send back waits under the pause until Resume sends it again. +it('a Stop whose withdrawal throws after landing still answers; Resume sends the draft it released again under a fresh id', async () => { const working = await rig.workingSend() const a = await queuedDraft('A') const { release } = holdDelivery() await rig.settleAccepted(working, 'working') await eventually(async () => expect(await rig.handoff(a)).toBeDefined()) const firstA = await rig.handoffId(a) - const withdraw = AgentSessionJournal.prototype.rejectQueuedSubmissions - const failing = vi - .spyOn(AgentSessionJournal.prototype, 'rejectQueuedSubmissions') - .mockImplementation(async function (this: AgentSessionJournal, ...args) { - const withdrawn = await withdraw.apply(this, args) - // Only the Stop's own withdrawal fails, after it landed; the delivery loop's pass through. - if (args[1].rejection.kind === 'cancelled') { - throw new Error('disk full') - } - return withdrawn - }) + // Only the Stop's own withdrawal fails, after it landed; every other hold passes through. + stopWithdrawal.run = async (settle) => { + await settle() + throw new Error('disk full') + } const warned = vi.spyOn(console, 'warn').mockImplementation(() => undefined) try { const stopped = rig.stop() @@ -129,12 +128,16 @@ it('a Stop whose withdrawal throws after landing still answers; the draft it rel expect.objectContaining({ step: 'withdrawal', error: new Error('disk full') }) ) } finally { - failing.mockRestore() + stopWithdrawal.run = undefined warned.mockRestore() release() } expect((await rig.submission(firstA))?.dispatchState).toBe('rejected') const before = new Set([working, firstA]) + expect(await rig.drafts()).toEqual([{ messageId: a, state: 'waiting' }]) + expect(await rig.queuePause()).toMatchObject({ reason: 'stopped' }) + expect((await submissionIds()).filter((id) => !before.has(id))).toEqual([]) + expect(await rig.resume()).toMatchObject({ ok: true }) await eventually(async () => { expect((await submissionIds()).filter((id) => !before.has(id))).toHaveLength(1) expect(await rig.drafts()).toEqual([]) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-stop-event-binding.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-stop-event-binding.test.ts index c09d2eba456c..d519f63058ff 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-stop-event-binding.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-stop-event-binding.test.ts @@ -104,19 +104,22 @@ async function queuedDraft(text: string): Promise { return queued.value.queued.messageId } -/** A person's Stop of a start that never landed, whose send opens no turn. `held`: a card queued - * behind the start, which the Stop holds. */ -async function stopOfStart(options: { held?: true } = {}): Promise { +/** A person's Stop of a start that never landed, whose send opens no turn: it resolves that send, + * which the Stop kept as a card at the head of the queue. `held`: a card queued behind the start, + * which the Stop holds too. */ +async function stopOfStart(options: { held?: true } = {}): Promise { rig = await createQueuedMessageTestRig({ starting: true, restartable: true }) // Held for a child that never proves its start, so it is never handed over. - rig.send('work on this') + const work = rig.send('work on this') await eventually(() => expect(childPhase()).toBe('starting')) - const held = options.held ? await queuedDraft('queued behind the start') : undefined + if (options.held) { + await queuedDraft('queued behind the start') + } expect(await rig.stop()).toMatchObject({ ok: true }) expect(stopEvents()).toEqual([expect.objectContaining({ reason: 'user-stop' })]) expect(stopEvents()[0]).not.toHaveProperty('turnId') await eventually(() => expect(childPhase()).toBeUndefined()) - return held + return work.id } /** Orchestration mail after the Stop starts a new child and its turn runs. */ @@ -200,7 +203,7 @@ describe('a Stop of a start that never landed binds no later turn', () => { }) it("writes the host's event when it evicts the turn of a card the Stop held, which Resume sent", async () => { - const held = (await stopOfStart({ held: true }))! + const held = await stopOfStart({ held: true }) expect(await rig.resume()).toMatchObject({ ok: true }) await eventually(async () => expect(await rig.handoff(held)).toBeDefined()) await turnOpenedBy(await rig.handoffId(held)) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-stop-idle-kept-card.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-stop-idle-kept-card.test.ts new file mode 100644 index 000000000000..3445b1490af9 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-stop-idle-kept-card.test.ts @@ -0,0 +1,104 @@ +// A person's Stop on a ready, idle agent, landing between a send's accept and its handover: the +// send comes back as a card, and every frame clients are published shows it under the Stop's +// pause, never as the next card to send. Against the real host, store and journal. + +import { afterEach, beforeEach, expect, it, vi } from 'vitest' +import type { AgentSessionJournal } from '../agent-session-journal/journal-store' +import { holdDelivery } from './structured-agent-session-delivery-hold.test-fixture' +import { + structuredQueueSendGate, + tryReadQueuePublication +} from './structured-agent-session-queued-publication' +import { + createQueuedMessageTestRig, + eventually, + type QueuedMessageTestRig +} from './structured-agent-session-queued-message-rig.test-fixture' +import { HOST_TEST_SESSION as SESSION } from './structured-agent-session-host-test-data' + +type Frame = { cards: string[]; pause: string | null; next: string | null } + +let rig: QueuedMessageTestRig + +beforeEach(async () => { + rig = await createQueuedMessageTestRig({ restartable: true }) +}) + +afterEach(async () => { + vi.restoreAllMocks() + await rig.dispose() +}) + +function journal(): AgentSessionJournal { + const open = rig.host.collaboratorsForTests().sessions.get(SESSION)?.journal + if (!open) { + throw new Error('the conversation is not open') + } + return open +} + +/** A send the delivery loop holds before its handover, on an agent that is ready and idle. */ +async function sendHeldOnIdleAgent(): Promise<{ id: string; release: () => void }> { + const working = await rig.workingSend() + await rig.settleAccepted(working, 'working') + await eventually(() => expect(journal().activeTurnId()).toBeNull()) + const hold = holdDelivery() + const typed = rig.send('typed then stopped') + await typed.result + await hold.held + return { id: typed.id, release: hold.release } +} + +/** Every queue publication clients are sent from now on. */ +function recordFrames(): Frame[] { + const frames: Frame[] = [] + const { subscribers } = rig.host.collaboratorsForTests() + const publish = subscribers.publish.bind(subscribers) + vi.spyOn(subscribers, 'publish').mockImplementation((sessionId, published, activity) => { + const queue = tryReadQueuePublication(published, structuredQueueSendGate(rig.store, SESSION)) + frames.push({ + cards: (queue?.queuedMessages ?? []).map((card) => card.messageId), + pause: queue?.queuePause?.reason ?? null, + next: queue?.nextQueuedMessageId ?? null + }) + publish(sessionId, published, activity) + }) + return frames +} + +it('publishes the kept card only under the Stop pause', async () => { + const typed = await sendHeldOnIdleAgent() + const dispatches = rig.dispatch.mock.calls.length + const frames = recordFrames() + + const stopped = rig.stop() + typed.release() + expect(await stopped).toMatchObject({ ok: true, value: { cancelled: true } }) + + const withCard = frames.filter((frame) => frame.cards.includes(typed.id)) + expect(withCard.length).toBeGreaterThan(0) + for (const frame of withCard) { + expect(frame).toMatchObject({ pause: 'stopped', next: null }) + } + expect(await rig.drafts()).toEqual([{ messageId: typed.id, state: 'waiting' }]) + expect(rig.dispatch.mock.calls.length).toBe(dispatches) +}) + +// A rare storage failure, accepted: with no Stop recorded nothing pauses the queue, so the kept +// card sends as any card would, once. +it("sends the kept card once when the Stop's own record fails to save", async () => { + const typed = await sendHeldOnIdleAgent() + const dispatches = rig.dispatch.mock.calls.length + vi.spyOn(journal(), 'appendStopEvent').mockRejectedValue(new Error('disk full')) + + const stopped = rig.stop() + typed.release() + expect(await stopped).toMatchObject({ ok: true, value: { cancelled: true } }) + + await eventually(async () => { + expect(await rig.drafts()).toEqual([]) + expect(rig.dispatch.mock.calls.length).toBe(dispatches + 1) + }) + expect(await rig.queuePause()).toBeNull() + expect((await rig.submission(typed.id))?.keptAsQueuedMessageId).toBe(typed.id) +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-stop-keeps-unsent.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-stop-keeps-unsent.test.ts new file mode 100644 index 000000000000..264d2288fc44 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-stop-keeps-unsent.test.ts @@ -0,0 +1,428 @@ +// A person's Stop settles a message the host still holds as a close does: the agent never got it, +// so the person's words come back as a card at the head of the queue, held by the Stop's pause, +// and the conversation shows no bubble for them. Against the real host, store and journal, with +// an agent that stays starting until the test proves its start. + +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { agentSessionFailureFact } from '../../../shared/agent-session-failure' +import { agentSessionFailureWords } from '../../../shared/agent-session-failure-words' +import { AGENT_JOURNAL_THREAD_SCOPE } from '../../../shared/agent-session-journal-types' +import type { AgentMessageSource } from '../../../shared/agent-session-message-source' +import { projectStructuredAgentSessionMessages } from '../../../shared/structured-agent-session-message-projection' +import { JournalQueuedMessages } from '../agent-session-journal/journal-queued-messages' +import type { AgentSessionJournal } from '../agent-session-journal/journal-store' +import { holdDelivery } from './structured-agent-session-delivery-hold.test-fixture' +import { structuredQueuePauses } from './structured-agent-session-queued-pause' +import { + createQueuedMessageTestRig, + eventually, + QUEUED_RIG_CALLER, + type QueuedMessageTestRig +} from './structured-agent-session-queued-message-rig.test-fixture' +import { + HOST_TEST_SESSION as SESSION, + hostTestMessage, + hostTestOperationId +} from './structured-agent-session-host-test-data' + +const CANCELLED = agentSessionFailureWords(agentSessionFailureFact('cancelled'), { + surface: 'rejection' +}) +/** A kept send: an ordinary waiting card, with no hold of its own. */ +const KEPT = { state: 'waiting' } + +/** Orchestration mail as the mailbox sends it: from another agent, naming its sender. */ +const MAIL_SOURCE: AgentMessageSource = { + kind: 'agent', + senders: [ + { + party: { address: 'agent:coordinator', terminalHandle: null, orcaSessionId: null }, + name: null + } + ], + orchestration: { message: 'mail-notice', mailbox: 'agent:worker', dispatchId: null, messages: [] } +} + +let rig: QueuedMessageTestRig + +beforeEach(async () => { + rig = await createQueuedMessageTestRig({ restartable: true, starting: true }) +}) + +afterEach(async () => { + vi.restoreAllMocks() + await rig.dispose() +}) + +function journal(): AgentSessionJournal { + const open = rig.host.collaboratorsForTests().sessions.get(SESSION)?.journal + if (!open) { + throw new Error('the conversation is not open') + } + return open +} + +function dispatchedTexts(): string[] { + return rig.dispatch.mock.calls.map(([input]) => + input.body.blocks.map((block) => (block.type === 'text' ? block.text : '')).join('') + ) +} + +/** The exact request a client sends, so a test can send the same one again. */ +function sendRequest(text: string) { + const body = hostTestMessage(text) + return { + envelope: rig.envelope({ body }, 'agentSession.send', hostTestOperationId()), + body, + userSend: true as const + } +} + +/** Sent while the agent starts, and held: none is handed over. */ +async function heldWhileStarting(...requests: ReturnType[]): Promise { + // No child, so the first send starts one, which stays starting. + await rig.host.close(SESSION, 'evict') + const ids: string[] = [] + for (const request of requests) { + expect(await rig.host.send(QUEUED_RIG_CALLER, request)).toMatchObject({ + ok: true, + value: { submission: { dispatchState: 'pending', handoverRecorded: true } } + }) + ids.push(request.envelope.clientOperationId) + } + await eventually(() => + expect(rig.host.collaboratorsForTests().sessions.get(SESSION)?.child?.phase).toBe('starting') + ) + for (const id of ids) { + expect((await rig.submission(id))?.handedOverAt).toBeUndefined() + } + return ids +} + +async function queuedCard(text: string): Promise { + const queued = await rig.send(text, 'queue-if-active').result + if (!queued.ok || !('queued' in queued.value)) { + throw new Error('expected a queued receipt') + } + return queued.value.queued.messageId +} + +/** The texts a client's transcript draws, desktop (`rejectedInPlace`) or phone. */ +async function transcriptTexts(rejectedInPlace: boolean): Promise { + const page = await rig.host.history({ sessionId: SESSION, direction: 'tail' }) + if (!page.ok) { + throw new Error('history refused') + } + return projectStructuredAgentSessionMessages(page.page.items, [], page.page.submissions, { + rejectedInPlace + }).map( + (message) => + `${message.blocks.map((block) => ('text' in block ? block.text : '')).join('')}${ + message.stoppedBeforeStart ? ' (stopped)' : '' + }` + ) +} + +async function stopKeepsBoth(): Promise<[string, string]> { + const [first, second] = await heldWhileStarting(sendRequest('first'), sendRequest('second')) + expect(await rig.stop()).toMatchObject({ ok: true, value: { cancelled: true } }) + await eventually(async () => + expect(await rig.drafts()).toEqual([ + { messageId: first, ...KEPT }, + { messageId: second, ...KEPT } + ]) + ) + return [first!, second!] +} + +describe('a Stop while the agent starts, with two messages held', () => { + it('keeps both as cards at the head of the queue, held by the Stop, with no bubble and no write', async () => { + const [first, second] = await heldWhileStarting(sendRequest('first'), sendRequest('second')) + // A card queued while the start held them stands behind them. + const queued = rig.send('queued behind the start', 'queue-if-active') + expect(await queued.result).toMatchObject({ ok: true, value: { queued: { state: 'waiting' } } }) + + expect(await rig.stop()).toMatchObject({ ok: true, value: { cancelled: true } }) + + await eventually(async () => + expect(await rig.drafts()).toEqual([ + { messageId: first, ...KEPT }, + { messageId: second, ...KEPT }, + { messageId: queued.id, state: 'waiting' } + ]) + ) + expect(rig.dispatch).not.toHaveBeenCalled() + expect(await rig.queuePause()).toEqual({ reason: 'stopped' }) + for (const [id, text] of [ + [first, 'first'], + [second, 'second'] + ] as const) { + expect(await rig.submission(id!)).toMatchObject({ + dispatchState: 'rejected', + ...CANCELLED, + keptAsQueuedMessageId: id + }) + expect(journal().queuedMessages.get(id!)).toMatchObject({ + holdReason: null, + body: hostTestMessage(text) + }) + } + // The card is the only surface: neither client draws a bubble or a stop row for them. + expect(await transcriptTexts(true)).toEqual([]) + expect(await transcriptTexts(false)).toEqual([]) + await new Promise((resolve) => setTimeout(resolve, 100)) + expect(rig.dispatch).not.toHaveBeenCalled() + }) + + it('Resume sends them in order, each exactly once', async () => { + const [first, second] = await stopKeepsBoth() + + expect(await rig.resume()).toMatchObject({ ok: true, value: { resumed: true } }) + await rig.proveStart() + await eventually(() => expect(dispatchedTexts()).toEqual(['first'])) + await rig.settleAccepted(await rig.handoffId(first), 'first') + await eventually(() => expect(dispatchedTexts()).toEqual(['first', 'second'])) + await rig.settleAccepted(await rig.handoffId(second), 'second') + await new Promise((resolve) => setTimeout(resolve, 100)) + expect(dispatchedTexts()).toEqual(['first', 'second']) + expect(await rig.drafts()).toEqual([]) + }) + + it("the person's next message sends first; its accepted turn lifts the pause, then the cards drain once, in order", async () => { + const [first, second] = await stopKeepsBoth() + + const next = rig.send('what was the last message I sent?') + await next.result + await rig.proveStart() + await eventually(() => expect(dispatchedTexts()).toEqual(['what was the last message I sent?'])) + expect(await rig.handoff(first)).toBeUndefined() + await rig.settleAccepted(next.id, 'next') + await eventually(() => + expect(dispatchedTexts()).toEqual(['what was the last message I sent?', 'first']) + ) + await rig.settleAccepted(await rig.handoffId(first), 'first') + await eventually(() => + expect(dispatchedTexts()).toEqual(['what was the last message I sent?', 'first', 'second']) + ) + await rig.settleAccepted(await rig.handoffId(second), 'second') + await new Promise((resolve) => setTimeout(resolve, 100)) + expect(dispatchedTexts()).toHaveLength(3) + }) + + it('a same-id Retry of a kept send replays its receipt and writes nothing to the agent', async () => { + const request = sendRequest('sent again') + const [id] = await heldWhileStarting(request) + expect(await rig.stop()).toMatchObject({ ok: true }) + await eventually(async () => expect(await rig.drafts()).toEqual([{ messageId: id, ...KEPT }])) + const submissionsBefore = (await rig.host.journalSnapshot(SESSION)).submissions.length + + expect(await rig.host.send(QUEUED_RIG_CALLER, request)).toMatchObject({ + ok: true, + replayed: true, + value: { + clientMessageId: id, + submission: { dispatchState: 'rejected', ...CANCELLED, keptAsQueuedMessageId: id } + } + }) + expect((await rig.host.journalSnapshot(SESSION)).submissions).toHaveLength(submissionsBefore) + expect(await rig.drafts()).toEqual([{ messageId: id, ...KEPT }]) + await new Promise((resolve) => setTimeout(resolve, 100)) + expect(rig.dispatch).not.toHaveBeenCalled() + }) +}) + +describe('the kept cards afterwards', () => { + it('a host restart keeps them, held, and sends nothing', async () => { + const [first, second] = await stopKeepsBoth() + + rig.crashRestartHostProcess() + + expect(await rig.drafts()).toEqual([ + { messageId: first, ...KEPT }, + { messageId: second, ...KEPT } + ]) + expect(structuredQueuePauses(journal()).map((pause) => pause.reason)).toContain('restarted') + await new Promise((resolve) => setTimeout(resolve, 100)) + expect(rig.dispatch).not.toHaveBeenCalled() + }) + + it('/clear carries them under its own pause', async () => { + const [first, second] = await stopKeepsBoth() + 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(await rig.drafts()).toEqual([ + { messageId: first, ...KEPT }, + { messageId: second, ...KEPT } + ]) + expect(structuredQueuePauses(journal()).map((pause) => pause.reason)).toContain('cleared') + expect(rig.dispatch).not.toHaveBeenCalled() + }) + + it("a rewind's new epoch restates the Stop's pause over them", async () => { + const [first, second] = await stopKeepsBoth() + + await journal().replaceEpochItems( + 'handle_forked', + rig.store.getRecord(SESSION)!.lease.runtimeFence, + [] + ) + + expect(structuredQueuePauses(journal()).map((pause) => pause.reason)).toEqual(['stopped']) + expect(await rig.queuePause()).toEqual({ reason: 'stopped' }) + expect(await rig.drafts()).toEqual([ + { messageId: first, ...KEPT }, + { messageId: second, ...KEPT } + ]) + await new Promise((resolve) => setTimeout(resolve, 100)) + expect(rig.dispatch).not.toHaveBeenCalled() + }) +}) + +describe('what a Stop does not keep', () => { + it('orchestration mail held behind the start is rejected as before, with no card', async () => { + await rig.host.close(SESSION, 'evict') + const mail = rig.send('mail notice', undefined, { internal: true, from: MAIL_SOURCE }) + await mail.result + await eventually(() => + expect(rig.host.collaboratorsForTests().sessions.get(SESSION)?.child?.phase).toBe('starting') + ) + + expect(await rig.stop()).toMatchObject({ ok: true }) + + await eventually(async () => + expect(await rig.submission(mail.id)).toMatchObject({ + dispatchState: 'rejected', + ...CANCELLED + }) + ) + expect(await rig.submission(mail.id)).not.toHaveProperty('keptAsQueuedMessageId') + expect(await rig.drafts()).toEqual([]) + }) + + it("a card's own hand-off is returned to its place under the pause, and makes no second card", async () => { + const working = await rig.workingSend() + const card = await queuedCard('a queued card') + const behind = await queuedCard('behind it') + // The turn ends and the drain hands the card off; its delivery is held, so it is not handed over. + const { release } = holdDelivery() + await rig.settleAccepted(working, 'working') + await eventually(async () => expect(await rig.handoff(card)).toBeDefined()) + const handoffId = await rig.handoffId(card) + expect((await rig.submission(handoffId))?.handedOverAt).toBeUndefined() + + const stopped = rig.stop() + release() + expect(await stopped).toMatchObject({ ok: true }) + + expect(await rig.submission(handoffId)).toMatchObject({ + dispatchState: 'rejected', + ...CANCELLED + }) + expect(await rig.submission(handoffId)).not.toHaveProperty('keptAsQueuedMessageId') + expect(await rig.drafts()).toEqual([ + { messageId: card, state: 'waiting' }, + { messageId: behind, state: 'waiting' } + ]) + expect(await rig.queuePause()).toEqual({ reason: 'stopped' }) + }) + + it('a Send now cut short by the Stop returns its card where it stood', async () => { + const working = await rig.workingSend() + const first = await queuedCard('first card') + const pushed = await queuedCard('pushed ahead') + const drain = holdDelivery() + await rig.settleAccepted(working, 'working') + await eventually(async () => expect(await rig.handoff(first)).toBeDefined()) + const stopped = rig.stop() + drain.release() + await stopped + // Send now on the second card; its hand-off is held, then a Stop takes it back. + const send = holdDelivery() + expect(await rig.sendNow(pushed)).toMatchObject({ ok: true }) + await eventually(async () => expect(await rig.handoff(pushed)).toBeDefined()) + expect((await rig.handoff(pushed))?.handedOverAt).toBeUndefined() + const again = rig.stop() + send.release() + expect(await again).toMatchObject({ ok: true }) + + expect(await rig.handoff(pushed)).toMatchObject({ dispatchState: 'rejected', origin: 'client' }) + expect(await rig.drafts()).toEqual([ + { messageId: first, state: 'waiting' }, + { messageId: pushed, state: 'waiting' } + ]) + }) + + it('a Stop with nothing held changes nothing and writes no Stop', async () => { + // An idle chat: no agent runs and nothing is queued. + await rig.host.close(SESSION, 'evict') + expect(await rig.stop()).toMatchObject({ ok: true, value: { cancelled: false } }) + expect(journal().stopMarks.latest()).toBeNull() + expect(await rig.drafts()).toEqual([]) + expect(await rig.queuePause()).toBeNull() + }) +}) + +describe('a card that cannot be written', () => { + it('falls back to the rejection, and the Stop still completes and is written', async () => { + const [id] = await heldWhileStarting(sendRequest('card write fails')) + const warned = vi.spyOn(console, 'warn').mockImplementation(() => undefined) + vi.spyOn(JournalQueuedMessages.prototype, 'holdInTransaction').mockImplementation(() => { + throw new Error('disk full') + }) + + expect(await rig.stop()).toMatchObject({ ok: true, value: { cancelled: true } }) + + await eventually(async () => + expect(await rig.submission(id!)).toMatchObject({ dispatchState: 'rejected', ...CANCELLED }) + ) + expect(await rig.submission(id!)).not.toHaveProperty('keptAsQueuedMessageId') + expect(await rig.drafts()).toEqual([]) + expect(journal().stopMarks.latest()?.event.reason).toBe('user-stop') + expect(warned).toHaveBeenCalledWith( + '[journal-hold] keeping an unsent send failed:', + expect.objectContaining({ clientMessageId: id, cause: 'cancelled' }) + ) + expect(rig.dispatch).not.toHaveBeenCalled() + }) +}) + +describe('a start that never answered, ended by the Stop', () => { + it("keeps a person's message the child was handed and never ran as a card", async () => { + await rig.dispose() + rig = await createQueuedMessageTestRig({ + restartable: true, + starting: true, + startUnanswered: true, + stopEndsSession: true + }) + const [id] = await heldWhileStarting(sendRequest('handed to a start that never answered')) + // Handed to the child before its start answered, as a build before held sends did. + await journal().resolveDispatch({ + clientMessageId: id!, + state: 'pending', + turnScope: AGENT_JOURNAL_THREAD_SCOPE, + fence: rig.store.getRecord(SESSION)!.lease.runtimeFence + }) + expect((await rig.submission(id!))?.handedOverAt).toBeDefined() + + expect(await rig.stop()).toMatchObject({ ok: true, value: { cancelled: true } }) + + await eventually(async () => + expect(await rig.submission(id!)).toMatchObject({ + dispatchState: 'rejected', + ...CANCELLED, + keptAsQueuedMessageId: id + }) + ) + expect(await rig.drafts()).toEqual([{ messageId: id, ...KEPT }]) + expect(structuredQueuePauses(journal()).map((pause) => pause.reason)).toEqual(['stopped']) + expect(await transcriptTexts(true)).toEqual([]) + }) +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-stop-never-waits.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-stop-never-waits.test.ts index 4cbc39073c17..e1e1a899a2c5 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-stop-never-waits.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-stop-never-waits.test.ts @@ -16,6 +16,12 @@ import { HOST_TEST_SESSION as SESSION, HOST_TEST_THREAD as THREAD } from './structured-agent-session-host-test-data' +import { stopWithdrawal } from './structured-agent-session-stop-withdrawal.test-fixture' + +vi.mock( + '../agent-session-journal/journal-unsent-send-hold', + () => import('./structured-agent-session-stop-withdrawal.test-fixture') +) let host: StructuredAgentSessionHost let acquire: Mock @@ -28,6 +34,7 @@ beforeEach(() => { }) afterEach(() => { + stopWithdrawal.run = undefined vi.restoreAllMocks() }) @@ -91,10 +98,8 @@ describe.each([ ['naming its turn', { turnId: 'turn-1' }] ])('a Stop %s', (_label, fields) => { it('interrupts when withdrawing the queued sends throws, and reports it', async () => { - const journal = await runningTurn() - vi.spyOn(journal, 'rejectQueuedSubmissions').mockImplementation(() => { - throw new Error(MALFORMED) - }) + await runningTurn() + stopWithdrawal.run = () => Promise.reject(new Error(MALFORMED)) expect(await stop(fields)).toMatchObject({ ok: true }) expect(cancelTurn).toHaveBeenCalledOnce() @@ -121,7 +126,7 @@ describe.each([ const order: string[] = [] const held = Promise.withResolvers() if (step === 'withdrawing the queued sends') { - vi.spyOn(journal, 'rejectQueuedSubmissions').mockImplementation(() => held.promise) + stopWithdrawal.run = () => held.promise } else { vi.spyOn(journal, 'appendStopEvent').mockImplementation(() => held.promise) } @@ -140,7 +145,8 @@ describe.each([ expect(await stopping).toMatchObject({ ok: true }) expect(order).toEqual(['interrupt', 'write fails']) - expectReported(failed) + // The answer never waits on the write, so its report may land after it. + await vi.waitFor(() => expectReported(failed)) }) it.each([ @@ -153,7 +159,7 @@ describe.each([ const order: string[] = [] const held = Promise.withResolvers() if (step === 'withdrawing the queued sends') { - vi.spyOn(journal, 'rejectQueuedSubmissions').mockImplementation(() => held.promise) + stopWithdrawal.run = () => held.promise } else { vi.spyOn(journal, 'appendStopEvent').mockImplementation(() => held.promise) } @@ -172,7 +178,7 @@ describe.each([ expect(await stopping).toMatchObject({ ok: true, value: { cancelled: true } }) expect(order).toEqual(['stop', 'write fails']) - expectReported(failed) + await vi.waitFor(() => expectReported(failed)) } ) @@ -193,7 +199,7 @@ describe.each([ Object.assign(host.deps.adapter, { stopEndsSession: () => true, closeSession }) const held = Promise.withResolvers() if (step === 'withdrawing the queued sends') { - vi.spyOn(journal, 'rejectQueuedSubmissions').mockImplementation(() => held.promise) + stopWithdrawal.run = () => held.promise } else { vi.spyOn(journal, 'appendStopEvent').mockImplementation(() => held.promise) } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-stop-withdrawal.test-fixture.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-stop-withdrawal.test-fixture.ts new file mode 100644 index 000000000000..31d727c5eb38 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-stop-withdrawal.test-fixture.ts @@ -0,0 +1,25 @@ +// Stands in for `journal-unsent-send-hold` in suites that make a Stop's withdrawal (its `cancelled` +// hold) fail or hang; every other hold runs as is. A suite installs it with +// `vi.mock('../agent-session-journal/journal-unsent-send-hold', () => import())`. + +import { vi } from 'vitest' +import type * as UnsentSendHold from '../agent-session-journal/journal-unsent-send-hold' + +type HoldModule = typeof UnsentSendHold +type Settled = ReturnType + +const actual = await vi.importActual( + '../agent-session-journal/journal-unsent-send-hold' +) + +/** While `run` is set, it answers the Stop's withdrawal; `settle` runs the real one. */ +export const stopWithdrawal: { run?: (settle: () => Settled) => Settled } = {} + +export const { unsentSendKeptAsCard, isUnansweredHandedOverSubmission } = actual + +export function holdUnsentSends(...args: Parameters): Settled { + const { run } = stopWithdrawal + return args[1].hold.cause === 'cancelled' && run + ? run(() => actual.holdUnsentSends(...args)) + : actual.holdUnsentSends(...args) +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-unrun-send-hold.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-unrun-send-hold.test.ts index e4de581bf067..daa5b18edba3 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-unrun-send-hold.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-unrun-send-hold.test.ts @@ -1,8 +1,8 @@ // A message sent while the agent's start never answers provably never ran: the host hands a child // nothing until it proves its start. So when the chat ends then, the message is settled as a -// queued one is for the same end (`journal-unsent-send-hold.ts`): a quit or a close keeps a -// person's words as an ordinary card that waits for the chat's next turn, a person's Stop -// withdraws it. Against the real host, store and journal, with an agent that stays starting. +// queued one is for the same end (`journal-unsent-send-hold.ts`): a quit, a close or a person's +// Stop keeps a person's words as an ordinary card, which waits out the pause in force. Against the +// real host, store and journal, with an agent that stays starting. import { afterEach, beforeEach, describe, expect, it } from 'vitest' import { agentSessionFailureFact } from '../../../shared/agent-session-failure' @@ -118,7 +118,7 @@ describe('a message held behind a start that never answered, then the chat ends' } ) - it("is withdrawn by a person's Stop, as a queued one is, and kept as no card", async () => { + it("is a card after a person's Stop, rejected as stopped, held by the Stop until a turn", async () => { const id = await heldBehindHungStart('stopped') expect(await rig.stop()).toMatchObject({ ok: true, value: { cancelled: true } }) @@ -126,9 +126,11 @@ describe('a message held behind a start that never answered, then the chat ends' await eventually(async () => expect(await rig.submission(id)).toMatchObject({ dispatchState: 'rejected', - ...words('cancelled') + ...words('cancelled'), + keptAsQueuedMessageId: id }) ) - expect(await rig.drafts()).toEqual([]) + expect(await rig.drafts()).toEqual([{ messageId: id, ...KEPT }]) + expect(derivedPauses()).toEqual(['stopped']) }) }) diff --git a/src/main/runtime/rpc/methods/structured-agent-session-create-message.test.ts b/src/main/runtime/rpc/methods/structured-agent-session-create-message.test.ts index 82307145453c..613f8319679e 100644 --- a/src/main/runtime/rpc/methods/structured-agent-session-create-message.test.ts +++ b/src/main/runtime/rpc/methods/structured-agent-session-create-message.test.ts @@ -111,7 +111,8 @@ it.each([ firstMessage: { clientMessageId: input.firstMessage?.clientMessageId, dispatchState: 'rejected', - rejection: { kind: 'cancelled' } + rejection: { kind: 'cancelled' }, + keptAsQueuedMessageId: input.firstMessage?.clientMessageId } } } diff --git a/src/main/runtime/runtime-worktree-agent-rows-verdict.test.ts b/src/main/runtime/runtime-worktree-agent-rows-verdict.test.ts index ca955e241c0d..a8e1bc7afc64 100644 --- a/src/main/runtime/runtime-worktree-agent-rows-verdict.test.ts +++ b/src/main/runtime/runtime-worktree-agent-rows-verdict.test.ts @@ -138,13 +138,16 @@ describe('a request that failed reads as failed through the feed, the ingest and fence: 1, handoverRecorded: true }) - await journal.rejectQueuedSubmissions( - 1, - agentSessionFailureWords(agentSessionFailureFact('notSignedIn'), { + await journal.resolveDispatch({ + clientMessageId: 'first', + state: 'rejected', + ...agentSessionFailureWords(agentSessionFailureFact('notSignedIn'), { surface: 'rejection', agentName: 'Claude' - }) - ) + }), + fence: 1, + recovered: true + }) const summary = publishedSummary(journal) expect(summary).toMatchObject({ status: 'idle', turnOutcome: 'failure', latestPrompt: 'hello' }) @@ -229,10 +232,13 @@ describe('a request that failed reads as failed through the feed, the ingest and fence: 1, handoverRecorded: true }) - await journal.rejectQueuedSubmissions( - 1, - agentSessionFailureWords(agentSessionFailureFact('cancelled'), { surface: 'rejection' }) - ) + await journal.resolveDispatch({ + clientMessageId: 'first', + state: 'rejected', + ...agentSessionFailureWords(agentSessionFailureFact('cancelled'), { surface: 'rejection' }), + fence: 1, + recovered: true + }) expect(publishedSummary(journal)).toMatchObject({ status: null }) expect(ingest(publishedSummary(journal)).ps).toBeUndefined() diff --git a/src/renderer/src/lib/structured-agent-session-launch-create-message.test.ts b/src/renderer/src/lib/structured-agent-session-launch-create-message.test.ts index 45613b5c7aaf..ba615d772267 100644 --- a/src/renderer/src/lib/structured-agent-session-launch-create-message.test.ts +++ b/src/renderer/src/lib/structured-agent-session-launch-create-message.test.ts @@ -59,7 +59,10 @@ import { findStructuredAgentSessionReadOwner, resetStructuredAgentSessionReadOwnersForTests } from '@/components/native-chat/structured-agent-session-read-owner' -import { resetStructuredAgentSessionSendsForTests } from '@/components/native-chat/structured-agent-session-message-sender' +import { + resetStructuredAgentSessionSendsForTests, + settleStructuredAgentSessionSendsFromJournal +} from '@/components/native-chat/structured-agent-session-message-sender' import { getStructuredAgentSessionPendingSends } from '@/components/native-chat/structured-agent-session-pending-sends' import { structuredAgentSessionStopControl } from '@/components/native-chat/structured-agent-session-stop-control' import { @@ -193,6 +196,32 @@ describe('the first message owned by create', () => { expect(mocks.call.mock.calls.map(([, method]) => method)).toEqual(['agentSession.create']) }) + // One surface: a host that keeps a Stopped first message as a card owns it; the composer stays + // empty even when create never answers. + it('leaves a first message a Stop kept as a card to the card, even when create never answers', async () => { + mocks.call.mockReturnValue(new Promise(() => {})) + const launch = start() + await vi.waitFor(() => expect(mocks.call).toHaveBeenCalledOnce()) + const id = createParams().firstMessage?.clientMessageId ?? '' + settleStructuredAgentSessionSendsFromJournal( + launch.sessionId, + [ + { + ...submission(id), + dispatchState: 'rejected', + rejection: { kind: 'cancelled' }, + keptAsQueuedMessageId: id + } + ], + [id] + ) + await expect(launch.promptDeliveryResult).resolves.toMatchObject({ delivered: true }) + await vi.advanceTimersByTimeAsync(30_000) + expect(draft(launch.sessionId)).toBe('') + expect(getStructuredAgentSessionPendingSends(launch.sessionId)).toEqual([]) + expect(mocks.call.mock.calls.map(([, method]) => method)).toEqual(['agentSession.create']) + }) + it('returns a refused create once after the new draft, with no later send', async () => { const create = deferred() mocks.call.mockReturnValue(create.promise) diff --git a/src/shared/native-chat-stopped-send-turn.test.ts b/src/shared/native-chat-stopped-send-turn.test.ts index c5004cae3cc2..d8aeb81734ac 100644 --- a/src/shared/native-chat-stopped-send-turn.test.ts +++ b/src/shared/native-chat-stopped-send-turn.test.ts @@ -399,3 +399,25 @@ describe('a send a Stop took back on a page that starts inside the turn it opene ]) }) }) + +describe('a send a Stop kept as a card', () => { + it('opens no turn, though it was held while the first send opened one and kept after', () => { + // The opener's fate is unknown once the Stop ended the agent, so nothing else names the turn. + const items = [ + sent('opener', 1), + turn('t1', 3, 'codex:thread-1:t1:0'), + stopNote('t1', 4), + sent('kept', 5) + ] + const submissions = [ + submission('opener', { dispatchState: 'unknown', submittedSequence: 1, resolvedAt: null }), + stopped('kept', { submittedSequence: 2, keptAsQueuedMessageId: 'kept' }) + ] + + expect(structuredAgentTurnAnchors(items, submissions).get('t1')).toBe('t1') + expect(drawn(items, submissions)).toEqual([ + { id: key('opener'), turn: key('opener') }, + { id: 'stop:t1', turn: 't1' } + ]) + }) +}) diff --git a/src/shared/native-chat-turn-membership.ts b/src/shared/native-chat-turn-membership.ts index 7eecbe0d3ea8..6e33321071fe 100644 --- a/src/shared/native-chat-turn-membership.ts +++ b/src/shared/native-chat-turn-membership.ts @@ -96,11 +96,13 @@ export function structuredAgentTurnAnchors( ) // A send a Stop took back after its turn opened but before the provider echoed it: the record // still names the provider's key, and the host names the turn the send was answered into, or - // none (null). A rejection written before that field is placed by journal order instead. + // none (null). A rejection written before that field is placed by journal order instead. One + // kept as a card opened no turn. const takenBack = submissions.filter( (submission) => !submission.providerItemId && submission.queuedMessageId === undefined && + submission.keptAsQueuedMessageId === undefined && submission.resolvedAt !== null && dispatchWasWithdrawn(submission) ) diff --git a/src/shared/protocol-version.ts b/src/shared/protocol-version.ts index 5d6c8cf0fa62..c44300a05fdb 100644 --- a/src/shared/protocol-version.ts +++ b/src/shared/protocol-version.ts @@ -240,17 +240,16 @@ export const AGENT_SESSION_SEND_ANSWERS_PROOF_RUNTIME_CAPABILITY = // Why: `agentSession.send`'s params are strict, so an older host rejects `delivery`; and only a // capable client can render the `queued` result arm. It gates `delivery: 'queue-if-active'` and // the client's queueing setting and chord, never the published draft list or its card actions: a -// host without it still publishes a message it kept unsent across a restart or a close, and both -// clients show that card and call its actions. DARK ON PURPOSE — not in RUNTIME_CAPABILITIES: +// host without it still publishes a message it kept unsent across a restart, a close or a Stop, +// and both clients show that card and call its actions. DARK ON PURPOSE — not in RUNTIME_CAPABILITIES: // advertising still requires the integrated Codex steer matrix (#21062) in the shipped host. // v1 includes `submission.queuedMessageId` on every draft hand-off: a client reads that link and // never compares a draft id with a submission id. It also publishes the queue's pause once, as // `queuePause` beside the list, lifted by `agentSession.queuedMessagesResume` or any turn sent // after it starting; while it is set every waiting card waits. `nextQueuedMessageId` beside the list // names the card the queue sends next once nothing runs, null while anything holds it; absent (an -// older host) reads as null. A card carries a hold of its own when its conversion failed -// (`send_failed`) or the host kept it unsent (`kept`). The host mechanism lands first; the constant -// gates the rollout. +// older host) reads as null. A card carries a hold of its own only when its conversion failed +// (`send_failed`). The host mechanism lands first; the constant gates the rollout. export const AGENT_SESSION_QUEUED_MESSAGES_RUNTIME_CAPABILITY = 'agent-session.queued-messages.v1' as const // Why: `agentSession.conversationCommand`'s params are strict, so an older host rejects diff --git a/src/shared/structured-agent-session-message-projection.ts b/src/shared/structured-agent-session-message-projection.ts index 43c3cf1ffee5..958fa8c11294 100644 --- a/src/shared/structured-agent-session-message-projection.ts +++ b/src/shared/structured-agent-session-message-projection.ts @@ -109,11 +109,15 @@ export function projectStructuredAgentSessionMessages( projectItems = projectStructuredItemsToNativeChat ): NativeChatMessage[] { // A send a Stop took back before the agent started it stays where it was sent, as the - // conversation's own history. A queued card's hand-off is left out: the card holds its text. + // conversation's own history. A queued card's hand-off, and a send the Stop kept as a card, are + // left out: the card holds its text. const stoppedBeforeStart = new Map( submissions .filter( - (submission) => dispatchWasWithdrawn(submission) && submission.queuedMessageId === undefined + (submission) => + dispatchWasWithdrawn(submission) && + submission.queuedMessageId === undefined && + submission.keptAsQueuedMessageId === undefined ) .map((submission) => [agentJournalSubmissionKey(submission.clientMessageId), submission]) )