diff --git a/ui/desktop/src/acp/__tests__/errors.test.ts b/ui/desktop/src/acp/__tests__/errors.test.ts new file mode 100644 index 000000000000..31c9b3b42fb3 --- /dev/null +++ b/ui/desktop/src/acp/__tests__/errors.test.ts @@ -0,0 +1,48 @@ +import { describe, expect, it } from 'vitest'; +import { parseAcpCreditsExhaustedError } from '../errors'; + +describe('parseAcpCreditsExhaustedError', () => { + it('parses structured ACP credits exhausted errors', () => { + expect( + parseAcpCreditsExhaustedError({ + code: -32603, + message: 'Please add credits to your account, then resend your message to continue.', + data: { + reason: 'credits_exhausted', + url: 'https://router.tetrate.ai/billing', + }, + }) + ).toEqual({ + message: 'Please add credits to your account, then resend your message to continue.', + url: 'https://router.tetrate.ai/billing', + }); + }); + + it('parses wrapped JSON-RPC errors', () => { + expect( + parseAcpCreditsExhaustedError({ + error: { + code: -32603, + message: 'Add credits to continue.', + data: { + reason: 'credits_exhausted', + }, + }, + }) + ).toEqual({ + message: 'Add credits to continue.', + }); + }); + + it('ignores non-credits-exhausted errors', () => { + expect( + parseAcpCreditsExhaustedError({ + code: -32603, + message: 'Something failed.', + data: { + reason: 'provider_error', + }, + }) + ).toBeNull(); + }); +}); diff --git a/ui/desktop/src/acp/__tests__/prompt.test.ts b/ui/desktop/src/acp/__tests__/prompt.test.ts new file mode 100644 index 000000000000..b758737ef2e6 --- /dev/null +++ b/ui/desktop/src/acp/__tests__/prompt.test.ts @@ -0,0 +1,42 @@ +import { describe, expect, it } from 'vitest'; +import type { Message } from '../../api'; +import { messageToAcpPromptContent } from '../prompt'; + +describe('messageToAcpPromptContent', () => { + it('converts text and image content into ACP prompt blocks', () => { + const message: Message = { + id: 'message-1', + role: 'user', + created: 123, + content: [ + { type: 'text', text: 'Describe this' }, + { type: 'image', data: 'abc123', mimeType: 'image/png' }, + ], + metadata: { userVisible: true, agentVisible: true }, + }; + + expect(messageToAcpPromptContent(message)).toEqual([ + { type: 'text', text: 'Describe this' }, + { type: 'image', data: 'abc123', mimeType: 'image/png' }, + ]); + }); + + it('omits empty text content and unsupported content blocks', () => { + const message: Message = { + id: 'message-1', + role: 'user', + created: 123, + content: [ + { type: 'text', text: ' ' }, + { + type: 'toolResponse', + id: 'tool-1', + toolResult: { status: 'success', value: [] }, + }, + ], + metadata: { userVisible: true, agentVisible: true }, + } as Message; + + expect(messageToAcpPromptContent(message)).toEqual([]); + }); +}); diff --git a/ui/desktop/src/acp/__tests__/sessionNotificationAdapter.test.ts b/ui/desktop/src/acp/__tests__/sessionNotificationAdapter.test.ts new file mode 100644 index 000000000000..99211e3f454b --- /dev/null +++ b/ui/desktop/src/acp/__tests__/sessionNotificationAdapter.test.ts @@ -0,0 +1,406 @@ +import type { GooseSessionNotification_unstable } from '@aaif/goose-sdk'; +import type { RequestPermissionRequest, SessionNotification } from '@agentclientprotocol/sdk'; +import { describe, expect, it } from 'vitest'; +import type { Message, MessageContent } from '../../api'; +import { + createAcpSessionNotificationAdapter, + type AcpChatStateChange, +} from '../sessionNotificationAdapter'; + +const SESSION_ID = 'session-1'; + +function acpUpdate(update: SessionNotification['update']): SessionNotification { + return { + sessionId: SESSION_ID, + update, + }; +} + +function gooseUpdate( + update: GooseSessionNotification_unstable['update'] +): GooseSessionNotification_unstable { + return { + sessionId: SESSION_ID, + update, + }; +} + +function agentText(text: string): SessionNotification { + return acpUpdate({ + sessionUpdate: 'agent_message_chunk', + content: { type: 'text', text }, + }); +} + +function userText(text: string): SessionNotification { + return acpUpdate({ + sessionUpdate: 'user_message_chunk', + content: { type: 'text', text }, + }); +} + +function agentThought(text: string): SessionNotification { + return acpUpdate({ + sessionUpdate: 'agent_thought_chunk', + content: { type: 'text', text }, + }); +} + +function agentImage(data: string, mimeType: string): SessionNotification { + return acpUpdate({ + sessionUpdate: 'agent_message_chunk', + content: { type: 'image', data, mimeType }, + }); +} + +function expectOnlyMessagesChange(chatStateChanges: AcpChatStateChange[]): Message[] { + expect(chatStateChanges).toHaveLength(1); + + const [chatStateChange] = chatStateChanges; + expect(chatStateChange.type).toBe('messages'); + + if (chatStateChange.type !== 'messages') { + throw new Error('expected messages state change'); + } + + return chatStateChange.messages; +} + +function firstContent(message: Message): MessageContent { + const content = message.content[0]; + expect(content).toBeDefined(); + return content; +} + +describe('createAcpSessionNotificationAdapter', () => { + describe('apply', () => { + describe('message chunks', () => { + it('maps and merges text chunks by role', () => { + const adapter = createAcpSessionNotificationAdapter(); + + adapter.apply(agentText('Hello ')); + + const secondChunkStateChanges = adapter.apply(agentText('world')); + let messages = expectOnlyMessagesChange(secondChunkStateChanges); + + expect(messages).toHaveLength(1); + expect(messages[0].role).toBe('assistant'); + expect(firstContent(messages[0])).toMatchObject({ type: 'text', text: 'Hello world' }); + + const userTextStateChanges = adapter.apply(userText('Question')); + messages = expectOnlyMessagesChange(userTextStateChanges); + + expect(messages).toHaveLength(2); + expect(messages[1].role).toBe('user'); + expect(firstContent(messages[1])).toMatchObject({ type: 'text', text: 'Question' }); + }); + + it('appends repeated adjacent text deltas', () => { + const adapter = createAcpSessionNotificationAdapter(); + + adapter.apply(agentText('Hel')); + const messages = expectOnlyMessagesChange(adapter.apply(agentText('l'))); + + expect(messages).toHaveLength(1); + expect(firstContent(messages[0])).toMatchObject({ type: 'text', text: 'Hell' }); + }); + + it('maps image and thinking chunks to existing message content shapes', () => { + const imageAdapter = createAcpSessionNotificationAdapter(); + + const imageStateChanges = imageAdapter.apply(agentImage('base64-image', 'image/png')); + const imageMessages = expectOnlyMessagesChange(imageStateChanges); + + expect(firstContent(imageMessages[0])).toMatchObject({ + type: 'image', + data: 'base64-image', + mimeType: 'image/png', + }); + + const thoughtAdapter = createAcpSessionNotificationAdapter(); + thoughtAdapter.apply(agentThought('Thinking ')); + + const thoughtStateChanges = thoughtAdapter.apply(agentThought('more')); + const thoughtMessages = expectOnlyMessagesChange(thoughtStateChanges); + + expect(thoughtMessages).toHaveLength(1); + expect(firstContent(thoughtMessages[0])).toMatchObject({ + type: 'thinking', + thinking: 'Thinking more', + signature: '', + }); + }); + }); + + describe('tools', () => { + it('maps tool calls and successful responses, including MCP app metadata', () => { + const adapter = createAcpSessionNotificationAdapter(); + + const toolCallStateChanges = adapter.apply( + acpUpdate({ + sessionUpdate: 'tool_call', + toolCallId: 'tool-1', + title: 'Read file', + kind: 'read', + status: 'in_progress', + rawInput: { path: 'README.md' }, + locations: [{ path: 'README.md', line: 1 }], + _meta: { + goose: { + toolCall: { + extensionName: 'developer', + toolName: 'read_file', + }, + }, + }, + }) + ); + let messages = expectOnlyMessagesChange(toolCallStateChanges); + + expect(messages).toHaveLength(1); + expect(messages[0].role).toBe('assistant'); + expect(firstContent(messages[0])).toMatchObject({ + type: 'toolRequest', + id: 'tool-1', + toolCall: { + status: 'success', + value: { + name: 'read_file', + arguments: { path: 'README.md' }, + }, + }, + metadata: { + title: 'Read file', + status: 'in_progress', + extensionName: 'developer', + kind: 'read', + locations: [{ path: 'README.md', line: 1 }], + }, + }); + + const toolResponseStateChanges = adapter.apply( + acpUpdate({ + sessionUpdate: 'tool_call_update', + toolCallId: 'tool-1', + status: 'completed', + rawOutput: 'raw result', + content: [ + { + type: 'content', + content: { type: 'text', text: 'rendered result' }, + }, + ], + _meta: { + goose: { + mcpApp: { + resourceUri: 'ui://app/resource', + extensionName: 'developer', + toolName: 'read_file', + }, + }, + }, + }) + ); + messages = expectOnlyMessagesChange(toolResponseStateChanges); + + expect(messages).toHaveLength(2); + expect(messages[1].role).toBe('user'); + expect(firstContent(messages[1])).toMatchObject({ + type: 'toolResponse', + id: 'tool-1', + toolResult: { + status: 'success', + value: { + content: [{ type: 'text', text: 'rendered result' }], + isError: false, + _meta: { + ui: { resourceUri: 'ui://app/resource' }, + extensionName: 'developer', + toolName: 'read_file', + }, + }, + }, + metadata: { + status: 'completed', + rawOutput: 'raw result', + }, + }); + }); + + it('maps failed tool responses to error results', () => { + const adapter = createAcpSessionNotificationAdapter(); + + const failedToolStateChanges = adapter.apply( + acpUpdate({ + sessionUpdate: 'tool_call_update', + toolCallId: 'tool-1', + status: 'failed', + title: 'Read file', + rawOutput: 'permission denied', + }) + ); + const messages = expectOnlyMessagesChange(failedToolStateChanges); + + expect(messages).toHaveLength(1); + expect(messages[0].role).toBe('user'); + expect(firstContent(messages[0])).toMatchObject({ + type: 'toolResponse', + id: 'tool-1', + toolResult: { + status: 'error', + error: 'permission denied', + }, + metadata: { + title: 'Read file', + status: 'failed', + rawOutput: 'permission denied', + }, + }); + }); + + it('uses failed tool response text content when raw output is absent', () => { + const adapter = createAcpSessionNotificationAdapter(); + + const failedToolStateChanges = adapter.apply( + acpUpdate({ + sessionUpdate: 'tool_call_update', + toolCallId: 'tool-1', + status: 'failed', + title: 'Read file', + content: [ + { + type: 'content', + content: { type: 'text', text: 'file not found' }, + }, + ], + }) + ); + const messages = expectOnlyMessagesChange(failedToolStateChanges); + + expect(firstContent(messages[0])).toMatchObject({ + type: 'toolResponse', + id: 'tool-1', + toolResult: { + status: 'error', + error: 'file not found', + }, + }); + }); + }); + }); + + describe('applyGoose', () => { + it('maps usage updates into token state', () => { + const adapter = createAcpSessionNotificationAdapter(); + + expect( + adapter.applyGoose( + gooseUpdate({ + sessionUpdate: 'usage_update', + used: 42, + contextLimit: 200, + accumulatedInputTokens: 10, + accumulatedOutputTokens: 15, + accumulatedCost: 0.12, + }) + ) + ).toEqual([ + { + type: 'tokenState', + tokenState: { + totalTokens: 42, + accumulatedInputTokens: 10, + accumulatedOutputTokens: 15, + accumulatedTotalTokens: 25, + accumulatedCost: 0.12, + }, + }, + ]); + }); + + it('maps status messages and keeps later id-less chunks separate', () => { + const adapter = createAcpSessionNotificationAdapter(); + + const noticeStateChanges = adapter.applyGoose( + gooseUpdate({ + sessionUpdate: 'status_message', + status: { type: 'notice', message: 'Checking files' }, + }) + ); + let messages = expectOnlyMessagesChange(noticeStateChanges); + + expect(messages).toHaveLength(1); + expect(messages[0].metadata).toMatchObject({ userVisible: true, agentVisible: false }); + expect(firstContent(messages[0])).toMatchObject({ + type: 'systemNotification', + notificationType: 'inlineMessage', + msg: 'Checking files', + }); + + const textStateChanges = adapter.apply(agentText('Result')); + messages = expectOnlyMessagesChange(textStateChanges); + + expect(messages).toHaveLength(2); + expect(firstContent(messages[1])).toMatchObject({ type: 'text', text: 'Result' }); + + const progressStateChanges = adapter.applyGoose( + gooseUpdate({ + sessionUpdate: 'status_message', + status: { type: 'progress', message: 'Still working' }, + }) + ); + messages = expectOnlyMessagesChange(progressStateChanges); + + expect(firstContent(messages[2])).toMatchObject({ + type: 'systemNotification', + notificationType: 'thinkingMessage', + msg: 'Still working', + }); + }); + }); + + describe('applyPermissionRequest', () => { + it('maps permission requests to action-required tool confirmations', () => { + const adapter = createAcpSessionNotificationAdapter(); + + const request: RequestPermissionRequest = { + sessionId: SESSION_ID, + options: [{ optionId: 'allow', name: 'Allow', kind: 'allow_once' }], + toolCall: { + toolCallId: 'tool-1', + title: 'Edit file', + rawInput: { path: 'README.md' }, + content: [ + { + type: 'content', + content: { type: 'text', text: 'Allow editing README.md?' }, + }, + ], + _meta: { + goose: { + toolCall: { + toolName: 'edit_file', + }, + }, + }, + }, + }; + + const permissionStateChanges = adapter.applyPermissionRequest(request); + const messages = expectOnlyMessagesChange(permissionStateChanges); + + expect(messages).toHaveLength(1); + expect(messages[0].role).toBe('assistant'); + expect(firstContent(messages[0])).toMatchObject({ + type: 'actionRequired', + data: { + actionType: 'toolConfirmation', + id: 'tool-1', + toolName: 'edit_file', + arguments: { path: 'README.md' }, + prompt: 'Allow editing README.md?', + }, + }); + }); + }); +}); diff --git a/ui/desktop/src/acp/acpConnection.ts b/ui/desktop/src/acp/acpConnection.ts index 59caba3b0f4d..e687f7047835 100644 --- a/ui/desktop/src/acp/acpConnection.ts +++ b/ui/desktop/src/acp/acpConnection.ts @@ -1,25 +1,22 @@ import { DEFAULT_GOOSE_MCP_HOST_CAPABILITIES, GooseClient, - type Client, + type GooseClientCallbacks, } from '@aaif/goose-sdk'; import { PROTOCOL_VERSION } from '@agentclientprotocol/sdk'; import packageJson from '../../package.json'; +import { routeAcpGooseSessionNotification, routeAcpSessionNotification } from './chatNotifications'; import { createWebSocketStream } from './createWebSocketStream'; +import { requestAcpPermission } from './permissionRequests'; let clientPromise: Promise | null = null; let resolvedClient: GooseClient | null = null; -function createClientCallbacks(): () => Client { +function createClientCallbacks(): () => GooseClientCallbacks { return () => ({ - requestPermission: async () => { - return { - outcome: { - outcome: 'cancelled', - }, - }; - }, - sessionUpdate: async () => {}, + requestPermission: requestAcpPermission, + sessionUpdate: routeAcpSessionNotification, + unstable_sessionUpdate: routeAcpGooseSessionNotification, }); } @@ -50,6 +47,7 @@ async function initializeConnection(): Promise { _meta: { goose: { mcpHostCapabilities: DEFAULT_GOOSE_MCP_HOST_CAPABILITIES, + customNotifications: true, }, }, }, diff --git a/ui/desktop/src/acp/adapter/gooseSessionNotifications.ts b/ui/desktop/src/acp/adapter/gooseSessionNotifications.ts new file mode 100644 index 000000000000..197b4f8a50cc --- /dev/null +++ b/ui/desktop/src/acp/adapter/gooseSessionNotifications.ts @@ -0,0 +1,58 @@ +import type { GooseSessionNotification_unstable } from '@aaif/goose-sdk'; +import { type AcpChatStateChange, type AdapterState, messagesChange } from './shared'; + +export function applyGooseSessionNotification( + state: AdapterState, + notification: GooseSessionNotification_unstable +): AcpChatStateChange[] { + const update = notification.update; + + switch (update.sessionUpdate) { + case 'usage_update': + return [ + { + type: 'tokenState', + tokenState: { + totalTokens: update.used, + accumulatedInputTokens: update.accumulatedInputTokens, + accumulatedOutputTokens: update.accumulatedOutputTokens, + accumulatedTotalTokens: update.accumulatedInputTokens + update.accumulatedOutputTokens, + ...(update.accumulatedCost !== undefined + ? { accumulatedCost: update.accumulatedCost } + : {}), + }, + }, + ]; + case 'status_message': + return applyStatusMessage(state, notification.sessionId, update); + default: + return []; + } +} + +function applyStatusMessage( + state: AdapterState, + sessionId: string, + update: Extract +): AcpChatStateChange[] { + const notificationType = update.status.type === 'notice' ? 'inlineMessage' : 'thinkingMessage'; + + state.messages.push({ + id: `acp_status_${sessionId}_${Date.now()}_${Math.random().toString(36).slice(2, 10)}`, + role: 'assistant', + created: Math.floor(Date.now() / 1000), + content: [ + { + type: 'systemNotification', + notificationType, + msg: update.status.message, + }, + ], + metadata: { + userVisible: true, + agentVisible: false, + }, + }); + + return messagesChange(state); +} diff --git a/ui/desktop/src/acp/adapter/messages.ts b/ui/desktop/src/acp/adapter/messages.ts new file mode 100644 index 000000000000..1b3b3caf0ce0 --- /dev/null +++ b/ui/desktop/src/acp/adapter/messages.ts @@ -0,0 +1,150 @@ +import type { + ContentBlock as AcpContentBlock, + SessionNotification, +} from '@agentclientprotocol/sdk'; +import type { Message, MessageContent } from '../../api'; +import { + type AcpChatStateChange, + type AdapterState, + DEFAULT_VISIBLE_MESSAGE_METADATA, + getGooseMessageMeta, + messagesChange, +} from './shared'; + +export function applyContentChunk( + state: AdapterState, + role: Message['role'], + update: Extract< + SessionNotification['update'], + { sessionUpdate: 'user_message_chunk' | 'agent_message_chunk' } + > +): AcpChatStateChange[] { + const content = messageContentFromAcpContentBlock(update.content); + if (!content) { + return []; + } + + const gooseMeta = getGooseMessageMeta(update); + const messageId = update.messageId ?? gooseMeta.messageId; + const existing = findMessageForChunk(state, role, messageId, gooseMeta.created); + + if (existing) { + const lastContent = existing.content[existing.content.length - 1]; + if (lastContent?.type === 'text' && content.type === 'text') { + lastContent.text += content.text; + } else if (content.type === 'image' && hasImageContent(existing, content)) { + return messagesChange(state); + } else { + existing.content.push(content); + } + } else { + state.messages.push({ + ...(messageId ? { id: messageId } : {}), + role, + created: gooseMeta.created ?? Math.floor(Date.now() / 1000), + content: [content], + metadata: { ...DEFAULT_VISIBLE_MESSAGE_METADATA }, + }); + } + + return messagesChange(state); +} + +export function applyThoughtChunk( + state: AdapterState, + update: Extract +): AcpChatStateChange[] { + if (update.content.type !== 'text') { + return []; + } + + const gooseMeta = getGooseMessageMeta(update); + const messageId = update.messageId ?? gooseMeta.messageId; + const existing = findMessageForChunk(state, 'assistant', messageId, gooseMeta.created); + + if (existing) { + const lastContent = existing.content[existing.content.length - 1]; + if (lastContent?.type === 'thinking') { + lastContent.thinking += update.content.text; + } else { + existing.content.push({ type: 'thinking', thinking: update.content.text, signature: '' }); + } + } else { + state.messages.push({ + ...(messageId ? { id: messageId } : {}), + role: 'assistant', + created: gooseMeta.created ?? Math.floor(Date.now() / 1000), + content: [{ type: 'thinking', thinking: update.content.text, signature: '' }], + metadata: { ...DEFAULT_VISIBLE_MESSAGE_METADATA }, + }); + } + + return messagesChange(state); +} + +function messageContentFromAcpContentBlock(content: AcpContentBlock): MessageContent | undefined { + switch (content.type) { + case 'text': + return { + type: 'text', + text: content.text, + ...(content._meta ? { _meta: content._meta } : {}), + ...(content.annotations ? { annotations: content.annotations } : {}), + }; + case 'image': + return { + type: 'image', + data: content.data, + mimeType: content.mimeType, + ...(content._meta ? { _meta: content._meta } : {}), + ...(content.annotations ? { annotations: content.annotations } : {}), + }; + default: + return undefined; + } +} + +export function findMessageForChunk( + state: AdapterState, + role: Message['role'], + messageId: string | undefined, + created: number | undefined +): Message | undefined { + if (!messageId) { + return lastMergeableMessageWithRole(state, role); + } + + const existing = state.messages.find( + (message) => message.id === messageId && message.role === role + ); + if (existing) { + return existing; + } + + const pending = lastMergeableMessageWithRole(state, role); + if (pending && !pending.id) { + pending.id = messageId; + pending.created = created ?? pending.created; + return pending; + } + + return undefined; +} + +function lastMergeableMessageWithRole( + state: AdapterState, + role: Message['role'] +): Message | undefined { + const lastMessage = state.messages[state.messages.length - 1]; + if (lastMessage?.role !== role || lastMessage.metadata.agentVisible === false) { + return undefined; + } + return lastMessage; +} + +function hasImageContent(message: Message, image: Extract) { + return message.content.some( + (content) => + content.type === 'image' && content.data === image.data && content.mimeType === image.mimeType + ); +} diff --git a/ui/desktop/src/acp/adapter/permissions.ts b/ui/desktop/src/acp/adapter/permissions.ts new file mode 100644 index 000000000000..e99212038ff1 --- /dev/null +++ b/ui/desktop/src/acp/adapter/permissions.ts @@ -0,0 +1,61 @@ +import type { RequestPermissionRequest } from '@agentclientprotocol/sdk'; +import { + type AcpChatStateChange, + type AdapterState, + DEFAULT_VISIBLE_MESSAGE_METADATA, + messagesChange, + rawInputToArguments, + toolIdentity, +} from './shared'; + +export function applyPermissionRequest( + state: AdapterState, + request: RequestPermissionRequest +): AcpChatStateChange[] { + const toolCallId = request.toolCall.toolCallId; + const existing = state.messages.some((message) => + message.content.some( + (content) => + content.type === 'actionRequired' && + content.data.actionType === 'toolConfirmation' && + content.data.id === toolCallId + ) + ); + if (existing) { + return messagesChange(state); + } + + const identity = toolIdentity(request.toolCall); + const prompt = permissionPrompt(request); + + state.messages.push({ + id: `acp_permission_${toolCallId}`, + role: 'assistant', + created: Math.floor(Date.now() / 1000), + content: [ + { + type: 'actionRequired', + data: { + actionType: 'toolConfirmation', + id: toolCallId, + toolName: identity.toolName ?? request.toolCall.title ?? toolCallId, + arguments: rawInputToArguments(request.toolCall.rawInput), + ...(prompt ? { prompt } : {}), + }, + }, + ], + metadata: { ...DEFAULT_VISIBLE_MESSAGE_METADATA }, + }); + + return messagesChange(state); +} + +function permissionPrompt(request: RequestPermissionRequest): string | undefined { + for (const content of request.toolCall.content ?? []) { + if (content.type === 'content' && content.content.type === 'text') { + return content.content.text; + } + } + + return undefined; +} diff --git a/ui/desktop/src/acp/adapter/shared.ts b/ui/desktop/src/acp/adapter/shared.ts new file mode 100644 index 000000000000..ded6e47f8961 --- /dev/null +++ b/ui/desktop/src/acp/adapter/shared.ts @@ -0,0 +1,79 @@ +import type { ToolCall, ToolCallUpdate } from '@agentclientprotocol/sdk'; +import type { Message, TokenState } from '../../api'; + +export type AcpChatStateChange = + | { type: 'messages'; messages: Message[] } + | { type: 'tokenState'; tokenState: Partial } + | { type: 'sessionInfo'; name?: string }; + +export interface AdapterState { + messages: Message[]; +} + +export interface GooseMessageMeta { + messageId?: string; + created?: number; +} + +export interface ToolIdentity { + toolName?: string; + extensionName?: string; +} + +export const DEFAULT_VISIBLE_MESSAGE_METADATA: Message['metadata'] = { + userVisible: true, + agentVisible: true, +}; + +export function messagesChange(state: AdapterState): AcpChatStateChange[] { + return [{ type: 'messages', messages: state.messages.map(cloneMessage) }]; +} + +export function cloneMessage(message: Message): Message { + return { + ...message, + content: message.content.map((content) => ({ ...content })), + metadata: { ...message.metadata }, + }; +} + +export function getGooseMessageMeta(update: { _meta?: unknown }): GooseMessageMeta { + if (!isRecord(update._meta)) { + return {}; + } + + const goose = update._meta.goose; + if (!isRecord(goose)) { + return {}; + } + + return { + created: typeof goose.created === 'number' ? goose.created : undefined, + messageId: typeof goose.messageId === 'string' ? goose.messageId : undefined, + }; +} + +export function rawInputToArguments(rawInput: unknown): Record { + return isRecord(rawInput) ? rawInput : {}; +} + +export function toolIdentity(update: ToolCall | ToolCallUpdate): ToolIdentity { + if (!isRecord(update._meta)) { + return {}; + } + + const goose = update._meta.goose; + if (!isRecord(goose) || !isRecord(goose.toolCall)) { + return {}; + } + + return { + toolName: typeof goose.toolCall.toolName === 'string' ? goose.toolCall.toolName : undefined, + extensionName: + typeof goose.toolCall.extensionName === 'string' ? goose.toolCall.extensionName : undefined, + }; +} + +export function isRecord(value: unknown): value is Record { + return typeof value === 'object' && value !== null; +} diff --git a/ui/desktop/src/acp/adapter/tools.ts b/ui/desktop/src/acp/adapter/tools.ts new file mode 100644 index 000000000000..b79cf07c8b83 --- /dev/null +++ b/ui/desktop/src/acp/adapter/tools.ts @@ -0,0 +1,326 @@ +import type { + ContentBlock as AcpContentBlock, + ToolCall, + ToolCallUpdate, +} from '@agentclientprotocol/sdk'; +import type { CallToolResponse, ContentBlock as ApiContentBlock, Message } from '../../api'; +import { findMessageForChunk } from './messages'; +import { + type AcpChatStateChange, + type AdapterState, + DEFAULT_VISIBLE_MESSAGE_METADATA, + type GooseMessageMeta, + getGooseMessageMeta, + isRecord, + messagesChange, + rawInputToArguments, + toolIdentity, + type ToolIdentity, +} from './shared'; + +export function applyToolCall(state: AdapterState, update: ToolCall): AcpChatStateChange[] { + const gooseMeta = getGooseMessageMeta(update); + const message = getOrCreateAssistantMessageForUpdate(state, gooseMeta); + + if ( + message.content.some( + (content) => content.type === 'toolRequest' && content.id === update.toolCallId + ) + ) { + return messagesChange(state); + } + + const identity = toolIdentity(update); + const metadata = toolRequestMetadata(update, identity); + + message.content.push({ + type: 'toolRequest', + id: update.toolCallId, + toolCall: { + status: 'success', + value: { + name: identity.toolName ?? update.title, + arguments: rawInputToArguments(update.rawInput), + }, + }, + ...(metadata ? { metadata } : {}), + ...(update._meta ? { _meta: update._meta } : {}), + }); + + return messagesChange(state); +} + +export function applyToolCallUpdate( + state: AdapterState, + update: ToolCallUpdate +): AcpChatStateChange[] { + if (update.status !== 'completed' && update.status !== 'failed') { + return []; + } + + if (hasToolResponse(state, update.toolCallId)) { + return messagesChange(state); + } + + const gooseMeta = getGooseMessageMeta(update); + const message = getOrCreateToolResponseMessageForUpdate(state, gooseMeta); + const identity = toolIdentity(update); + const metadata = toolResponseMetadata(update, identity); + + message.content.push({ + type: 'toolResponse', + id: update.toolCallId, + toolResult: + update.status === 'failed' + ? { status: 'error', error: toolError(update) } + : { status: 'success', value: toolResultValue(update, mcpAppMetadata(update)) }, + ...(metadata ? { metadata } : {}), + }); + + return messagesChange(state); +} + +function getOrCreateAssistantMessageForUpdate( + state: AdapterState, + gooseMeta: GooseMessageMeta +): Message { + const existing = findMessageForChunk(state, 'assistant', gooseMeta.messageId, gooseMeta.created); + if (existing) { + return existing; + } + + const message: Message = { + ...(gooseMeta.messageId ? { id: gooseMeta.messageId } : {}), + role: 'assistant', + created: gooseMeta.created ?? Math.floor(Date.now() / 1000), + content: [], + metadata: { ...DEFAULT_VISIBLE_MESSAGE_METADATA }, + }; + state.messages.push(message); + return message; +} + +function getOrCreateToolResponseMessageForUpdate( + state: AdapterState, + gooseMeta: GooseMessageMeta +): Message { + if (gooseMeta.messageId) { + const existing = state.messages.find( + (message) => message.id === gooseMeta.messageId && message.role === 'user' + ); + if (existing) { + return existing; + } + } + + const message: Message = { + ...(gooseMeta.messageId ? { id: gooseMeta.messageId } : {}), + role: 'user', + created: gooseMeta.created ?? Math.floor(Date.now() / 1000), + content: [], + metadata: { ...DEFAULT_VISIBLE_MESSAGE_METADATA }, + }; + state.messages.push(message); + return message; +} + +function hasToolResponse(state: AdapterState, toolCallId: string): boolean { + return state.messages.some((message) => + message.content.some((content) => content.type === 'toolResponse' && content.id === toolCallId) + ); +} + +function toolRequestMetadata( + update: ToolCall, + identity: ToolIdentity +): Record | undefined { + return baseToolMetadata(update, identity); +} + +function toolResponseMetadata( + update: ToolCallUpdate, + identity: ToolIdentity +): Record | undefined { + const metadata = baseToolMetadata(update, identity) ?? {}; + if (update.rawOutput !== undefined) { + metadata.rawOutput = update.rawOutput; + } + if (update.content) { + metadata.content = update.content; + } + + return Object.keys(metadata).length > 0 ? metadata : undefined; +} + +function baseToolMetadata( + update: ToolCall | ToolCallUpdate, + identity: ToolIdentity +): Record | undefined { + const metadata: Record = {}; + + if (update.title) { + metadata.title = update.title; + } + if (update.status) { + metadata.status = update.status; + } + if (identity.extensionName) { + metadata.extensionName = identity.extensionName; + } + if (update.kind) { + metadata.kind = update.kind; + } + if (update.locations) { + metadata.locations = update.locations; + } + + return Object.keys(metadata).length > 0 ? metadata : undefined; +} + +function toolResultValue( + update: ToolCallUpdate, + mcpAppMeta: DesktopMcpAppMeta | undefined +): CallToolResponse { + return { + content: toolResultContent(update), + isError: false, + ...(mcpAppMeta ? { _meta: mcpAppMeta } : {}), + }; +} + +function toolResultContent(update: ToolCallUpdate): ApiContentBlock[] { + const content: ApiContentBlock[] = []; + + for (const item of update.content ?? []) { + if (item.type !== 'content') { + continue; + } + + const block = apiContentBlockFromAcpContentBlock(item.content); + if (block) { + content.push(block); + } + } + + if (content.length > 0) { + return content; + } + + if (typeof update.rawOutput === 'string') { + return [{ type: 'text', text: update.rawOutput }]; + } + + return []; +} + +function apiContentBlockFromAcpContentBlock(content: AcpContentBlock): ApiContentBlock | undefined { + switch (content.type) { + case 'text': + return { + type: 'text', + text: content.text, + ...(content._meta ? { _meta: content._meta } : {}), + }; + case 'image': + return { + type: 'image', + data: content.data, + mimeType: content.mimeType, + ...(content._meta ? { _meta: content._meta } : {}), + }; + case 'audio': + return { + type: 'audio', + data: content.data, + mimeType: content.mimeType, + }; + case 'resource_link': + return { + type: 'resource_link', + uri: content.uri, + name: content.name, + ...(content.description ? { description: content.description } : {}), + ...(content.mimeType ? { mimeType: content.mimeType } : {}), + ...(content.size !== undefined && content.size !== null ? { size: content.size } : {}), + ...(content.title ? { title: content.title } : {}), + ...(content._meta ? { _meta: content._meta } : {}), + }; + case 'resource': + return { + type: 'resource', + resource: apiResourceContentsFromAcpResource(content.resource), + ...(content._meta ? { _meta: content._meta } : {}), + }; + default: + return undefined; + } +} + +function apiResourceContentsFromAcpResource( + resource: Extract['resource'] +): Extract['resource'] { + if ('text' in resource) { + return { + uri: resource.uri, + text: resource.text, + ...(resource.mimeType ? { mimeType: resource.mimeType } : {}), + ...(resource._meta ? { _meta: resource._meta } : {}), + }; + } + + return { + uri: resource.uri, + blob: resource.blob, + ...(resource.mimeType ? { mimeType: resource.mimeType } : {}), + ...(resource._meta ? { _meta: resource._meta } : {}), + }; +} + +function toolError(update: ToolCallUpdate): string { + if (typeof update.rawOutput === 'string' && update.rawOutput.trim()) { + return update.rawOutput; + } + + const contentText = toolResultContent(update) + .flatMap((content) => (content.type === 'text' ? [content.text] : [])) + .filter((text) => text.trim().length > 0) + .join('\n'); + if (contentText) { + return contentText; + } + + return update.title ?? 'Tool call failed'; +} + +interface DesktopMcpAppMeta extends Record { + ui: { + resourceUri: string; + }; + extensionName?: string; + toolName?: string; +} + +function mcpAppMetadata(update: ToolCallUpdate): DesktopMcpAppMeta | undefined { + if (!isRecord(update._meta)) { + return undefined; + } + + const goose = update._meta.goose; + if (!isRecord(goose) || !isRecord(goose.mcpApp)) { + return undefined; + } + + const resourceUri = goose.mcpApp.resourceUri; + if (typeof resourceUri !== 'string') { + return undefined; + } + + return { + ui: { + resourceUri, + }, + extensionName: + typeof goose.mcpApp.extensionName === 'string' ? goose.mcpApp.extensionName : undefined, + toolName: typeof goose.mcpApp.toolName === 'string' ? goose.mcpApp.toolName : undefined, + }; +} diff --git a/ui/desktop/src/acp/chatNotifications.ts b/ui/desktop/src/acp/chatNotifications.ts new file mode 100644 index 000000000000..c79caa5c24cb --- /dev/null +++ b/ui/desktop/src/acp/chatNotifications.ts @@ -0,0 +1,21 @@ +import type { GooseSessionNotification_unstable } from '@aaif/goose-sdk'; +import type { SessionNotification } from '@agentclientprotocol/sdk'; +import { createSessionScopedNotificationRouter } from './sessionScopedNotificationRouter'; + +const acpSessionRouter = createSessionScopedNotificationRouter(); +const gooseSessionRouter = + createSessionScopedNotificationRouter(); + +export const subscribeToAcpSession = acpSessionRouter.subscribe; +export const routeAcpSessionNotification = async ( + notification: SessionNotification +): Promise => { + await acpSessionRouter.route(notification); +}; + +export const subscribeToAcpGooseSession = gooseSessionRouter.subscribe; +export const routeAcpGooseSessionNotification = async ( + notification: GooseSessionNotification_unstable +): Promise => { + await gooseSessionRouter.route(notification); +}; diff --git a/ui/desktop/src/acp/errors.ts b/ui/desktop/src/acp/errors.ts new file mode 100644 index 000000000000..aec28b2ba8aa --- /dev/null +++ b/ui/desktop/src/acp/errors.ts @@ -0,0 +1,45 @@ +export interface AcpCreditsExhaustedError { + message: string; + url?: string; +} + +const CREDITS_EXHAUSTED_REASON = 'credits_exhausted'; + +export function parseAcpCreditsExhaustedError(error: unknown): AcpCreditsExhaustedError | null { + const jsonRpcError = asAcpJsonRpcError(error); + if (jsonRpcError?.data?.reason !== CREDITS_EXHAUSTED_REASON) { + return null; + } + + const url = typeof jsonRpcError.data.url === 'string' ? jsonRpcError.data.url : undefined; + + return { + message: jsonRpcError.message, + ...(url ? { url } : {}), + }; +} + +interface AcpJsonRpcError { + message: string; + data: Record; +} + +function asAcpJsonRpcError(error: unknown): AcpJsonRpcError | null { + if (!isRecord(error)) { + return null; + } + + const candidate = isRecord(error.error) ? error.error : error; + if (typeof candidate.message !== 'string' || !isRecord(candidate.data)) { + return null; + } + + return { + message: candidate.message, + data: candidate.data, + }; +} + +function isRecord(value: unknown): value is Record { + return typeof value === 'object' && value !== null; +} diff --git a/ui/desktop/src/acp/permissionRequests.ts b/ui/desktop/src/acp/permissionRequests.ts new file mode 100644 index 000000000000..eddd0745ad92 --- /dev/null +++ b/ui/desktop/src/acp/permissionRequests.ts @@ -0,0 +1,132 @@ +import type { RequestPermissionRequest, RequestPermissionResponse } from '@agentclientprotocol/sdk'; +import type { Permission } from '../api'; +import { createSessionScopedNotificationRouter } from './sessionScopedNotificationRouter'; + +interface PendingPermissionRequest { + request: RequestPermissionRequest; + resolve: (response: RequestPermissionResponse) => void; +} + +const permissionRequestRouter = createSessionScopedNotificationRouter(); +const pendingRequests = new Map(); + +export const subscribeToAcpPermissionRequests = permissionRequestRouter.subscribe; + +export async function requestAcpPermission( + request: RequestPermissionRequest +): Promise { + const key = permissionRequestKey(request.sessionId, request.toolCall.toolCallId); + const previous = pendingRequests.get(key); + if (previous) { + previous.resolve(cancelledPermissionResponse()); + } + + return new Promise((resolve) => { + pendingRequests.set(key, { request, resolve }); + + permissionRequestRouter + .route(request) + .then((routed) => { + if (!routed) { + const pending = pendingRequests.get(key); + if (pending?.resolve === resolve) { + pendingRequests.delete(key); + resolve(cancelledPermissionResponse()); + } + } + }) + .catch((error) => { + console.warn('Failed to route ACP permission request:', error); + const pending = pendingRequests.get(key); + if (pending?.resolve === resolve) { + pendingRequests.delete(key); + resolve(cancelledPermissionResponse()); + } + }); + }); +} + +export function resolveAcpPermissionRequest( + sessionId: string, + toolCallId: string, + action: Permission +): boolean { + const key = permissionRequestKey(sessionId, toolCallId); + const pending = pendingRequests.get(key); + if (!pending) { + return false; + } + + pendingRequests.delete(key); + pending.resolve(permissionResponseForAction(pending.request, action)); + return true; +} + +export function cancelAcpPermissionRequestsForSession(sessionId: string): void { + for (const [key, pending] of pendingRequests) { + if (pending.request.sessionId === sessionId) { + pendingRequests.delete(key); + pending.resolve(cancelledPermissionResponse()); + } + } +} + +function permissionResponseForAction( + request: RequestPermissionRequest, + action: Permission +): RequestPermissionResponse { + if (action === 'cancel') { + return cancelledPermissionResponse(); + } + + const optionId = permissionOptionIdForAction(request, action); + if (!optionId) { + return cancelledPermissionResponse(); + } + + return { + outcome: { + outcome: 'selected', + optionId, + }, + }; +} + +function permissionOptionIdForAction( + request: RequestPermissionRequest, + action: Permission +): string | undefined { + const kind = permissionOptionKindForAction(action); + if (!kind) { + return undefined; + } + + return request.options.find((candidate) => candidate.kind === kind)?.optionId; +} + +function permissionOptionKindForAction(action: Permission) { + switch (action) { + case 'allow_once': + return 'allow_once'; + case 'always_allow': + return 'allow_always'; + case 'deny_once': + return 'reject_once'; + case 'always_deny': + return 'reject_always'; + case 'cancel': + return undefined; + } +} + +function cancelledPermissionResponse(): RequestPermissionResponse { + return { + outcome: { + outcome: 'cancelled', + }, + }; +} + +function permissionRequestKey(sessionId: string, toolCallId: string): string { + return `${sessionId}\u0000${toolCallId}`; +} diff --git a/ui/desktop/src/acp/prompt.ts b/ui/desktop/src/acp/prompt.ts new file mode 100644 index 000000000000..780fba242267 --- /dev/null +++ b/ui/desktop/src/acp/prompt.ts @@ -0,0 +1,45 @@ +import type { ContentBlock, PromptResponse } from '@agentclientprotocol/sdk'; +import type { Message } from '../api'; +import { getAcpClient } from './acpConnection'; + +export async function acpPromptSession( + sessionId: string, + message: Message +): Promise { + const client = await getAcpClient(); + return client.prompt({ + sessionId, + prompt: messageToAcpPromptContent(message), + }); +} + +export async function acpCancelPrompt(sessionId: string): Promise { + const client = await getAcpClient(); + await client.cancel({ sessionId }); +} + +export function messageToAcpPromptContent(message: Message): ContentBlock[] { + const prompt: ContentBlock[] = []; + + for (const content of message.content) { + switch (content.type) { + case 'text': + if (content.text.trim()) { + prompt.push({ + type: 'text', + text: content.text, + }); + } + break; + case 'image': + prompt.push({ + type: 'image', + data: content.data, + mimeType: content.mimeType, + }); + break; + } + } + + return prompt; +} diff --git a/ui/desktop/src/acp/sessionNotificationAdapter.ts b/ui/desktop/src/acp/sessionNotificationAdapter.ts new file mode 100644 index 000000000000..ccb0ad0cab1c --- /dev/null +++ b/ui/desktop/src/acp/sessionNotificationAdapter.ts @@ -0,0 +1,71 @@ +import type { GooseSessionNotification_unstable } from '@aaif/goose-sdk'; +import type { RequestPermissionRequest, SessionNotification } from '@agentclientprotocol/sdk'; +import type { Message } from '../api'; +import { applyGooseSessionNotification } from './adapter/gooseSessionNotifications'; +import { applyContentChunk, applyThoughtChunk } from './adapter/messages'; +import { applyPermissionRequest as applyPermissionRequestToState } from './adapter/permissions'; +import { type AcpChatStateChange, type AdapterState, cloneMessage } from './adapter/shared'; +import { applyToolCall, applyToolCallUpdate } from './adapter/tools'; + +export type { AcpChatStateChange } from './adapter/shared'; + +export interface AcpSessionNotificationAdapter { + apply(notification: SessionNotification): AcpChatStateChange[]; + applyGoose(notification: GooseSessionNotification_unstable): AcpChatStateChange[]; + applyPermissionRequest(request: RequestPermissionRequest): AcpChatStateChange[]; + getMessages(): Message[]; +} + +export function createAcpSessionNotificationAdapter( + initialMessages: Message[] = [] +): AcpSessionNotificationAdapter { + const state: AdapterState = { + messages: initialMessages.map(cloneMessage), + }; + + return { + apply(notification) { + return applyAcpSessionNotification(state, notification); + }, + applyGoose(notification) { + return applyGooseSessionNotification(state, notification); + }, + applyPermissionRequest(request) { + return applyPermissionRequestToState(state, request); + }, + getMessages() { + return state.messages.map(cloneMessage); + }, + }; +} + +function applyAcpSessionNotification( + state: AdapterState, + notification: SessionNotification +): AcpChatStateChange[] { + const update = notification.update; + + switch (update.sessionUpdate) { + case 'user_message_chunk': + return applyContentChunk(state, 'user', update); + case 'agent_message_chunk': + return applyContentChunk(state, 'assistant', update); + case 'agent_thought_chunk': + return applyThoughtChunk(state, update); + case 'tool_call': + return applyToolCall(state, update); + case 'tool_call_update': + return applyToolCallUpdate(state, update); + case 'session_info_update': + return [ + { + type: 'sessionInfo', + ...(update.title ? { name: update.title } : {}), + }, + ]; + case 'usage_update': + return []; + default: + return []; + } +} diff --git a/ui/desktop/src/acp/sessionScopedNotificationRouter.ts b/ui/desktop/src/acp/sessionScopedNotificationRouter.ts new file mode 100644 index 000000000000..e8973a8e3e8c --- /dev/null +++ b/ui/desktop/src/acp/sessionScopedNotificationRouter.ts @@ -0,0 +1,59 @@ +type SessionScopedNotificationListener = ( + notification: TNotification +) => Promise | void; + +interface SessionScopedNotification { + sessionId: string; +} + +export function createSessionScopedNotificationRouter< + TNotification extends SessionScopedNotification, +>() { + const listenersBySessionId = new Map< + string, + Set> + >(); + + const subscribe = ( + sessionId: string, + listener: SessionScopedNotificationListener + ): (() => void) => { + const listeners = listenersBySessionId.get(sessionId) ?? new Set(); + listeners.add(listener); + listenersBySessionId.set(sessionId, listeners); + + let subscribed = true; + + return () => { + if (!subscribed) { + return; + } + + subscribed = false; + const currentListeners = listenersBySessionId.get(sessionId); + if (!currentListeners) { + return; + } + + currentListeners.delete(listener); + if (currentListeners.size === 0) { + listenersBySessionId.delete(sessionId); + } + }; + }; + + const route = async (notification: TNotification): Promise => { + const listeners = listenersBySessionId.get(notification.sessionId); + if (!listeners) { + return false; + } + + await Promise.all([...listeners].map((listener) => listener(notification))); + return true; + }; + + return { + route, + subscribe, + }; +} diff --git a/ui/desktop/src/acpChatFeatureFlag.ts b/ui/desktop/src/acpChatFeatureFlag.ts new file mode 100644 index 000000000000..34ebb17537a1 --- /dev/null +++ b/ui/desktop/src/acpChatFeatureFlag.ts @@ -0,0 +1 @@ +export const USE_ACP_CHAT = false; diff --git a/ui/desktop/src/components/BaseChat.tsx b/ui/desktop/src/components/BaseChat.tsx index 0be8abcc28c6..4f0a836f9546 100644 --- a/ui/desktop/src/components/BaseChat.tsx +++ b/ui/desktop/src/components/BaseChat.tsx @@ -16,7 +16,7 @@ import { ChatType } from '../types/chat'; import { useIsMobile } from '../hooks/use-mobile'; import { useNavigationContextSafe } from './Layout/NavigationContext'; import { cn } from '../utils'; -import { useChatStream } from '../hooks/useChatStream'; +import { useChatSession } from '../hooks/useChatSession'; import { useNavigation } from '../hooks/useNavigation'; import { RecipeHeader } from './RecipeHeader'; import { RecipeWarningModal } from './ui/RecipeWarningModal'; @@ -114,7 +114,7 @@ export default function BaseChat({ tokenState, notifications: toolCallNotifications, onMessageUpdate, - } = useChatStream({ + } = useChatSession({ sessionId, onStreamFinish, }); diff --git a/ui/desktop/src/components/ToolApprovalButtons.test.tsx b/ui/desktop/src/components/ToolApprovalButtons.test.tsx new file mode 100644 index 000000000000..c7a9890c3a5d --- /dev/null +++ b/ui/desktop/src/components/ToolApprovalButtons.test.tsx @@ -0,0 +1,89 @@ +import { render, type RenderOptions, screen } from '@testing-library/react'; +import userEvent from '@testing-library/user-event'; +import { beforeEach, describe, expect, it, vi } from 'vitest'; +import { confirmToolAction } from '../api'; +import { resolveAcpPermissionRequest } from '../acp/permissionRequests'; +import { IntlTestWrapper } from '../i18n/test-utils'; +import ToolApprovalButtons from './ToolApprovalButtons'; + +vi.mock('../api', () => ({ + confirmToolAction: vi.fn(), +})); + +vi.mock('../acp/permissionRequests', () => ({ + resolveAcpPermissionRequest: vi.fn(), +})); + +vi.mock('../acpChatFeatureFlag', () => ({ + USE_ACP_CHAT: true, +})); + +const renderWithIntl = (ui: React.ReactElement, options?: RenderOptions) => + render(ui, { wrapper: IntlTestWrapper, ...options }); + +const confirmToolActionMock = vi.mocked(confirmToolAction); +const resolveAcpPermissionRequestMock = vi.mocked(resolveAcpPermissionRequest); + +describe('ToolApprovalButtons', () => { + beforeEach(() => { + vi.clearAllMocks(); + }); + + it('marks the approval accepted when the ACP request resolves', async () => { + resolveAcpPermissionRequestMock.mockReturnValueOnce(true); + + renderWithIntl( + + ); + + await userEvent.click(screen.getByRole('button', { name: 'Allow Once' })); + + expect(resolveAcpPermissionRequestMock).toHaveBeenCalledWith( + 'session-1', + 'tool-call-approved', + 'allow_once' + ); + expect(confirmToolActionMock).not.toHaveBeenCalled(); + expect(screen.getByText('developer__shell - Allowed once')).toBeInTheDocument(); + }); + + it('falls back to the REST confirmation when no ACP request is pending', async () => { + resolveAcpPermissionRequestMock.mockReturnValueOnce(false); + confirmToolActionMock.mockResolvedValueOnce({ error: undefined } as Awaited< + ReturnType + >); + + renderWithIntl( + + ); + + await userEvent.click(screen.getByRole('button', { name: 'Allow Once' })); + + expect(resolveAcpPermissionRequestMock).toHaveBeenCalledWith( + 'session-1', + 'tool-call-rerun', + 'allow_once' + ); + expect(confirmToolActionMock).toHaveBeenCalledWith({ + body: { + sessionId: 'session-1', + id: 'tool-call-rerun', + action: 'allow_once', + principalType: 'Tool', + }, + }); + expect(screen.getByText('developer__shell - Allowed once')).toBeInTheDocument(); + }); +}); diff --git a/ui/desktop/src/components/ToolApprovalButtons.tsx b/ui/desktop/src/components/ToolApprovalButtons.tsx index b83e18600d6c..e6472c0b690e 100644 --- a/ui/desktop/src/components/ToolApprovalButtons.tsx +++ b/ui/desktop/src/components/ToolApprovalButtons.tsx @@ -1,6 +1,8 @@ import { useState, useEffect } from 'react'; import { Button } from './ui/button'; import { confirmToolAction, Permission } from '../api'; +import { resolveAcpPermissionRequest } from '../acp/permissionRequests'; +import { USE_ACP_CHAT } from '../acpChatFeatureFlag'; import { defineMessages, useIntl } from '../i18n'; const i18n = defineMessages({ @@ -75,10 +77,18 @@ export default function ToolApprovalButtons({ data }: { data: ToolApprovalData } }, [id, decision, isClicked]); const handleAction = async (action: Permission) => { - setDecision(action); - setIsClicked(true); - try { + // Edit-in-place reruns go through the legacy REST path even when ACP chat is + // enabled, so fall back to confirmToolAction when no ACP request is pending. + if (USE_ACP_CHAT && resolveAcpPermissionRequest(sessionId, id, action)) { + setDecision(action); + setIsClicked(true); + return; + } + + setDecision(action); + setIsClicked(true); + const response = await confirmToolAction({ body: { sessionId, diff --git a/ui/desktop/src/hooks/useAcpChatSession.ts b/ui/desktop/src/hooks/useAcpChatSession.ts new file mode 100644 index 000000000000..4cbc33436d7c --- /dev/null +++ b/ui/desktop/src/hooks/useAcpChatSession.ts @@ -0,0 +1,1264 @@ +import { useCallback, useEffect, useMemo, useReducer, useRef } from 'react'; +import { defineMessages, useIntl } from '../i18n'; +import { v7 as uuidv7 } from 'uuid'; +import { AppEvents } from '../constants/events'; +import { ChatState } from '../types/chatState'; + +import { + getSession, + Message, + resumeAgent, + Session, + sessionCancel, + sessionReply, + TokenState, + updateFromSession, + updateSessionUserRecipeValues, + listApps, +} from '../api'; + +import { + createUserMessage, + createElicitationResponseMessage, + getCompactingMessage, + getThinkingMessage, + NotificationEvent, + UserInput, +} from '../types/message'; +import { errorMessage } from '../utils/conversionUtils'; +import { showExtensionLoadResults } from '../utils/extensionErrorUtils'; +import { maybeHandlePlatformEvent } from '../utils/platform_events'; +import { useSessionEvents, type SessionEvent } from './useSessionEvents'; +import type { UseChatSessionParams, UseChatSessionResult } from './useChatSessionTypes'; +import { subscribeToAcpGooseSession, subscribeToAcpSession } from '../acp/chatNotifications'; +import { + cancelAcpPermissionRequestsForSession, + subscribeToAcpPermissionRequests, +} from '../acp/permissionRequests'; +import { parseAcpCreditsExhaustedError, type AcpCreditsExhaustedError } from '../acp/errors'; +import { acpCancelPrompt, acpPromptSession } from '../acp/prompt'; +import { + createAcpSessionNotificationAdapter, + type AcpChatStateChange, + type AcpSessionNotificationAdapter, +} from '../acp/sessionNotificationAdapter'; + +const resultsCache = new Map(); + +interface StreamState { + messages: Message[]; + session: Session | undefined; + chatState: ChatState; + sessionLoadError: string | undefined; + tokenState: TokenState; + notifications: NotificationEvent[]; +} + +type StreamAction = + | { type: 'SET_MESSAGES'; payload: Message[] } + | { type: 'SET_SESSION'; payload: Session | undefined } + | { type: 'SET_CHAT_STATE'; payload: ChatState } + | { type: 'SET_SESSION_LOAD_ERROR'; payload: string | undefined } + | { type: 'SET_TOKEN_STATE'; payload: TokenState } + | { type: 'ADD_NOTIFICATION'; payload: NotificationEvent } + | { type: 'CLEAR_NOTIFICATIONS' } + | { type: 'APPLY_ACP_CHAT_STATE_CHANGE'; payload: AcpChatStateChange } + | { + type: 'SESSION_LOADED'; + payload: { + session: Session; + messages: Message[]; + tokenState: TokenState; + }; + } + | { type: 'RESET_FOR_NEW_SESSION' } + | { type: 'START_STREAMING' } + | { type: 'STREAM_ERROR'; payload: string } + | { type: 'STREAM_FINISH'; payload?: string }; + +const initialTokenState: TokenState = { + inputTokens: 0, + outputTokens: 0, + totalTokens: 0, + accumulatedInputTokens: 0, + accumulatedOutputTokens: 0, + accumulatedTotalTokens: 0, +}; + +const initialState: StreamState = { + messages: [], + session: undefined, + chatState: ChatState.Idle, + sessionLoadError: undefined, + tokenState: initialTokenState, + notifications: [], +}; + +function streamReducer(state: StreamState, action: StreamAction): StreamState { + switch (action.type) { + case 'SET_MESSAGES': + return { ...state, messages: action.payload }; + + case 'SET_SESSION': + return { ...state, session: action.payload }; + + case 'SET_CHAT_STATE': + return { ...state, chatState: action.payload }; + + case 'SET_SESSION_LOAD_ERROR': + return { ...state, sessionLoadError: action.payload }; + + case 'SET_TOKEN_STATE': + return { ...state, tokenState: action.payload }; + + case 'ADD_NOTIFICATION': + return { ...state, notifications: [...state.notifications, action.payload] }; + + case 'CLEAR_NOTIFICATIONS': + return { ...state, notifications: [] }; + + case 'APPLY_ACP_CHAT_STATE_CHANGE': { + const update = action.payload; + switch (update.type) { + case 'messages': + return { ...state, messages: update.messages }; + case 'tokenState': + return { ...state, tokenState: { ...state.tokenState, ...update.tokenState } }; + case 'sessionInfo': + return update.name + ? { + ...state, + session: state.session ? { ...state.session, name: update.name } : undefined, + } + : state; + } + return state; + } + + case 'SESSION_LOADED': + return { + ...state, + session: action.payload.session, + messages: action.payload.messages, + tokenState: action.payload.tokenState, + chatState: ChatState.Idle, + sessionLoadError: undefined, + }; + + case 'RESET_FOR_NEW_SESSION': + return { + ...state, + messages: [], + session: undefined, + sessionLoadError: undefined, + chatState: ChatState.LoadingConversation, + }; + + case 'START_STREAMING': + return { + ...state, + chatState: ChatState.Streaming, + notifications: [], + }; + + case 'STREAM_ERROR': + return { + ...state, + sessionLoadError: action.payload, + chatState: ChatState.Idle, + }; + + case 'STREAM_FINISH': + return { + ...state, + sessionLoadError: action.payload, + chatState: ChatState.Idle, + }; + + default: + return state; + } +} + +function pushMessage(currentMessages: Message[], incomingMsg: Message): Message[] { + const lastMsg = currentMessages[currentMessages.length - 1]; + + if (lastMsg?.id && lastMsg.id === incomingMsg.id) { + const lastContent = lastMsg.content[lastMsg.content.length - 1]; + const newContent = incomingMsg.content[incomingMsg.content.length - 1]; + + if (incomingMsg.metadata?.inference) { + lastMsg.metadata = { + ...lastMsg.metadata, + inference: incomingMsg.metadata.inference, + }; + } + + if ( + lastContent?.type === 'text' && + newContent?.type === 'text' && + incomingMsg.content.length === 1 + ) { + lastContent.text += newContent.text; + } else if ( + lastContent?.type === 'thinking' && + newContent?.type === 'thinking' && + incomingMsg.content.length === 1 && + 'thinking' in lastContent && + 'thinking' in newContent + ) { + // For thinking blocks: if the new block has a signature, it's the complete + // block from content_block_stop — replace entirely. Otherwise append the delta. + if ('signature' in newContent && newContent.signature) { + lastContent.thinking = newContent.thinking; + lastContent.signature = newContent.signature; + } else { + lastContent.thinking += newContent.thinking; + } + } else { + lastMsg.content.push(...incomingMsg.content); + } + return [...currentMessages]; + } else { + return [...currentMessages, incomingMsg]; + } +} + +function prefersReducedMotion(): boolean { + return window.matchMedia('(prefers-reduced-motion: reduce)').matches; +} + +function createAcpCreditsExhaustedMessage(error: AcpCreditsExhaustedError): Message { + return { + id: uuidv7(), + role: 'assistant', + created: Math.floor(Date.now() / 1000), + content: [ + { + type: 'systemNotification', + notificationType: 'creditsExhausted', + msg: error.message, + ...(error.url ? { data: { top_up_url: error.url } } : {}), + }, + ], + metadata: { userVisible: true, agentVisible: false }, + }; +} + +const REDUCED_MOTION_BATCH_INTERVAL = 1000; + +/** + * Creates an event processor that handles individual SSE events for a request. + * Returns an unsubscribe function and a handler to process events. + */ +function createEventProcessor( + initialMessages: Message[], + dispatch: React.Dispatch, + onFinish: (error?: string) => void, + sessionId: string, + onReloadNeeded?: () => void +) { + let currentMessages = initialMessages; + const reduceMotion = prefersReducedMotion(); + let latestTokenState: TokenState | null = null; + let latestChatState: ChatState = ChatState.Streaming; + let lastBatchUpdate = Date.now(); + let hasPendingUpdate = false; + let pendingInference: Message['metadata']['inference'] | undefined; + + const flushBatchedUpdates = () => { + if (reduceMotion && hasPendingUpdate) { + if (latestTokenState) { + dispatch({ type: 'SET_TOKEN_STATE', payload: latestTokenState }); + } + dispatch({ type: 'SET_MESSAGES', payload: currentMessages }); + dispatch({ type: 'SET_CHAT_STATE', payload: latestChatState }); + hasPendingUpdate = false; + lastBatchUpdate = Date.now(); + } + }; + + const maybeUpdateUI = (tokenState: TokenState, chatState: ChatState, forceImmediate = false) => { + if (!reduceMotion) { + dispatch({ type: 'SET_TOKEN_STATE', payload: tokenState }); + dispatch({ type: 'SET_MESSAGES', payload: currentMessages }); + dispatch({ type: 'SET_CHAT_STATE', payload: chatState }); + } else if (forceImmediate) { + dispatch({ type: 'SET_TOKEN_STATE', payload: tokenState }); + dispatch({ type: 'SET_MESSAGES', payload: currentMessages }); + dispatch({ type: 'SET_CHAT_STATE', payload: chatState }); + hasPendingUpdate = false; + lastBatchUpdate = Date.now(); + } else { + latestTokenState = tokenState; + latestChatState = chatState; + hasPendingUpdate = true; + const now = Date.now(); + if (now - lastBatchUpdate >= REDUCED_MOTION_BATCH_INTERVAL) { + flushBatchedUpdates(); + } + } + }; + + const flushPendingInference = () => { + if (!pendingInference) { + return; + } + + for (let i = currentMessages.length - 1; i >= 0; i--) { + const message = currentMessages[i]; + if (message.role === 'assistant' && message.metadata.userVisible) { + currentMessages = [ + ...currentMessages.slice(0, i), + { + ...message, + metadata: { + ...message.metadata, + inference: message.metadata.inference ?? pendingInference, + }, + }, + ...currentMessages.slice(i + 1), + ]; + break; + } + } + pendingInference = undefined; + }; + + // Returns true if the event is terminal (Finish or Error) + const processEvent = (event: SessionEvent): boolean => { + switch (event.type) { + case 'Message': { + let msg = (event as Record).message as Message; + const tokenState = (event as Record).token_state as TokenState; + + if (msg.content.length === 0 && msg.metadata?.inference) { + pendingInference = msg.metadata.inference; + return false; + } + + if (pendingInference && msg.role === 'assistant' && msg.metadata.userVisible) { + msg = { + ...msg, + metadata: { + ...msg.metadata, + inference: msg.metadata.inference ?? pendingInference, + }, + }; + pendingInference = undefined; + } + + currentMessages = pushMessage(currentMessages, msg); + + const hasToolConfirmation = msg.content.some( + (content) => + content.type === 'actionRequired' && content.data.actionType === 'toolConfirmation' + ); + + const hasElicitation = msg.content.some( + (content) => + content.type === 'actionRequired' && content.data.actionType === 'elicitation' + ); + + if (hasToolConfirmation || hasElicitation) { + maybeUpdateUI(tokenState, ChatState.WaitingForUserInput, true); + } else if (getCompactingMessage(msg)) { + maybeUpdateUI(tokenState, ChatState.Compacting); + } else if (getThinkingMessage(msg)) { + maybeUpdateUI(tokenState, ChatState.Thinking); + } else { + maybeUpdateUI(tokenState, ChatState.Streaming); + } + return false; + } + case 'Error': { + flushPendingInference(); + flushBatchedUpdates(); + dispatch({ type: 'SET_MESSAGES', payload: currentMessages }); + const errorMsg = String((event as Record).error ?? ''); + if (errorMsg.includes('too far behind') && onReloadNeeded) { + // Server indicated we missed events — end streaming without setting + // an error (which would show a blocking error screen), then reload + // the full conversation so the UI reflects the actual state. + onFinish(); + onReloadNeeded(); + } else { + onFinish('Stream error: ' + errorMsg); + } + return true; + } + case 'Finish': { + flushPendingInference(); + flushBatchedUpdates(); + dispatch({ type: 'SET_MESSAGES', payload: currentMessages }); + onFinish(); + return true; + } + case 'UpdateConversation': { + const conversation = (event as Record).conversation as Message[]; + currentMessages = conversation; + if (!reduceMotion) { + dispatch({ type: 'SET_MESSAGES', payload: conversation }); + } else { + hasPendingUpdate = true; + } + return false; + } + case 'Notification': { + dispatch({ type: 'ADD_NOTIFICATION', payload: event as unknown as NotificationEvent }); + maybeHandlePlatformEvent((event as Record).message, sessionId); + return false; + } + case 'Ping': + return false; + default: + return false; + } + }; + + return processEvent; +} + +const i18n = defineMessages({ + notificationTitle: { + id: 'chat.notification.taskComplete.title', + defaultMessage: 'Goose finished the task.', + }, + notificationBody: { + id: 'chat.notification.taskComplete.body', + defaultMessage: 'Click here to bring Goose back into focus.', + }, +}); + +export function useAcpChatSession({ + sessionId, + onStreamFinish, + onSessionLoaded, +}: UseChatSessionParams): UseChatSessionResult { + const intl = useIntl(); + const [state, dispatch] = useReducer(streamReducer, initialState); + + // Long-lived SSE connection for this session + const { addListener, setActiveRequestsHandler } = useSessionEvents(sessionId); + + // Track the active request for cancellation (includes the session that started it) + const activeRequestIdRef = useRef(null); + const activeRequestSessionIdRef = useRef(null); + const activeAbortRef = useRef(null); + const activeUnsubscribeRef = useRef<(() => void) | null>(null); + // When ActiveRequests fires before resumeAgent populates messages (cold mount), + // defer the reattach until the session is loaded so the event processor has + // the full conversation history. Events are buffered in the meantime. + const pendingReattachRequestIdRef = useRef(null); + const pendingReattachBufferRef = useRef([]); + const namePollingRef = useRef | null>(null); + + // Ref to access latest state in callbacks (avoids stale closures) + const stateRef = useRef(state); + stateRef.current = state; + const doReattachRef = useRef<((requestId: string, messages: Message[]) => void) | null>(null); + const acpAdapterRef = useRef( + createAcpSessionNotificationAdapter() + ); + + const dispatchAcpChatStateChanges = useCallback((chatStateChanges: AcpChatStateChange[]) => { + for (const chatStateChange of chatStateChanges) { + dispatch({ type: 'APPLY_ACP_CHAT_STATE_CHANGE', payload: chatStateChange }); + } + }, []); + + useEffect(() => { + const messages = state.session?.id === sessionId ? state.messages : []; + acpAdapterRef.current = createAcpSessionNotificationAdapter(messages); + }, [sessionId, state.messages, state.session?.id]); + + useEffect(() => { + if (!sessionId) { + return; + } + + const unsubscribeAcp = subscribeToAcpSession(sessionId, (notification) => { + dispatchAcpChatStateChanges(acpAdapterRef.current.apply(notification)); + }); + const unsubscribeGoose = subscribeToAcpGooseSession(sessionId, (notification) => { + dispatchAcpChatStateChanges(acpAdapterRef.current.applyGoose(notification)); + }); + const unsubscribePermissionRequests = subscribeToAcpPermissionRequests(sessionId, (request) => { + dispatchAcpChatStateChanges(acpAdapterRef.current.applyPermissionRequest(request)); + dispatch({ type: 'SET_CHAT_STATE', payload: ChatState.WaitingForUserInput }); + }); + + return () => { + unsubscribeAcp(); + unsubscribeGoose(); + unsubscribePermissionRequests(); + cancelAcpPermissionRequestsForSession(sessionId); + }; + }, [dispatchAcpChatStateChanges, sessionId]); + + useEffect(() => { + return () => { + if (namePollingRef.current) { + clearTimeout(namePollingRef.current); + namePollingRef.current = null; + } + }; + }, [sessionId]); + + useEffect(() => { + if (state.session) { + resultsCache.set(sessionId, { session: state.session, messages: state.messages }); + } + }, [sessionId, state.session, state.messages]); + + const onFinish = useCallback( + async (error?: string): Promise => { + // Note: SSE listener/ref cleanup is handled by the terminal-event + // handler in each listener closure (which guards on requestId) so + // that overlapping requests don't clobber each other's state. + + if (namePollingRef.current) { + clearTimeout(namePollingRef.current); + namePollingRef.current = null; + } + + dispatch({ type: 'STREAM_FINISH', payload: error }); + + if (!error) { + try { + const [notificationsEnabled, anyWindowFocused] = await Promise.all([ + window.electron.getSetting('enableNotifications'), + window.electron.isAnyWindowFocused(), + ]); + if (notificationsEnabled === true && !anyWindowFocused) { + window.electron.showNotification({ + title: intl.formatMessage(i18n.notificationTitle), + body: intl.formatMessage(i18n.notificationBody), + }); + } + } catch (notifyError) { + console.warn('Failed to show task completion notification:', notifyError); + } + } + + const isNewSession = sessionId && sessionId.match(/^\d{8}_\d{6}$/); + if (isNewSession) { + window.dispatchEvent(new CustomEvent(AppEvents.MESSAGE_STREAM_FINISHED)); + } + + // Refresh session name after each reply for the first 3 user messages + if (!error && sessionId) { + const currentState = stateRef.current; + const userMessageCount = currentState.messages.filter((m) => m.role === 'user').length; + + if (userMessageCount <= 3) { + try { + const response = await getSession({ + path: { session_id: sessionId }, + throwOnError: true, + }); + if (response.data?.name) { + dispatch({ + type: 'SET_SESSION', + payload: currentState.session + ? { ...currentState.session, name: response.data.name } + : undefined, + }); + window.dispatchEvent( + new CustomEvent(AppEvents.SESSION_RENAMED, { + detail: { sessionId, newName: response.data.name }, + }) + ); + } + } catch (refreshError) { + console.warn('Failed to refresh session name:', refreshError); + } + } + } + + onStreamFinish(); + }, + [intl, onStreamFinish, sessionId] + ); + + // Reload the full conversation from the server, e.g. after the SSE + // stream indicates the client fell too far behind the replay buffer. + const reloadConversation = useCallback(() => { + getSession({ + path: { session_id: sessionId }, + throwOnError: true, + }) + .then((response) => { + const session = response.data as Session; + if (session?.conversation) { + dispatch({ type: 'SET_MESSAGES', payload: session.conversation }); + } + }) + .catch((e) => { + console.warn('Failed to reload conversation after buffer overflow:', e); + }); + }, [sessionId]); + + // Perform the actual reattach: wire up an event processor and listener + // for a request that is already in-flight on the server. + const doReattach = useCallback( + (requestId: string, messages: Message[]) => { + activeRequestIdRef.current = requestId; + activeRequestSessionIdRef.current = sessionId; + pendingReattachRequestIdRef.current = null; + + dispatch({ type: 'SET_CHAT_STATE', payload: ChatState.Streaming }); + dispatch({ type: 'SET_SESSION_LOAD_ERROR', payload: undefined }); + + const processEvent = createEventProcessor( + messages, + dispatch, + onFinish, + sessionId, + reloadConversation + ); + + // Replay any events that were buffered during cold-mount wait + const buffered = pendingReattachBufferRef.current; + pendingReattachBufferRef.current = []; + let finished = false; + for (const event of buffered) { + if (processEvent(event)) { + finished = true; + break; + } + } + + if (finished) { + // The reply already completed while we were waiting for session load. + // Clean up — the buffering listener will be replaced below but the + // old one captured into activeUnsubscribeRef should be removed. + if (activeUnsubscribeRef.current) { + activeUnsubscribeRef.current(); + activeUnsubscribeRef.current = null; + } + activeRequestIdRef.current = null; + activeRequestSessionIdRef.current = null; + return; + } + + // Replace the buffering listener with a real processing listener + if (activeUnsubscribeRef.current) { + activeUnsubscribeRef.current(); + } + const unsubscribe = addListener(requestId, (event) => { + const isTerminal = processEvent(event); + if (isTerminal) { + unsubscribe(); + if (activeRequestIdRef.current === requestId) { + activeUnsubscribeRef.current = null; + activeRequestIdRef.current = null; + activeRequestSessionIdRef.current = null; + } + } + }); + activeUnsubscribeRef.current = unsubscribe; + }, + [sessionId, addListener, onFinish, reloadConversation] + ); + doReattachRef.current = doReattach; + + // Reattach to in-flight replies discovered via the SSE ActiveRequests event. + // This handles the case where the chat view remounts while a reply is still + // running on the server — the new hook instance picks up the existing request + // and starts processing its events. + useEffect(() => { + setActiveRequestsHandler((requestIds: string[]) => { + // Only reattach if we don't already have an active request + if (activeRequestIdRef.current) return; + if (requestIds.length === 0) return; + + // Reattach to the first (most recent) active request. + // Multiple concurrent requests per session aren't supported in the UI. + const requestId = requestIds[0]; + const currentMessages = stateRef.current.messages; + + if (currentMessages.length === 0) { + // Cold mount: resumeAgent hasn't populated messages yet. + // Defer event processing until session load completes so the + // processor starts with the full conversation history. + // Register a buffering listener NOW so replayed events aren't + // lost while we wait. + pendingReattachRequestIdRef.current = requestId; + pendingReattachBufferRef.current = []; + activeRequestIdRef.current = requestId; + activeRequestSessionIdRef.current = sessionId; + dispatch({ type: 'SET_CHAT_STATE', payload: ChatState.Streaming }); + dispatch({ type: 'SET_SESSION_LOAD_ERROR', payload: undefined }); + + const unsubscribe = addListener(requestId, (event) => { + pendingReattachBufferRef.current.push(event); + }); + activeUnsubscribeRef.current = unsubscribe; + return; + } + + doReattach(requestId, currentMessages); + }); + + return () => { + setActiveRequestsHandler(null); + }; + }, [sessionId, addListener, onFinish, reloadConversation, setActiveRequestsHandler, doReattach]); + + /** + * Submit a message via the new POST+SSE pattern. + * 1. Generate request_id + * 2. Register SSE listener BEFORE POST (no race condition) + * 3. POST to /sessions/{id}/reply + * 4. Events arrive on the long-lived SSE connection + */ + const submitToSession = useCallback( + async ( + targetSessionId: string, + userMessage: Message, + currentMessages: Message[], + overrideConversation?: Message[] + ) => { + const requestId = uuidv7(); + const abortController = new AbortController(); + activeRequestIdRef.current = requestId; + activeRequestSessionIdRef.current = targetSessionId; + activeAbortRef.current = abortController; + + // Create event processor and register listener BEFORE the POST + const processEvent = createEventProcessor( + currentMessages, + dispatch, + onFinish, + targetSessionId, + reloadConversation + ); + + const unsubscribe = addListener(requestId, (event) => { + const isTerminal = processEvent(event); + if (isTerminal) { + unsubscribe(); + // Only clear global refs if this request is still the active one. + // A newer request may have already replaced them. + if (activeRequestIdRef.current === requestId) { + activeUnsubscribeRef.current = null; + activeRequestIdRef.current = null; + activeRequestSessionIdRef.current = null; + activeAbortRef.current = null; + } + } + }); + activeUnsubscribeRef.current = unsubscribe; + + try { + await sessionReply({ + path: { id: targetSessionId }, + body: { + request_id: requestId, + user_message: userMessage, + override_conversation: overrideConversation, + }, + signal: abortController.signal, + throwOnError: true, + }); + } catch (error) { + // Abort is expected when stopStreaming races with the POST + if (abortController.signal.aborted) return; + // POST failed — clean up listener and report error. + // Only clear global refs if this request is still the active one; + // a newer request may have already replaced them. + unsubscribe(); + if (activeRequestIdRef.current === requestId) { + activeUnsubscribeRef.current = null; + activeRequestIdRef.current = null; + activeRequestSessionIdRef.current = null; + activeAbortRef.current = null; + } + const msg = errorMessage(error); + if (msg.includes('already has an active request')) { + dispatch({ type: 'SET_CHAT_STATE', payload: ChatState.Idle }); + } else { + onFinish('Submit error: ' + msg); + } + } + }, + [addListener, onFinish, reloadConversation] + ); + + const submitToAcpSession = useCallback( + async (targetSessionId: string, userMessage: Message) => { + activeRequestSessionIdRef.current = targetSessionId; + + try { + await acpPromptSession(targetSessionId, userMessage); + onFinish(); + } catch (error) { + const creditsExhaustedError = parseAcpCreditsExhaustedError(error); + if (creditsExhaustedError) { + dispatch({ + type: 'SET_MESSAGES', + payload: [ + ...stateRef.current.messages, + createAcpCreditsExhaustedMessage(creditsExhaustedError), + ], + }); + onFinish(); + return; + } + + onFinish('Submit error: ' + errorMessage(error)); + } finally { + if (activeRequestSessionIdRef.current === targetSessionId) { + activeRequestSessionIdRef.current = null; + } + } + }, + [onFinish] + ); + + // Load session on mount or sessionId change + useEffect(() => { + if (!sessionId) return; + + const cached = resultsCache.get(sessionId); + if (cached) { + dispatch({ + type: 'SESSION_LOADED', + payload: { + session: cached.session, + messages: cached.messages, + tokenState: { + inputTokens: cached.session?.input_tokens ?? 0, + outputTokens: cached.session?.output_tokens ?? 0, + totalTokens: cached.session?.total_tokens ?? 0, + accumulatedInputTokens: cached.session?.accumulated_input_tokens ?? 0, + accumulatedOutputTokens: cached.session?.accumulated_output_tokens ?? 0, + accumulatedTotalTokens: cached.session?.accumulated_total_tokens ?? 0, + }, + }, + }); + window.dispatchEvent( + new CustomEvent(AppEvents.SESSION_EXTENSIONS_LOADED, { detail: { sessionId } }) + ); + onSessionLoaded?.(); + return; + } + + dispatch({ type: 'RESET_FOR_NEW_SESSION' }); + + let cancelled = false; + + (async () => { + try { + const response = await resumeAgent({ + body: { + session_id: sessionId, + load_model_and_extensions: true, + }, + throwOnError: true, + }); + + if (cancelled) { + return; + } + + const resumeData = response.data; + const loadedSession = resumeData?.session; + const extensionResults = resumeData?.extension_results; + + showExtensionLoadResults(extensionResults); + window.dispatchEvent( + new CustomEvent(AppEvents.SESSION_EXTENSIONS_LOADED, { detail: { sessionId } }) + ); + + const pendingRequestId = pendingReattachRequestIdRef.current; + const reattachedToActiveRequest = activeRequestIdRef.current !== null; + + if (pendingRequestId) { + // Cold-mount reattach: ActiveRequests arrived before resumeAgent + // returned. Load session state first, then complete the reattach + // with the full conversation so the event processor has context. + dispatch({ + type: 'SESSION_LOADED', + payload: { + session: loadedSession!, + messages: loadedSession?.conversation || [], + tokenState: { + inputTokens: loadedSession?.input_tokens ?? 0, + outputTokens: loadedSession?.output_tokens ?? 0, + totalTokens: loadedSession?.total_tokens ?? 0, + accumulatedInputTokens: loadedSession?.accumulated_input_tokens ?? 0, + accumulatedOutputTokens: loadedSession?.accumulated_output_tokens ?? 0, + accumulatedTotalTokens: loadedSession?.accumulated_total_tokens ?? 0, + }, + }, + }); + // Now complete the deferred reattach with the loaded messages + doReattachRef.current?.(pendingRequestId, loadedSession?.conversation || []); + } else if (reattachedToActiveRequest) { + // ActiveRequests already wired up an event processor with existing + // messages — only load session metadata, don't overwrite messages + // with the stale DB snapshot. + dispatch({ type: 'SET_SESSION', payload: loadedSession }); + dispatch({ + type: 'SET_TOKEN_STATE', + payload: { + inputTokens: loadedSession?.input_tokens ?? 0, + outputTokens: loadedSession?.output_tokens ?? 0, + totalTokens: loadedSession?.total_tokens ?? 0, + accumulatedInputTokens: loadedSession?.accumulated_input_tokens ?? 0, + accumulatedOutputTokens: loadedSession?.accumulated_output_tokens ?? 0, + accumulatedTotalTokens: loadedSession?.accumulated_total_tokens ?? 0, + }, + }); + } else { + dispatch({ + type: 'SESSION_LOADED', + payload: { + session: loadedSession!, + messages: loadedSession?.conversation || [], + tokenState: { + inputTokens: loadedSession?.input_tokens ?? 0, + outputTokens: loadedSession?.output_tokens ?? 0, + totalTokens: loadedSession?.total_tokens ?? 0, + accumulatedInputTokens: loadedSession?.accumulated_input_tokens ?? 0, + accumulatedOutputTokens: loadedSession?.accumulated_output_tokens ?? 0, + accumulatedTotalTokens: loadedSession?.accumulated_total_tokens ?? 0, + }, + }, + }); + } + + listApps({ + throwOnError: true, + query: { session_id: sessionId }, + }).catch((err) => { + console.warn('Failed to populate apps cache:', err); + }); + + onSessionLoaded?.(); + } catch (error) { + if (cancelled) return; + + dispatch({ type: 'STREAM_ERROR', payload: errorMessage(error) }); + } + })(); + + return () => { + cancelled = true; + }; + }, [sessionId, onSessionLoaded]); + + const handleSubmit = useCallback( + async (input: UserInput) => { + const { msg: userMessage, images } = input; + const currentState = stateRef.current; + + if ( + !currentState.session || + currentState.chatState === ChatState.LoadingConversation || + currentState.chatState === ChatState.Streaming || + currentState.chatState === ChatState.Thinking || + currentState.chatState === ChatState.Compacting + ) { + return; + } + + const hasExistingMessages = currentState.messages.length > 0; + const hasNewMessage = userMessage.trim().length > 0 || images.length > 0; + + if (!hasNewMessage && !hasExistingMessages) { + return; + } + + // Emit session-created event for first message in a new session + if (!hasExistingMessages && hasNewMessage) { + window.dispatchEvent(new CustomEvent(AppEvents.SESSION_CREATED)); + + const pollForName = async (attempts = 0) => { + if (attempts >= 20) return; + + try { + const response = await getSession({ + path: { session_id: sessionId }, + throwOnError: true, + }); + const currentState = stateRef.current; + const currentName = currentState.session?.name; + const newName = response.data?.name; + + if (newName && newName !== currentName) { + dispatch({ + type: 'SET_SESSION', + payload: currentState.session + ? { ...currentState.session, name: newName } + : undefined, + }); + window.dispatchEvent( + new CustomEvent(AppEvents.SESSION_RENAMED, { + detail: { sessionId, newName }, + }) + ); + return; + } + } catch { + // Silently continue polling + } + + const latestState = stateRef.current; + if ( + latestState.chatState === ChatState.Streaming || + latestState.chatState === ChatState.Thinking || + latestState.chatState === ChatState.Compacting + ) { + namePollingRef.current = setTimeout(() => pollForName(attempts + 1), 500); + } + }; + + namePollingRef.current = setTimeout(() => pollForName(0), 1000); + } + + const newMessage = hasNewMessage + ? createUserMessage(userMessage, images) + : currentState.messages[currentState.messages.length - 1]; + const currentMessages = hasNewMessage + ? [...currentState.messages, newMessage] + : [...currentState.messages]; + + if (hasNewMessage) { + dispatch({ type: 'SET_MESSAGES', payload: currentMessages }); + } + + dispatch({ type: 'START_STREAMING' }); + + await submitToAcpSession(sessionId, newMessage); + }, + [sessionId, submitToAcpSession] + ); + + const submitElicitationResponse = useCallback( + async (elicitationId: string, userData: Record) => { + const currentState = stateRef.current; + + if (!currentState.session || currentState.chatState === ChatState.LoadingConversation) { + return; + } + + // An elicitation response unblocks an in-flight tool call on the original + // request's SSE stream — don't start a new stream or flip chat state. + const responseMessage = createElicitationResponseMessage(elicitationId, userData); + const nextMessages = [...currentState.messages, responseMessage]; + dispatch({ type: 'SET_MESSAGES', payload: nextMessages }); + + try { + await sessionReply({ + path: { id: sessionId }, + body: { + request_id: uuidv7(), + user_message: responseMessage, + }, + throwOnError: true, + }); + } catch (error) { + onFinish('Submit error: ' + errorMessage(error)); + } + }, + [sessionId, onFinish] + ); + + const setRecipeUserParams = useCallback( + async (user_recipe_values: Record) => { + const currentState = stateRef.current; + + if (currentState.session) { + await updateSessionUserRecipeValues({ + path: { + session_id: sessionId, + }, + body: { + userRecipeValues: user_recipe_values, + }, + throwOnError: true, + }); + dispatch({ + type: 'SET_SESSION', + payload: { + ...currentState.session, + user_recipe_values, + }, + }); + } else { + dispatch({ + type: 'SET_SESSION_LOAD_ERROR', + payload: "can't call setRecipeParams without a session", + }); + } + }, + [sessionId] + ); + + useEffect(() => { + if (state.session) { + updateFromSession({ + body: { + session_id: state.session.id, + }, + throwOnError: true, + }); + } + }, [state.session]); + + const stopStreaming = useCallback(() => { + const requestId = activeRequestIdRef.current; + const requestSessionId = activeRequestSessionIdRef.current; + + // Abort the in-flight POST so the reply never starts if cancel wins the race + if (activeAbortRef.current) { + activeAbortRef.current.abort(); + activeAbortRef.current = null; + } + + if (requestId && requestSessionId) { + // Cancel against the session that originally started the request, + // not the current sessionId (which may have changed if user navigated). + sessionCancel({ + path: { id: requestSessionId }, + body: { request_id: requestId }, + }).catch((e) => { + console.warn('Failed to cancel request:', e); + }); + } else if (requestSessionId) { + cancelAcpPermissionRequestsForSession(requestSessionId); + acpCancelPrompt(requestSessionId).catch((e) => { + console.warn('Failed to cancel ACP prompt:', e); + }); + } + + // Clean up listener + if (activeUnsubscribeRef.current) { + activeUnsubscribeRef.current(); + activeUnsubscribeRef.current = null; + } + activeRequestIdRef.current = null; + activeRequestSessionIdRef.current = null; + + dispatch({ type: 'SET_CHAT_STATE', payload: ChatState.Idle }); + }, []); + + const onMessageUpdate = useCallback( + async (messageId: string, newContent: string, editType: 'fork' | 'edit' = 'fork') => { + const currentState = stateRef.current; + + dispatch({ type: 'SET_CHAT_STATE', payload: ChatState.Thinking }); + + try { + const { forkSession } = await import('../api'); + const message = currentState.messages.find((m) => m.id === messageId); + + if (!message) { + throw new Error(`Message with id ${messageId} not found in current messages`); + } + + const response = await forkSession({ + path: { + session_id: sessionId, + }, + body: { + timestamp: message.created, + truncate: true, + copy: editType === 'fork', + }, + throwOnError: true, + }); + + const targetSessionId = response.data?.sessionId; + if (!targetSessionId) { + throw new Error('No session ID returned from fork'); + } + + if (editType === 'fork') { + dispatch({ type: 'SET_CHAT_STATE', payload: ChatState.Idle }); + const event = new CustomEvent(AppEvents.SESSION_FORKED, { + detail: { + newSessionId: targetSessionId, + shouldStartAgent: true, + editedMessage: newContent, + }, + }); + window.dispatchEvent(event); + window.electron.logInfo(`Dispatched session-forked event for session ${targetSessionId}`); + } else { + const { getSession } = await import('../api'); + const sessionResponse = await getSession({ + path: { session_id: targetSessionId }, + throwOnError: true, + }); + + if (sessionResponse.data?.conversation) { + const truncatedMessages = [...sessionResponse.data.conversation]; + const updatedUserMessage = createUserMessage(newContent); + + for (const content of message.content) { + if (content.type === 'image') { + updatedUserMessage.content.push(content); + } + } + + const messagesForUI = [...truncatedMessages, updatedUserMessage]; + dispatch({ type: 'SET_MESSAGES', payload: messagesForUI }); + dispatch({ type: 'START_STREAMING' }); + + await submitToSession(targetSessionId, updatedUserMessage, messagesForUI); + } else { + await handleSubmit({ msg: newContent, images: [] }); + } + } + } catch (error) { + dispatch({ type: 'SET_CHAT_STATE', payload: ChatState.Idle }); + const errorMsg = errorMessage(error); + console.error('Failed to edit message:', error); + const { toastError } = await import('../toasts'); + toastError({ + title: 'Failed to edit message', + msg: errorMsg, + }); + } + }, + [sessionId, handleSubmit, submitToSession] + ); + + const setChatState = useCallback((newState: ChatState) => { + dispatch({ type: 'SET_CHAT_STATE', payload: newState }); + }, []); + + const cached = resultsCache.get(sessionId); + const maybe_cached_messages = state.session ? state.messages : cached?.messages || []; + const maybe_cached_session = state.session ?? cached?.session; + + const notificationsMap = useMemo(() => { + return state.notifications.reduce((map, notification) => { + const key = notification.request_id; + if (!map.has(key)) { + map.set(key, []); + } + map.get(key)!.push(notification); + return map; + }, new Map()); + }, [state.notifications]); + + return { + sessionLoadError: state.sessionLoadError, + messages: maybe_cached_messages, + session: maybe_cached_session, + chatState: state.chatState, + setChatState, + handleSubmit, + submitElicitationResponse, + stopStreaming, + setRecipeUserParams, + tokenState: state.tokenState, + notifications: notificationsMap, + onMessageUpdate, + }; +} diff --git a/ui/desktop/src/hooks/useChatSession.ts b/ui/desktop/src/hooks/useChatSession.ts new file mode 100644 index 000000000000..f1f8ba1ca4a5 --- /dev/null +++ b/ui/desktop/src/hooks/useChatSession.ts @@ -0,0 +1,8 @@ +import { USE_ACP_CHAT } from '../acpChatFeatureFlag'; +import { useAcpChatSession } from './useAcpChatSession'; +import { useChatStream } from './useChatStream'; +import type { UseChatSessionHook } from './useChatSessionTypes'; + +export const useChatSession: UseChatSessionHook = USE_ACP_CHAT + ? useAcpChatSession + : useChatStream; diff --git a/ui/desktop/src/hooks/useChatSessionTypes.ts b/ui/desktop/src/hooks/useChatSessionTypes.ts new file mode 100644 index 000000000000..85ee82599511 --- /dev/null +++ b/ui/desktop/src/hooks/useChatSessionTypes.ts @@ -0,0 +1,33 @@ +import type { Message, Session, TokenState } from '../api'; +import type { ChatState } from '../types/chatState'; +import type { NotificationEvent, UserInput } from '../types/message'; + +export interface UseChatSessionParams { + sessionId: string; + onStreamFinish: () => void; + onSessionLoaded?: () => void; +} + +export interface UseChatSessionResult { + session?: Session; + messages: Message[]; + chatState: ChatState; + setChatState: (state: ChatState) => void; + handleSubmit: (input: UserInput) => Promise; + submitElicitationResponse: ( + elicitationId: string, + userData: Record + ) => Promise; + setRecipeUserParams: (values: Record) => Promise; + stopStreaming: () => void; + sessionLoadError?: string; + tokenState: TokenState; + notifications: Map; + onMessageUpdate: ( + messageId: string, + newContent: string, + editType?: 'fork' | 'edit' + ) => Promise; +} + +export type UseChatSessionHook = (params: UseChatSessionParams) => UseChatSessionResult; diff --git a/ui/desktop/src/hooks/useChatStream.ts b/ui/desktop/src/hooks/useChatStream.ts index c1eb21aa208c..69cb0a7781b4 100644 --- a/ui/desktop/src/hooks/useChatStream.ts +++ b/ui/desktop/src/hooks/useChatStream.ts @@ -29,6 +29,7 @@ import { errorMessage } from '../utils/conversionUtils'; import { showExtensionLoadResults } from '../utils/extensionErrorUtils'; import { maybeHandlePlatformEvent } from '../utils/platform_events'; import { useSessionEvents, type SessionEvent } from './useSessionEvents'; +import type { UseChatSessionParams, UseChatSessionResult } from './useChatSessionTypes'; const resultsCache = new Map(); @@ -36,34 +37,6 @@ export function clearSessionCache(sessionId: string): void { resultsCache.delete(sessionId); } -interface UseChatStreamProps { - sessionId: string; - onStreamFinish: () => void; - onSessionLoaded?: () => void; -} - -interface UseChatStreamReturn { - session?: Session; - messages: Message[]; - chatState: ChatState; - setChatState: (state: ChatState) => void; - handleSubmit: (input: UserInput) => Promise; - submitElicitationResponse: ( - elicitationId: string, - userData: Record - ) => Promise; - setRecipeUserParams: (values: Record) => Promise; - stopStreaming: () => void; - sessionLoadError?: string; - tokenState: TokenState; - notifications: Map; - onMessageUpdate: ( - messageId: string, - newContent: string, - editType?: 'fork' | 'edit' - ) => Promise; -} - interface StreamState { messages: Message[]; session: Session | undefined; @@ -417,7 +390,7 @@ export function useChatStream({ sessionId, onStreamFinish, onSessionLoaded, -}: UseChatStreamProps): UseChatStreamReturn { +}: UseChatSessionParams): UseChatSessionResult { const intl = useIntl(); const [state, dispatch] = useReducer(streamReducer, initialState);