diff --git a/mobile/src/session/MobileNativeChatView.test.ts b/mobile/src/session/MobileNativeChatView.test.ts index 5dc4e59a7abf..290074c01464 100644 --- a/mobile/src/session/MobileNativeChatView.test.ts +++ b/mobile/src/session/MobileNativeChatView.test.ts @@ -170,10 +170,10 @@ describe('MobileNativeChatView', () => { }) it.each([ - ['queue', 'Queue a message to run after the stop'], + ['queue', 'Message, @files, /commands'], ['send', 'Send a message to run after the stop'] ] as const)( - 'tells the composer a message sent now runs after the stop (%s)', + 'tells the composer whether a message sent now runs after the stop (%s)', async (afterStop, placeholder) => { await render({ structuredActivityUi: true, diff --git a/mobile/src/session/mobile-structured-queued-message-cards.test.ts b/mobile/src/session/mobile-structured-queued-message-cards.test.ts index 3341b7a80d9c..83a1c56eaf29 100644 --- a/mobile/src/session/mobile-structured-queued-message-cards.test.ts +++ b/mobile/src/session/mobile-structured-queued-message-cards.test.ts @@ -7,6 +7,7 @@ import { QUEUED_MESSAGE_PAUSED_SEND_FAILED } from '../../../src/shared/agent-session-wire' import type { AgentSessionQueuedMessage } from '../../../src/shared/agent-session-wire' +import type { AgentMessageSource } from '../../../src/shared/agent-session-message-source' import { mobileQueueHasResumableCard, mobileQueuePauseLabel, @@ -19,6 +20,15 @@ function returnedAs(fact: Parameters[0]) { return { state: 'returned' as const, returnedReason: reason, returnedRejection: rejection } } +/** Another agent's message, which names its sender on its body. */ +const AGENT_FROM: AgentMessageSource = { + kind: 'agent', + senders: [], + orchestration: { message: 'mail-notice', mailbox: 'run:r1', dispatchId: null, messages: [] } +} + +const STOPPED = { reason: 'stopped' } as const + function draft(overrides: Partial & { messageId: string }) { return { position: 1, @@ -32,6 +42,11 @@ function draft(overrides: Partial & { messageId: stri } } +function mailDraft(messageId: string, position: number) { + const base = draft({ messageId, position }) + return { ...base, body: { ...base.body, from: AGENT_FROM } } +} + describe('mobileQueuedMessageCards', () => { it('renders nothing without a published list', () => { expect(mobileQueuedMessageCards(null, [], { pendingPrompt: false })).toEqual([]) @@ -63,13 +78,23 @@ describe('mobileQueuedMessageCards', () => { it('captions nothing on a waiting card of a paused queue; its pause row explains', () => { const [card] = mobileQueuedMessageCards([draft({ messageId: 'a' })], [], { pendingPrompt: false, - queuePaused: true + queuePause: STOPPED }) expect(card?.caption).toBeNull() expect(card?.paused).toBe(false) expect(card?.needsAttention).toBe(false) }) + // Another agent's mail runs when the stop lands: only a person's card waits on the Stop's pause. + it("captions mail as an ordinary waiting card under a person's Stop", () => { + const cards = mobileQueuedMessageCards( + [draft({ messageId: 'typed' }), mailDraft('mail', 2)], + [], + { pendingPrompt: true, queuePause: STOPPED } + ) + expect(cards.map((card) => card.caption)).toEqual([null, 'Waiting for your answer']) + }) + it("keeps a card's own failed send and reads a prompt's wait as queued under a paused queue", () => { const cards = mobileQueuedMessageCards( [ @@ -77,7 +102,7 @@ describe('mobileQueuedMessageCards', () => { draft({ messageId: 'b', position: 2 }) ], [], - { pendingPrompt: true, queuePaused: true } + { pendingPrompt: true, queuePause: STOPPED } ) expect(cards.map((card) => card.caption)).toEqual(["Couldn't send — tap Send to retry", null]) expect(cards[0]?.paused).toBe(true) @@ -86,7 +111,7 @@ describe('mobileQueuedMessageCards', () => { it('finds something for Resume to send only in a waiting card with no hold, ahead of a returned one', () => { const resumable = (drafts: AgentSessionQueuedMessage[]) => mobileQueueHasResumableCard( - mobileQueuedMessageCards(drafts, [], { pendingPrompt: false, queuePaused: true }) + mobileQueuedMessageCards(drafts, [], { pendingPrompt: false, queuePause: STOPPED }) ) const returned = draft({ messageId: 'r', diff --git a/mobile/src/session/mobile-structured-queued-message-cards.ts b/mobile/src/session/mobile-structured-queued-message-cards.ts index 6388efbf4123..097ac1c43e23 100644 --- a/mobile/src/session/mobile-structured-queued-message-cards.ts +++ b/mobile/src/session/mobile-structured-queued-message-cards.ts @@ -10,6 +10,7 @@ import { structuredAgentSessionAttemptFailureParts } from '../../../src/shared/s import { QUEUED_MESSAGE_PAUSED_KEPT, QUEUED_MESSAGE_PAUSED_SEND_FAILED, + queuePauseHoldsCard, type AgentSessionQueuedMessage, type AgentSessionQueuePause } from '../../../src/shared/agent-session-wire' @@ -99,7 +100,7 @@ export function mobileQueuePauseLabel(pause: Pick[], - facts: { pendingPrompt: boolean; queuePaused?: boolean } + facts: { pendingPrompt: boolean; queuePause?: AgentSessionQueuePause | null } ): MobileQueuedMessageCard[] { if (!queuedMessages || queuedMessages.length === 0) { return [] @@ -125,7 +126,7 @@ export function mobileQueuedMessageCards( ? pausedCaption(draft.pausedReason) : behindReturned ? 'Waiting — a message ahead needs attention' - : facts.queuePaused + : queuePauseHoldsCard(facts.queuePause, draft) ? // The pause row says why and offers Resume; the card promises no send time. null : facts.pendingPrompt diff --git a/mobile/src/session/use-mobile-native-chat-input-lease.test.ts b/mobile/src/session/use-mobile-native-chat-input-lease.test.ts index 847fbe3d517d..75681d84f25b 100644 --- a/mobile/src/session/use-mobile-native-chat-input-lease.test.ts +++ b/mobile/src/session/use-mobile-native-chat-input-lease.test.ts @@ -114,16 +114,18 @@ describe('useSettledMobileNativeChatInputLock', () => { }) describe("the phone chat composer's placeholder", () => { - it('says a message runs after the stop while the chat reads Stopping', () => { - expect(mobileNativeChatComposerPlaceholder(null, 'queue')).toBe( - 'Queue a message to run after the stop' - ) - // Where the host does not queue sends, it holds the message until the stop lands. + // Where the host does not queue sends, it holds the message until the stop lands. + it('says a message runs after the stop while the chat reads Stopping and sends are not queued', () => { expect(mobileNativeChatComposerPlaceholder(null, 'send')).toBe( 'Send a message to run after the stop' ) }) + // A queued one is a card the Stop holds until Resume: nothing promises it runs after the stop. + it('reads as usual while Stopping where a send is queued', () => { + expect(mobileNativeChatComposerPlaceholder(null, 'queue')).toBe('Message, @files, /commands') + }) + it('reads as usual otherwise', () => { expect(mobileNativeChatComposerPlaceholder(null, undefined)).toBe('Message, @files, /commands') }) diff --git a/mobile/src/session/use-mobile-native-chat-input-lease.ts b/mobile/src/session/use-mobile-native-chat-input-lease.ts index 02cb57849d4d..c881c888cd34 100644 --- a/mobile/src/session/use-mobile-native-chat-input-lease.ts +++ b/mobile/src/session/use-mobile-native-chat-input-lease.ts @@ -95,7 +95,8 @@ export function useSettledMobileNativeChatInputLock( } /** What the composer says while the input lease blocks it, else that a message sent now runs - * after a Stop the chat reads as stopping, else its normal prompt. */ + * after a Stop the chat reads as stopping, else its normal prompt. A queued one is a card the + * Stop holds until Resume, so it gets the normal prompt. */ export function mobileNativeChatComposerPlaceholder( lockReason: MobileNativeChatInputLockReason | null, /** While stopping: whether a message sent now is queued, or sent for the host to hold. */ @@ -107,10 +108,8 @@ export function mobileNativeChatComposerPlaceholder( if (lockReason === 'waiting') { return 'Waiting for terminal…' } - if (afterStop) { - return afterStop === 'queue' - ? NATIVE_CHAT_TURN_STATUS_COPY.queueAfterStop - : NATIVE_CHAT_TURN_STATUS_COPY.sendAfterStop + if (afterStop === 'send') { + return NATIVE_CHAT_TURN_STATUS_COPY.sendAfterStop } return 'Message, @files, /commands' } diff --git a/mobile/src/session/use-mobile-structured-agent-session-queued.test.tsx b/mobile/src/session/use-mobile-structured-agent-session-queued.test.tsx index c14e823ab401..4332913c83fb 100644 --- a/mobile/src/session/use-mobile-structured-agent-session-queued.test.tsx +++ b/mobile/src/session/use-mobile-structured-agent-session-queued.test.tsx @@ -173,6 +173,32 @@ describe('mobile structured queued messages', () => { ) }) + // Nothing is left to steer into while a Stop ends the turn: the send runs after it, as a card. + it('queues a send made while this phone reads Stopping', async () => { + sendRequest.mockImplementation(async (method) => { + if (method === 'agentSession.cancel') { + return new Promise(() => {}) + } + if (method === 'agentSession.send') { + return mutationOk({ + clientMessageId: 'client-1', + queued: { messageId: 'client-1', position: 1, state: 'waiting' } + }) + } + return method === 'agentSession.options' ? ok({ models: [], current: {} }) : ok({}) + }) + await mountSession(CAPABLE, snapshotEvent({ runningTurn: true })) + act(() => hook!.cancel()) + await vi.waitFor(() => expect(hook!.turnIndicator.stopping).toBe(true)) + + await act(async () => { + expect(await hook!.sendWithOutcome('run this after the stop')).toBe('queued') + }) + + expect(calls('agentSession.send')).toHaveLength(1) + expect(requestOf('agentSession.send').params.delivery).toBe('queue-if-active') + }) + it('keeps today’s request exactly against an incapable host', async () => { sendRequest.mockImplementation(async (method) => { if (method === 'agentSession.send') { diff --git a/mobile/src/session/use-mobile-structured-queued-message-controls.ts b/mobile/src/session/use-mobile-structured-queued-message-controls.ts index f65ecf2d46e3..5d872ae2ce77 100644 --- a/mobile/src/session/use-mobile-structured-queued-message-controls.ts +++ b/mobile/src/session/use-mobile-structured-queued-message-controls.ts @@ -73,7 +73,7 @@ export function useMobileStructuredQueuedMessageControls(args: { () => mobileQueuedMessageCards(queuedMessages, submissions, { pendingPrompt, - queuePaused: queuePause !== null + queuePause }), [pendingPrompt, queuePause, queuedMessages, submissions] ) diff --git a/src/main/native-chat/agent-session-journal/queued-message-pause.test.ts b/src/main/native-chat/agent-session-journal/queued-message-pause.test.ts index 83570726945d..106859731cd2 100644 --- a/src/main/native-chat/agent-session-journal/queued-message-pause.test.ts +++ b/src/main/native-chat/agent-session-journal/queued-message-pause.test.ts @@ -31,6 +31,7 @@ import { agentSessionFailureFact } from '../../../shared/agent-session-failure' import { agentSessionFailureWords } from '../../../shared/agent-session-failure-words' import { QUEUED_MESSAGE_PAUSED_KEPT } from '../../../shared/agent-session-queued-message-wire' import { claudeProviderHandle } from '../../../shared/agent-session-provider-handle-encoding' +import type { AgentMessageSource } from '../../../shared/agent-session-message-source' const IDENTITY: AgentSessionJournalIdentity = { sessionId: 'session-p', @@ -58,10 +59,22 @@ function open(): Promise { }) } -function queueDraft(journal: AgentSessionJournal, messageId: string, carriedFrom?: string) { +/** Orchestration mail's sender, as another agent's message names it on its body. */ +const MAIL: AgentMessageSource = { + kind: 'agent', + senders: [], + orchestration: { message: 'mail-notice', mailbox: 'inbox', dispatchId: null, messages: [] } +} + +function queueDraft( + journal: AgentSessionJournal, + messageId: string, + carriedFrom?: string, + from?: AgentMessageSource +) { return journal.queuedMessages.insert({ messageId, - body: message(messageId), + body: { ...message(messageId), ...(from ? { from } : {}) }, fingerprint: `fp-${messageId}`, hostInstance: HOST, ...(carriedFrom ? { carriedFrom } : {}) @@ -397,23 +410,59 @@ describe("the queue's pause, derived from the journal", () => { }) describe('which cards a pause holds', () => { - it('only cards queued before the Stop event: one queued after it is a new instruction', async () => { + // Sent while the stopped turn winds down, a card waits with the rest for Resume or a new turn. + it("a person's Stop holds every card, one queued after it too, across a reopen", async () => { let journal = await open() await queueDraft(journal, 'before') await userStop(journal) await queueDraft(journal, 'after') expect(held(journal)).toEqual([ ['before', true], - ['after', false] + ['after', true] ]) await journal.close() journal = await open() expect(held(journal)).toEqual([ ['before', true], - ['after', false] + ['after', true] ]) }) + it('holds a lone card queued after it, and publishes its pause over it', async () => { + const journal = await open() + await userStop(journal) + await queueDraft(journal, 'alone') + const pauses = journal.queuedMessages.pauses(HOST) + const cards = journal.queuedMessages.list() + expect(held(journal)).toEqual([['alone', true]]) + expect(nextSendableQueuedCard(pauses, cards)).toBeNull() + expect(resumableQueuePause(pauses, cards)?.reason).toBe('stopped') + }) + + // Mail never waits on a person's Resume: it runs past the cards the Stop holds, and its turn, + // once accepted, lifts the pause for them. + it('holds no mail, queued before the Stop or after, and mail passes the cards it holds', async () => { + const journal = await open() + await queueDraft(journal, 'mail-before', undefined, MAIL) + await queueDraft(journal, 'person') + await userStop(journal) + await queueDraft(journal, 'mail-after', undefined, MAIL) + expect(held(journal)).toEqual([ + ['mail-before', false], + ['person', true], + ['mail-after', false] + ]) + const next = () => + nextSendableQueuedCard(journal.queuedMessages.pauses(HOST), journal.queuedMessages.list()) + expect(next()?.messageId).toBe('mail-before') + await journal.queuedMessages.withdraw({ messageIds: ['mail-before'], settledByOp: 'op-1' }) + expect(next()?.messageId).toBe('mail-after') + expect( + resumableQueuePause(journal.queuedMessages.pauses(HOST), journal.queuedMessages.list()) + ?.reason + ).toBe('stopped') + }) + it('a steer the Stop withdrew comes back in its own place, and is held', async () => { const journal = await open() await queueDraft(journal, 'steered') @@ -437,7 +486,7 @@ describe('which cards a pause holds', () => { }) expect(held(journal)).toEqual([ ['steered', true], - ['after', false] + ['after', true] ]) }) @@ -467,6 +516,31 @@ describe('which cards a pause holds', () => { expect(journal.queuedMessages.get('newer')?.state).toBe('waiting') }) + it("the queue's own consume takes mail past the cards a person's Stop holds", async () => { + const journal = await open() + await queueDraft(journal, 'held') + await userStop(journal) + await queueDraft(journal, 'mail', undefined, MAIL) + await journal.appendSubmission( + { + clientMessageId: 'drain-mail', + payloadFingerprint: 'fp-mail', + body: message('mail'), + fence: 0, + handoverRecorded: true + }, + { + messageId: 'mail', + expect: 'waiting', + settledByOp: null, + hostInstance: HOST, + yieldsToPause: { hostInstance: HOST } + } + ) + expect(journal.queuedMessages.get('mail')?.state).toBe('dispatched') + expect(journal.queuedMessages.get('held')?.state).toBe('waiting') + }) + it("'cleared' holds the carried cards, not one typed after them", async () => { const journal = await open() await queueDraft(journal, 'carried-1', 'source-session') @@ -540,10 +614,19 @@ describe('which cards the pauses in force hold', () => { type Card = Parameters[1] & { messageId: string } const DEAD = 'proc-0' - function card(messageId: string, queuedAfter: number, fields: Partial = {}): Card { - const queuedAt = { epoch: 'epoch-1', sequence: queuedAfter + 1 } - const base = { state: 'waiting', holdReason: null, hostInstance: HOST, carriedFrom: null } - return { messageId, ...base, queuedAt, ...fields } + function card(messageId: string, fields: Partial = {}): Card { + const base = { + state: 'waiting', + holdReason: null, + hostInstance: HOST, + carriedFrom: null, + body: {} + } + return { messageId, ...base, ...fields } + } + + function mail(messageId: string, fields: Partial = {}): Card { + return card(messageId, { body: { from: MAIL }, ...fields }) } /** A Stop at sequence 5 unless `stopped` is 0; no turn or Resume since. */ @@ -566,55 +649,54 @@ describe('which cards the pauses in force hold', () => { return cards.map((each) => [each.messageId, queuePauseHolding(pauses, each)?.reason ?? null]) } - it('a Stop holds a card with no recorded position, and one queued before a rewind', () => { - const cards = [ - card('unrecorded', 5, { queuedAt: null }), - // Later in its own epoch than the Stop is in this one: only the epoch says it came first. - card('before-rewind', 5, { queuedAt: { epoch: 'epoch-0', sequence: 9 } }), - card('after', 5) - ] - expect(holding(cards)).toEqual([ - ['unrecorded', 'stopped'], - ['before-rewind', 'stopped'], - ['after', null] - ]) - }) - - it("a Stop that holds nothing never hides a restart's: a dead process's card queued after it waits", () => { - const after = card('after', 5, { hostInstance: DEAD }) - expect(pausesOver([after]).map((pause) => pause.reason)).toEqual(['stopped', 'restarted']) - expect(holding([after])).toEqual([['after', 'restarted']]) - expect(nextSendableQueuedCard(pausesOver([after]), [after])).toBeNull() - expect(resumableQueuePause(pausesOver([after]), [after])?.reason).toBe('restarted') - // Written by this process, nothing holds it: a card queued after a Stop sends normally. - const live = card('after', 5) + it("a Stop that holds no mail never hides a restart's: a dead process's mail waits", () => { + const deadMail = mail('mail', { hostInstance: DEAD }) + expect(pausesOver([deadMail]).map((pause) => pause.reason)).toEqual(['stopped', 'restarted']) + expect(holding([deadMail])).toEqual([['mail', 'restarted']]) + expect(nextSendableQueuedCard(pausesOver([deadMail]), [deadMail])).toBeNull() + expect(resumableQueuePause(pausesOver([deadMail]), [deadMail])?.reason).toBe('restarted') + // Written by this process, nothing holds it: mail runs as the stop lands. + const live = mail('mail') expect(nextSendableQueuedCard(pausesOver([live]), [live])).toBe(live) }) it("a card names the first pause holding it, and the header names the first held card's", () => { - const cards = [ - card('before', 3, { hostInstance: DEAD }), - card('after', 5, { hostInstance: DEAD }) - ] + const cards = [card('person', { hostInstance: DEAD }), mail('mail', { hostInstance: DEAD })] expect(holding(cards)).toEqual([ - ['before', 'stopped'], - ['after', 'restarted'] + ['person', 'stopped'], + ['mail', 'restarted'] ]) expect(resumableQueuePause(pausesOver(cards), cards)?.reason).toBe('stopped') }) + // Mail passes only the cards a person's Stop alone holds: never a returned card, and never one + // another pause holds too. + it("mail passes the cards held only by a person's Stop, and nothing else", () => { + const passing = [card('person-1'), card('person-2'), mail('mail'), card('person-3')] + expect(nextSendableQueuedCard(pausesOver(passing), passing)?.messageId).toBe('mail') + expect(nextSendableQueuedCard(pausesOver(passing, 0), passing)?.messageId).toBe('person-1') + const returned = [card('refused', { state: 'returned' }), mail('mail')] + expect(nextSendableQueuedCard(pausesOver(returned), returned)).toBeNull() + const carried = [card('carried', { carriedFrom: 'source-session' }), mail('mail')] + expect(holding(carried)).toEqual([ + ['carried', 'stopped'], + ['mail', null] + ]) + expect(nextSendableQueuedCard(pausesOver(carried), carried)).toBeNull() + }) + // Held on its own, as a card whose send failed: the queue goes past it, and Resume is offered // over the cards a pause holds behind it. it('a kept card is skipped like a send_failed one; the cards behind it still send', () => { - const behind = card('behind', 2) - const kept = [card('kept', 1, { holdReason: QUEUED_MESSAGE_PAUSED_KEPT }), behind] + const behind = card('behind') + const kept = [card('kept', { holdReason: QUEUED_MESSAGE_PAUSED_KEPT }), behind] expect(holding(kept, 0)).toEqual([ ['kept', null], ['behind', null] ]) expect(nextSendableQueuedCard(pausesOver(kept, 0), kept)).toBe(behind) expect(resumableQueuePause(pausesOver(kept, 0), kept)).toBeNull() - const restarted = [kept[0]!, card('dead', 2, { hostInstance: DEAD })] + const restarted = [kept[0]!, card('dead', { hostInstance: DEAD })] expect(nextSendableQueuedCard(pausesOver(restarted, 0), restarted)).toBeNull() expect(resumableQueuePause(pausesOver(restarted, 0), restarted)?.reason).toBe('restarted') }) @@ -622,7 +704,7 @@ describe('which cards the pauses in force hold', () => { // A dead process's card held on its own waits for its own Send, so it pauses no other card. it('a card held on its own from a dead process starts no restart pause', () => { for (const holdReason of [QUEUED_MESSAGE_PAUSED_KEPT, 'send_failed']) { - const cards = [card('held', 1, { hostInstance: DEAD, holdReason }), card('live', 2)] + const cards = [card('held', { hostInstance: DEAD, holdReason }), card('live')] expect(pausesOver(cards, 0)).toEqual([]) expect(nextSendableQueuedCard(pausesOver(cards, 0), cards)?.messageId).toBe('live') } @@ -630,8 +712,8 @@ describe('which cards the pauses in force hold', () => { it("a /clear's pause that holds nothing never hides a restart's", () => { const cards = [ - card('carried', 1, { carriedFrom: 'source-session', holdReason: 'send_failed' }), - card('typed', 2, { hostInstance: DEAD }) + card('carried', { carriedFrom: 'source-session', holdReason: 'send_failed' }), + card('typed', { hostInstance: DEAD }) ] expect(pausesOver(cards, 0).map((pause) => pause.reason)).toEqual(['cleared', 'restarted']) expect(holding(cards, 0)).toEqual([ diff --git a/src/main/native-chat/agent-session-journal/queued-message-pause.ts b/src/main/native-chat/agent-session-journal/queued-message-pause.ts index 5b13856b8f0a..645080b688f9 100644 --- a/src/main/native-chat/agent-session-journal/queued-message-pause.ts +++ b/src/main/native-chat/agent-session-journal/queued-message-pause.ts @@ -2,7 +2,8 @@ // nothing has to retire them. Each holds its own cards (`queuePauseHolding`). In force when: // - 'stopped': the latest Stop event is a person's (reason `user-stop`), with no later Resume row // and no turn sent after it and accepted. A later Stop of any reason supersedes it; only a -// person's pauses. +// person's pauses. It holds every card but orchestration mail, which never waits on a +// person's Resume: it runs past the cards this pause holds, and its turn lifts the pause. // - 'cleared': a card /clear carried into this conversation waits, and no turn or Resume has // happened here since. // - 'restarted': a waiting card with no hold of its own was written by another host process, @@ -14,6 +15,7 @@ import type { AgentJournalCursor, + AgentJournalMessageItem, AgentJournalSubmission } from '../../../shared/agent-session-journal-types' import type { JournalStopEvent, JournalTombstoneRow } from './journal-row-schema' @@ -42,7 +44,7 @@ export type JournalQueuePauseMarks = { export type DerivedQueuePause = { reason: QueuePauseReason - /** Where a Stop's pause began: a card queued at or after it is newer. Null for the others. */ + /** Where a Stop's pause began: only a turn sent after it lifts it. Null for the others. */ since: AgentJournalCursor | null } @@ -51,7 +53,7 @@ type QueueCard = { holdReason: string | null hostInstance: string carriedFrom: string | null - queuedAt: AgentJournalCursor | null + body: Pick } export function createJournalQueuePauseMarks(): JournalQueuePauseMarks { @@ -153,20 +155,22 @@ export function deriveQueuePauses(input: { return pauses } -/** Queued before the pause began: for /clear, a card it carried; for a restart, every card. For a - * Stop, a card queued before its row; one from another epoch (before a rewind) or from a build - * that recorded no position counts as before. A withdrawn steer keeps its position, so is held. */ -function queuedBeforePause(pause: DerivedQueuePause, card: QueueCard): boolean { - if (pause.reason === 'cleared') { - return card.carriedFrom !== null +/** Another agent's message names its sender on its body; a client's send cannot set it. */ +function isOrchestrationMail(card: QueueCard): boolean { + return card.body.from !== undefined +} + +/** A person's Stop holds every card but mail, one sent while it winds down too; /clear the cards + * it carried; a restart every card. */ +function pauseHolds(pause: DerivedQueuePause, card: QueueCard): boolean { + switch (pause.reason) { + case 'stopped': + return !isOrchestrationMail(card) + case 'cleared': + return card.carriedFrom !== null + case 'restarted': + return true } - const { since } = pause - return ( - since === null || - card.queuedAt === null || - card.queuedAt.epoch !== since.epoch || - card.queuedAt.sequence < since.sequence - ) } /** THE rule for which cards are held: by ANY pause in force, named by the first that holds it, so @@ -179,29 +183,40 @@ export function queuePauseHolding( if (card.state !== 'waiting' || card.holdReason !== null) { return undefined } - // A card queued AFTER a Stop is a new instruction and is not held; it still waits behind a held one. - return pauses.find((pause) => queuedBeforePause(pause, card)) + return pauses.find((pause) => pauseHolds(pause, card)) } /** The card the queue sends next: the oldest waiting one with no hold of its own, unless a - * returned card or one a pause holds comes first. The queue never reorders, so a newer card never - * overtakes one a pause holds; a card held on its own is passed over. The drain's pick and its - * consume both read this. */ + * returned card or one a pause holds comes first. The queue never reorders, with one exception: + * mail passes cards held only by a person's Stop, which it lifts once its turn is accepted. A + * card held on its own is passed over. The drain's pick, its consume and admission all read this. */ export function nextSendableQueuedCard( pauses: readonly DerivedQueuePause[], cards: readonly T[] ): T | null { + let pastStopHeld = false for (const card of cards) { - if (card.state === 'returned' || queuePauseHolding(pauses, card)) { + if (card.state === 'returned') { return null } + if (queuePauseHolding(pauses, card)) { + if (!heldOnlyByStop(pauses, card)) { + return null + } + pastStopHeld = true + continue + } if (card.state === 'waiting' && card.holdReason === null) { - return card + return !pastStopHeld || isOrchestrationMail(card) ? card : null } } return null } +function heldOnlyByStop(pauses: readonly DerivedQueuePause[], card: QueueCard): boolean { + return pauses.every((pause) => pause.reason === 'stopped' || !pauseHolds(pause, card)) +} + /** The pause to PUBLISH: the one holding the first card Resume would send, not behind a returned * card, which blocks everything after it until the user acts. None otherwise, so its header * never offers a Resume that sends nothing. */ diff --git a/src/main/native-chat/agent-session-journal/queued-message-table.ts b/src/main/native-chat/agent-session-journal/queued-message-table.ts index f3e82613b1f3..7b58cdf8fd5e 100644 --- a/src/main/native-chat/agent-session-journal/queued-message-table.ts +++ b/src/main/native-chat/agent-session-journal/queued-message-table.ts @@ -63,8 +63,8 @@ export type QueuedMessageRow = { /** The conversation /clear carried this card from; null for a card written here. What the * replacement's 'cleared' pause is derived from. */ carriedFrom: string | null - /** Where the journal stood when it was queued: a Stop's pause holds only cards queued before - * it. Null on rows from builds before it was recorded, which read as queued before any Stop. */ + /** Where the journal stood when it was queued. Still written for older builds, whose Stop pause + * holds only cards queued before it; null on rows from builds before it was recorded. */ queuedAt: AgentJournalCursor | null } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-claude-stop-ends-session.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-claude-stop-ends-session.test.ts index ed06f809c6d8..50a02567ee4c 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-claude-stop-ends-session.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-claude-stop-ends-session.test.ts @@ -150,7 +150,7 @@ function eventually(assertion: () => T | Promise): Promise { } function envelope( - method: 'agentSession.send' | 'agentSession.cancel', + method: 'agentSession.send' | 'agentSession.cancel' | 'agentSession.queuedMessagesResume', fields: Record, fence = store.getRecord(SESSION)!.lease.runtimeFence ) { @@ -568,17 +568,10 @@ it('delivers a send issued with the pre-Stop fence during the Stop to the resume expect((await statusTexts()).filter((text) => text !== 'Cancellation requested.')).toEqual([]) }) -it('sends a queue-if-active message issued while the Stop ends the child directly to the resumed child', async () => { +it('holds a queue-if-active message issued while the Stop ends the child as a card; Resume sends it to the resumed child', async () => { const connection = claude.connections[0]! await openTurn(connection) const fence = store.getRecord(SESSION)!.lease.runtimeFence - let childrenWhenClosed: number | undefined - const close = connection.close - connection.close = async () => { - const proven = await close() - childrenWhenClosed = claude.connections.length - return proven - } await expect(stop()).resolves.toMatchObject({ ok: true, value: { cancelled: true } }) // Issued while the Stop's second step waits for the stopped turn, with the fence it moves. @@ -590,15 +583,23 @@ it('sends a queue-if-active message issued while the Stop ends the child directl }) frame(connection, INTERRUPTED_RESULT) - // Nothing runs after the rest, so it goes out as a direct submission, not a queued card. - expect(await sent).toMatchObject({ ok: true, value: { submission: expect.any(Object) } }) + // Sent while Stopping, it is a card the Stop holds: the stop landing runs nothing. + expect(await sent).toMatchObject({ ok: true, value: { queued: { state: 'waiting' } } }) + await eventually(() => expect(connection.closed).toBe(true)) + await laneDrained() + expect(claude.connections).toHaveLength(1) + expect(wrote(connection, 'Queued during the Stop.')).toBe(false) + + expect( + await host.queuedMessagesResume(CALLER, { + envelope: envelope('agentSession.queuedMessagesResume', {}) + }) + ).toMatchObject({ ok: true, value: { resumed: true } }) await eventually(() => { const started = claude.connections.at(-1)! expect(started).not.toBe(connection) expect(wrote(started, 'Queued during the Stop.')).toBe(true) }) - expect(childrenWhenClosed).toBe(1) - expect(wrote(connection, 'Queued during the Stop.')).toBe(false) }) it("runs nothing queued during the Stop's first step before the child's end", async () => { diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host-mutations.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host-mutations.ts index 8055af641583..409b2ab5a272 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host-mutations.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host-mutations.ts @@ -22,8 +22,11 @@ import type { AgentSessionPromptRequest } from './structured-agent-session-turns import { threadGoalPlan } from './structured-agent-session-thread-goal' import { mutateStructuredAgentSession, + mutateStructuredAgentSessionOffLane, type StructuredAgentSessionMutationContext } from './structured-agent-session-mutation-context' +import { queuedMessageBodyIsTextOnly } from './structured-agent-session-queued-messages' +import { AgentSessionJournalError } from '../agent-session-journal/journal-write-guards' import { openForProviderWrite, openWithAgent, @@ -36,7 +39,10 @@ import { sendPlan, setOptionPlan } from './structured-agent-session-mutation-plans' -import { runQueueableStructuredAgentSessionSend } from './structured-agent-session-queued-send' +import { + queueStructuredAgentSessionSendAfterStop, + runQueueableStructuredAgentSessionSend +} from './structured-agent-session-queued-send' import { cancelStructuredAgentSessionPrompt } from './structured-agent-session-prompt-cancel' import { mutateWithChatStop } from './structured-agent-session-chat-stop' export type { StructuredAgentSessionMutationContext } from './structured-agent-session-mutation-context' @@ -68,6 +74,21 @@ export function sendStructuredAgentSessionTurn( arrival?: Parameters[2] ): Promise> { const plan = sendPlan(params) + if (sendQueuesAfterStop(context, params, arrival)) { + // A person's Stop holds the lane until its provider answers; this send is only a card write, + // ordered with the Stop's own writes by the journal's writer, so it is a card at once. + return mutateStructuredAgentSessionOffLane(context, caller, params.envelope, { + ...plan, + run: (ctx) => { + if (context.sessions.get(ctx.sessionId)?.journal !== ctx.journal) { + // Closed or reopened since admission: nothing is written, so the operation stays + // pending and a resend runs it afresh. + throw new AgentSessionJournalError('journal_closed', 'the conversation was replaced') + } + return queueStructuredAgentSessionSendAfterStop(context, ctx, params) + } + }) + } return mutateStructuredAgentSession( context, caller, @@ -88,6 +109,25 @@ export function sendStructuredAgentSessionTurn( ) } +/** A text send asking to be queued while the host reads that a person's Stop is ending the work, + * with nothing the lane would answer first: a /clear in flight, a rewind it recovers or refuses, + * or a check made at acceptance. */ +function sendQueuesAfterStop( + context: StructuredAgentSessionMutationContext, + params: Parameters[2], + arrival: Parameters[2] +): boolean { + const { sessionId } = params.envelope + return ( + params.delivery === 'queue-if-active' && + queuedMessageBodyIsTextOnly(params.body) && + !params.beforeRun && + !arrival?.clearInFlight && + context.readStopping?.(sessionId) === true && + structuredAgentSessionSendBlock(context.deps.store.getRecord(sessionId)) === null + ) +} + export function cancelStructuredAgentSessionTurn( context: StructuredAgentSessionMutationContext, caller: StructuredAgentSessionCaller, diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts index 786577a37998..17da3afef99d 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts @@ -295,6 +295,7 @@ export class StructuredAgentSessionHost { wakeDelivery: (sessionId) => this.conversationDelivery.loop.wake(sessionId), stopAgent: (sessionId, ending) => this.lifetime.stopAgent(sessionId, ending), wakeQueuedDrain: (sessionId) => this.queued.drain.schedule(sessionId), + readStopping: this.clientDelivery.readStopping, now: () => this.now() } } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-mutation-context.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-mutation-context.ts index fb12399c2370..0af758ccabd1 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-mutation-context.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-mutation-context.ts @@ -1,5 +1,6 @@ // The context every client mutation of a session runs with, and the one path each takes: admit the -// envelope against the lease, then run its plan inside the session's serialize. +// envelope against the lease, then run its plan inside the session's serialize (or, for a plan +// that only writes the journal, beside it). import type { AgentSessionMutationEnvelope, @@ -44,6 +45,9 @@ export type StructuredAgentSessionMutationContext = { * journal commit (a conversation command). Draft-table changes need no call: * the draft store notifies through the journal's own commit listener. */ wakeQueuedDrain?: (sessionId: string) => void + /** The host's reading that a person's Stop is still ending the session's work, as the status + * feed projects it from the journal. */ + readStopping?: (sessionId: string) => boolean now: () => number } @@ -56,19 +60,31 @@ export function mutateStructuredAgentSession( prepareSession?: AgentSessionMutationRequest['prepareSession'] ): Promise> { return context.serialize(envelope.sessionId, () => - admitAndRunAgentSessionMutation({ - store: context.deps.store, - adapter: context.deps.adapter, - agents: context.deps.agents, - logger: context.deps.logger, - callerKey: caller.callerKey, - envelope, - plan, - journal: () => context.sessions.get(envelope.sessionId)?.journal, - prepareSession, - publish: (journal) => context.publish(envelope.sessionId, journal), - providerChildPhase: () => context.sessions.get(envelope.sessionId)?.child?.phase, - now: () => context.now() - }) + mutateStructuredAgentSessionOffLane(context, caller, envelope, plan, prepareSession) ) } + +/** Admits the envelope and runs the plan without the session's serialize: for a plan that only + * writes the journal, through its own ordered writer, and touches nothing the lane orders. */ +export function mutateStructuredAgentSessionOffLane( + context: StructuredAgentSessionMutationContext, + caller: StructuredAgentSessionCaller, + envelope: AgentSessionMutationEnvelope, + plan: MutationPlan, + prepareSession?: AgentSessionMutationRequest['prepareSession'] +): Promise> { + return admitAndRunAgentSessionMutation({ + store: context.deps.store, + adapter: context.deps.adapter, + agents: context.deps.agents, + logger: context.deps.logger, + callerKey: caller.callerKey, + envelope, + plan, + journal: () => context.sessions.get(envelope.sessionId)?.journal, + prepareSession, + publish: (journal) => context.publish(envelope.sessionId, journal), + providerChildPhase: () => context.sessions.get(envelope.sessionId)?.child?.phase, + now: () => context.now() + }) +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-card-holds.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-card-holds.test.ts index b169bbe4e5e7..6247cd07fae7 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-card-holds.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-card-holds.test.ts @@ -153,7 +153,7 @@ async function watchClient(): Promise { const working = hostWorking || (nextQueuedMessageId !== null && !hostWorking) const cards = projectQueuedMessageCards(queued, all, { hasPendingPrompt: false, - queuePaused: queuePause !== null + queuePause }) // As use-structured-agent-session-queued-messages.ts derives them. const header = queuedMessagesQueuePause(cards, queuePause) !== null diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-messages.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-messages.test.ts index 7418e7eb8df0..92f1788da98e 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-messages.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-messages.test.ts @@ -649,7 +649,8 @@ describe('/clear', () => { const working = await workingSend() await send('pointer', 'queue-if-active', { internal: true, from }).result await send('typed', 'queue-if-active').result - await stop() + // A restart holds both: mail waits on no person's Stop. + await rig.restartHostProcess() await settleAccepted(working, 'a') const cleared = await clear(hostTestOperationId()) const replacementId = cleared.ok ? cleared.value.replacementSessionId : undefined diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-messages.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-messages.ts index 3acc45552aa6..3a13cf1f4538 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-messages.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-messages.ts @@ -1,6 +1,7 @@ -// Mid-turn queueing: the accept decision that turns a send into a host-held -// draft, the serialized drain that converts one draft into an ordinary -// submission when the session stops owing work, and the published draft list. +// Mid-turn queueing: the accept gate that turns a send into a host-held draft +// (written by `structured-agent-session-queued-send.ts`), the serialized drain that +// converts one draft into an ordinary submission when the session stops owing +// work, and the published draft list. // // Drafts are never owed work: they feed no reducer, no working status, no // teardown and no idle sweep. The drain re-reads every gate inside its own @@ -10,18 +11,14 @@ import { randomUUID } from 'node:crypto' import type { AgentJournalMessageItem } from '../../../shared/agent-session-journal-types' import { QUEUED_MESSAGE_PAUSED_SEND_FAILED, - type AgentSessionSendResult, type AgentSessionWireRefusal } from '../../../shared/agent-session-wire' import { createStructuredAgentSessionOperationId } from '../../../shared/structured-agent-session-mutation' -import { agentSessionSendBodyFingerprint } from '../../../shared/structured-agent-session-send-mutation' -import { queuedSendAnswer } from './structured-agent-session-queued-send-answer' import { structuredAgentSessionSendBlock } from './structured-agent-session-send-preparation' import { isUnsettledQueuedMessage } from '../agent-session-journal/queued-message-table' import type { AgentSessionRecord } from '../../../shared/agent-session-record' import { isStructuredAgentSessionMainAgentWorking } from '../../../shared/structured-agent-session-main-agent-working' import type { AgentSessionJournal } from '../agent-session-journal/journal-store' -import type { AgentSessionTurnContext } from './structured-agent-session-turns' import { QueuedMessageNotConsumableError } from '../agent-session-journal/journal-queued-messages' import type { QueuedMessageRow } from '../agent-session-journal/queued-message-table' import type { StructuredAgentSessionHostSession } from './structured-agent-session-host-types' @@ -59,8 +56,9 @@ export function pendingPromptExists(journal: Pick ): QueuedMessageRow | null { @@ -185,75 +183,6 @@ export function queuedMessageBudgetRefusal( return null } -/** - * The accept branch: a capable send while the session is working (or behind an - * actionable backlog) becomes a draft instead of a submission. Returns null for - * the immediate path — an incapable client, an image body (text-only v1), a - * replayed id the journal already answers, or an idle session. - */ -export async function maybeQueueStructuredAgentSessionSend( - context: { - deps: { store: { getRecord: (sessionId: string) => AgentSessionRecord | null } } - }, - ctx: Pick, - params: { - envelope: { clientOperationId: string } - body: AgentJournalMessageItem - delivery?: 'queue-if-active' - } -): Promise< - | { ok: true; value: AgentSessionSendResult } - | { ok: false; refusal: AgentSessionWireRefusal } - | null -> { - const clientMessageId = params.envelope.clientOperationId - if (params.delivery !== 'queue-if-active' || !queuedMessageBodyIsTextOnly(params.body)) { - return null - } - // Asked again with no ledger answer: a send this host queued answers as its replay would — - // its hand-off goes out under a fresh id, so no submission under this id guards it. - const queuedBefore = queuedSendAnswer(ctx.journal, clientMessageId) - if (queuedBefore) { - return { ok: true, value: queuedBefore } - } - // A recorded direct submission under this id replays through today's path. - if (ctx.journal.submissions().some((entry) => entry.clientMessageId === clientMessageId)) { - return null - } - if ( - !shouldQueueStructuredAgentSessionSend({ - journal: ctx.journal, - record: context.deps.store.getRecord(ctx.sessionId), - fence: ctx.fence - }) - ) { - return null - } - const refusal = queuedMessageBudgetRefusal(ctx.journal, params.body) - if (refusal) { - return { ok: false, refusal } - } - // The insert notifies through the journal's commit listener: publication and - // the drain re-derive with no call here to forget. - const row = await ctx.journal.queuedMessages.insert( - { - messageId: clientMessageId, - body: params.body, - // In the session that will send it: the reducer aliases the provider's echo by exactly this. - fingerprint: agentSessionSendBodyFingerprint(ctx.sessionId, params.body), - hostInstance: structuredAgentSessionHostInstance() - }, - ctx.operationReceipt - ) - return { - ok: true, - value: { - clientMessageId, - queued: { messageId: row.messageId, position: row.position, state: row.state } - } - } -} - export type QueuedMessageDrainDeps = { sessions: ReadonlyMap getRecord: (sessionId: string) => AgentSessionRecord | null 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 cf1746eaaf0a..3edd62f627a7 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 @@ -226,12 +226,15 @@ describe('the pause read', () => { }) }) -describe('a card queued after a Stop is a new instruction', () => { - it('a correction typed after a Stop over an empty queue sends when the stopped turn ends', async () => { +describe('a card queued after a Stop', () => { + // Sent while the stopped turn winds down, it waits with the rest: the Stop's pause holds it. + it('waits for Resume even over an empty queue: the stopped turn ending sends nothing', async () => { const working = await rig.workingSend() await rig.stop() const correction = await queuedDraft('typed right after the stop') await rig.settleAccepted(working, 'stopped') + await expectPaused(correction) + expect(await rig.resume()).toMatchObject({ ok: true, value: { resumed: true } }) await eventually(async () => expect(await rig.handoff(correction)).toBeDefined()) }) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-send-while-stopping.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-send-while-stopping.test.ts new file mode 100644 index 000000000000..64d914377c04 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-send-while-stopping.test.ts @@ -0,0 +1,270 @@ +// A send asking to be queued while a person's Stop ends the work is a card at the END of the queue +// at once, written beside the session's lane, which that Stop holds until its provider answers. +// The Stop's pause holds it, even alone, until Resume or a turn accepted after the Stop; +// orchestration mail is the exception, running as the stop lands. Through the real host and journal. + +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { AGENT_JOURNAL_THREAD_SCOPE } from '../../../shared/agent-session-journal-types' +import type { AgentMessageSource } from '../../../shared/agent-session-message-source' +import type { AgentSessionRecord } from '../../../shared/agent-session-record' +import { + HOST_TEST_SESSION as SESSION, + hostTestMessage, + hostTestOperationId +} from './structured-agent-session-host-test-data' +import { + createQueuedMessageTestRig, + eventually, + QUEUED_RIG_CALLER, + type QueuedMessageTestRig +} from './structured-agent-session-queued-message-rig.test-fixture' + +const MAIL: AgentMessageSource = { + kind: 'agent', + senders: [], + orchestration: { message: 'mail-notice', mailbox: 'run:r1', dispatchId: null, messages: [] } +} + +let rig: QueuedMessageTestRig + +beforeEach(async () => { + rig = await createQueuedMessageTestRig() +}) + +afterEach(async () => { + vi.restoreAllMocks() + await rig.dispose() +}) + +function journal() { + const open = rig.host.collaboratorsForTests().sessions.get(SESSION)?.journal + if (!open) { + throw new Error('expected the conversation open') + } + return open +} + +/** The provider's row for the working send's turn. */ +function turnRow(state: 'running' | 'interrupted') { + return journal().appendItem( + { provider: 'codex', threadId: 'thread-1', turnId: 'turn-1', ordinal: 999 }, + { kind: 'turn', turnId: 'turn-1', state, startedAt: 1 }, + { fence: 1, turnScope: AGENT_JOURNAL_THREAD_SCOPE } + ) +} + +/** A send whose turn runs until the Stop `stop` presses lands. */ +async function runningTurn(): Promise { + const working = await rig.workingSend() + await turnRow('running') + return working +} + +/** A person's Stop of the running turn: the chat reads Stopping until the returned land ends it. */ +async function stop(working: string): Promise<() => Promise> { + expect(await rig.stop()).toMatchObject({ ok: true }) + expect(readStopping()).toBe(true) + return async () => { + await rig.settleAccepted(working, 'stopped') + await turnRow('interrupted') + expect(readStopping()).toBe(false) + } +} + +function readStopping(): boolean { + return rig.host['clientDelivery'].readStopping(SESSION) +} + +/** Holds the session's lane until released: a send that answers meanwhile never waited on it. */ +function holdLane(): () => void { + const held = Promise.withResolvers() + void rig.host['tasks'].serialize(SESSION, () => held.promise) + return () => held.resolve() +} + +async function queuedId(sent: ReturnType['result']): Promise { + const queued = await sent + if (!queued.ok || !('queued' in queued.value)) { + throw new Error(`expected a queued receipt: ${JSON.stringify(queued)}`) + } + return queued.value.queued.messageId +} + +/** Nothing sends any of them by itself, and the queue reads paused by the Stop. */ +async function expectHeld(...draftIds: string[]): Promise { + await new Promise((resolve) => setTimeout(resolve, 250)) + for (const draftId of draftIds) { + expect(await rig.handoff(draftId)).toBeUndefined() + } + expect(await rig.queuePause()).toEqual({ reason: 'stopped' }) +} + +/** The next admission waits until released, once it has read the journal it will write to. */ +function holdNextAdmission(): { reached: Promise; release: () => void } { + const reached = Promise.withResolvers() + const gate = Promise.withResolvers() + const admit = rig.store.admitMutationOperation.bind(rig.store) + vi.spyOn(rig.store, 'admitMutationOperation').mockImplementationOnce(async (operation) => { + reached.resolve() + await gate.promise + return admit(operation) + }) + return { reached: reached.promise, release: () => gate.resolve() } +} + +describe('a queued send while Stopping', () => { + it('is a card at once, at the end, with the lane held; Resume sends the cards in order', async () => { + const working = await runningTurn() + const before = await queuedId(rig.send('queued before the stop', 'queue-if-active').result) + const land = await stop(working) + const release = holdLane() + + const during = await queuedId(rig.send('sent while stopping', 'queue-if-active').result) + + release() + expect(await rig.drafts()).toEqual([ + { messageId: before, state: 'waiting' }, + { messageId: during, state: 'waiting' } + ]) + await land() + await expectHeld(before, during) + expect(await rig.resume()).toMatchObject({ ok: true, value: { resumed: true } }) + await eventually(async () => expect((await rig.handoff(before))?.handedOverAt).toBeDefined()) + expect(await rig.handoff(during)).toBeUndefined() + await rig.settleAccepted(await rig.handoffId(before), 'before') + await eventually(async () => expect(await rig.handoff(during)).toBeDefined()) + }) + + it('alone, waits too; a turn sent and accepted after the Stop releases it after that turn', async () => { + const land = await stop(await runningTurn()) + const alone = await queuedId(rig.send('sent while stopping', 'queue-if-active').result) + await land() + await expectHeld(alone) + // What "Send message" does: a plain send, past the card, that ends the pause once accepted. + const next = rig.send('sent now') + expect(await next.result).toMatchObject({ ok: true, value: { submission: expect.anything() } }) + await eventually(async () => + expect((await rig.submission(next.id))?.handedOverAt).toBeDefined() + ) + expect(await rig.handoff(alone)).toBeUndefined() + await rig.settleAccepted(next.id, 'next') + await eventually(async () => expect(await rig.handoff(alone)).toBeDefined()) + }) +}) + +// Mail never waits on a person's Resume: queued while Stopping it runs as the stop lands, ahead of +// the person's cards, and its accepted turn lifts the pause so they follow in order. +describe('orchestration mail sent while Stopping', () => { + it("is a card at the end the person's Stop does not hold: it runs first, then the cards follow", async () => { + const working = await runningTurn() + const person = await queuedId(rig.send('queued before the stop', 'queue-if-active').result) + const land = await stop(working) + const mail = await queuedId( + rig.send('mail sent while stopping', 'queue-if-active', { internal: true, from: MAIL }).result + ) + const typed = await queuedId(rig.send('typed while stopping', 'queue-if-active').result) + expect((await rig.drafts()).map((draft) => draft.messageId)).toEqual([person, mail, typed]) + // Only the mail's body names a sender, so a client labels it as waiting, not paused. + const page = await rig.host.history({ sessionId: SESSION, direction: 'tail' }) + expect( + page.ok && page.page.queuedMessages?.map((card) => card.body.from !== undefined) + ).toEqual([false, true, false]) + + await land() + await eventually(async () => expect((await rig.handoff(mail))?.handedOverAt).toBeDefined()) + expect(await rig.handoff(person)).toBeUndefined() + await rig.settleAccepted(await rig.handoffId(mail), 'mail') + await eventually(async () => expect((await rig.handoff(person))?.handedOverAt).toBeDefined()) + expect(await rig.handoff(typed)).toBeUndefined() + await rig.settleAccepted(await rig.handoffId(person), 'person') + await eventually(async () => expect(await rig.handoff(typed)).toBeDefined()) + }) +}) + +describe('the card written beside the lane', () => { + // Read once, before admission: a stop that lands meanwhile still gets the card, held, and never + // a fallback onto the lane. + it('is still a held card at the end when the stop lands between the read and the write', async () => { + const working = await runningTurn() + const before = await queuedId(rig.send('queued before the stop', 'queue-if-active').result) + const land = await stop(working) + const admission = holdNextAdmission() + const sent = rig.send('sent as the stop lands', 'queue-if-active') + await admission.reached + await land() + admission.release() + + const during = await queuedId(sent.result) + expect(await rig.drafts()).toEqual([ + { messageId: before, state: 'waiting' }, + { messageId: during, state: 'waiting' } + ]) + expect(await rig.submission(sent.id)).toBeUndefined() + await expectHeld(before, during) + }) + + it('writes nothing when the conversation closes meanwhile; the operation stays pending and a resend answers once', async () => { + const land = await stop(await runningTurn()) + const admission = holdNextAdmission() + const clientOperationId = hostTestOperationId() + const fields = { body: hostTestMessage('sent as the chat closes'), delivery: 'queue-if-active' } + const resend = () => + rig.host.send(QUEUED_RIG_CALLER, { + envelope: rig.envelope(fields, 'agentSession.send', clientOperationId), + body: fields.body, + delivery: 'queue-if-active' + }) + const sent = resend() + await admission.reached + await land() + await rig.host.close(SESSION, 'evict') + admission.release() + + await expect(sent).rejects.toMatchObject({ code: 'journal_closed' }) + expect(rig.store.getOperationRow(QUEUED_RIG_CALLER.callerKey, clientOperationId)).toMatchObject( + { outcome: { status: 'pending' } } + ) + // The chat reopens idle: the resend runs for the first time, once. + const first = await resend() + expect(first).toMatchObject({ ok: true, value: { submission: expect.anything() } }) + expect(await resend()).toMatchObject({ + ok: true, + replayed: true, + value: first.ok ? first.value : {} + }) + expect(await rig.drafts()).toEqual([]) + }) + + it('refuses as the lane does when the conversation is cleared meanwhile', async () => { + await stop(await runningTurn()) + const admission = holdNextAdmission() + const sent = rig.send('sent as the chat clears', 'queue-if-active') + await admission.reached + const getRecord = rig.store.getRecord.bind(rig.store) + vi.spyOn(rig.store, 'getRecord').mockImplementation((sessionId): AgentSessionRecord | null => { + const record = getRecord(sessionId) + return ( + record && { + ...record, + conversationCommand: { + command: 'clear', + runtimeFence: 1, + operationId: 'clear-1', + callerKey: QUEUED_RIG_CALLER.callerKey, + phase: 'committed', + state: 'completed', + replacementSessionId: 'clear-replacement-1' + } + } + ) + }) + admission.release() + + expect(await sent.result).toMatchObject({ + ok: false, + refusal: { details: { reason: 'conversationCleared' } } + }) + vi.mocked(rig.store.getRecord).mockRestore() + expect(await rig.drafts()).toEqual([]) + }) +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-send.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-send.ts index 0fa49460d5e9..543e9d99b00a 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-send.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-send.ts @@ -1,10 +1,22 @@ -// A send's queue step around its immediate path: the queue decision before it. Its -// accepted turn lifts a paused queue (`queued-message-pause.ts`), not anything here. +// A send's queue step around its immediate path: the queue decision before it, and the card it +// writes. Its accepted turn lifts a paused queue (`queued-message-pause.ts`), not anything here. -import type { AgentSessionSendResult } from '../../../shared/agent-session-wire' +import type { + AgentSessionSendResult, + AgentSessionWireRefusal +} from '../../../shared/agent-session-wire' import type { AgentJournalMessageItem } from '../../../shared/agent-session-journal-types' +import type { AgentSessionRecord } from '../../../shared/agent-session-record' +import { agentSessionSendBodyFingerprint } from '../../../shared/structured-agent-session-send-mutation' import type { StructuredAgentSessionMutationContext } from './structured-agent-session-host-mutations' -import { maybeQueueStructuredAgentSessionSend } from './structured-agent-session-queued-messages' +import { + queuedMessageBodyIsTextOnly, + queuedMessageBudgetRefusal, + shouldQueueStructuredAgentSessionSend +} from './structured-agent-session-queued-messages' +import { queuedSendAnswer } from './structured-agent-session-queued-send-answer' +import { structuredAgentSessionHostInstance } from './structured-agent-session-queued-pause' +import { structuredAgentSessionSendBlock } from './structured-agent-session-send-preparation' import type { AgentSessionTurnContext, TurnOutcome } from './structured-agent-session-turns' export async function runQueueableStructuredAgentSessionSend( @@ -30,3 +42,111 @@ export async function runQueueableStructuredAgentSessionSend( } return accepted } + +/** + * The accept branch: a capable send while the session is working (or behind an + * actionable backlog) becomes a draft instead of a submission. Returns null for + * the immediate path — an incapable client, an image body (text-only v1), a + * replayed id the journal already answers, or an idle session. + */ +export async function maybeQueueStructuredAgentSessionSend( + context: { + deps: { store: { getRecord: (sessionId: string) => AgentSessionRecord | null } } + }, + ctx: Pick, + params: { + envelope: { clientOperationId: string } + body: AgentJournalMessageItem + delivery?: 'queue-if-active' + } +): Promise< + | { ok: true; value: AgentSessionSendResult } + | { ok: false; refusal: AgentSessionWireRefusal } + | null +> { + const clientMessageId = params.envelope.clientOperationId + if (params.delivery !== 'queue-if-active' || !queuedMessageBodyIsTextOnly(params.body)) { + return null + } + // Asked again with no ledger answer: a send this host queued answers as its replay would — + // its hand-off goes out under a fresh id, so no submission under this id guards it. + const queuedBefore = queuedSendAnswer(ctx.journal, clientMessageId) + if (queuedBefore) { + return { ok: true, value: queuedBefore } + } + // A recorded direct submission under this id replays through today's path. + if (ctx.journal.submissions().some((entry) => entry.clientMessageId === clientMessageId)) { + return null + } + if ( + !shouldQueueStructuredAgentSessionSend({ + journal: ctx.journal, + record: context.deps.store.getRecord(ctx.sessionId), + fence: ctx.fence + }) + ) { + return null + } + return insertQueuedSend(ctx, params) +} + +type QueuedSendParams = Parameters[2] + +/** + * A text send asking to be queued while a person's Stop ends the work: always a card at the end, + * whatever the session reads by now, which that Stop's pause holds until Resume or an accepted + * turn. A replayed id answers as it was recorded, and `blocked` refuses as the immediate path does. + */ +export async function queueStructuredAgentSessionSendAfterStop( + context: { + deps: { store: { getRecord: (sessionId: string) => AgentSessionRecord | null } } + }, + ctx: Pick, + params: QueuedSendParams +): Promise> { + const clientMessageId = params.envelope.clientOperationId + const queuedBefore = queuedSendAnswer(ctx.journal, clientMessageId) + if (queuedBefore) { + return { ok: true, value: queuedBefore } + } + const submission = ctx.journal.submission(clientMessageId) + if (submission) { + return { ok: true, value: { clientMessageId, submission } } + } + return ( + structuredAgentSessionSendBlock(context.deps.store.getRecord(ctx.sessionId)) ?? + insertQueuedSend(ctx, params) + ) +} + +async function insertQueuedSend( + ctx: Pick, + params: QueuedSendParams +): Promise< + { ok: true; value: AgentSessionSendResult } | { ok: false; refusal: AgentSessionWireRefusal } +> { + const clientMessageId = params.envelope.clientOperationId + const refusal = queuedMessageBudgetRefusal(ctx.journal, params.body) + if (refusal) { + return { ok: false, refusal } + } + // The insert notifies through the journal's commit listener: publication and + // the drain re-derive with no call here to forget. + const row = await ctx.journal.queuedMessages.insert( + { + messageId: clientMessageId, + body: params.body, + // In the session that will send it: the reducer aliases the provider's echo by exactly this. + fingerprint: agentSessionSendBodyFingerprint(ctx.sessionId, params.body), + hostInstance: structuredAgentSessionHostInstance() + }, + ctx.operationReceipt + ) + return { + ok: true, + value: { + clientMessageId, + queued: { messageId: row.messageId, position: row.position, state: row.state } + } + } +} 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 4e45ab4a6cc3..3b9e0dc4f0e6 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 @@ -1,10 +1,11 @@ // Stop writes one event row before it interrupts, and the queue's pause is derived from it: -// through the real host, the cards queued before a Stop wait, a card queued after it sends -// normally but never ahead of them, a withdrawn card comes back under it, a crash keeps it, any -// turn sent after it ends it, and no stored pause is ever written. +// through the real host, every card waits under a person's Stop, one queued while it winds down +// too, a withdrawn card comes back under it, a crash keeps it, any turn sent after it ends it, and +// no stored pause is ever written. import { afterEach, beforeEach, describe, expect, it } from 'vitest' import { agentSessionFailureFact } from '../../../shared/agent-session-failure' +import type { AgentMessageSource } from '../../../shared/agent-session-message-source' import { agentSessionFailureWords } from '../../../shared/agent-session-failure-words' import Database from '../../sqlite/sync-database' import { journalDatabasePath } from '../agent-session-journal/journal-host-database' @@ -23,6 +24,12 @@ import { structuredQueuePauses } from './structured-agent-session-queued-pause' let rig: QueuedMessageTestRig +const RIG_MAIL: AgentMessageSource = { + kind: 'agent', + senders: [], + orchestration: { message: 'mail-notice', mailbox: 'run:r1', dispatchId: null, messages: [] } +} + beforeEach(async () => { rig = await createQueuedMessageTestRig({ restartable: true }) }) @@ -94,7 +101,7 @@ describe("Stop's event", () => { expect(pausedAtInterrupt).toBe('stopped') }) - it('over an EMPTY queue holds nothing: a card queued during a later mail turn sends normally', async () => { + it('over an EMPTY queue, a later mail turn ends it: a card queued during that turn sends after it', async () => { const working = await rig.workingSend() await rig.stop() await rig.settleAccepted(working, 'stopped') @@ -105,13 +112,13 @@ describe("Stop's event", () => { await eventually(async () => expect(await rig.handoff(typed)).toBeDefined()) }) - it('a card queued after the Stop waits behind the cards it holds, then all send in order', async () => { + it('holds a card queued after the Stop at the end, behind the cards before it; Resume sends all in order', async () => { const working = await rig.workingSend() const held = await queuedDraft('queued before the stop') await rig.stop() const later = await queuedDraft('typed while the stopped turn winds down') await rig.settleAccepted(working, 'stopped') - // The queue never reorders: the newer card waits behind the held one, with no caption of its own. + // Both held, the newer one at the end, with no caption of its own. await expectHeld('stopped', held, later) expect(await rig.drafts()).toEqual([ { messageId: held, state: 'waiting' }, @@ -188,16 +195,24 @@ describe("Stop's event", () => { }) describe("a restart's hold over a card queued after a Stop", () => { - it('a card queued after a Stop over an empty queue, before a restart, waits unshown', async () => { + // Mail, which a person's Stop never holds: only the restart can hold it. + it('mail queued after a Stop over an empty queue, before a restart, waits unshown', async () => { const working = await rig.workingSend() await rig.stop() await rig.settleAccepted(working, 'stopped') const mail = await mailTurn() - const typed = await queuedDraft('typed during the mail turn') + const queued = await rig.send('mail queued during the mail turn', 'queue-if-active', { + internal: true, + from: RIG_MAIL + }).result + if (!queued.ok || !('queued' in queued.value)) { + throw new Error('expected a queued receipt') + } + const typed = queued.value.queued.messageId // The process dies with no close: a quit writes no Stop event, so only a turn ends the pauses. rig.crashRestartHostProcess() await rig.settleAccepted(mail, 'mail') - // The new host opens the conversation for its first reader. The card came after the Stop, so + // The new host opens the conversation for its first reader. The Stop's pause passes mail, so // only the restart holds it, and a restart's hold is never published. await rig.queuePause() expect(structuredQueuePauses(journal()).map((pause) => pause.reason)).toEqual([ @@ -207,17 +222,23 @@ describe("a restart's hold over a card queued after a Stop", () => { await expectHeld(null, typed) }) - it("a card queued during the queue's own send after a Stop, before a restart, waits unshown", async () => { + // Queued mail is the queue's one send a person's Stop does not hold: it runs as the stop lands. + it("a card queued during the queue's own send of mail after a Stop, before a restart, waits unshown", async () => { const working = await rig.workingSend() await rig.stop() - const correction = await queuedDraft('typed while the interrupt lands') + const mail = await rig.send('mail sent while the interrupt lands', 'queue-if-active', { + internal: true, + from: RIG_MAIL + }).result + if (!mail.ok || !('queued' in mail.value)) { + throw new Error('expected a queued receipt') + } + const mailId = mail.value.queued.messageId await rig.settleAccepted(working, 'stopped') - await eventually(async () => - expect((await rig.handoff(correction))?.handedOverAt).toBeDefined() - ) + await eventually(async () => expect((await rig.handoff(mailId))?.handedOverAt).toBeDefined()) const typed = await queuedDraft('typed during that send') rig.crashRestartHostProcess() - await rig.settleAccepted(await rig.handoffId(correction), 'correction') + await rig.settleAccepted(await rig.handoffId(mailId), 'mail') await rig.queuePause() expect(structuredQueuePauses(journal()).map((pause) => pause.reason)).toEqual(['restarted']) await expectHeld(null, typed) 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 964cefc1b2f5..7a5b427cc8d6 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 @@ -61,7 +61,7 @@ export async function runRecordedStop( * What a Stop reaching a running agent stops: work no Stop event records yet (`unrecorded`), or * only what the Stop still in force already records, which it repeats with nothing sent since, on * the same turn or one that opened after a Stop pressed before any turn showed (`repeat`): a card - * queued between the presses then sends normally, as after one Stop. `late`: it names a turn + * queued between the presses then waits under that one Stop. `late`: it names a turn * already over, as a late Stop from a phone can. */ export function stopReachesUnrecordedWork( diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-stop-event.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-stop-event.test.ts index 2a8c496e8b2a..a7e54d5f9fbe 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-stop-event.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-stop-event.test.ts @@ -55,6 +55,15 @@ async function turnRow(turnId: string, state: 'running' | 'interrupted') { ) } +/** Held by the one Stop once its turn ends, then sent by Resume. */ +async function expectReleasedByResume(draftId: string): Promise { + await new Promise((resolve) => setTimeout(resolve, 250)) + expect(await rig.handoff(draftId)).toBeUndefined() + expect(await rig.queuePause()).toEqual({ reason: 'stopped' }) + expect(await rig.resume()).toMatchObject({ ok: true, value: { resumed: true } }) + await eventually(async () => expect(await rig.handoff(draftId)).toBeDefined()) +} + function withdraw(clientMessageId: string) { return rig.host.settleLateDispatch({ sessionId: HOST_TEST_SESSION, @@ -169,7 +178,7 @@ describe("a Stop's event", () => { ]) }) - it('a second press while the first interrupt lands writes nothing: a card queued between them sends normally', async () => { + it('a second press while the first interrupt lands writes nothing: a card queued between them waits as after one Stop', async () => { rig = await createQueuedMessageTestRig() const working = await rig.workingSend() expect(await rig.stop()).toMatchObject({ ok: true, value: { cancelled: true } }) @@ -177,10 +186,10 @@ describe("a Stop's event", () => { expect(await rig.stop()).toMatchObject({ ok: true }) expect(stopEvents()).toHaveLength(1) await rig.settleAccepted(working, 'stopped') - await eventually(async () => expect(await rig.handoff(between)).toBeDefined()) + await expectReleasedByResume(between) }) - it('a second press once the turn shows, after a first before it did, writes nothing: a card queued between sends', async () => { + it('a second press once the turn shows, after a first before it did, writes nothing: a card queued between waits', async () => { rig = await createQueuedMessageTestRig() const working = await rig.workingSend() await rig.stop() @@ -190,7 +199,7 @@ describe("a Stop's event", () => { expect(stopEvents()).toHaveLength(1) await rig.settleAccepted(working, 'stopped') await turnRow('turn-1', 'interrupted') - await eventually(async () => expect(await rig.handoff(between)).toBeDefined()) + await expectReleasedByResume(between) }) it('a second press after a card sent into the turn settled unknown writes again', async () => { diff --git a/src/main/runtime/structured-agent-session-codex-interrupt-order.test.ts b/src/main/runtime/structured-agent-session-codex-interrupt-order.test.ts index e28f7f689006..e21cbdd48def 100644 --- a/src/main/runtime/structured-agent-session-codex-interrupt-order.test.ts +++ b/src/main/runtime/structured-agent-session-codex-interrupt-order.test.ts @@ -88,6 +88,38 @@ async function send(text: string): Promise { return sent.value.clientMessageId } +/** A send asking to be queued while the agent works, under `clientOperationId` when it retries. */ +async function sendQueued(text: string, clientOperationId?: string) { + const body = hostTestMessage(text) + const delivery = 'queue-if-active' as const + const fields = { body, delivery } + const sent = await host.send(CALLER, { + envelope: { + ...envelope('agentSession.send', fields), + ...(clientOperationId ? { clientOperationId } : {}) + }, + body, + delivery, + userSend: true + }) + if (!sent.ok) { + throw new Error(JSON.stringify(sent.refusal)) + } + return sent.value +} + +async function queuePage() { + const page = await host.history({ sessionId: SESSION, direction: 'tail' }) + if (!page.ok) { + throw new Error('history refused') + } + return { cards: page.page.queuedMessages ?? [], pause: page.page.queuePause ?? null } +} + +async function queuedCards() { + return (await queuePage()).cards +} + async function stop(turnId?: string): Promise { const stopped = await host.cancel(CALLER, { envelope: envelope('agentSession.cancel', turnId === undefined ? {} : { turnId }), @@ -311,3 +343,148 @@ describe('a send behind a Codex Stop answered before its turn ends', () => { expect(submission?.dispatchState).not.toBe('rejected') }) }) + +async function resume(): Promise { + const resumed = await host.queuedMessagesResume(CALLER, { + envelope: envelope('agentSession.queuedMessagesResume', {}) + }) + expect(resumed).toMatchObject({ ok: true, value: { resumed: true } }) +} + +/** The cards each queue hand-off sent, in order. */ +async function handOffs(): Promise { + return (await host.journalSnapshot(SESSION)).submissions.flatMap((entry) => + entry.queuedMessageId ? [entry.queuedMessageId] : [] + ) +} + +function queuedMessageId(sent: Awaited>): string { + if (!('queued' in sent)) { + throw new Error('expected a queued receipt') + } + return sent.queued.messageId +} + +/** A running turn-1, and a person's Stop of it whose interrupt Codex has yet to answer. */ +async function stopAwaitingCodex(beforeStop?: () => Promise) { + const first = await send('look around') + await vi.waitFor(() => expect(answers).toBe(1)) + turns.start() + turns.echo(first) + await beforeStop?.() + const answer = Promise.withResolvers() + interruptAnswer = answer.promise + let stopped = false + const stopping = stop().then(() => { + stopped = true + }) + await vi.waitFor(() => expect(statuses.at(-1)?.stopping).toBe(true)) + return { + stopped: () => stopped, + /** Codex answers the interrupt, then ends the turn; nothing then runs by itself. */ + land: async () => { + answer.resolve() + await stopping + turns.end('interrupted') + await vi.waitFor(async () => expect(await owesWork()).toBe(false)) + await new Promise((resolve) => setTimeout(resolve, 250)) + expect(answers).toBe(1) + expect(startedTurns).toEqual(['turn-1']) + } + } +} + +/** The turn the last turn/start opened runs, takes the latest send, and completes. */ +async function completeTurn(): Promise { + const latest = (await host.journalSnapshot(SESSION)).submissions.at(-1) + turns.start() + turns.echo(latest?.clientMessageId ?? '') + turns.end('completed') +} + +// A message sent asking to be queued while a person's Stop ends the turn is a card at once, at the +// end, while the Stop still waits on Codex. The Stop's pause holds it, even alone: Resume, or a turn +// accepted after the Stop, sends the cards, each as its own turn. +describe('a queued send while a Codex Stop waits on its answer', () => { + it('is a card at once that the Stop holds even alone; Resume sends it as its own turn', async () => { + const stopping = await stopAwaitingCodex() + + const queued = await sendQueued('run this after the stop') + + expect(stopping.stopped()).toBe(false) + const messageId = queuedMessageId(queued) + expect(await queuePage()).toEqual({ + cards: [expect.objectContaining({ messageId, state: 'waiting' })], + pause: expect.objectContaining({ reason: 'stopped' }) + }) + // A retry of the same operation answers the same card. + expect(await sendQueued('run this after the stop', messageId)).toEqual(queued) + + await stopping.land() + expect(await queuedCards()).toEqual([expect.objectContaining({ messageId, state: 'waiting' })]) + + await resume() + await vi.waitFor(() => expect(answers).toBe(2)) + expect(startedTurns).toEqual(['turn-1', 'turn-2']) + expect(steers).toBe(0) + // Retried once handed off, it answers with the card's hand-off, as any queued send. + expect(await sendQueued('run this after the stop', messageId)).toMatchObject({ + submission: { queuedMessageId: messageId } + }) + const handOff = (await host.journalSnapshot(SESSION)).submissions.find( + (entry) => entry.queuedMessageId === messageId + ) + expect(handOff?.dispatchState).not.toBe('rejected') + expect(await queuedCards()).toEqual([]) + }) +}) + +describe('a queued send while Stopping, behind a card the Stop paused', () => { + let paused = '' + const queueBeforeStop = async () => { + paused = queuedMessageId(await sendQueued('queued before the stop')) + } + + it('waits at the end with it; Resume sends both in order, each its own turn', async () => { + const stopping = await stopAwaitingCodex(queueBeforeStop) + const during = queuedMessageId(await sendQueued('sent while stopping')) + expect((await queuedCards()).map((card) => card.messageId)).toEqual([paused, during]) + + await stopping.land() + expect(await queuePage()).toEqual({ + cards: [ + expect.objectContaining({ messageId: paused, state: 'waiting' }), + expect.objectContaining({ messageId: during, state: 'waiting' }) + ], + pause: expect.objectContaining({ reason: 'stopped' }) + }) + + await resume() + await vi.waitFor(() => expect(answers).toBe(2)) + expect(await handOffs()).toEqual([paused]) + await completeTurn() + await vi.waitFor(() => expect(answers).toBe(3)) + expect(await handOffs()).toEqual([paused, during]) + expect(startedTurns).toEqual(['turn-1', 'turn-2', 'turn-3']) + expect(steers).toBe(0) + }) + + it('the next turn sent after the stop releases them: it runs first, then both in order', async () => { + const stopping = await stopAwaitingCodex(queueBeforeStop) + const during = queuedMessageId(await sendQueued('sent while stopping')) + await stopping.land() + + // What the "Send message?" dialog's Send does: a plain send past the paused cards. + await send('sent now, past the paused cards') + await vi.waitFor(() => expect(answers).toBe(2)) + expect(await handOffs()).toEqual([]) + await completeTurn() + await vi.waitFor(() => expect(answers).toBe(3)) + expect(await handOffs()).toEqual([paused]) + await completeTurn() + await vi.waitFor(() => expect(answers).toBe(4)) + expect(await handOffs()).toEqual([paused, during]) + expect(startedTurns).toEqual(['turn-1', 'turn-2', 'turn-3', 'turn-4']) + expect(steers).toBe(0) + }) +}) diff --git a/src/renderer/src/components/native-chat/NativeChatStructuredSession.stopping.test.tsx b/src/renderer/src/components/native-chat/NativeChatStructuredSession.stopping.test.tsx index 1299f43f2950..8a561390ed0e 100644 --- a/src/renderer/src/components/native-chat/NativeChatStructuredSession.stopping.test.tsx +++ b/src/renderer/src/components/native-chat/NativeChatStructuredSession.stopping.test.tsx @@ -93,7 +93,7 @@ describe("the chat pane while a person's Stop ends the turn", () => { expect(mocks.stop).toHaveBeenCalledOnce() }) - it('says a message is queued to run after the stop only where the host queues sends', async () => { + it('tells the composer a send while Stopping becomes a card where the host queues sends', async () => { mocks.turnId = 'turn-1' mocks.sendsQueue = true renderPane() diff --git a/src/renderer/src/components/native-chat/NativeChatStructuredSession.test-harness.tsx b/src/renderer/src/components/native-chat/NativeChatStructuredSession.test-harness.tsx index 5f4fdb95e608..62fa9371e229 100644 --- a/src/renderer/src/components/native-chat/NativeChatStructuredSession.test-harness.tsx +++ b/src/renderer/src/components/native-chat/NativeChatStructuredSession.test-harness.tsx @@ -260,9 +260,9 @@ export function createStructuredSessionMocks() { epoch: 'epoch-1', rewind: { surface: undefined }, canStop: mocks.canStop ?? mocks.turnId !== null, - queueSendsNext: mocks.queueSendsNext, stopPressed: mocks.stopPressed, sendsQueue: mocks.sendsQueue, + queueSendsNext: mocks.queueSendsNext, stop: mocks.stop, queuedMessages: { cards: mocks.queuedCards, diff --git a/src/renderer/src/components/native-chat/native-chat-composer-placeholder.test.ts b/src/renderer/src/components/native-chat/native-chat-composer-placeholder.test.ts index 0fbb8a9cec79..0eb56c05423f 100644 --- a/src/renderer/src/components/native-chat/native-chat-composer-placeholder.test.ts +++ b/src/renderer/src/components/native-chat/native-chat-composer-placeholder.test.ts @@ -6,16 +6,18 @@ vi.mock('@/runtime/runtime-terminal-inspection', () => ({ isRemoteRuntimePtyId: import { nativeChatComposerPlaceholder } from './native-chat-composer-target' describe("the composer's placeholder", () => { - it('says a message runs after the stop while the chat reads Stopping', () => { - expect(nativeChatComposerPlaceholder(true, true, 'queue')).toBe( - 'Queue a message to run after the stop' - ) - // Where the host does not queue sends, it holds the message until the stop lands. + // Where a send is not queued, the host holds it until the stop lands. + it('says a message runs after the stop while the chat reads Stopping and sends are not queued', () => { expect(nativeChatComposerPlaceholder(true, true, 'send')).toBe( 'Send a message to run after the stop' ) }) + // A queued one is a card the Stop holds until Resume: nothing promises it runs after the stop. + it('reads as usual while Stopping where a send is queued', () => { + expect(nativeChatComposerPlaceholder(true, true, 'queue')).toBe('Send a message…') + }) + it('reads as usual otherwise', () => { expect(nativeChatComposerPlaceholder(true, true)).toBe('Send a message…') }) diff --git a/src/renderer/src/components/native-chat/native-chat-composer-target.ts b/src/renderer/src/components/native-chat/native-chat-composer-target.ts index ff4e64bc6abb..5997d2a3382d 100644 --- a/src/renderer/src/components/native-chat/native-chat-composer-target.ts +++ b/src/renderer/src/components/native-chat/native-chat-composer-target.ts @@ -30,12 +30,7 @@ export function nativeChatComposerPlaceholder( if (!canSend) { return translate('components.native-chat.composer.locked', 'Input is held by another device.') } - if (afterStop === 'queue') { - return translate( - 'components.native-chat.status.queueAfterStop', - NATIVE_CHAT_TURN_STATUS_COPY.queueAfterStop - ) - } + // A queued one is a card the Stop holds until Resume, so it gets the normal prompt. if (afterStop === 'send') { return translate( 'components.native-chat.status.sendAfterStop', diff --git a/src/renderer/src/components/native-chat/structured-agent-session-queued-cards.test.ts b/src/renderer/src/components/native-chat/structured-agent-session-queued-cards.test.ts index 28df16cb67b0..912bb53c5d63 100644 --- a/src/renderer/src/components/native-chat/structured-agent-session-queued-cards.test.ts +++ b/src/renderer/src/components/native-chat/structured-agent-session-queued-cards.test.ts @@ -4,7 +4,11 @@ import { describe, expect, it } from 'vitest' import type { AgentJournalSubmission } from '../../../../shared/agent-session-journal-types' -import type { AgentSessionQueuedMessage } from '../../../../shared/agent-session-wire' +import type { AgentMessageSource } from '../../../../shared/agent-session-message-source' +import type { + AgentSessionQueuedMessage, + AgentSessionQueuePause +} from '../../../../shared/agent-session-wire' import type { StructuredAgentSessionOutboxEntry } from '../../../../shared/structured-agent-session-outbox' import { newestSteerableQueuedMessageCard, @@ -55,7 +59,19 @@ function handOff( return submission(id, dispatchState, draftId) } +/** Another agent's message, which names its sender on its body. */ +const AGENT_FROM: AgentMessageSource = { + kind: 'agent', + senders: [], + orchestration: { message: 'mail-notice', mailbox: 'run:r1', dispatchId: null, messages: [] } +} + +function mailDraft(id: string, position: number): AgentSessionQueuedMessage { + const base = draft(id, position) + return { ...base, body: { ...base.body, from: AGENT_FROM } } +} const IDLE = { hasPendingPrompt: false } +const STOPPED = { reason: 'stopped' } as const const QUEUEING = { capability: 'supported', enabled: true } as const describe('queued message cards', () => { @@ -148,7 +164,7 @@ describe('queued message cards', () => { draft('behind', 4) ], [], - { hasPendingPrompt: true, queuePaused: true } + { hasPendingPrompt: true, queuePause: STOPPED } ) expect(cards.map((card) => card.hold)).toEqual([ 'queue-paused', @@ -161,10 +177,22 @@ describe('queued message cards', () => { ).toBe('awaiting-answer') }) + // Another agent's mail runs when the stop lands, so only a person's card reads paused. + it("a person's Stop holds every card but mail; any other pause holds mail too", () => { + const mail = mailDraft('mail', 2) + const project = (queuePause: AgentSessionQueuePause) => + projectQueuedMessageCards([draft('typed', 1), mail], [], { + hasPendingPrompt: false, + queuePause + }).map((card) => card.hold) + expect(project(STOPPED)).toEqual(['queue-paused', 'turn']) + expect(project({ reason: 'cleared' })).toEqual(['queue-paused', 'queue-paused']) + }) + it('a paused queue holds every waiting card, in order: an answer does not drain it', () => { const cards = projectQueuedMessageCards([draft('held', 1), draft('typed-after', 2)], [], { hasPendingPrompt: true, - queuePaused: true + queuePause: STOPPED }) expect(cards.map((card) => card.hold)).toEqual(['queue-paused', 'queue-paused']) // Still Steer: the header row, not the card, says it waits. @@ -173,8 +201,11 @@ describe('queued message cards', () => { it("the header names the queue's pause while it holds a card, and none over cards Resume would not send", () => { const stopped = { reason: 'stopped' } as const - const project = (messages: AgentSessionQueuedMessage[], queuePaused = true) => - projectQueuedMessageCards(messages, [], { hasPendingPrompt: false, queuePaused }) + const project = (messages: AgentSessionQueuedMessage[], paused = true) => + projectQueuedMessageCards(messages, [], { + hasPendingPrompt: false, + queuePause: paused ? stopped : null + }) expect(queuedMessagesQueuePause(project([draft('held', 1)]), stopped)).toEqual(stopped) expect(queuedMessagesQueuePause(project([draft('held', 1)]), { reason: 'cleared' })).toEqual({ reason: 'cleared' diff --git a/src/renderer/src/components/native-chat/structured-agent-session-queued-cards.ts b/src/renderer/src/components/native-chat/structured-agent-session-queued-cards.ts index 4720d9eefbc3..fc01915faab7 100644 --- a/src/renderer/src/components/native-chat/structured-agent-session-queued-cards.ts +++ b/src/renderer/src/components/native-chat/structured-agent-session-queued-cards.ts @@ -3,9 +3,10 @@ import type { UnreadAgentSessionFailureFact } from '../../../../shared/agent-session-failure' import type { AgentJournalSubmission } from '../../../../shared/agent-session-journal-types' -import type { - AgentSessionQueuedMessage, - AgentSessionQueuePause +import { + queuePauseHoldsCard, + type AgentSessionQueuedMessage, + type AgentSessionQueuePause } from '../../../../shared/agent-session-wire' import { readAgentMessageSource, @@ -62,7 +63,7 @@ function queuedMessageCardText(body: AgentSessionQueuedMessage['body']): string export function projectQueuedMessageCards( queuedMessages: readonly AgentSessionQueuedMessage[] | null | undefined, submissions: readonly AgentJournalSubmission[], - session: { hasPendingPrompt: boolean; queuePaused?: boolean } + session: { hasPendingPrompt: boolean; queuePause?: AgentSessionQueuePause | null } ): QueuedMessageCard[] { const handedOff = handedOffQueuedMessageIds( submissions.filter((submission) => submission.dispatchState !== 'rejected') @@ -79,7 +80,7 @@ export function projectQueuedMessageCards( ? 'paused' : behindReturned ? 'behind-returned' - : session.queuePaused + : queuePauseHoldsCard(session.queuePause, message) ? 'queue-paused' : session.hasPendingPrompt ? 'awaiting-answer' diff --git a/src/renderer/src/components/native-chat/use-structured-agent-session-queued-messages.resume.test.tsx b/src/renderer/src/components/native-chat/use-structured-agent-session-queued-messages.resume.test.tsx index bfe2599e6572..58b50a4ceb74 100644 --- a/src/renderer/src/components/native-chat/use-structured-agent-session-queued-messages.resume.test.tsx +++ b/src/renderer/src/components/native-chat/use-structured-agent-session-queued-messages.resume.test.tsx @@ -142,11 +142,13 @@ describe('whether Resume is offered', () => { }) describe('while a turn runs over held cards', () => { - it('the header row still names the pause and offers Resume; the composer does not', () => { + it('the header row still names the pause and offers Resume; the composer offers neither it nor "Send message?"', () => { // `isWorking` counts the queue's coming send, which the host names. const { result } = renderController({ isWorking: true }) expect(result.current.pause).toEqual({ reason: 'stopped' }) expect(result.current.queueResume).toBeUndefined() + // Stopping reads as working: a message sent then joins the queue, with no dialog. + expect(result.current.queueHold).toBeUndefined() }) }) diff --git a/src/renderer/src/components/native-chat/use-structured-agent-session-queued-messages.ts b/src/renderer/src/components/native-chat/use-structured-agent-session-queued-messages.ts index dfb5a211c5c8..af3b90d88ed0 100644 --- a/src/renderer/src/components/native-chat/use-structured-agent-session-queued-messages.ts +++ b/src/renderer/src/components/native-chat/use-structured-agent-session-queued-messages.ts @@ -86,7 +86,7 @@ export function useStructuredAgentSessionQueuedMessages(args: { () => projectQueuedMessageCards(queuedMessages, submissions, { hasPendingPrompt, - queuePaused: queuePause !== null + queuePause }), [hasPendingPrompt, queuePause, queuedMessages, submissions] ) diff --git a/src/renderer/src/components/native-chat/use-structured-agent-session.queue-while-stopping.test.tsx b/src/renderer/src/components/native-chat/use-structured-agent-session.queue-while-stopping.test.tsx index 397c3a422bfe..d319c9442b80 100644 --- a/src/renderer/src/components/native-chat/use-structured-agent-session.queue-while-stopping.test.tsx +++ b/src/renderer/src/components/native-chat/use-structured-agent-session.queue-while-stopping.test.tsx @@ -1,8 +1,9 @@ // @vitest-environment happy-dom -// A message sent while the chat reads Stopping, through the chat's own send and its real outbox: -// where the host does not queue sends, it goes out plain, for the host to hold until the stop lands, -// and is marked as sent while stopping, so it draws after the Stopping line. +// A message sent while the chat reads Stopping, through the chat's own send and its real outbox. +// With queueing on, where the host queues sends, it asks to be queued and draws no bubble: the host +// makes it a card the Stop holds. Otherwise it goes out plain, for the host to hold until the stop +// lands, and is marked as sent while stopping, so it draws after the Stopping line. import { cleanup, renderHook, waitFor } from '@testing-library/react' import { afterEach, beforeEach, expect, it, vi } from 'vitest' @@ -28,7 +29,10 @@ vi.mock('./use-structured-agent-session-read', () => ({ }) })) -import { AGENT_SESSION_CONVERSATION_STOP_RUNTIME_CAPABILITY } from '../../../../shared/protocol-version' +import { + AGENT_SESSION_CONVERSATION_STOP_RUNTIME_CAPABILITY, + AGENT_SESSION_QUEUED_MESSAGES_RUNTIME_CAPABILITY +} from '../../../../shared/protocol-version' import { setLocalRuntimeCapabilitiesForTests } from '@/runtime/local-runtime-capabilities' import { clearNativeChatDraftCacheForTests } from './native-chat-draft-cache' import { useStructuredAgentSession } from './use-structured-agent-session' @@ -63,7 +67,7 @@ afterEach(() => { mocks.call.mockReset() }) -function renderStopping() { +function renderStopping(queueFollowUps = false) { return renderHook(() => useStructuredAgentSession({ sessionId: 'session-1', @@ -71,24 +75,55 @@ function renderStopping() { target: { kind: 'local' }, isVisible: true, composerScopeKey: 'scope-1', - queueFollowUps: false, + queueFollowUps, hostStopping: true }) ) } +function expectPlainSentWhileStopping( + messages: ReturnType['result']['current']['messages'] +) { + expect(sends()[0]).not.toHaveProperty('delivery') + expect( + messages.find((message) => JSON.stringify(message.blocks).includes('run this after the stop')) + ).toMatchObject({ role: 'user', sentWhileStopping: true }) +} + +it('with queueing on, asks a queueing host to queue it, and draws no bubble', async () => { + setLocalRuntimeCapabilitiesForTests([ + AGENT_SESSION_CONVERSATION_STOP_RUNTIME_CAPABILITY, + AGENT_SESSION_QUEUED_MESSAGES_RUNTIME_CAPABILITY + ]) + const { result } = renderStopping(true) + + expect(result.current.send('run this after the stop')).toBe('queued') + + await waitFor(() => expect(sends()).toHaveLength(1)) + expect(sends()[0]?.delivery).toBe('queue-if-active') + expect(JSON.stringify(result.current.messages)).not.toContain('run this after the stop') +}) + +it('with queueing off, sends plain to a queueing host, marked as sent while stopping', async () => { + setLocalRuntimeCapabilitiesForTests([ + AGENT_SESSION_CONVERSATION_STOP_RUNTIME_CAPABILITY, + AGENT_SESSION_QUEUED_MESSAGES_RUNTIME_CAPABILITY + ]) + const { result } = renderStopping(false) + + expect(result.current.send('run this after the stop')).toBe(true) + + await waitFor(() => expect(sends()).toHaveLength(1)) + expectPlainSentWhileStopping(result.current.messages) +}) + it('sends plain where the host does not queue sends, marked as sent while stopping', async () => { const { result } = renderStopping() expect(result.current.send('run this after the stop')).toBe(true) await waitFor(() => expect(sends()).toHaveLength(1)) - expect(sends()[0]).not.toHaveProperty('delivery') - expect( - result.current.messages.find((message) => - JSON.stringify(message.blocks).includes('run this after the stop') - ) - ).toMatchObject({ role: 'user', sentWhileStopping: true }) + expectPlainSentWhileStopping(result.current.messages) }) // Sent before the Stop, the host steers it into the turn: it is not one held behind the Stop. diff --git a/src/renderer/src/i18n/en-runtime-required.json b/src/renderer/src/i18n/en-runtime-required.json index a36edc792f1c..42aa0c9b63e4 100644 --- a/src/renderer/src/i18n/en-runtime-required.json +++ b/src/renderer/src/i18n/en-runtime-required.json @@ -2856,7 +2856,6 @@ "status": { "failedAfter": "Failed after {{value0}}", "interruptedAfter": "Interrupted after {{value0}}", - "queueAfterStop": "Queue a message to run after the stop", "responding": "Agent is responding", "sendAfterStop": "Send a message to run after the stop", "toggleDetails": "Toggle turn details", diff --git a/src/renderer/src/i18n/locales/en.json b/src/renderer/src/i18n/locales/en.json index 89549f52ed1f..fa1c3e1a7f26 100644 --- a/src/renderer/src/i18n/locales/en.json +++ b/src/renderer/src/i18n/locales/en.json @@ -18124,7 +18124,6 @@ "responding": "Agent is responding", "working": "Working…", "stopping": "Stopping…", - "queueAfterStop": "Queue a message to run after the stop", "sendAfterStop": "Send a message to run after the stop", "thinking": "Thinking", "workingFor": "Working for {{value0}}", diff --git a/src/renderer/src/i18n/locales/es.json b/src/renderer/src/i18n/locales/es.json index 44cf0a1dc9e3..7365772ef223 100644 --- a/src/renderer/src/i18n/locales/es.json +++ b/src/renderer/src/i18n/locales/es.json @@ -17857,7 +17857,6 @@ "status": { "responding": "El agente está respondiendo", "stopping": "Deteniendo…", - "queueAfterStop": "Pon en cola un mensaje para que se ejecute después de detenerse", "sendAfterStop": "Envía un mensaje para que se ejecute después de detenerse", "interruptedAfter": "Interrumpido tras {{value0}}", "failedAfter": "Falló tras {{value0}}", diff --git a/src/renderer/src/i18n/locales/fr.json b/src/renderer/src/i18n/locales/fr.json index 3e8e74252196..b0ea5dde55a2 100644 --- a/src/renderer/src/i18n/locales/fr.json +++ b/src/renderer/src/i18n/locales/fr.json @@ -17959,7 +17959,6 @@ "responding": "L'agent répond", "working": "Fonctionnement…", "stopping": "Arrêt en cours…", - "queueAfterStop": "Mettez un message en file d'attente pour l'exécuter après l'arrêt", "sendAfterStop": "Envoyez un message pour l'exécuter après l'arrêt", "thinking": "Réflexion", "workingFor": "En cours depuis {{value0}}", diff --git a/src/renderer/src/i18n/locales/ja.json b/src/renderer/src/i18n/locales/ja.json index bdb0027a161f..2cc9a4561da5 100644 --- a/src/renderer/src/i18n/locales/ja.json +++ b/src/renderer/src/i18n/locales/ja.json @@ -17893,7 +17893,6 @@ "responding": "Agent が応答中です", "working": "処理中…", "stopping": "停止中…", - "queueAfterStop": "停止後に実行するメッセージをキューに追加", "sendAfterStop": "停止後に実行するメッセージを送信", "thinking": "考え", "workingFor": "{{value0}} 処理中", diff --git a/src/renderer/src/i18n/locales/ko.json b/src/renderer/src/i18n/locales/ko.json index aedeff54d6cd..be908e0b523a 100644 --- a/src/renderer/src/i18n/locales/ko.json +++ b/src/renderer/src/i18n/locales/ko.json @@ -17893,7 +17893,6 @@ "responding": "Agent이 응답 중입니다.", "working": "작업 중…", "stopping": "중지하는 중…", - "queueAfterStop": "중지 후에 실행할 메시지를 대기열에 추가", "sendAfterStop": "중지 후에 실행할 메시지 보내기", "thinking": "생각", "workingFor": "{{value0}} 작업 중", diff --git a/src/renderer/src/i18n/locales/zh.json b/src/renderer/src/i18n/locales/zh.json index 824d16f0be59..50ad261fc78b 100644 --- a/src/renderer/src/i18n/locales/zh.json +++ b/src/renderer/src/i18n/locales/zh.json @@ -17858,7 +17858,6 @@ "responding": "智能体正在回复", "working": "处理中…", "stopping": "正在停止…", - "queueAfterStop": "将消息加入队列，在停止后运行", "sendAfterStop": "发送消息，在停止后运行", "thinking": "思维", "workingFor": "已工作 {{value0}}", diff --git a/src/shared/agent-session-queued-message-wire.ts b/src/shared/agent-session-queued-message-wire.ts index a1df60b85f6f..7aeb38d60178 100644 --- a/src/shared/agent-session-queued-message-wire.ts +++ b/src/shared/agent-session-queued-message-wire.ts @@ -15,9 +15,10 @@ export type AgentSessionQueuedMessagePausedReason = | typeof QUEUED_MESSAGE_PAUSED_SEND_FAILED | typeof QUEUED_MESSAGE_PAUSED_KEPT -/** The whole queue is paused and sends nothing on its own: 'stopped' — the user - * interrupted ("Queue paused because you interrupted") — or 'cleared' — a /clear - * carried the cards into a fresh conversation. Resume (`agentSession.queuedMessagesResume`), +/** The queue is paused and sends nothing on its own: 'stopped' — the user + * interrupted ("Queue paused because you interrupted"); it holds every card but + * orchestration mail — or 'cleared' — a /clear carried the cards into a fresh + * conversation. Resume (`agentSession.queuedMessagesResume`), * or any turn starting, lifts it; Send-now on one card sends that card and leaves the * rest paused until its turn starts. A client treats an unknown reason as a * plain pause, so a newer host can add one. */ @@ -59,6 +60,20 @@ export type AgentSessionQueuedMessage = { returnedRejection?: UnreadAgentSessionFailureFact } +/** Whether the queue's pause holds this card: a person's Stop holds every card but another + * agent's mail, which runs when the stop lands; any other pause, an unknown one too, holds all. + * Mail is a body naming its sender (`from`), whatever its shape, as the host reads it; a client's + * own send cannot set it. */ +export function queuePauseHoldsCard( + pause: AgentSessionQueuePause | null | undefined, + card: Pick +): boolean { + if (!pause) { + return false + } + return pause.reason !== 'stopped' || card.body.from === undefined +} + /** No body: the card leaving the published list IS the outcome, so a lost * answer needs no re-ask and no text ever rides the wire back. */ export type AgentSessionQueuedMessageDeleteResult = diff --git a/src/shared/native-chat-turn-status.ts b/src/shared/native-chat-turn-status.ts index 83da06a82702..26c4e45f8494 100644 --- a/src/shared/native-chat-turn-status.ts +++ b/src/shared/native-chat-turn-status.ts @@ -10,9 +10,8 @@ export const NATIVE_CHAT_TURN_STATUS_COPY = { thinking: 'Thinking', working: 'Working…', stopping: 'Stopping…', - /** The composer's placeholder while the chat reads Stopping, where the host queues sends. */ - queueAfterStop: 'Queue a message to run after the stop', - /** The same where it does not: the host holds the send until the stop lands. */ + /** The composer's placeholder while the chat reads Stopping and a send is not queued: the host + * holds it until the stop lands. */ sendAfterStop: 'Send a message to run after the stop', workingFor: 'Working for {{value0}}', workedFor: 'Worked for {{value0}}', diff --git a/src/shared/protocol-version.ts b/src/shared/protocol-version.ts index 56fe9decc2e7..942f84f93086 100644 --- a/src/shared/protocol-version.ts +++ b/src/shared/protocol-version.ts @@ -236,7 +236,9 @@ export const AGENT_SESSION_SEND_ANSWERS_PROOF_RUNTIME_CAPABILITY = // 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 +// after it starting; while it is set every waiting card waits, but under a person's Stop another +// agent's mail, which runs as the stop lands (its body names its sender, `from`, so a client +// labels it). `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 diff --git a/tests/e2e/cross-version-wire/queued-card-sender-downgrade.unit.test.ts b/tests/e2e/cross-version-wire/queued-card-sender-downgrade.unit.test.ts new file mode 100644 index 000000000000..45dca6e1b959 --- /dev/null +++ b/tests/e2e/cross-version-wire/queued-card-sender-downgrade.unit.test.ts @@ -0,0 +1,74 @@ +import { expect, test } from 'vitest' +import type { AgentMessageSource } from '../../../src/shared/agent-session-message-source' +import type { + AgentSessionQueuedMessage, + AgentSessionQueuePause +} from '../../../src/shared/agent-session-wire' +import { projectQueuedMessageCards } from '../../../src/renderer/src/components/native-chat/structured-agent-session-queued-cards' +import { importReleaseCheckoutModule, materializeReleaseCheckout } from './release-checkout' + +// A person's Stop no longer holds another agent's mail, so a client labels a card whose body names +// its sender (`from`) as waiting rather than paused. No wire change: `from` already rides on each +// card's body. A build without this change labels every card under a pause as before, and this build +// reads a card from a host without `from` (one that also predates this rule, so holds mail) as the +// person's. The main commit this change branched from, which has the queue and `from`; move it to +// the newest release that has both and predates this change. A baseline holding this change tests +// no downgrade. +const BASELINE_REF = '0c96550ee9ab44c86b770518cc0deb56a0c130d0' +const CARDS = 'src/renderer/src/components/native-chat/structured-agent-session-queued-cards.ts' +const STOPPED: AgentSessionQueuePause = { reason: 'stopped' } +const AGENT_FROM: AgentMessageSource = { + kind: 'agent', + senders: [], + orchestration: { message: 'mail-notice', mailbox: 'run:r1', dispatchId: null, messages: [] } +} + +function card(messageId: string, position: number, from?: AgentMessageSource) { + return { + messageId, + position, + body: { + kind: 'message' as const, + role: 'user' as const, + blocks: [{ type: 'text' as const, text: messageId }], + ...(from ? { from } : {}) + }, + state: 'waiting' as const + } satisfies AgentSessionQueuedMessage +} + +type OlderProjection = ( + queuedMessages: readonly unknown[], + submissions: readonly unknown[], + session: { hasPendingPrompt: boolean; queuePaused?: boolean } +) => readonly { messageId: string; hold: string }[] + +async function olderProjection(): Promise { + const checkout = await materializeReleaseCheckout(BASELINE_REF) + const cards = await importReleaseCheckoutModule(checkout, CARDS) + const project = cards.projectQueuedMessageCards + if (typeof project !== 'function') { + throw new Error('the pinned build exports no projectQueuedMessageCards') + } + return (queuedMessages, submissions, session) => project(queuedMessages, submissions, session) +} + +test("an older client reads this host's cards, mail included, as paused under a person's Stop, as before", async () => { + const project = await olderProjection() + const held = project([card('typed', 1), card('mail', 2, AGENT_FROM)], [], { + hasPendingPrompt: false, + queuePaused: true + }) + expect(held.map((entry) => [entry.messageId, entry.hold])).toEqual([ + ['typed', 'queue-paused'], + ['mail', 'queue-paused'] + ]) +}) + +test("this client reads a card whose body names no sender as the person's", () => { + const held = projectQueuedMessageCards([card('typed', 1), card('unnamed', 2)], [], { + hasPendingPrompt: false, + queuePause: STOPPED + }) + expect(held.map((entry) => entry.hold)).toEqual(['queue-paused', 'queue-paused']) +})