diff --git a/src/renderer/src/components/native-chat/structured-agent-session-entry-endings.ts b/src/renderer/src/components/native-chat/structured-agent-session-entry-endings.ts new file mode 100644 index 00000000000..9cc8a568ce4 --- /dev/null +++ b/src/renderer/src/components/native-chat/structured-agent-session-entry-endings.ts @@ -0,0 +1,83 @@ +// How each outbox entry finally ended, for whoever waits on one (a launch prompt's caller): the +// host holds it, or it left without reaching the host. A send with no answer yet has not ended; +// the open chat keeps sending it under its id. + +export type StructuredAgentSessionEntryEnding = 'delivered' | 'notDelivered' + +type Watcher = (ending: StructuredAgentSessionEntryEnding) => void + +const watchers = new Map>>() + +/** Says how an entry ended to everyone watching it. */ +export function endStructuredAgentSessionEntry( + sessionId: string, + clientMessageId: string, + ending: StructuredAgentSessionEntryEnding +): void { + const session = watchers.get(sessionId) + const watching = session?.get(clientMessageId) + if (!session || !watching) { + return + } + session.delete(clientMessageId) + if (session.size === 0) { + watchers.delete(sessionId) + } + for (const watcher of watching) { + watcher(ending) + } +} + +/** An outbox committed without a watched entry: it left without a settlement saying how (a Stop + * took it back, a closed chat discarded it), so it was not delivered. */ +export function noteStructuredAgentSessionOutboxCommitted( + sessionId: string, + entries: readonly { clientMessageId: string }[] +): void { + const session = watchers.get(sessionId) + if (!session) { + return + } + for (const clientMessageId of session.keys()) { + if (!entries.some((entry) => entry.clientMessageId === clientMessageId)) { + endStructuredAgentSessionEntry(sessionId, clientMessageId, 'notDelivered') + } + } +} + +/** Watches one entry until it ends. Watch before sending, so an answer that settles at once is + * not missed; `cancel` when the ending is no longer wanted. */ +export function watchStructuredAgentSessionEntryEnding( + sessionId: string, + clientMessageId: string +): { + ending: Promise + /** How it ended, if it already has. */ + endedAs: () => StructuredAgentSessionEntryEnding | null + cancel: () => void +} { + const ended = Promise.withResolvers() + let endedAs: StructuredAgentSessionEntryEnding | null = null + const watcher: Watcher = (ending) => { + endedAs = ending + ended.resolve(ending) + } + const session = watchers.get(sessionId) ?? new Map>() + watchers.set(sessionId, session) + const watching = session.get(clientMessageId) ?? new Set() + session.set(clientMessageId, watching) + watching.add(watcher) + return { + ending: ended.promise, + endedAs: () => endedAs, + cancel: () => { + watching.delete(watcher) + if (watching.size === 0 && session.get(clientMessageId) === watching) { + session.delete(clientMessageId) + } + if (session.size === 0 && watchers.get(sessionId) === session) { + watchers.delete(sessionId) + } + } + } +} diff --git a/src/renderer/src/components/native-chat/structured-agent-session-outbox-dispatch.ts b/src/renderer/src/components/native-chat/structured-agent-session-outbox-dispatch.ts index 906a5a62938..f926820ae72 100644 --- a/src/renderer/src/components/native-chat/structured-agent-session-outbox-dispatch.ts +++ b/src/renderer/src/components/native-chat/structured-agent-session-outbox-dispatch.ts @@ -36,6 +36,10 @@ import { getStructuredAgentLaunchPromptDispatch, shareStructuredAgentLaunchPromptDispatch } from '@/lib/structured-agent-launch-prompt-in-flight-dispatches' +import { + endStructuredAgentSessionEntry, + type StructuredAgentSessionEntryEnding +} from './structured-agent-session-entry-endings' import { clearStructuredAgentSessionChatLineHeldBy, returnStructuredAgentSessionMessage, @@ -122,6 +126,22 @@ export function sayStructuredAgentSessionSettlement( } } +/** How a settlement ends its entry for whoever waits on it, or null while it has not ended. */ +export function structuredAgentSessionSettlementEnding( + settlement: StructuredAgentSessionSendSettlement +): StructuredAgentSessionEntryEnding | null { + switch (settlement.kind) { + case 'recorded': + case 'pending': + return 'delivered' + case 'returned': + case 'withdrawn': + return 'notDelivered' + case 'unanswered': + return null + } +} + /** Settles one entry against the current outbox and commits it. */ export function settleStructuredAgentSessionOutboxEntry( sessionId: string, @@ -133,6 +153,10 @@ export function settleStructuredAgentSessionOutboxEntry( if (!entry) { return } + const ending = structuredAgentSessionSettlementEnding(settlement) + if (ending) { + endStructuredAgentSessionEntry(sessionId, clientMessageId, ending) + } // A send a Stop outran never goes again: no answer leaves it waiting for the Stop's. const kept = entry.stoppedBy !== undefined && settlement.kind === 'unanswered' diff --git a/src/renderer/src/components/native-chat/structured-agent-session-outbox-journal-settlement.ts b/src/renderer/src/components/native-chat/structured-agent-session-outbox-journal-settlement.ts index 3cb40f155d1..5842e94b85c 100644 --- a/src/renderer/src/components/native-chat/structured-agent-session-outbox-journal-settlement.ts +++ b/src/renderer/src/components/native-chat/structured-agent-session-outbox-journal-settlement.ts @@ -15,7 +15,11 @@ import { commitStructuredAgentSessionOutbox, getStructuredAgentSessionOutbox } from './structured-agent-session-outbox-storage' -import { sayStructuredAgentSessionSettlement } from './structured-agent-session-outbox-dispatch' +import { + sayStructuredAgentSessionSettlement, + structuredAgentSessionSettlementEnding +} from './structured-agent-session-outbox-dispatch' +import { endStructuredAgentSessionEntry } from './structured-agent-session-entry-endings' import { returnStructuredAgentSessionMessage, setStructuredAgentSessionChatLine @@ -49,6 +53,12 @@ export function settleStructuredAgentSessionOutboxFromJournal( if (entries === current) { return entries } + for (const { clientMessageId, settlement } of settled) { + const ending = structuredAgentSessionSettlementEnding(settlement) + if (ending) { + endStructuredAgentSessionEntry(sessionId, clientMessageId, ending) + } + } // Each returned message goes to the draft before the outbox that drops it is saved. for (const back of returned) { returnStructuredAgentSessionMessage(back.entry) diff --git a/src/renderer/src/components/native-chat/structured-agent-session-outbox-storage.ts b/src/renderer/src/components/native-chat/structured-agent-session-outbox-storage.ts index c75cdb1f654..e240e220a02 100644 --- a/src/renderer/src/components/native-chat/structured-agent-session-outbox-storage.ts +++ b/src/renderer/src/components/native-chat/structured-agent-session-outbox-storage.ts @@ -6,6 +6,7 @@ import { } from '../../../../shared/structured-agent-session-outbox' import { createStructuredAgentSessionOperationId } from '../../../../shared/structured-agent-session-mutation' import { createBrowserUuid } from '@/lib/browser-uuid' +import { noteStructuredAgentSessionOutboxCommitted } from './structured-agent-session-entry-endings' const OUTBOX_PREFIX = 'orca:desktopStructuredAgentSessionOutbox:v1:' @@ -171,6 +172,7 @@ export function commitStructuredAgentSessionOutbox( listener() } } + noteStructuredAgentSessionOutboxCommitted(sessionId, entries) return saved } diff --git a/src/renderer/src/components/native-chat/use-structured-agent-session-outbox-fence.test.tsx b/src/renderer/src/components/native-chat/use-structured-agent-session-outbox-fence.test.tsx index 9957a552fe5..716ba260e0d 100644 --- a/src/renderer/src/components/native-chat/use-structured-agent-session-outbox-fence.test.tsx +++ b/src/renderer/src/components/native-chat/use-structured-agent-session-outbox-fence.test.tsx @@ -134,19 +134,23 @@ describe('an outbox on a host that accepts a send before any agent has it', () = const setItem = vi.spyOn(localStorage, 'setItem').mockImplementation(() => { throw new Error('quota exceeded') }) - const delivery = await act(async () => - settleStructuredAgentLaunchPrompt({ - launchResult: Promise.resolve({ sessionId: 'session-1', fence: 1 }), - target: { kind: 'local' }, - options: { prompt: 'launch notes' }, - stagedEntry: staged - }) - ) - setItem.mockRestore() - // Unsaved, it never goes out; the chat holds it and tries again, so the caller says nothing. - expect(delivery).toEqual({ delivered: false, failureNotified: true }) + let delivery: unknown = 'unsettled' + void settleStructuredAgentLaunchPrompt({ + launchResult: Promise.resolve({ sessionId: 'session-1', fence: 1 }), + target: { kind: 'local' }, + options: { prompt: 'launch notes' }, + stagedEntry: staged + })?.then((settled) => { + delivery = settled + }) + try { + await waitFor(() => expect(result.current.outbox).toMatchObject([{ state: 'unconfirmed' }])) + } finally { + setItem.mockRestore() + } + // Unsaved, it never goes out; the chat holds it and tries again, and the caller waits for that. + expect(delivery).toBe('unsettled') expect(mocks.call).not.toHaveBeenCalled() - expect(result.current.outbox).toMatchObject([{ state: 'unconfirmed' }]) expect(readNativeChatDraftCache(structuredAgentSessionDraftScopeKey('session-1'))).toBe('') expect(result.current.error).toBe("Couldn't save your message.") }) diff --git a/src/renderer/src/lib/structured-agent-session-launch-prompt.test.ts b/src/renderer/src/lib/structured-agent-session-launch-prompt.test.ts index 6a67fae390a..b6182e621f0 100644 --- a/src/renderer/src/lib/structured-agent-session-launch-prompt.test.ts +++ b/src/renderer/src/lib/structured-agent-session-launch-prompt.test.ts @@ -1,12 +1,18 @@ // @vitest-environment happy-dom -import { beforeEach, describe, expect, it, vi } from 'vitest' +import { cleanup, renderHook } from '@testing-library/react' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { AGENT_SESSION_SEND_ANSWERS_PROOF_RUNTIME_CAPABILITY } from '../../../shared/protocol-version' import { enqueueStructuredAgentSessionLaunchPrompt, getStructuredAgentSessionOutbox } from '@/components/native-chat/structured-agent-session-outbox-storage' import { structuredAgentSessionDraftScopeKey } from '@/components/native-chat/native-chat-composer-draft-store' -import { readNativeChatDraftCache } from '@/components/native-chat/native-chat-draft-cache' +import { + clearNativeChatDraftCacheForTests, + readNativeChatDraftCache +} from '@/components/native-chat/native-chat-draft-cache' +import { resetStructuredAgentSessionChatLinesForTests } from '@/components/native-chat/structured-agent-session-returned-send' const mocks = vi.hoisted(() => ({ call: vi.fn() })) @@ -14,12 +20,59 @@ vi.mock('@/runtime/structured-agent-session-client', () => ({ callStructuredAgentSession: mocks.call })) +import { setLocalRuntimeCapabilitiesForTests } from '@/runtime/local-runtime-capabilities' +import { useStructuredAgentSessionOutbox } from '@/components/native-chat/use-structured-agent-session-outbox' import { settleStructuredAgentLaunchPrompt } from './structured-agent-session-launch-prompt' +type SentParams = { envelope: { clientOperationId: string } } + +function accepted(clientMessageId: string) { + return { + ok: true, + replayed: false, + fence: 1, + cursor: { epoch: 'epoch-1', sequence: 2 }, + value: { + clientMessageId, + submission: { + clientMessageId, + fence: 1, + payloadFingerprint: 'fingerprint', + dispatchState: 'accepted', + providerItemId: null, + reason: null, + submittedAt: 1, + resolvedAt: 1 + } + } + } +} + +/** The chat open on the session: its outbox resends a send with no answer under the same id. */ +function mountChat(): void { + renderHook(() => + useStructuredAgentSessionOutbox({ + sessionId: 'session-1', + target: { kind: 'local' }, + fence: 1, + submissions: [], + journalCursor: { epoch: 'epoch-1', sequence: 1 } + }) + ) +} + +afterEach(() => { + cleanup() + setLocalRuntimeCapabilitiesForTests(null) +}) + describe('settleStructuredAgentLaunchPrompt', () => { beforeEach(() => { vi.clearAllMocks() localStorage.clear() + clearNativeChatDraftCacheForTests() + resetStructuredAgentSessionChatLinesForTests() + setLocalRuntimeCapabilitiesForTests([AGENT_SESSION_SEND_ANSWERS_PROOF_RUNTIME_CAPABILITY]) vi.spyOn(globalThis.crypto, 'randomUUID').mockReturnValue( '11111111-1111-4111-8111-111111111111' ) @@ -91,20 +144,70 @@ describe('settleStructuredAgentLaunchPrompt', () => { 'review this' ) }) - // The chat keeps it and sends it again: a caller offering it to copy would invite a duplicate. - it('reports a launch prompt with no answer yet as handled by the chat', async () => { + // With no answer the open chat keeps sending it, so the caller waits for how it finally ends: + // offering the prompt again meanwhile could send it twice. + it('waits through a resend when the first send throws, and reports the delivery once', async () => { const stagedEntry = enqueueStructuredAgentSessionLaunchPrompt('session-1', 'review this') - mocks.call.mockRejectedValue(new Error('socket closed')) + const onPromptDelivered = vi.fn() + let calls = 0 + mocks.call.mockImplementation(async (_target, _method, params: SentParams) => { + calls += 1 + if (calls === 1) { + throw new Error('socket closed') + } + return accepted(params.envelope.clientOperationId) + }) + let result: unknown = 'unsettled' + void settleStructuredAgentLaunchPrompt({ + launchResult: Promise.resolve({ sessionId: 'session-1', fence: 1 }), + target: { kind: 'local' }, + options: { prompt: 'review this', onPromptDelivered }, + stagedEntry + })?.then((settled) => { + result = settled + }) + await vi.waitFor(() => + expect(getStructuredAgentSessionOutbox('session-1')).toMatchObject([{ state: 'unconfirmed' }]) + ) + expect(onPromptDelivered).not.toHaveBeenCalled() + expect(result).toBe('unsettled') - await expect( - settleStructuredAgentLaunchPrompt({ - launchResult: Promise.resolve({ sessionId: 'session-1', fence: 1 }), - target: { kind: 'local' }, - options: { prompt: 'review this' }, - stagedEntry - }) - ).resolves.toEqual({ delivered: false, failureNotified: true }) - expect(getStructuredAgentSessionOutbox('session-1')).toMatchObject([{ state: 'unconfirmed' }]) + // The open chat resends it under its id. + mountChat() + await vi.waitFor(() => expect(result).toEqual({ delivered: true, failureNotified: false }), { + timeout: 3000 + }) + expect(onPromptDelivered).toHaveBeenCalledOnce() + expect(calls).toBe(2) + }) + + it('reports a prompt whose resend was refused as given back, and never as delivered', async () => { + const stagedEntry = enqueueStructuredAgentSessionLaunchPrompt('session-1', 'review this') + const onPromptDelivered = vi.fn() + let calls = 0 + mocks.call.mockImplementation(async () => { + calls += 1 + if (calls === 1) { + throw new Error('socket closed') + } + return { ok: false, refusal: { code: 'agent_session_journal_unreadable', message: 'x' } } + }) + const settled = settleStructuredAgentLaunchPrompt({ + launchResult: Promise.resolve({ sessionId: 'session-1', fence: 1 }), + target: { kind: 'local' }, + options: { prompt: 'review this', onPromptDelivered }, + stagedEntry + }) + await vi.waitFor(() => + expect(getStructuredAgentSessionOutbox('session-1')).toMatchObject([{ state: 'unconfirmed' }]) + ) + mountChat() + + await expect(settled).resolves.toEqual({ delivered: false, failureNotified: true }) + expect(onPromptDelivered).not.toHaveBeenCalled() + expect(readNativeChatDraftCache(structuredAgentSessionDraftScopeKey('session-1'))).toBe( + 'review this' + ) }) it('reports a launch prompt that never went out as not handled', async () => { diff --git a/src/renderer/src/lib/structured-agent-session-launch-prompt.ts b/src/renderer/src/lib/structured-agent-session-launch-prompt.ts index b6d019419e6..c74fe094b70 100644 --- a/src/renderer/src/lib/structured-agent-session-launch-prompt.ts +++ b/src/renderer/src/lib/structured-agent-session-launch-prompt.ts @@ -2,6 +2,7 @@ import type { StructuredAgentSessionOutboxEntry } from '../../../shared/structur import { getStructuredAgentSessionOutbox } from '@/components/native-chat/structured-agent-session-outbox-storage' import { sendStructuredAgentSessionOutboxEntry } from '@/components/native-chat/structured-agent-session-outbox-dispatch' import type { RuntimeClientTarget } from '@/runtime/runtime-client-target' +import { watchStructuredAgentSessionEntryEnding } from '@/components/native-chat/structured-agent-session-entry-endings' import { shareStructuredAgentLaunchPromptDispatch, type StructuredAgentLaunchPromptDispatch @@ -57,6 +58,8 @@ export function settleStructuredAgentLaunchPrompt(args: { return { delivered: false, failureNotified: true } } const entry = args.stagedEntry + // Watched before it goes, so an answer that settles it at once is not missed. + const watch = watchStructuredAgentSessionEntryEnding(entry.sessionId, entry.clientMessageId) const dispatch = shareStructuredAgentLaunchPromptDispatch( entry.sessionId, entry.clientMessageId, @@ -64,12 +67,21 @@ export function settleStructuredAgentLaunchPrompt(args: { () => dispatchStructuredLaunchPrompt(entry, receipt, args.target) ) const settlement = await dispatch.promise - const delivered = settlement?.kind === 'recorded' || settlement?.kind === 'pending' + const held = getStructuredAgentSessionOutbox(entry.sessionId).some( + (candidate) => candidate.clientMessageId === entry.clientMessageId + ) + if (settlement === null && !held && watch.endedAs() === null) { + // Settled before this ran: nothing here sent it or can say how it ended. + watch.cancel() + return { delivered: false, failureNotified: false } + } + // With no answer yet the open chat keeps sending it, so this waits for how it finally ends, + // never offering the prompt again while the chat may still deliver it. + const delivered = (await watch.ending) === 'delivered' if (delivered) { args.options.onPromptDelivered?.() } - // Once sent, the chat holds the prompt: it says why it came back, or keeps sending it, so the - // caller must not offer it again beside the chat. - return { delivered, failureNotified: !delivered && settlement !== null } + // Not delivered, it came back to the chat's composer, which says why. + return { delivered, failureNotified: !delivered } }) }