From cac586dd2a2ddeefc6d35c5e9867951b094c403c Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Thu, 8 Oct 2026 20:13:27 -0700 Subject: [PATCH 1/8] Let /clear wait in line as a queued card while the agent works A /clear sent while the agent is working (a turn, an unanswered message, a pending question) was refused. Behind a host advertising agent-session.queued-clear.v1, it now waits as a /clear card like a queued /compact. When its turn comes the host runs it itself through the same-chat clear transaction: the card is spent in the clear's own write, cards sent after it run in the fresh context in order and unpaused, background tasks or a handoff keep it waiting, and any other failure returns the card with its reason so Send retries and Delete takes it back. --- .../MobileNativeChatQueuedMessages.tsx | 2 +- ...chat-merged-snapshot-parity-hooks.test.tsx | 1 + ...uctured-agent-session-host-support.test.ts | 5 + ...e-structured-agent-session-host-support.ts | 4 + ...mobile-structured-composer-command.test.ts | 24 +- ...le-structured-queued-message-cards.test.ts | 91 +++++ .../mobile-structured-queued-message-cards.ts | 35 +- ...ed-agent-session-background-tasks.test.tsx | 34 +- ...tured-agent-session-prompt-cancel.test.tsx | 1 + ...tured-agent-session-queued.test-fixture.ts | 1 + .../use-mobile-structured-agent-session.ts | 5 +- ...bile-structured-queued-message-controls.ts | 5 + ...use-mobile-structured-send-with-outcome.ts | 7 +- ...-mobile-structured-turn-indicator.test.tsx | 1 + .../journal-context-clear.ts | 21 +- .../journal-context-controller.ts | 11 +- .../journal-queued-messages.ts | 16 +- .../queued-message-return.ts | 43 +++ .../structured-agent-session-host.ts | 3 +- ...uctured-agent-session-queued-clear.test.ts | 311 ++++++++++++++++++ ...tured-agent-session-queued-compact.test.ts | 12 +- ...tructured-agent-session-queued-messages.ts | 13 +- ...ructured-agent-session-queued-mutations.ts | 39 ++- .../structured-agent-session-queued-wiring.ts | 34 +- .../structured-conversation-clear.ts | 158 +++++++++ .../structured-conversation-command.ts | 131 ++++---- .../agent-session-conversation-receipts.ts | 4 + .../agent-session-record-lease-events.test.ts | 40 +++ .../agent-session-record-lease-events.ts | 53 +++ .../runtime/agent-session-record-store.ts | 25 +- .../NativeChatQueuedMessageCard.tsx | 10 +- .../NativeChatQueuedMessageList.test.tsx | 18 + ...uctured-agent-session-queued-cards.test.ts | 88 +++++ .../structured-agent-session-queued-cards.ts | 39 ++- ...ructured-conversation-command-send.test.ts | 71 ++++ .../structured-conversation-command-send.ts | 22 +- ...tructured-agent-session-queued-messages.ts | 27 +- ...agent-session.queued-clear-gating.test.tsx | 222 +++++++++++++ .../use-structured-agent-session.ts | 9 +- src/renderer/src/i18n/locales/en.json | 1 + src/renderer/src/i18n/locales/es.json | 1 + src/renderer/src/i18n/locales/fr.json | 1 + src/renderer/src/i18n/locales/ja.json | 1 + src/renderer/src/i18n/locales/ko.json | 1 + src/renderer/src/i18n/locales/zh.json | 1 + ...tructured-agent-session-host-capability.ts | 9 + src/shared/protocol-version.ts | 6 + .../structured-agent-session-params.ts | 8 +- ...tructured-agent-session-queue-selection.ts | 37 +++ 49 files changed, 1559 insertions(+), 143 deletions(-) create mode 100644 src/main/native-chat/agent-session-journal/queued-message-return.ts create mode 100644 src/main/native-chat/agent-session-wire/structured-agent-session-queued-clear.test.ts create mode 100644 src/main/native-chat/agent-session-wire/structured-conversation-clear.ts create mode 100644 src/main/runtime/agent-session-record-lease-events.test.ts create mode 100644 src/main/runtime/agent-session-record-lease-events.ts create mode 100644 src/renderer/src/components/native-chat/use-structured-agent-session.queued-clear-gating.test.tsx create mode 100644 src/shared/structured-agent-session-queue-selection.ts diff --git a/mobile/src/session/MobileNativeChatQueuedMessages.tsx b/mobile/src/session/MobileNativeChatQueuedMessages.tsx index 3aff7ff1bd80..b1025e698c51 100644 --- a/mobile/src/session/MobileNativeChatQueuedMessages.tsx +++ b/mobile/src/session/MobileNativeChatQueuedMessages.tsx @@ -143,7 +143,7 @@ export function MobileNativeChatQueuedMessages({ ) : null} {/* A command never steers: its Send shows only while the agent is idle. */} - {card.waitsForAgent ? null : ( + {card.waitsForAgent || card.runsOnItsOwn ? null : ( { questionAnswers: false, queuedMessages: false, queuedCommands: false, + queuedClear: false, quietRepeatedStop: false, statusFeed: false } @@ -32,6 +34,9 @@ describe('structuredAgentSessionHostSupport', () => { expect( structuredAgentSessionHostSupport([AGENT_SESSION_QUEUED_COMMANDS_RUNTIME_CAPABILITY]) ).toEqual({ ...none, queuedCommands: true }) + expect( + structuredAgentSessionHostSupport([AGENT_SESSION_QUEUED_CLEAR_RUNTIME_CAPABILITY]) + ).toEqual({ ...none, queuedClear: true }) expect( structuredAgentSessionHostSupport([AGENT_SESSION_REPEATED_STOP_RUNTIME_CAPABILITY]) ).toEqual({ ...none, quietRepeatedStop: true }) diff --git a/mobile/src/session/mobile-structured-agent-session-host-support.ts b/mobile/src/session/mobile-structured-agent-session-host-support.ts index 84c329e63738..f3500c6c4b7c 100644 --- a/mobile/src/session/mobile-structured-agent-session-host-support.ts +++ b/mobile/src/session/mobile-structured-agent-session-host-support.ts @@ -1,6 +1,7 @@ import { AGENT_SESSION_PROMPT_CANCEL_RUNTIME_CAPABILITY, AGENT_SESSION_QUESTION_ANSWERS_RUNTIME_CAPABILITY, + AGENT_SESSION_QUEUED_CLEAR_RUNTIME_CAPABILITY, AGENT_SESSION_QUEUED_COMMANDS_RUNTIME_CAPABILITY, AGENT_SESSION_QUEUED_MESSAGES_RUNTIME_CAPABILITY, AGENT_SESSION_STATUS_FEED_RUNTIME_CAPABILITY, @@ -15,6 +16,8 @@ export type StructuredAgentSessionHostSupport = { queuedMessages: boolean /** A /compact sent while the agent works waits as a card; an older host refuses it. */ queuedCommands: boolean + /** The same for a /clear, which the host runs itself when its card's turn comes. */ + queuedClear: boolean /** The host publishes every session's status on one stream. A host from before phones could * read it refuses the call, which reads the same as its absence. */ statusFeed: boolean @@ -30,6 +33,7 @@ export function structuredAgentSessionHostSupport( questionAnswers: capabilities.includes(AGENT_SESSION_QUESTION_ANSWERS_RUNTIME_CAPABILITY), queuedMessages: capabilities.includes(AGENT_SESSION_QUEUED_MESSAGES_RUNTIME_CAPABILITY), queuedCommands: capabilities.includes(AGENT_SESSION_QUEUED_COMMANDS_RUNTIME_CAPABILITY), + queuedClear: capabilities.includes(AGENT_SESSION_QUEUED_CLEAR_RUNTIME_CAPABILITY), statusFeed: capabilities.includes(AGENT_SESSION_STATUS_FEED_RUNTIME_CAPABILITY), quietRepeatedStop: capabilities.includes(AGENT_SESSION_REPEATED_STOP_RUNTIME_CAPABILITY) } diff --git a/mobile/src/session/mobile-structured-composer-command.test.ts b/mobile/src/session/mobile-structured-composer-command.test.ts index ec36a48dfe2c..b18d02d770a4 100644 --- a/mobile/src/session/mobile-structured-composer-command.test.ts +++ b/mobile/src/session/mobile-structured-composer-command.test.ts @@ -139,10 +139,32 @@ describe('mobile structured conversation commands', () => { const fields = requestFields(sendRequest.mock.calls[0]) expect(fields).toMatchObject({ command: 'compact', delivery: 'queue-if-active' }) expect(input.onError).not.toHaveBeenCalled() - // A /clear never waits: the busy check still answers it. + // A host that holds only /compact: the busy check still answers a /clear. expect(await dispatchMobileStructuredCommand({ ...input, text: '/clear' })).toBe('rejected') expect(sendRequest).toHaveBeenCalledOnce() }) + it('a /clear the host holds in line asks to wait the same way, and its card is the answer', async () => { + const { input, sendRequest } = setup() + sendRequest.mockResolvedValueOnce({ + ok: true, + result: { + ok: true, + value: { + command: 'clear', + state: 'completed', + queued: { messageId: 'op', position: 1, state: 'waiting' } + } + } + }) + input.busy = () => 'working' + input.waitsInLine = () => true + expect(await dispatchMobileStructuredCommand({ ...input, text: '/clear' })).toBe('accepted') + expect(requestFields(sendRequest.mock.calls[0])).toMatchObject({ + command: 'clear', + delivery: 'queue-if-active' + }) + expect(input.onError).not.toHaveBeenCalled() + }) it('a /clear while the agent works says so in plain words', async () => { const { input, sendRequest } = setup() input.busy = () => 'working' 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 ce3eda3c57b2..b94083b78156 100644 --- a/mobile/src/session/mobile-structured-queued-message-cards.test.ts +++ b/mobile/src/session/mobile-structured-queued-message-cards.test.ts @@ -102,6 +102,97 @@ describe('mobileQueuedMessageCards', () => { ).toMatchObject({ command: true, waitsForAgent: true }) }) + it('a /clear next in line that nothing holds offers no Send: the queue is about to run it', () => { + const clear = (overrides: Partial = {}) => + draft({ + messageId: 'c', + body: { + kind: 'message', + role: 'user', + blocks: [{ type: 'text', text: '/clear' }], + command: { name: 'clear' } + }, + ...overrides + }) + const idle = { pendingPrompt: false } + expect(mobileQueuedMessageCards([clear()], [], idle)[0]).toMatchObject({ runsOnItsOwn: true }) + // Anything the person must act on keeps its Send, as does a /clear behind another card. + for (const [cards, facts] of [ + [[clear()], { ...idle, queuePaused: true }], + [[clear()], { pendingPrompt: true }], + [[clear({ paused: true, pausedReason: QUEUED_MESSAGE_PAUSED_SEND_FAILED })], idle], + [[draft({ messageId: 'a' }), clear()], idle] + ] as const) { + expect(mobileQueuedMessageCards(cards, [], facts).some((card) => card.runsOnItsOwn)).toBe( + false + ) + } + }) + + it('a /clear next in line that only background tasks hold says so; nothing else does', () => { + const clear = draft({ + messageId: 'c', + body: { + kind: 'message', + role: 'user', + blocks: [{ type: 'text', text: '/clear' }], + command: { name: 'clear' } + } + }) + const tasks = { pendingPrompt: false, backgroundTasksRunning: true } + const [first, second] = mobileQueuedMessageCards( + [clear, draft({ messageId: 'm', position: 2 })], + [], + tasks + ) + expect(first).toMatchObject({ + caption: 'Waiting for background tasks to finish', + runsOnItsOwn: true + }) + expect(second?.caption).toBeNull() + // Behind a turn or an answer, that is what it waits on; behind another card, its turn. + expect( + mobileQueuedMessageCards([clear], [], { ...tasks, agentWorking: true })[0]?.caption + ).toBeNull() + expect( + mobileQueuedMessageCards([clear], [], { ...tasks, pendingPrompt: true })[0]?.caption + ).toBe('Waiting for your answer') + expect( + mobileQueuedMessageCards([draft({ messageId: 'm' }), { ...clear, position: 2 }], [], tasks)[1] + ?.caption + ).toBeNull() + }) + + it('skips a failed card ahead of clear, respecting returned and queue-pause barriers', () => { + const clear = draft({ + messageId: 'clear', + position: 2, + body: { + kind: 'message', + role: 'user', + blocks: [{ type: 'text', text: '/clear' }], + command: { name: 'clear' } + } + }) + const held = draft({ + messageId: 'held', + paused: true, + pausedReason: QUEUED_MESSAGE_PAUSED_SEND_FAILED + }) + expect( + mobileQueuedMessageCards([held, clear], [], { pendingPrompt: false })[1]?.runsOnItsOwn + ).toBe(true) + expect( + mobileQueuedMessageCards([{ ...held, state: 'returned' }, clear], [], { + pendingPrompt: false + })[1]?.runsOnItsOwn + ).toBeUndefined() + expect( + mobileQueuedMessageCards([held, clear], [], { pendingPrompt: false, queuePaused: true })[1] + ?.runsOnItsOwn + ).toBeUndefined() + }) + it("a send-failed command card's caption names Send only when Send is there", () => { const failed = draft({ messageId: 'c', diff --git a/mobile/src/session/mobile-structured-queued-message-cards.ts b/mobile/src/session/mobile-structured-queued-message-cards.ts index 42883b4ed4df..3e1450a7158f 100644 --- a/mobile/src/session/mobile-structured-queued-message-cards.ts +++ b/mobile/src/session/mobile-structured-queued-message-cards.ts @@ -2,6 +2,10 @@ // The wire carries no hold copy on purpose: the caption is derived here from the // draft's own state plus the live facts the client already holds. +import { + nextActionableQueuedMessage, + queuedClearWaitsOnBackgroundTasks +} from '../../../src/shared/structured-agent-session-queue-selection' import { readWholeAgentSessionFailureFact, type AgentSessionFailureFact @@ -38,6 +42,9 @@ export type MobileQueuedMessageCard = { command?: true /** A command card while the agent works: it offers no send until the agent is idle. */ waitsForAgent?: true + /** A /clear next in line that nothing but background tasks holds: the queue runs it without a + * press, so it offers no Send. */ + runsOnItsOwn?: true } function queuedMessageBodyText(body: AgentSessionQueuedMessage['body']): string { @@ -115,6 +122,8 @@ export function mobileQueuedMessageCards( pendingPrompt: boolean queuePaused?: boolean agentWorking?: boolean + /** Background tasks run: a /clear next in line waits them out on the host. */ + backgroundTasksRunning?: boolean agentName?: string statedFailures?: readonly AgentSessionFailureFact[] } @@ -129,12 +138,17 @@ export function mobileQueuedMessageCards( : [] ) ) + const shown = queuedMessages.filter( + (draft) => draft.state === 'returned' || !handedOff.has(draft.messageId) + ) + const next = nextActionableQueuedMessage( + shown, + (draft) => draft.paused === true, + () => facts.queuePaused === true + ) let behindReturned = false const cards: MobileQueuedMessageCard[] = [] - for (const draft of queuedMessages) { - if (draft.state !== 'returned' && handedOff.has(draft.messageId)) { - continue - } + for (const draft of shown) { const paused = draft.paused === true const caption = draft.state === 'returned' @@ -151,7 +165,15 @@ export function mobileQueuedMessageCards( null : facts.pendingPrompt ? 'Waiting for your answer' - : null + : queuedClearWaitsOnBackgroundTasks(draft, next, facts) + ? 'Waiting for background tasks to finish' + : null + const runsOnItsOwn = + draft === next && + draft.body.command?.name === 'clear' && + !behindReturned && + facts.queuePaused !== true && + !facts.pendingPrompt cards.push({ messageId: draft.messageId, text: queuedMessageBodyText(draft.body), @@ -165,7 +187,8 @@ export function mobileQueuedMessageCards( ...(draft.body.command !== undefined ? { command: true as const, - ...(facts.agentWorking ? { waitsForAgent: true as const } : {}) + ...(facts.agentWorking ? { waitsForAgent: true as const } : {}), + ...(runsOnItsOwn ? { runsOnItsOwn: true as const } : {}) } : {}) }) diff --git a/mobile/src/session/use-mobile-structured-agent-session-background-tasks.test.tsx b/mobile/src/session/use-mobile-structured-agent-session-background-tasks.test.tsx index aef38e82e6e9..36b11b416fc2 100644 --- a/mobile/src/session/use-mobile-structured-agent-session-background-tasks.test.tsx +++ b/mobile/src/session/use-mobile-structured-agent-session-background-tasks.test.tsx @@ -6,7 +6,10 @@ import { createElement } from 'react' import { act, create, type ReactTestRenderer } from 'react-test-renderer' import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import type { AgentChildWorkView } from '../../../src/shared/agent-status-child-work-view' -import type { AgentSessionSubscribeEvent } from '../../../src/shared/agent-session-wire' +import type { + AgentSessionQueuedMessage, + AgentSessionSubscribeEvent +} from '../../../src/shared/agent-session-wire' import type { RpcClient } from '../transport/rpc-client' import { backgroundTaskRowMeta, @@ -41,8 +44,11 @@ function childView(id: string): AgentChildWorkView { } } -function withChildren(hostNow?: number): AgentSessionSubscribeEvent { - const snapshot = snapshotEvent() +function withChildren( + hostNow?: number, + queuedMessages?: AgentSessionQueuedMessage[] +): AgentSessionSubscribeEvent { + const snapshot = snapshotEvent(queuedMessages ? { queuedMessages } : undefined) return snapshot.type === 'snapshot' ? { ...snapshot, @@ -169,6 +175,28 @@ describe('mobile structured session background tasks', () => { clock.mockRestore() }) + it('a /clear next in line once the turn ended says it waits on the tasks, with no Send', async () => { + const card = (messageId: string, text: string, position: number) => ({ + messageId, + position, + state: 'waiting' as const, + body: { + kind: 'message' as const, + role: 'user' as const, + blocks: [{ type: 'text' as const, text }], + ...(text === '/clear' ? { command: { name: 'clear' as const } } : {}) + } + }) + await mount(withChildren(undefined, [card('clear-1', '/clear', 1), card('after', 'After', 2)])) + expect(hook?.isWorking).toBe(false) + expect(hook?.queued.cards[0]).toMatchObject({ + messageId: 'clear-1', + caption: 'Waiting for background tasks to finish', + runsOnItsOwn: true + }) + expect(hook?.queued.cards[1]?.caption).toBeNull() + }) + it("stops one child through the host's background-task cancel", async () => { await mount(withChildren()) await act(async () => { diff --git a/mobile/src/session/use-mobile-structured-agent-session-prompt-cancel.test.tsx b/mobile/src/session/use-mobile-structured-agent-session-prompt-cancel.test.tsx index 7a14895fd66e..e61f6381abce 100644 --- a/mobile/src/session/use-mobile-structured-agent-session-prompt-cancel.test.tsx +++ b/mobile/src/session/use-mobile-structured-agent-session-prompt-cancel.test.tsx @@ -123,6 +123,7 @@ function Harness({ questionAnswers: questionAnswersSupported, queuedMessages: false, queuedCommands: false, + queuedClear: false, statusFeed: false, quietRepeatedStop: false }, diff --git a/mobile/src/session/use-mobile-structured-agent-session-queued.test-fixture.ts b/mobile/src/session/use-mobile-structured-agent-session-queued.test-fixture.ts index 7a0c57b9b352..b5980b61d300 100644 --- a/mobile/src/session/use-mobile-structured-agent-session-queued.test-fixture.ts +++ b/mobile/src/session/use-mobile-structured-agent-session-queued.test-fixture.ts @@ -17,6 +17,7 @@ export const CAPABLE: StructuredAgentSessionHostSupport = { questionAnswers: false, queuedMessages: true, queuedCommands: false, + queuedClear: false, statusFeed: false, quietRepeatedStop: false } diff --git a/mobile/src/session/use-mobile-structured-agent-session.ts b/mobile/src/session/use-mobile-structured-agent-session.ts index fb3c69c51aa6..b390f7e9b8b3 100644 --- a/mobile/src/session/use-mobile-structured-agent-session.ts +++ b/mobile/src/session/use-mobile-structured-agent-session.ts @@ -71,8 +71,9 @@ export function useMobileStructuredAgentSession(args: { } = args // Only a host that queues sends gets the delivery field; any host's published cards show. const queueCapable = hostSupport?.queuedMessages === true - // A /compact waits in line only where its card renders. + // A command waits in line only where its card renders. const commandsWait = queueCapable && hostSupport?.queuedCommands === true + const clearWaits = queueCapable && hostSupport?.queuedClear === true const promptCancelSupported = hostSupport?.promptCancel ?? null const hostAnswersRepeatedStops = hostSupport?.quietRepeatedStop ?? null const sessionKey = encodeNativeChatTranscriptIdentity([sourceIdentity, agent, sessionId]) @@ -119,6 +120,7 @@ export function useMobileStructuredAgentSession(args: { enabled, queueCapable, commandsWait, + clearWaits, stateRef, commandPending: commandPendingRef, controller: sendController, @@ -221,6 +223,7 @@ export function useMobileStructuredAgentSession(args: { submissions: state.submissions, pendingPrompt: approvalPrompt !== null || questionPrompt !== null, agentWorking: isStructuredAgentSessionMainAgentWorking(turnId, state.submissions, state.fence), + backgroundTasksRunning: backgroundTasks.view.isMonitoring, mutate, appendComposerText, onSendError, 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 1a73756841f1..d7cbcd6c2cad 100644 --- a/mobile/src/session/use-mobile-structured-queued-message-controls.ts +++ b/mobile/src/session/use-mobile-structured-queued-message-controls.ts @@ -58,6 +58,8 @@ export function useMobileStructuredQueuedMessageControls(args: { pendingPrompt: boolean /** The chat shows the agent working: a command card offers no send then. */ agentWorking?: boolean + /** Background tasks run: a /clear next in line says it waits them out. */ + backgroundTasksRunning?: boolean mutate: MobileStructuredAgentMutate /** The active pane's live composer, Edit's copy target; absent = Edit refuses. False when * nothing was copied. */ @@ -78,11 +80,13 @@ export function useMobileStructuredQueuedMessageControls(args: { submissions } = args const agentWorking = args.agentWorking === true + const backgroundTasksRunning = args.backgroundTasksRunning === true const cards = useMemo( () => mobileQueuedMessageCards(queuedMessages, submissions, { pendingPrompt, agentWorking, + backgroundTasksRunning, agentName: args.agentName, statedFailures: queuedMessages?.some((draft) => draft.state === 'returned') ? agentSessionVisibleFailureFacts(args.journalItems ?? []) @@ -91,6 +95,7 @@ export function useMobileStructuredQueuedMessageControls(args: { }), [ agentWorking, + backgroundTasksRunning, args.agentName, args.journalItems, pendingPrompt, diff --git a/mobile/src/session/use-mobile-structured-send-with-outcome.ts b/mobile/src/session/use-mobile-structured-send-with-outcome.ts index c6c8e9604e07..c3072c305344 100644 --- a/mobile/src/session/use-mobile-structured-send-with-outcome.ts +++ b/mobile/src/session/use-mobile-structured-send-with-outcome.ts @@ -44,6 +44,8 @@ export function useMobileStructuredSendWithOutcome(args: { queueCapable: boolean /** The host holds a /compact sent while the agent works as a card. */ commandsWait: boolean + /** The same for a /clear. */ + clearWaits: boolean stateRef: { readonly current: StructuredAgentSessionState } commandPending: { current: boolean } controller: Pick< @@ -62,6 +64,7 @@ export function useMobileStructuredSendWithOutcome(args: { client, commandPending, commandsWait, + clearWaits, controller, enabled, onSendError, @@ -112,8 +115,7 @@ export function useMobileStructuredSendWithOutcome(args: { : null, // A card waiting on a prompt nothing here can answer would hold it forever. waitsInLine: (command) => - command === 'compact' && - commandsWait && + (command === 'clear' ? clearWaits : commandsWait) && !pendingPromptsAllUnanswerableHere(stateRef.current.items), onError: onSendError, timeoutMs @@ -143,6 +145,7 @@ export function useMobileStructuredSendWithOutcome(args: { client, commandPending, commandsWait, + clearWaits, controller, enabled, onSendError, diff --git a/mobile/src/session/use-mobile-structured-turn-indicator.test.tsx b/mobile/src/session/use-mobile-structured-turn-indicator.test.tsx index 79f761f5511e..604dbfb0ff22 100644 --- a/mobile/src/session/use-mobile-structured-turn-indicator.test.tsx +++ b/mobile/src/session/use-mobile-structured-turn-indicator.test.tsx @@ -266,6 +266,7 @@ describe("useMobileStructuredAgentSession and the host's Stopping", () => { questionAnswers: false, queuedMessages: false, queuedCommands: false, + queuedClear: false, quietRepeatedStop: false, statusFeed }, diff --git a/src/main/native-chat/agent-session-journal/journal-context-clear.ts b/src/main/native-chat/agent-session-journal/journal-context-clear.ts index 8630fe4d704f..ba090054c1f6 100644 --- a/src/main/native-chat/agent-session-journal/journal-context-clear.ts +++ b/src/main/native-chat/agent-session-journal/journal-context-clear.ts @@ -5,6 +5,7 @@ import type { JournalOperationReceipt, JournalRowWriter } from './journal-row-wr import type { JournalQueuedMessages } from './journal-queued-messages' import type { JournalTombstoneRow } from './journal-row-schema' import { buildJournalQueueClearRow } from './journal-stop-and-resume-rows' +import { QueuedMessageNotConsumableError } from './queued-message-consume-error' export function appendJournalContextClear(input: { state: () => JournalReducerState @@ -13,18 +14,28 @@ export function appendJournalContextClear(input: { boundary: AgentSessionProviderContextBoundary receipt: JournalOperationReceipt settledByOp: string + /** A /clear card the queue runs: spent here, and only the cards queued before it were written + * for the context it discards. Those behind it were sent for the fresh one and stay as they are. */ + ranFrom?: string }) { - const { boundary, state } = input + const { boundary, state, ranFrom } = input let commandIds: string[] = [] return input.writer .enqueueRows( () => { - const cards = input.cards.list() + const all = input.cards.list() + const at = + ranFrom === undefined ? all.length : all.findIndex((c) => c.messageId === ranFrom) + if (at < 0) { + throw new QueuedMessageNotConsumableError(ranFrom ?? '', 'waiting') + } + const cards = all.slice(0, at) commandIds = cards .filter( (card) => card.body.command && (card.state === 'waiting' || card.state === 'returned') ) .map((card) => card.messageId) + .concat(ranFrom === undefined ? [] : [ranFrom]) const messageIds = cards .filter((card) => card.state === 'waiting' && !card.body.command) .map((card) => card.messageId) @@ -52,10 +63,14 @@ export function appendJournalContextClear(input: { }, { write: (db) => { - input.cards.withdrawInTransaction(db, { + const withdrawn = input.cards.withdrawInTransaction(db, { messageIds: commandIds, settledByOp: input.settledByOp }) + // The card and its clear commit together or not at all, so it can never run twice. + if (ranFrom !== undefined && !withdrawn.includes(ranFrom)) { + throw new QueuedMessageNotConsumableError(ranFrom, 'waiting') + } input.receipt.write(db) }, committed: input.receipt.committed diff --git a/src/main/native-chat/agent-session-journal/journal-context-controller.ts b/src/main/native-chat/agent-session-journal/journal-context-controller.ts index 7403760e8e91..d37ab0e7530d 100644 --- a/src/main/native-chat/agent-session-journal/journal-context-controller.ts +++ b/src/main/native-chat/agent-session-journal/journal-context-controller.ts @@ -26,9 +26,16 @@ export class JournalContextController { clear( boundary: AgentSessionProviderContextBoundary, receipt: JournalOperationReceipt, - settledByOp: string + settledByOp: string, + ranFrom?: string ) { - return appendJournalContextClear({ ...this.deps, boundary, receipt, settledByOp }) + return appendJournalContextClear({ + ...this.deps, + boundary, + receipt, + settledByOp, + ...(ranFrom === undefined ? {} : { ranFrom }) + }) } rewind( diff --git a/src/main/native-chat/agent-session-journal/journal-queued-messages.ts b/src/main/native-chat/agent-session-journal/journal-queued-messages.ts index 3337727689cd..9a1658064150 100644 --- a/src/main/native-chat/agent-session-journal/journal-queued-messages.ts +++ b/src/main/native-chat/agent-session-journal/journal-queued-messages.ts @@ -243,20 +243,20 @@ export class JournalQueuedMessages { withdrawInTransaction( db: Database.Database, input: { messageIds: readonly string[]; settledByOp: string } - ): void { + ): string[] { if (this.deps.database().db !== db) { throw new AgentSessionJournalError('journal_closed', 'withdraw crossed database handles') } - this.changeRevision += withdrawQueuedMessages(db, { - ...input, - sessionId: this.deps.sessionId, - now: this.deps.now() - }).length + const { sessionId, now } = this.deps + const withdrawn = withdrawQueuedMessages(db, { ...input, sessionId, now: now() }) + this.changeRevision += withdrawn.length + return withdrawn.map((row) => row.messageId) } /** One standalone draft-table transaction on the journal's queue; one that - * changed rows bumps the revision and notifies after COMMIT, `adopted` first. */ - private transact( + * changed rows bumps the revision and notifies after COMMIT, `adopted` first. Open to the + * draft-table writers outside this file (`queued-message-return.ts`), which take no other path. */ + transact( run: (db: Database.Database) => JournalWriteResult, changed: (result: T) => boolean, adopted?: () => void diff --git a/src/main/native-chat/agent-session-journal/queued-message-return.ts b/src/main/native-chat/agent-session-journal/queued-message-return.ts new file mode 100644 index 000000000000..fc2d30beb3c2 --- /dev/null +++ b/src/main/native-chat/agent-session-journal/queued-message-return.ts @@ -0,0 +1,43 @@ +// A card the host runs itself (a queued /clear) that could not run: returned with why, never handed +// off. Like a refused hand-off's card it blocks the cards behind it until its own Send or Delete. + +import type Database from '../../sqlite/sync-database' +import type { UnreadAgentSessionFailureFact } from '../../../shared/agent-session-failure' +import type { JournalQueuedMessages } from './journal-queued-messages' + +/** waiting → returned with no hand-off; a returned card asked again keeps its place with the newer + * reason. False when the card is no longer unsettled. */ +export function returnUnsentQueuedCard( + queued: Pick, + input: { + messageId: string + reason: string | null + rejection: UnreadAgentSessionFailureFact + now: number + } +): Promise { + return queued.transact( + (db) => returnUnsentQueuedMessage(db, { ...input, sessionId: queued.sessionId }), + (changed) => changed + ) +} + +function returnUnsentQueuedMessage( + db: Database.Database, + input: { + sessionId: string + messageId: string + reason: string | null + rejection: UnreadAgentSessionFailureFact + now: number + } +): boolean { + const changed = db + .prepare( + `UPDATE queued_messages + SET state = 'returned', hold_reason = NULL, returned_reason = ?, returned_rejection = ?, settled_at = ? + WHERE session_id = ? AND message_id = ? AND state IN ('waiting', 'returned')` + ) + .run(input.reason, JSON.stringify(input.rejection), input.now, input.sessionId, input.messageId) + return Number(changed.changes ?? 0) > 0 +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts index 251ccc88e32b..634d9e5d7cf3 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 @@ -78,7 +78,7 @@ export class StructuredAgentSessionHost { () => this.deps, (sessionId) => this.queued.onJournalActivity(sessionId), (sessionId) => this.restartResume.onAgentStarted(sessionId), - (sessionId) => this.backgroundTasks.publish(sessionId), + (sessionId) => this.queued.onChildWorkChanged(sessionId, this.backgroundTasks), (sessionId) => this.backgroundTasks.read(sessionId) ) private readonly subscribers = this.clientDelivery.subscribers @@ -97,6 +97,7 @@ export class StructuredAgentSessionHost { // Every collaborator reads this copy, so a logger that throws cannot fail what it reports. this.deps = deps = sessionLogger.withNeverThrowingLogger(deps) this.clientDelivery.watchAtRestCommands(deps.adapter) + this.queued.wakeOnHandoffEnded(deps.store) this.backgroundTasks = new StructuredAgentSessionBackgroundTaskChannel( deps, this.sessions, diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-clear.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-clear.test.ts new file mode 100644 index 000000000000..37a72c606199 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-clear.test.ts @@ -0,0 +1,311 @@ +// A /clear asked to wait (`delivery`) while the agent works: held as a card like a queued send, +// answered at once, and run by the host itself when its turn comes, in the same conversation. The +// cards sent after it run in the fresh context, in order. Without the opt-in, today's refusal. + +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import type { AgentChildWorkView } from '../../../shared/agent-status-child-work-view' +import { agentSessionOperationKey } from '../../../shared/agent-session-operation-ledger' +import { + createQueuedMessageTestRig, + eventually, + QUEUED_RIG_CALLER as CALLER, + type QueuedMessageTestRig +} from './structured-agent-session-queued-message-rig.test-fixture' +import { + HOST_TEST_SESSION as SESSION, + hostTestOperationId +} from './structured-agent-session-host-test-data' +import { QUEUED_CLEAR_CALLER_KEY } from './structured-conversation-clear' + +let rig: QueuedMessageTestRig + +beforeEach(async () => { + rig = await createQueuedMessageTestRig() +}) + +afterEach(() => rig.dispose()) + +function clear(delivery?: 'queue-if-active', clientOperationId = hostTestOperationId()) { + const fields = { command: 'clear' as const, ...(delivery ? { delivery } : {}) } + return rig.host.conversationCommand(CALLER, { + envelope: rig.envelope(fields, 'agentSession.conversationCommand', clientOperationId), + ...fields, + userSend: true + }) +} + +async function queuedClear(): Promise { + const id = hostTestOperationId() + expect(await clear('queue-if-active', id)).toMatchObject({ + ok: true, + value: { command: 'clear', state: 'completed', queued: { messageId: id, state: 'waiting' } } + }) + return id +} + +async function queuedSend(text: string): Promise { + const sent = await rig.send(text, 'queue-if-active').result + if (!sent.ok || !('queued' in sent.value)) { + throw new Error('expected a queued receipt') + } + return sent.value.queued.messageId +} + +function journal() { + return rig.host.collaboratorsForTests().sessions.get(SESSION)!.journal +} + +async function dividers(): Promise { + return (await rig.host.journalSnapshot(SESSION)).items.filter( + (item) => item.body.kind === 'status' && item.body.presentation === 'context-cleared' + ).length +} + +const BACKGROUND_TASK: AgentChildWorkView = { + id: 'child-dev', + providerId: 'task-dev', + kind: 'command', + description: 'npm run dev', + state: 'working', + membership: 'live', + firstObservedAt: 1, + observedAt: 1, + stoppable: true, + invocation: { invocationId: 'spawn-dev', generation: 1 } +} + +const settleMs = () => new Promise((resolve) => setTimeout(resolve, 150)) + +describe('a /clear that waits in line', () => { + it('behind an unanswered message: a card at once, nothing stopped or cleared yet', async () => { + await rig.workingSend() + const clearId = await queuedClear() + expect(await rig.drafts()).toEqual([{ messageId: clearId, state: 'waiting' }]) + expect(journal().queuedMessages.get(clearId)?.body).toMatchObject({ + blocks: [{ type: 'text', text: '/clear' }], + command: { name: 'clear' } + }) + await settleMs() + expect(rig.closeSession).not.toHaveBeenCalled() + expect(rig.store.getRecord(SESSION)?.providerContextBoundary).toBeUndefined() + expect(await dividers()).toBe(0) + }) + + it('runs in the same chat once the turn ends: one divider, spent in that write, never sent', async () => { + const working = await rig.workingSend() + const clearId = await queuedClear() + await rig.settleAccepted(working, 'a') + await eventually(async () => expect(await dividers()).toBe(1)) + expect(rig.closeSession).toHaveBeenCalled() + expect(rig.store.getRecord(SESSION)).toMatchObject({ + providerHandleChain: [], + conversationCommand: { + command: 'clear', + operationId: clearId, + callerKey: QUEUED_CLEAR_CALLER_KEY, + phase: 'committed' + } + }) + expect(journal().queuedMessages.get(clearId)).toMatchObject({ + state: 'withdrawn', + settledByOp: agentSessionOperationKey(QUEUED_CLEAR_CALLER_KEY, clearId) + }) + expect(await rig.drafts()).toEqual([]) + expect(await rig.handoff(clearId)).toBeUndefined() + expect(rig.dispatch).toHaveBeenCalledOnce() + }) + + it('cards sent after it run in the fresh context, in order, unpaused', async () => { + const working = await rig.workingSend() + const clearId = await queuedClear() + const first = await queuedSend('first after the clear') + const second = await queuedSend('second after the clear') + await rig.settleAccepted(working, 'a') + await eventually(async () => expect(await rig.handoff(first)).toBeDefined()) + expect(await dividers()).toBe(1) + expect(await rig.queuePause()).toBeNull() + expect(journal().queuedMessages.pauses()).toEqual([]) + // The first is the agent's work now, so the second waits for it. + await settleMs() + expect(await rig.handoff(second)).toBeUndefined() + expect(await rig.drafts()).toEqual([{ messageId: second, state: 'waiting' }]) + const sent = (await rig.host.journalSnapshot(SESSION)).submissions + expect(sent.flatMap((entry) => (entry.queuedMessageId ? [entry.queuedMessageId] : []))).toEqual( + [first] + ) + expect(sent.some((entry) => entry.queuedMessageId === clearId)).toBe(false) + }) + + it('a resent id answers from its card, and after it ran as withdrawn; one clear', async () => { + const working = await rig.workingSend() + const clearId = await queuedClear() + expect(await clear('queue-if-active', clearId)).toMatchObject({ + ok: true, + value: { queued: { messageId: clearId, state: 'waiting' } } + }) + await rig.settleAccepted(working, 'a') + await eventually(async () => expect(await dividers()).toBe(1)) + expect(await clear('queue-if-active', clearId)).toMatchObject({ + ok: true, + value: { command: 'clear', queued: { messageId: clearId, state: 'withdrawn' } } + }) + await settleMs() + expect(await dividers()).toBe(1) + }) + + it('a restart after it ran never runs it again', async () => { + const working = await rig.workingSend() + await queuedClear() + const after = await queuedSend('after the clear') + await rig.settleAccepted(working, 'a') + await eventually(async () => expect(await rig.handoff(after)).toBeDefined()) + await rig.quitRestartHostProcess() + await rig.host.journalSnapshot(SESSION) + await settleMs() + expect(await dividers()).toBe(1) + }) + + it('never steers: its Send mid-turn is refused and it keeps waiting', async () => { + const working = await rig.workingSend() + const clearId = await queuedClear() + expect(await rig.sendNow(clearId)).toMatchObject({ + ok: false, + refusal: { message: "A command can't be sent while the agent is working." } + }) + expect(await dividers()).toBe(0) + await rig.settleAccepted(working, 'a') + await eventually(async () => expect(await dividers()).toBe(1)) + }) + + it('Delete takes it back: it never runs, and the cards behind it send as usual', async () => { + const working = await rig.workingSend() + const clearId = await queuedClear() + const after = await queuedSend('after') + expect(await rig.deleteQueued(clearId)).toMatchObject({ ok: true, value: { deleted: true } }) + await rig.settleAccepted(working, 'a') + await eventually(async () => expect(await rig.handoff(after)).toBeDefined()) + expect(await dividers()).toBe(0) + }) + + it('at rest, clears at once as before', async () => { + const answer = await clear('queue-if-active') + expect(answer).toMatchObject({ ok: true, value: { command: 'clear', state: 'completed' } }) + expect(answer.ok && answer.value.queued).toBeFalsy() + expect(await dividers()).toBe(1) + expect(await rig.drafts()).toEqual([]) + }) + + it('without the opt-in (an older client), is refused while a message is unanswered, as today', async () => { + await rig.workingSend() + expect(await clear()).toMatchObject({ + ok: false, + refusal: { details: { reason: 'messagesUnsettled' } } + }) + expect(await rig.drafts()).toEqual([]) + }) +}) + +describe('a /clear card the queue cannot run yet', () => { + it('waits out background tasks, then runs when they end; the cards behind it wait too', async () => { + let tasks: AgentChildWorkView[] = [] + Object.assign(rig.host.deps, { + statusSink: { publish: () => {}, forget: () => {}, readChildWork: () => tasks } + }) + const working = await rig.workingSend() + const clearId = await queuedClear() + const after = await queuedSend('after the clear') + tasks = [BACKGROUND_TASK] + await rig.settleAccepted(working, 'a') + await settleMs() + expect(await rig.drafts()).toEqual([ + { messageId: clearId, state: 'waiting' }, + { messageId: after, state: 'waiting' } + ]) + expect(await dividers()).toBe(0) + // Its Send says why it waits, and changes nothing on the card. + expect(await rig.sendNow(clearId)).toMatchObject({ + ok: false, + refusal: { details: { reason: 'backgroundTasksRunning' } } + }) + tasks = [{ ...BACKGROUND_TASK, state: 'done', membership: 'settled', outcome: 'succeeded' }] + rig.host.publishChildWorkEvidence(SESSION, []) + await eventually(async () => expect(await rig.handoff(after)).toBeDefined()) + expect(await dividers()).toBe(1) + }) + + it('a failed commit returns it with why once; nothing behind it runs; Send retries', async () => { + const working = await rig.workingSend() + const clearId = await queuedClear() + const after = await queuedSend('after the clear') + vi.spyOn(console, 'warn').mockImplementation(() => undefined) + vi.spyOn(rig.store.conversationReceipts, 'queuedClear').mockReturnValueOnce({ + write: () => { + throw new Error('disk full') + }, + committed: () => {} + }) + await rig.settleAccepted(working, 'a') + await eventually(async () => + expect(await rig.drafts()).toEqual([ + { messageId: clearId, state: 'returned' }, + { messageId: after, state: 'waiting' } + ]) + ) + // Rolled back whole: no divider, no fresh context, and the card was not spent. + expect(await dividers()).toBe(0) + expect(rig.store.getRecord(SESSION)?.providerContextBoundary).toBeUndefined() + expect(journal().queuedMessages.get(clearId)).toMatchObject({ + state: 'returned', + returnedRejection: { kind: 'commandRefused' } + }) + await settleMs() + expect(await rig.handoff(after)).toBeUndefined() + + expect(await rig.sendNow(clearId)).toMatchObject({ + ok: true, + value: { queued: { messageId: clearId, state: 'withdrawn' } } + }) + expect(await dividers()).toBe(1) + await eventually(async () => expect(await rig.handoff(after)).toBeDefined()) + }) + + it('a returned /clear deleted: the cards behind it run in the uncleared chat', async () => { + const working = await rig.workingSend() + const clearId = await queuedClear() + const after = await queuedSend('after the clear') + vi.spyOn(console, 'warn').mockImplementation(() => undefined) + vi.spyOn(rig.store.conversationReceipts, 'queuedClear').mockReturnValueOnce({ + write: () => { + throw new Error('disk full') + }, + committed: () => {} + }) + await rig.settleAccepted(working, 'a') + await eventually(async () => + expect(journal().queuedMessages.get(clearId)?.state).toBe('returned') + ) + expect(await rig.deleteQueued(clearId)).toMatchObject({ ok: true, value: { deleted: true } }) + await eventually(async () => expect(await rig.handoff(after)).toBeDefined()) + expect(await dividers()).toBe(0) + }) + + it('its Send over a paused queue holds the cards queued before it; Resume sends them', async () => { + const working = await rig.workingSend() + const before = await queuedSend('before the clear') + const clearId = await queuedClear() + await rig.stop() + const after = await queuedSend('after the clear') + await rig.settleAccepted(working, 'a') + expect(await rig.sendNow(clearId)).toMatchObject({ ok: true }) + expect(await dividers()).toBe(1) + expect(journal().queuedMessages.pauses()).toContainEqual( + expect.objectContaining({ reason: 'cleared', messageIds: [before] }) + ) + // The queue never reorders: the card after it waits behind the one it holds. + await settleMs() + expect(await rig.handoff(before)).toBeUndefined() + expect(await rig.handoff(after)).toBeUndefined() + expect(await rig.resume()).toMatchObject({ ok: true, value: { resumed: true } }) + await eventually(async () => expect(await rig.handoff(before)).toBeDefined()) + }) +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-compact.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-compact.test.ts index ee7a26af161e..9500a8346463 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-compact.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-compact.test.ts @@ -539,19 +539,11 @@ describe('a /compact that waits in line', () => { }) }) -it('only a /compact may ask to wait; a /clear sent asking to is refused, never queued', async () => { +it('either command may ask to wait, only with the one delivery the host knows', () => { const base = { envelope: rig.envelope({}, 'agentSession.conversationCommand', 'op-schema') } const parse = (fields: Record) => ConversationCommandParams.safeParse({ ...base, ...fields }).success expect(parse({ command: 'compact', delivery: 'queue-if-active' })).toBe(true) + expect(parse({ command: 'clear', delivery: 'queue-if-active' })).toBe(true) expect(parse({ command: 'compact', delivery: 'now' })).toBe(false) - await rig.workingSend() - const fields = { command: 'clear' as const, delivery: 'queue-if-active' as const } - expect( - await rig.host.conversationCommand(CALLER, { - envelope: rig.envelope(fields, 'agentSession.conversationCommand', hostTestOperationId()), - ...fields - }) - ).toMatchObject({ ok: false }) - expect(await rig.drafts()).toEqual([]) }) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-messages.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-messages.ts index 249396d38b5d..e7db29c27be5 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. +// submission when the session stops owing work (or runs a /clear card itself), +// 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 @@ -31,6 +32,7 @@ import { } from './structured-agent-session-queued-pause' import { nextSendableQueuedCard } from '../agent-session-journal/queued-message-pause' import type { StructuredAgentSessionLogger } from './structured-agent-session-logger' +import { isQueuedClearCard } from './structured-conversation-clear' import { agentSessionAttachmentExpiredRefusal } from './structured-agent-session-turns' import { isAgentSessionAttachmentExpiredError } from '../agent-session-attachments/agent-session-attachment-claims' @@ -257,13 +259,16 @@ export type QueuedMessageDrainDeps = { /** The consumed submission is ordinary #22821 work from here on. */ wakeDelivery: (sessionId: string) => void logger: StructuredAgentSessionLogger + /** Runs a /clear card itself, inside the step's serialize (`runQueuedConversationClear`). */ + runClear: (sessionId: string, card: QueuedMessageRow) => Promise } /** * The serialized drain. Woken by every journal commit (turn, submission, prompt, * command and Stop settlements are all commits), by draft mutations, and by the * conversation opening; each step re-derives everything and consumes at most one - * draft — the consumed submission then owes work, which gates the next. + * draft — the consumed submission then owes work, which gates the next. A /clear + * card is run, never consumed: its commit wakes the step that sends the card behind it. */ export class StructuredAgentSessionQueuedMessageDrain { private readonly scheduled = new Set() @@ -344,6 +349,10 @@ export class StructuredAgentSessionQueuedMessageDrain { if (this.disposed || !next) { return } + if (isQueuedClearCard(next)) { + await this.deps.runClear(sessionId, next) + return + } // Always a fresh id: the submission names its draft by `queuedMessageId`, never by id equality. const submissionId = createStructuredAgentSessionOperationId(randomUUID) try { diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-mutations.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-mutations.ts index 28be32c15759..9eef8e59ebc0 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-mutations.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-mutations.ts @@ -37,6 +37,7 @@ import { structuredAgentSessionSendBlock } from './structured-agent-session-send-preparation' import type { AgentSessionTurnContext, TurnOutcome } from './structured-agent-session-turns' +import { isQueuedClearCard, runQueuedConversationClear } from './structured-conversation-clear' function invalid(message: string): { ok: false @@ -70,7 +71,6 @@ export async function withdrawQueuedMessagesForOperation( }) } - /** Draft actions run like any mutation: admitted on the session's lane, the * conversation opened for the write. */ function mutateQueued( @@ -107,7 +107,8 @@ async function refusingNewerOrcaJournal( * wait — through the same send block and pending-prompt gates as any send; * supersession, Stop and prepared commands are never overridden. The card goes * out under this operation's id, never its own, and the submission names it by - * `queuedMessageId`; one id still means one delivery. + * `queuedMessageId`; one id still means one delivery. A /clear card is run by + * the host instead, spent in its clear's own transaction, so it runs once. */ export function sendQueuedStructuredAgentMessage( context: StructuredAgentSessionMutationContext, @@ -116,6 +117,8 @@ export function sendQueuedStructuredAgentMessage( ): Promise> { const { messageId } = params const operationId = params.envelope.clientOperationId + const settledByOp = (ctx: AgentSessionTurnContext) => + agentSessionOperationKey(ctx.resolvedBy, operationId) const plan: MutationPlan = { method: 'agentSession.queuedMessageSend', fields: { messageId }, @@ -156,6 +159,15 @@ export function sendQueuedStructuredAgentMessage( if (hold === 'working' && row.body.command) { return invalid("A command can't be sent while the agent is working.") } + if (isQueuedClearCard(row)) { + // Run by the host, as the drain runs it: never a submission. A failure lands on the card + // itself, so this answers with the card rather than saying it twice; a wait (background + // tasks, a handoff) changes nothing on the card, so this Send says it. + const ran = await runQueuedConversationClear(context, ctx, row, settledByOp(ctx)) + return ran.kind === 'waiting' + ? { ok: false, refusal: ran.refusal } + : { ok: true, value: queuedClearSendAnswer(ctx, operationId, row) } + } const submissionId = operationId try { await ctx.journal.appendSubmission( @@ -171,7 +183,7 @@ export function sendQueuedStructuredAgentMessage( { messageId, expect: row.state, - settledByOp: agentSessionOperationKey(ctx.resolvedBy, operationId), + settledByOp: settledByOp(ctx), hostInstance: structuredAgentSessionHostInstance() } ) @@ -189,9 +201,12 @@ export function sendQueuedStructuredAgentMessage( return { ok: true, value: { clientMessageId: submissionId, submission } } }, replay: (ctx) => { - const opKey = agentSessionOperationKey(ctx.resolvedBy, operationId) + const card = ctx.journal.queuedMessages.get(messageId) + if (card && isQueuedClearCard(card)) { + return queuedClearSendAnswer(ctx, operationId, card) + } const row = ctx.journal.queuedMessages - .receipts(opKey) + .receipts(settledByOp(ctx)) .find((receipt) => receipt.messageId === messageId) if (!row || row.state !== 'dispatched') { return null @@ -203,6 +218,20 @@ export function sendQueuedStructuredAgentMessage( return mutateQueued(context, caller, params.envelope, plan) } +/** A /clear card's Send answers with the card as it now stands: withdrawn once its clear ran, + * returned with why when it could not. */ +function queuedClearSendAnswer( + ctx: AgentSessionTurnContext, + operationId: string, + card: QueuedMessageRow +): AgentSessionSendResult { + const { state } = ctx.journal.queuedMessages.get(card.messageId) ?? card + return { + clientMessageId: operationId, + queued: { messageId: card.messageId, position: card.position, state } + } +} + /** Delete = discard, with no body in the answer: the card leaving the published * list IS the outcome, so a lost answer needs no re-ask. An Edit is the client * copying the text it already renders, then this Delete. */ diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-wiring.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-wiring.ts index a5ddfeb64748..b5ab60fdde00 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-wiring.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-wiring.ts @@ -1,6 +1,6 @@ // Host wiring for mid-turn queueing: builds the serialized drain from the -// host's mutation context and exposes the draft actions (Send, Delete, Resume), so the host -// class stays a description of its surface. +// host's mutation context, with the /clear runner its /clear cards use, and exposes the draft +// actions (Send, Delete, Resume), so the host class stays a description of its surface. import type { StructuredAgentSessionMutationContext } from './structured-agent-session-host-mutations' import type { @@ -15,6 +15,12 @@ import { sendQueuedStructuredAgentMessage } from './structured-agent-session-queued-mutations' import { deferredStructuredAgentSessionLogger } from './structured-agent-session-logger' +import { agentSessionOperationKey } from '../../../shared/agent-session-operation-ledger' +import type { AgentSessionRecordStore } from '../../runtime/agent-session-record-store' +import { + QUEUED_CLEAR_CALLER_KEY, + runQueuedConversationClear +} from './structured-conversation-clear' /** `sessions` are the live conversations (their `touch` is the idle sweep's activity renewal, * which the drain's schedule rides); everything else comes from the host's mutation context, @@ -33,10 +39,32 @@ export function wireStructuredAgentSessionQueuedMessages( structuredAgentSessionConversationFence(context().deps.store, sessionId), wakeDelivery: (sessionId) => context().wakeDelivery(sessionId), // Read lazily, like the rest of this wiring: the host's deps are not assigned yet. - logger: deferredStructuredAgentSessionLogger(() => context().deps.logger) + logger: deferredStructuredAgentSessionLogger(() => context().deps.logger), + runClear: (sessionId, card) => { + const { store, adapter } = context().deps + return runQueuedConversationClear( + context(), + { + sessionId, + journal: sessions.get(sessionId)!.journal, + fence: structuredAgentSessionConversationFence(store, sessionId), + adapter + }, + card, + agentSessionOperationKey(QUEUED_CLEAR_CALLER_KEY, card.messageId) + ) + } }) return { drain, + /** Child work changed: the strip republishes, and a /clear card waiting on it may run. */ + onChildWorkChanged: (sessionId: string, strip: { publish: (sessionId: string) => void }) => { + strip.publish(sessionId) + drain.schedule(sessionId) + }, + /** A /clear card waits on a handoff, which ends in the record store, not the journal. */ + wakeOnHandoffEnded: (store: Pick) => + store.onHandoffEnded((sessionId) => drain.schedule(sessionId)), /** Every journal publish: turn, submission, prompt, command and Stop * settlements are all commits, and each re-derives the drain's gates. */ onJournalActivity: (sessionId: string) => { diff --git a/src/main/native-chat/agent-session-wire/structured-conversation-clear.ts b/src/main/native-chat/agent-session-wire/structured-conversation-clear.ts new file mode 100644 index 000000000000..05ee7870d314 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-conversation-clear.ts @@ -0,0 +1,158 @@ +// `/clear` under the session's serialize: the agent stopped, then a fresh provider context committed +// in the same conversation. The command RPC runs it at once, and the queue runs a /clear card when +// its turn comes; neither ever hands it to the agent. + +import { agentSessionFailureFact } from '../../../shared/agent-session-failure' +import { agentSessionFailureWords } from '../../../shared/agent-session-failure-words' +import type { AgentJournalMessageItem } from '../../../shared/agent-session-journal-types' +import { + agentSessionRefusalReference, + type AgentSessionWireRefusal +} from '../../../shared/agent-session-wire-refusals' +import { + providerContextBoundaryForClear, + type AgentSessionConversationClear +} from '../../runtime/agent-session-conversation-command-record' +import type { AgentSessionJournal } from '../agent-session-journal/journal-store' +import type { QueuedMessageRow } from '../agent-session-journal/queued-message-table' +import { returnUnsentQueuedCard } from '../agent-session-journal/queued-message-return' +import type { StructuredAgentSessionMutationContext } from './structured-agent-session-mutation-context' +import type { TurnOutcome } from './structured-agent-session-turns' +import { + conversationCommandBlocked, + type ConversationCommandAdmissionContext +} from './structured-conversation-command-admission' + +export const STRUCTURED_AGENT_SESSION_CLEAR_COMMAND = 'clear' + +/** Who a clear the queue ran is recorded as: no client pressed it then, and which one queued it + * is not kept, so no caller's later /clear replays it. */ +export const QUEUED_CLEAR_CALLER_KEY = 'orca:queued-clear' + +/** What the user sent for `/clear`, as its card shows it. */ +export function structuredAgentSessionClearBody(): AgentJournalMessageItem { + return { + kind: 'message', + role: 'user', + blocks: [{ type: 'text', text: '/clear' }], + command: { name: STRUCTURED_AGENT_SESSION_CLEAR_COMMAND } + } +} + +export function isQueuedClearCard(card: Pick): boolean { + return card.body.command?.name === STRUCTURED_AGENT_SESSION_CLEAR_COMMAND +} + +export type ConversationClearContext = ConversationCommandAdmissionContext & { + journal: AgentSessionJournal +} + +/** + * Refused while anything it would cut off is in flight; else the agent is stopped, admission is + * read again at the stop's fence, and `commit` writes the clear. A refusal or a throw before the + * commit changed nothing. + */ +export async function clearConversationUnderSerialize( + context: StructuredAgentSessionMutationContext, + ctx: ConversationClearContext, + marker: { operationId: string; callerKey: string }, + commit: (clear: AgentSessionConversationClear) => Promise +): Promise> { + const { sessionId } = ctx + const store = context.deps.store + const blocked = conversationCommandBlocked( + ctx, + store.getRecord(sessionId)!, + context.readChildWork(sessionId), + context.sessions.get(sessionId)?.child ? undefined : 'at-rest' + ) + if (blocked) { + return { ok: false, refusal: blocked } + } + await context.stopAgent(sessionId, { cause: 'context-clear' }) + const stopped = store.getRecord(sessionId)! + const fence = stopped.lease.runtimeFence + const rechecked = conversationCommandBlocked( + { ...ctx, fence }, + stopped, + context.readChildWork(sessionId), + 'at-rest' + ) + if (rechecked) { + return { ok: false, refusal: rechecked } + } + const command: AgentSessionConversationClear['command'] = { + command: 'clear', + runtimeFence: fence, + ...marker, + phase: 'committed', + state: 'completed' + } + await commit({ sessionId, fence, command, now: context.now() }) + return { ok: true, value: command } +} + +/** Refusals that end on their own: the card keeps waiting, and the drain runs it once they end + * (it wakes on child work and on a handoff ending), as it waits for a turn. */ +const WAITS_FOR = new Set(['backgroundTasksRunning', 'handoffInFlight']) + +export type QueuedClearOutcome = + | { kind: 'cleared' } + /** Still waiting, on something that ends on its own; nothing was written. */ + | { kind: 'waiting'; refusal: AgentSessionWireRefusal } + | { kind: 'returned' } + +/** + * A /clear card's turn, from the drain or the card's own Send, inside the session's serialize. The + * card is spent in the clear's own transaction, so it runs once. Refused for a wait it stays + * waiting; refused otherwise, or failed before its commit, the card is returned with why (the card + * is where that is said, once), and the cards behind it wait: they were written for the cleared chat. + */ +export async function runQueuedConversationClear( + context: StructuredAgentSessionMutationContext, + ctx: ConversationClearContext, + card: Pick, + /** The card's receipt: the drain's own key, or the Send that ran it. */ + settledByOp: string +): Promise { + let fact = agentSessionFailureFact('commandRefused') + try { + const cleared = await clearConversationUnderSerialize( + context, + ctx, + { operationId: card.messageId, callerKey: QUEUED_CLEAR_CALLER_KEY }, + (clear) => + ctx.journal.context.clear( + providerContextBoundaryForClear(clear), + context.deps.store.conversationReceipts.queuedClear(() => clear), + settledByOp, + card.messageId + ) + ) + if (cleared.ok) { + return { kind: 'cleared' } + } + const { details } = cleared.refusal + const reason = details && 'reason' in details ? details.reason : undefined + if (reason !== undefined && WAITS_FOR.has(reason)) { + return { kind: 'waiting', refusal: cleared.refusal } + } + fact = agentSessionFailureFact('commandRefused', { + refusal: agentSessionRefusalReference(cleared.refusal) + }) + } catch (error) { + context.deps.logger.warn('running a queued /clear failed', { + scope: 'queued-clear', + sessionId: ctx.sessionId, + error + }) + } + const words = agentSessionFailureWords(fact, { command: 'clear', surface: 'rejection' }) + await returnUnsentQueuedCard(ctx.journal.queuedMessages, { + messageId: card.messageId, + reason: words.reason, + rejection: words.rejection, + now: context.now() + }) + return { kind: 'returned' } +} diff --git a/src/main/native-chat/agent-session-wire/structured-conversation-command.ts b/src/main/native-chat/agent-session-wire/structured-conversation-command.ts index ab6df634d13f..06dbaff3d86f 100644 --- a/src/main/native-chat/agent-session-wire/structured-conversation-command.ts +++ b/src/main/native-chat/agent-session-wire/structured-conversation-command.ts @@ -4,14 +4,20 @@ import type { } from '../../../shared/agent-session-conversation-command' import type { AgentSessionMutationEnvelope, - AgentSessionMutationResult + AgentSessionMutationResult, + AgentSessionQueuedSendReceipt } from '../../../shared/agent-session-wire' import { admitAndRunAgentSessionMutation } from './structured-agent-session-mutation-admission' import { agentSessionOperationKey } from '../../../shared/agent-session-operation-ledger' import type { StructuredAgentSessionMutationContext } from './structured-agent-session-host-mutations' import { sendPreparation } from './structured-agent-session-send-preparation' import type { StructuredAgentSessionCaller } from './structured-agent-session-host-types' -import { conversationCommandBlocked } from './structured-conversation-command-admission' +import { + clearConversationUnderSerialize, + structuredAgentSessionClearBody +} from './structured-conversation-clear' +import { maybeQueueStructuredAgentSessionSend } from './structured-agent-session-queued-messages' +import { queuedSendAnswer } from './structured-agent-session-queued-send-answer' import type { AgentSessionFailureFact } from '../../../shared/agent-session-failure' import { agentSessionFailureWords, @@ -37,20 +43,29 @@ export function conversationCommandFailure( export type ConversationCommandParams = { envelope: AgentSessionMutationEnvelope command: AgentSessionConversationCommand - /** /compact only: wait as a card while the agent works, as a queued send does. */ + /** Wait as a card while the agent works, as a queued send does. */ delivery?: 'queue-if-active' /** Host-local, set by the client-facing command RPC as for an ordinary send. */ userSend?: true } -/** Stop the provider and record a fresh-context boundary in the same conversation. */ +/** A /clear waiting as a card answers at once; the card is the one surface from here. */ +function queuedClearAnswer( + queued: AgentSessionQueuedSendReceipt +): AgentSessionConversationCommandResult { + return { command: 'clear', state: 'completed', queued } +} + +/** Stop the provider and record a fresh-context boundary in the same conversation; asked with + * `delivery` while the agent works, wait as a /clear card the queue runs when its turn comes. */ export function runStructuredConversationCommand( context: StructuredAgentSessionMutationContext, caller: StructuredAgentSessionCaller, params: ConversationCommandParams ): Promise> { - const { envelope, command } = params + const { envelope, command, delivery } = params const { sessionId, clientOperationId } = envelope const store = context.deps.store + const operation = { callerKey: caller.callerKey, operationId: clientOperationId } const matching = () => { const record = store.getRecord(sessionId)?.conversationCommand return record?.operationId === clientOperationId && record.callerKey === caller.callerKey @@ -73,71 +88,77 @@ export function runStructuredConversationCommand( now: context.now, plan: { method: 'agentSession.conversationCommand', - fields: { command }, + fields: { command, ...(delivery ? { delivery } : {}) }, // Written to the conversation, not the agent, so whoever owns the agent does not matter. conversationWrite: true, recoverUnknownFromDurableState: true, settlesWithWrite: true, - successReceipt: () => - store.conversationReceipts.clear( - () => { - if (!clear) { - throw new Error('agent_session_clear_not_completed') - } - return clear - }, - { callerKey: caller.callerKey, operationId: clientOperationId } - ), - replay: (_ctx, outcome) => { + // The clear's own record write, or for a card, the ledger's plain acceptance. + successReceipt: () => { + const cleared = store.conversationReceipts.clear(() => { + if (!clear) { + throw new Error('agent_session_clear_not_completed') + } + return clear + }, operation) + const queued = store.operationOutcomeReceipt({ + ...operation, + outcome: { status: 'succeeded', sessionId } + }) + return { + write: (db) => (clear ? cleared : queued).write(db), + committed: () => (clear ? cleared : queued).committed() + } + }, + replay: (ctx, outcome) => { if (outcome.status === 'succeeded' && outcome.conversationCommand) { return outcome.conversationCommand } + // A card answers from itself; one pruned after it ran or was deleted, as withdrawn. + if (delivery) { + const card = queuedSendAnswer(ctx.journal, clientOperationId) + if (card && 'queued' in card) { + return queuedClearAnswer(card.queued) + } + if (outcome.status === 'succeeded') { + return queuedClearAnswer({ + messageId: clientOperationId, + position: 0, + state: 'withdrawn' + }) + } + } const prior = matching() return prior?.phase === 'committed' ? prior : null }, // The commit is the only write, so a clear with no committed answer changed nothing. rerunWhenReplayMissing: () => true, run: async (ctx) => { - const record = store.getRecord(sessionId)! - const blocked = conversationCommandBlocked( - ctx, - record, - context.readChildWork(sessionId), - context.sessions.get(sessionId)?.child ? undefined : 'at-rest' - ) - if (blocked) { - return { ok: false, refusal: blocked } - } - await context.stopAgent(sessionId, { cause: 'context-clear' }) - const stopped = store.getRecord(sessionId)! - const fence = stopped.lease.runtimeFence - const rechecked = conversationCommandBlocked( - { ...ctx, fence }, - stopped, - context.readChildWork(sessionId), - 'at-rest' - ) - if (rechecked) { - return { ok: false, refusal: rechecked } - } - const completed: AgentSessionConversationClear['command'] = { - command: 'clear', - runtimeFence: fence, - operationId: clientOperationId, - callerKey: caller.callerKey, - phase: 'committed', - state: 'completed' + // The queue's own accept rule decides first, as for /compact: whatever a queued send + // waits behind, the /clear waits behind too, and the queue runs it when its turn comes. + const card = await maybeQueueStructuredAgentSessionSend(context, ctx, { + envelope, + body: structuredAgentSessionClearBody(), + ...(params.userSend ? { userSend: params.userSend } : {}), + ...(delivery ? { delivery } : {}) + }) + if (card && !card.ok) { + return card } - clear = { sessionId, fence, command: completed, now: context.now() } - if (!ctx.operationReceipt) { - throw new Error('agent_session_clear_receipt_missing') + if (card && 'queued' in card.value) { + return { ok: true, value: queuedClearAnswer(card.value.queued) } } - await ctx.journal.context.clear( - providerContextBoundaryForClear(clear), - ctx.operationReceipt, - agentSessionOperationKey(caller.callerKey, clientOperationId) - ) - return { ok: true, value: completed } + return clearConversationUnderSerialize(context, ctx, operation, (completed) => { + clear = completed + if (!ctx.operationReceipt) { + throw new Error('agent_session_clear_receipt_missing') + } + return ctx.journal.context.clear( + providerContextBoundaryForClear(completed), + ctx.operationReceipt, + agentSessionOperationKey(caller.callerKey, clientOperationId) + ) + }) } } }) diff --git a/src/main/runtime/agent-session-conversation-receipts.ts b/src/main/runtime/agent-session-conversation-receipts.ts index 6935f8bb0836..007b6f47881d 100644 --- a/src/main/runtime/agent-session-conversation-receipts.ts +++ b/src/main/runtime/agent-session-conversation-receipts.ts @@ -26,6 +26,10 @@ export function createAgentSessionConversationReceipts( } }) }), + /** A clear the queue ran from a card: no operation of its own to settle, as the card's own + * withdrawal in the same transaction is what answers for it. */ + queuedClear: (clear: () => AgentSessionConversationClear) => + transactions.receipt((draft) => commitConversationClearRecord(draft, clear())), rewind: ( ...args: Parameters extends [unknown, ...infer A] ? A diff --git a/src/main/runtime/agent-session-record-lease-events.test.ts b/src/main/runtime/agent-session-record-lease-events.test.ts new file mode 100644 index 000000000000..a44582015fc6 --- /dev/null +++ b/src/main/runtime/agent-session-record-lease-events.test.ts @@ -0,0 +1,40 @@ +import { describe, expect, it } from 'vitest' +import type { AgentSessionRecord } from '../../shared/agent-session-record' +import { AgentSessionLeaseEventListeners } from './agent-session-record-lease-events' + +type Lease = AgentSessionRecord['lease'] + +function record(lease: Partial): AgentSessionRecord { + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: collect reads only `lease`; the rest of the record is irrelevant here. + return { lease: { handoffStage: null, handoffOperationId: null, ...lease } } as AgentSessionRecord +} + +describe('lease events a transaction raises', () => { + it('tells handoff listeners only of a handoff that ended, after the commit', () => { + const listeners = new AgentSessionLeaseEventListeners() + expect(listeners.empty).toBe(true) + const ended: string[] = [] + const stop = listeners.onHandoffEnded((sessionId) => ended.push(sessionId)) + expect(listeners.empty).toBe(false) + const before = new Map([ + ['ending', record({ handoffStage: 'recovering' }).lease], + ['still', record({ handoffOperationId: 'op-1' }).lease], + ['idle', record({}).lease] + ]) + const events = listeners.collect( + before, + new Map([ + ['ending', record({})], + ['still', record({ handoffOperationId: 'op-1' })], + ['idle', record({})], + ['starting', record({ handoffStage: 'recovering' })] + ]) + ) + expect(ended).toEqual([]) + listeners.notify(events) + expect(ended).toEqual(['ending']) + stop() + listeners.notify(events) + expect(ended).toEqual(['ending']) + }) +}) diff --git a/src/main/runtime/agent-session-record-lease-events.ts b/src/main/runtime/agent-session-record-lease-events.ts new file mode 100644 index 000000000000..2c07b9e420ec --- /dev/null +++ b/src/main/runtime/agent-session-record-lease-events.ts @@ -0,0 +1,53 @@ +// What a record-store transaction did to leases that listeners outside the store wait on. + +import type { AgentSessionRecord } from '../../shared/agent-session-record' + +type Lease = AgentSessionRecord['lease'] + +/** Listeners keyed by event, each told the session the event happened to. */ +export class AgentSessionLeaseEventListeners { + readonly deathEvidence = new Set<(sessionId: string) => void>() + private readonly handoffEnded = new Set<(sessionId: string) => void>() + + onHandoffEnded = (listener: (sessionId: string) => void): (() => void) => { + this.handoffEnded.add(listener) + return () => this.handoffEnded.delete(listener) + } + + get empty(): boolean { + return this.deathEvidence.size === 0 && this.handoffEnded.size === 0 + } + + /** Read inside the transaction: each session it wrote a new proof of death for, and each whose + * handoff it ended. */ + collect(before: ReadonlyMap, records: ReadonlyMap) { + const events: { deathEvidence: string[]; handoffEnded: string[] } = { + deathEvidence: [], + handoffEnded: [] + } + for (const [sessionId, { lease }] of records) { + const prior = before.get(sessionId) + if (lease.deathEvidence && lease.deathEvidence !== prior?.deathEvidence) { + events.deathEvidence.push(sessionId) + } + if (handoffInFlight(prior) && !handoffInFlight(lease)) { + events.handoffEnded.push(sessionId) + } + } + return events + } + + /** Once the transaction committed, so a listener may read the store. */ + notify(events: ReturnType): void { + for (const sessionId of events.deathEvidence) { + this.deathEvidence.forEach((listener) => listener(sessionId)) + } + for (const sessionId of events.handoffEnded) { + this.handoffEnded.forEach((listener) => listener(sessionId)) + } + } +} + +function handoffInFlight(lease: Lease | undefined): boolean { + return Boolean(lease?.handoffStage || lease?.handoffOperationId) +} diff --git a/src/main/runtime/agent-session-record-store.ts b/src/main/runtime/agent-session-record-store.ts index 1b4dd7bd9d60..f2d7a26930ec 100644 --- a/src/main/runtime/agent-session-record-store.ts +++ b/src/main/runtime/agent-session-record-store.ts @@ -57,6 +57,7 @@ import { type AgentSessionRestartProbeArgs } from './agent-session-restart-reconciliation' import { replaceAgentSessionRecordOptions } from './agent-session-record-options' +import { AgentSessionLeaseEventListeners } from './agent-session-record-lease-events' import { commitAgentSessionReservation, type AgentSessionReserveRequest, @@ -85,7 +86,7 @@ export const AGENT_SESSION_LEASE_TTL_MS = 30_000, AGENT_SESSION_LEASE_RENEW_INTERVAL_MS = 10_000 export class AgentSessionRecordStore { - private readonly deathEvidenceListeners = new Set<(sessionId: string) => void>() + private readonly leaseListeners = new AgentSessionLeaseEventListeners() private readonly firstRecordListeners = new Set<() => void>() readonly conversationReceipts: ReturnType @@ -342,9 +343,11 @@ export class AgentSessionRecordStore { /** Told, once committed, of each session a transaction wrote a proof of death for — whichever * transition wrote it, since every one lands here. Must not throw. */ onDeathEvidence(listener: (sessionId: string) => void): () => void { - this.deathEvidenceListeners.add(listener) - return () => this.deathEvidenceListeners.delete(listener) + this.leaseListeners.deathEvidence.add(listener) + return () => this.leaseListeners.deathEvidence.delete(listener) } + /** A session's lease handoff ended: what waited on it may run. */ + onHandoffEnded = this.leaseListeners.onHandoffEnded /** Told, once committed, when the store records its first chat. Must not throw. */ onFirstRecord(listener: () => void): () => void { @@ -358,24 +361,20 @@ export class AgentSessionRecordStore { apply: (draft: AgentSessionStoreState) => T, options?: { inMemoryWhenReadOnly?: boolean } ): Promise => { - let proven: string[] = [] + let events: ReturnType | null = null let heldBefore = true const result = await this.transactions.transact((draft) => { heldBefore = this.holdsRecords() - if (this.deathEvidenceListeners.size === 0) { + if (this.leaseListeners.empty) { return apply(draft) } - const before = new Map( - [...draft.records].map(([id, record]) => [id, record.lease.deathEvidence]) - ) + const before = new Map([...draft.records].map(([id, record]) => [id, record.lease])) const applied = apply(draft) - proven = [...draft.records] - .filter(([id, { lease }]) => lease.deathEvidence && lease.deathEvidence !== before.get(id)) - .map(([id]) => id) + events = this.leaseListeners.collect(before, draft.records) return applied }, options) - for (const sessionId of proven) { - this.deathEvidenceListeners.forEach((listener) => listener(sessionId)) + if (events) { + this.leaseListeners.notify(events) } if (!heldBefore && this.holdsRecords()) { this.firstRecordListeners.forEach((listener) => listener()) diff --git a/src/renderer/src/components/native-chat/NativeChatQueuedMessageCard.tsx b/src/renderer/src/components/native-chat/NativeChatQueuedMessageCard.tsx index 0b83b794b055..66e5dcc7f0f2 100644 --- a/src/renderer/src/components/native-chat/NativeChatQueuedMessageCard.tsx +++ b/src/renderer/src/components/native-chat/NativeChatQueuedMessageCard.tsx @@ -86,6 +86,11 @@ export function queuedMessageCardCaption( 'components.native-chat.queuedMessages.behindReturnedHold', 'Waiting — a message ahead needs attention' ) + case 'background-tasks': + return translate( + 'components.native-chat.queuedMessages.waitingForBackgroundTasks', + 'Waiting for background tasks to finish' + ) case 'awaiting-answer': return translate( 'components.native-chat.queuedMessages.awaitingAnswerHold', @@ -187,8 +192,9 @@ export function NativeChatQueuedMessageCard({ {/* Nothing acts on a send still on its way: the host holds no card for it yet. */} {card.hold === 'sending' ? null : ( <> - {/* A command never steers: its Send shows only while the agent is idle. */} - {card.waitsForAgent ? null : ( + {/* A command never steers: its Send shows only while the agent is idle, and never on + a /clear the queue is about to run. */} + {card.waitsForAgent || card.runsOnItsOwn ? null : (