From 601cbea66e627d26518230283fb95c975d7eaef4 Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Fri, 9 Oct 2026 00:58:04 -0700 Subject: [PATCH 1/3] fix(native-chat): Stop keeps a never-delivered message as a paused queued card A person's Stop pressed before the agent had a message rejected it and left the bubble in the conversation with "Stopped manually", even though the agent never received it, so the screen and the agent disagreed about what was sent. A restart and a chat close already settle such a host-held send by keeping a person's text as a card at the head of the queue. Stop now does the same: holdUnsentSends gains the `cancelled` cause and the Stop's withdrawal uses it, issuing every row at once so the fold shows nothing queued when it returns. The kept card waits under the existing Stop pause; Resume or the person's next accepted turn releases it. Images, commands, mail and a card's own hand-off are rejected as before. A child that dies in its start after a Stop keeps its unrun person's send the same way, so unrunRejection and the dead rejectPendingSubmissions go. The projection no longer draws a kept send as a stopped bubble; the card is its only surface. --- ...native-chat-merged-snapshot-parity.test.ts | 41 +- ...e-native-chat-stop-journal.test-fixture.ts | 24 +- .../journal-pending-submission-recovery.ts | 15 - .../agent-session-journal/journal-store.ts | 9 - .../journal-unsent-send-hold.test.ts | 62 +++ .../journal-unsent-send-hold.ts | 42 +- .../structured-agent-session-chat-stop.ts | 20 +- .../structured-agent-session-child-exit.ts | 24 +- ...ent-session-claude-hung-start-stop.test.ts | 22 + ...gent-session-dead-generation-settlement.ts | 12 +- ...ed-agent-session-queued-pause-lift.test.ts | 12 +- ...ured-agent-session-queued-stop-row.test.ts | 4 + .../structured-agent-session-queued-stop.ts | 7 +- ...ent-session-queued-withdrawn-draft.test.ts | 25 +- ...d-agent-session-stop-event-binding.test.ts | 17 +- ...ed-agent-session-stop-keeps-unsent.test.ts | 428 ++++++++++++++++++ ...red-agent-session-stop-never-waits.test.ts | 24 +- ...nt-session-stop-withdrawal.test-fixture.ts | 25 + ...ured-agent-session-unrun-send-hold.test.ts | 14 +- src/shared/native-chat-turn-membership.ts | 4 +- ...ctured-agent-session-message-projection.ts | 8 +- 21 files changed, 712 insertions(+), 127 deletions(-) create mode 100644 src/main/native-chat/agent-session-wire/structured-agent-session-stop-keeps-unsent.test.ts create mode 100644 src/main/native-chat/agent-session-wire/structured-agent-session-stop-withdrawal.test-fixture.ts 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..9ba8b12433ac 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,21 +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, 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..38d79aa57a5a 100644 --- a/src/main/native-chat/agent-session-journal/journal-store.ts +++ b/src/main/native-chat/agent-session-journal/journal-store.ts @@ -31,7 +31,6 @@ import type { JournalHostDatabase } from './journal-host-database' import { journalRowsAfterReader, type JournalLoad } from './journal-open' import { markJournalPendingSubmissionsUnknown, - rejectJournalPendingSubmissions, rejectJournalQueuedSubmissions } from './journal-pending-submission-recovery' import { @@ -355,14 +354,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, 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-wire/structured-agent-session-chat-stop.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-chat-stop.ts index b6cd948fa4dc..3f857746b521 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 { 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-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..ca51e65c9a1e 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 @@ -108,17 +113,11 @@ it('a Stop whose withdrawal throws after landing still answers; the draft it rel 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,7 +128,7 @@ 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() } 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-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/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/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]) ) From 62761dbd9ff6a972148ac416652a4e3255a7173c Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Fri, 9 Oct 2026 01:48:13 -0700 Subject: [PATCH 2/3] fix(native-chat): publish a Stop-kept card only under its pause On a ready, idle agent the Stop awaited its withdrawal before issuing its event, so clients got one frame with the kept card waiting, unpaused and next to send. The event is now issued in the hold's tick, so one frame carries both. Also: a turn-membership test for the kept-send exclusion, a test pinning that a failed Stop record leaves the kept card unpaused, the dead rejectQueuedSubmissions helpers removed (their tests ported to holdUnsentSends or resolveDispatch), and the queued-messages capability comment corrected. --- .../journal-pending-submission-recovery.ts | 26 ----- .../journal-queued-rejection-batch.test.ts | 7 +- .../agent-session-journal/journal-store.ts | 15 +-- .../queued-message-delivered-echo.test.ts | 7 +- .../queued-message-store.test.ts | 49 ++++++--- .../structured-agent-session-chat-stop.ts | 8 +- ...session-dead-generation-settlement.test.ts | 5 +- ...ent-session-queued-withdrawn-draft.test.ts | 10 +- ...-agent-session-stop-idle-kept-card.test.ts | 104 ++++++++++++++++++ ...untime-worktree-agent-rows-verdict.test.ts | 24 ++-- .../native-chat-stopped-send-turn.test.ts | 22 ++++ src/shared/protocol-version.ts | 9 +- 12 files changed, 198 insertions(+), 88 deletions(-) create mode 100644 src/main/native-chat/agent-session-wire/structured-agent-session-stop-idle-kept-card.test.ts 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 9ba8b12433ac..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,32 +63,6 @@ export async function markJournalPendingSubmissionsUnknown( return unresolved.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 38d79aa57a5a..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,10 +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, - rejectJournalQueuedSubmissions -} from './journal-pending-submission-recovery' +import { markJournalPendingSubmissionsUnknown } from './journal-pending-submission-recovery' import { applyJournalRow, createJournalReducerState, @@ -354,15 +350,6 @@ export class AgentSessionJournal { return markJournalPendingSubmissionsUnknown(this, fence, reason) } - /** 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/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 3f857746b521..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 @@ -121,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-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-queued-withdrawn-draft.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-withdrawn-draft.test.ts index ca51e65c9a1e..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 @@ -104,9 +104,9 @@ 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() @@ -134,6 +134,10 @@ it('a Stop whose withdrawal throws after landing still answers; the draft it rel } 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-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/runtime/runtime-worktree-agent-rows-verdict.test.ts b/src/main/runtime/runtime-worktree-agent-rows-verdict.test.ts index ff82a61406e7..a890e4c25577 100644 --- a/src/main/runtime/runtime-worktree-agent-rows-verdict.test.ts +++ b/src/main/runtime/runtime-worktree-agent-rows-verdict.test.ts @@ -140,13 +140,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' }) @@ -231,10 +234,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/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/protocol-version.ts b/src/shared/protocol-version.ts index 95da4aba339d..7930fd2d0fb7 100644 --- a/src/shared/protocol-version.ts +++ b/src/shared/protocol-version.ts @@ -237,17 +237,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 From 29c0d57901d217a85904fcf745a9a0321963e2cd Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Fri, 9 Oct 2026 13:36:52 -0700 Subject: [PATCH 3/3] test(native-chat): a new chat's first message stopped before hand-over is kept as a card Main now carries a new chat's first message in its create request. A Stop before the agent is handed that message keeps it as a card at the head of the queue under the Stop's pause, like any person's held send: no provider write, no bubble, sent once on Resume or after the next message. The create flow puts nothing back in the composer for it, so the card is the one surface. --- ...tured-agent-session-create-test-fixture.ts | 15 +++ .../structured-agent-session-create.test.ts | 110 +++++++++++------- ...tured-agent-session-create-message.test.ts | 3 +- ...gent-session-launch-create-message.test.ts | 31 ++++- 4 files changed, 118 insertions(+), 41 deletions(-) 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/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/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)