diff --git a/ui/desktop/src/App.test.tsx b/ui/desktop/src/App.test.tsx index c5b72dd9a362..f951518f8cac 100644 --- a/ui/desktop/src/App.test.tsx +++ b/ui/desktop/src/App.test.tsx @@ -9,6 +9,7 @@ import { vi, describe, it, expect, beforeEach, afterEach } from 'vitest'; import { AppInner, resolveSessionInitialMessage } from './App'; import { IntlTestWrapper } from './i18n/test-utils'; import { FeaturesProvider } from './contexts/FeaturesContext'; +import { reconnectAcpAfterSystemResume } from './acp/acpConnection'; // Set up globals for jsdom Object.defineProperty(window, 'location', { @@ -51,6 +52,11 @@ vi.mock('./acp/capabilities', () => ({ getAcpFeatureCapabilities: vi.fn().mockResolvedValue({ localInference: true }), })); +vi.mock('./acp/acpConnection', async (importOriginal) => ({ + ...(await importOriginal()), + reconnectAcpAfterSystemResume: vi.fn(), +})); + // Mock the ACP providers module used by OnboardingGuard so it doesn't try to // open a real ACP client connection during tests. Returning null defaults // keeps the app in the "brand new" (no provider configured) onboarding state. @@ -301,6 +307,23 @@ describe('App Component - Brand New State', () => { expect(mockNavigate).toHaveBeenCalledWith('/'); }); + it('should reconnect ACP when the main process emits system-resume', async () => { + render(, { wrapper: AppInnerTestWrapper }); + + await waitFor(() => { + expect(mockElectron.reactReady).toHaveBeenCalled(); + }); + + const systemResumeHandler = mockElectron.on.mock.calls.find( + ([channel]) => channel === 'system-resume' + )?.[1]; + expect(systemResumeHandler).toBeDefined(); + + systemResumeHandler?.({} as any); + + expect(reconnectAcpAfterSystemResume).toHaveBeenCalledOnce(); + }); + it('should seed recipe sessions with the recipe prompt when no initial message is provided', () => { expect( resolveSessionInitialMessage( diff --git a/ui/desktop/src/App.tsx b/ui/desktop/src/App.tsx index 509f43c1e26d..334e93e6918f 100644 --- a/ui/desktop/src/App.tsx +++ b/ui/desktop/src/App.tsx @@ -58,6 +58,7 @@ import { usePageViewTracking } from './hooks/useAnalytics'; import { trackErrorWithContext } from './utils/analytics'; import { AppEvents } from './constants/events'; import { registerPlatformEventHandlers } from './utils/platform_events'; +import { reconnectAcpAfterSystemResume } from './acp/acpConnection'; function PageViewTracker() { usePageViewTracking(); @@ -396,6 +397,12 @@ export function AppInner() { } }, []); + useEffect(() => { + const handleSystemResume = () => reconnectAcpAfterSystemResume(); + window.electron.on('system-resume', handleSystemResume); + return () => window.electron.off('system-resume', handleSystemResume); + }, []); + useEffect(() => { acpListSessions() .then(({ sessions }) => { diff --git a/ui/desktop/src/acp/__tests__/acpConnection.test.ts b/ui/desktop/src/acp/__tests__/acpConnection.test.ts new file mode 100644 index 000000000000..85d5ade8e829 --- /dev/null +++ b/ui/desktop/src/acp/__tests__/acpConnection.test.ts @@ -0,0 +1,214 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { GOOSE_SERVE_EXITED_USER_MESSAGE } from '../../gooseServeLeaseRegistry'; + +const sdk = vi.hoisted(() => { + const initialize = vi.fn(); + const instances: MockGooseClient[] = []; + + class MockGooseClient { + readonly initialize = initialize; + readonly closed: Promise; + resolveClosed: () => void = () => undefined; + + constructor() { + this.closed = new Promise((resolve) => { + this.resolveClosed = resolve; + }); + instances.push(this); + } + } + + return { GooseClient: MockGooseClient, initialize, instances }; +}); + +const transport = vi.hoisted(() => ({ + createWebSocketStream: vi.fn(), +})); + +vi.mock('@aaif/goose-sdk', () => ({ + DEFAULT_GOOSE_MCP_HOST_CAPABILITIES: {}, + GooseClient: sdk.GooseClient, +})); + +vi.mock('../createWebSocketStream', () => ({ + createWebSocketStream: transport.createWebSocketStream, +})); + +describe('ACP connection ownership', () => { + beforeEach(() => { + vi.useFakeTimers(); + vi.resetModules(); + vi.spyOn(Math, 'random').mockReturnValue(0.5); + sdk.initialize.mockReset().mockResolvedValue({}); + sdk.instances.length = 0; + transport.createWebSocketStream.mockReset().mockImplementation(() => ({ + readable: {}, + writable: {}, + close: vi.fn(), + })); + window.electron.getAcpUrl = vi.fn().mockResolvedValue('ws://localhost/acp'); + }); + + afterEach(() => { + vi.useRealTimers(); + vi.restoreAllMocks(); + }); + + it('shares one initialization between concurrent callers', async () => { + const { getAcpClient } = await import('../acpConnection'); + + const [first, second] = await Promise.all([getAcpClient(), getAcpClient()]); + + expect(first).toBe(second); + expect(sdk.instances).toHaveLength(1); + expect(sdk.initialize).toHaveBeenCalledTimes(1); + expect(transport.createWebSocketStream).toHaveBeenCalledTimes(1); + }); + + it('automatically reconnects after close and shares the result between callers', async () => { + const { getAcpClient } = await import('../acpConnection'); + const firstClient = await getAcpClient(); + const firstStream = transport.createWebSocketStream.mock.results[0].value; + + sdk.instances[0].resolveClosed(); + await Promise.resolve(); + const firstCaller = getAcpClient(); + const secondCaller = getAcpClient(); + + await vi.advanceTimersByTimeAsync(249); + expect(sdk.instances).toHaveLength(1); + + await vi.advanceTimersByTimeAsync(1); + const [firstResult, secondResult] = await Promise.all([firstCaller, secondCaller]); + + expect(firstStream.close).toHaveBeenCalledOnce(); + expect(firstResult).toBe(secondResult); + expect(firstResult).not.toBe(firstClient); + expect(sdk.instances).toHaveLength(2); + expect(transport.createWebSocketStream).toHaveBeenCalledTimes(2); + }); + + it('increases the backoff after a failed reconnect attempt', async () => { + sdk.initialize + .mockResolvedValueOnce({}) + .mockRejectedValueOnce(new Error('server unavailable')) + .mockResolvedValueOnce({}); + const { getAcpClient } = await import('../acpConnection'); + await getAcpClient(); + + sdk.instances[0].resolveClosed(); + await Promise.resolve(); + const reconnected = getAcpClient(); + + await vi.advanceTimersByTimeAsync(250); + expect(sdk.instances).toHaveLength(2); + + await vi.advanceTimersByTimeAsync(499); + expect(sdk.instances).toHaveLength(2); + + await vi.advanceTimersByTimeAsync(1); + await reconnected; + + expect(sdk.instances).toHaveLength(3); + }); + + it('stops reconnecting when the Goose backend has exited', async () => { + const { getAcpClient, subscribeToAcpRecovery } = await import('../acpConnection'); + const listener = vi.fn(); + subscribeToAcpRecovery(listener); + await getAcpClient(); + + const getAcpUrl = vi + .fn() + .mockRejectedValue( + new Error(`Error invoking remote method 'get-acp-url': ${GOOSE_SERVE_EXITED_USER_MESSAGE}`) + ); + window.electron.getAcpUrl = getAcpUrl; + sdk.instances[0].resolveClosed(); + await Promise.resolve(); + + const connection = expect(getAcpClient()).rejects.toThrow(GOOSE_SERVE_EXITED_USER_MESSAGE); + await vi.advanceTimersByTimeAsync(250); + await connection; + + expect(listener.mock.calls).toEqual([[true], [false]]); + await vi.advanceTimersByTimeAsync(60_000); + expect(getAcpUrl).toHaveBeenCalledOnce(); + }); + + it('reconnects immediately after system resume', async () => { + const { getAcpClient, reconnectAcpAfterSystemResume } = await import('../acpConnection'); + await getAcpClient(); + const firstStream = transport.createWebSocketStream.mock.results[0].value; + + reconnectAcpAfterSystemResume(); + const reconnected = getAcpClient(); + await reconnected; + + expect(firstStream.close).toHaveBeenCalledOnce(); + expect(sdk.instances).toHaveLength(2); + }); + + it('does nothing on system resume before ACP has been used', async () => { + const { reconnectAcpAfterSystemResume } = await import('../acpConnection'); + + reconnectAcpAfterSystemResume(); + await Promise.resolve(); + + expect(sdk.instances).toHaveLength(0); + expect(transport.createWebSocketStream).not.toHaveBeenCalled(); + }); + + it('uses normal backoff when the immediate resume attempt fails', async () => { + sdk.initialize + .mockResolvedValueOnce({}) + .mockRejectedValueOnce(new Error('network is not ready')) + .mockResolvedValueOnce({}); + const { getAcpClient, reconnectAcpAfterSystemResume } = await import('../acpConnection'); + await getAcpClient(); + + reconnectAcpAfterSystemResume(); + const reconnected = getAcpClient(); + await Promise.resolve(); + expect(sdk.instances).toHaveLength(2); + + await vi.advanceTimersByTimeAsync(249); + expect(sdk.instances).toHaveLength(2); + + await vi.advanceTimersByTimeAsync(1); + await reconnected; + + expect(sdk.instances).toHaveLength(3); + }); + + it('supersedes an older retry loop after system resume', async () => { + const { getAcpClient, reconnectAcpAfterSystemResume } = await import('../acpConnection'); + await getAcpClient(); + + sdk.instances[0].resolveClosed(); + await Promise.resolve(); + reconnectAcpAfterSystemResume(); + await getAcpClient(); + expect(sdk.instances).toHaveLength(2); + + await vi.advanceTimersByTimeAsync(250); + + expect(sdk.instances).toHaveLength(2); + }); + + it('notifies subscribers while reconnecting and after recovery', async () => { + const { getAcpClient, subscribeToAcpRecovery } = await import('../acpConnection'); + const listener = vi.fn(); + subscribeToAcpRecovery(listener); + await getAcpClient(); + + sdk.instances[0].resolveClosed(); + await Promise.resolve(); + expect(listener).toHaveBeenCalledWith(true); + + await vi.advanceTimersByTimeAsync(250); + await getAcpClient(); + + expect(listener).toHaveBeenLastCalledWith(false); + }); +}); diff --git a/ui/desktop/src/acp/__tests__/chatSessionController.test.ts b/ui/desktop/src/acp/__tests__/chatSessionController.test.ts index 20afdb1b0df9..77c165bd3cb6 100644 --- a/ui/desktop/src/acp/__tests__/chatSessionController.test.ts +++ b/ui/desktop/src/acp/__tests__/chatSessionController.test.ts @@ -168,6 +168,23 @@ describe('acpChatSessionController.loadSession', () => { loadedSession() ); }); + + it('restores a cached session from the server', async () => { + vi.mocked(acpChatSessionStore.getSnapshot).mockReturnValue({ + ...snapshotWithActivePrompt(null), + session: loadedSession(), + }); + vi.mocked(isAcpSessionLoadInFlight).mockReturnValue(false); + + await acpChatSessionController.restoreSession(SESSION_ID); + + expect(acpChatSessionActions.startSessionLoad).toHaveBeenCalledWith(SESSION_ID); + expect(acpLoadSession).toHaveBeenCalledWith(SESSION_ID); + expect(acpChatSessionActions.finishSessionLoad).toHaveBeenCalledWith( + SESSION_ID, + loadedSession() + ); + }); }); describe('acpChatSessionController.stop', () => { @@ -353,7 +370,10 @@ describe('acpChatSessionController.updateMessage', () => { resolvePromptCancellation!(); await updatePromise; - expect(acpTruncateSessionConversation).toHaveBeenCalledWith(SESSION_ID, existingMessage.created); + expect(acpTruncateSessionConversation).toHaveBeenCalledWith( + SESSION_ID, + existingMessage.created + ); expect(acpPromptSession).toHaveBeenCalled(); expect(acpChatSessionActions.clearPromptCancellation).not.toHaveBeenCalledWith( SESSION_ID, diff --git a/ui/desktop/src/acp/__tests__/createWebSocketStream.test.ts b/ui/desktop/src/acp/__tests__/createWebSocketStream.test.ts new file mode 100644 index 000000000000..071edf9e9319 --- /dev/null +++ b/ui/desktop/src/acp/__tests__/createWebSocketStream.test.ts @@ -0,0 +1,128 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { createWebSocketStream } from '../createWebSocketStream'; + +class FakeWebSocket extends window.EventTarget { + static readonly CONNECTING = 0; + static readonly OPEN = 1; + static readonly CLOSING = 2; + static readonly CLOSED = 3; + + readonly sent: string[] = []; + readyState = FakeWebSocket.CONNECTING; + + constructor(readonly url: string) { + super(); + fakeWebSockets.push(this); + } + + open(): void { + this.readyState = FakeWebSocket.OPEN; + this.dispatchEvent(new Event('open')); + } + + send(message: string): void { + this.sent.push(message); + } + + close(): void { + if (this.readyState === FakeWebSocket.CLOSED) { + return; + } + this.readyState = FakeWebSocket.CLOSED; + this.dispatchEvent(new Event('close')); + } + + fail(): void { + this.dispatchEvent(new Event('error')); + this.close(); + } +} + +const fakeWebSockets: FakeWebSocket[] = []; + +function latestWebSocket(): FakeWebSocket { + const ws = fakeWebSockets[fakeWebSockets.length - 1]; + if (!ws) { + throw new Error('Expected a WebSocket to be created'); + } + return ws; +} + +function testRequest() { + return { + jsonrpc: '2.0' as const, + id: 1, + method: 'test', + }; +} + +describe('createWebSocketStream', () => { + beforeEach(() => { + fakeWebSockets.length = 0; + vi.stubGlobal('WebSocket', FakeWebSocket); + }); + + afterEach(() => { + vi.unstubAllGlobals(); + }); + + it('waits for the socket to open before sending JSON', async () => { + const stream = createWebSocketStream('ws://localhost/acp'); + const writer = stream.writable.getWriter(); + const write = writer.write(testRequest()); + const ws = latestWebSocket(); + + expect(ws.sent).toEqual([]); + + ws.open(); + await write; + + expect(ws.sent).toEqual(['{"jsonrpc":"2.0","id":1,"method":"test"}']); + }); + + it('closes the readable stream when the socket closes', async () => { + const stream = createWebSocketStream('ws://localhost/acp'); + const reader = stream.readable.getReader(); + const ws = latestWebSocket(); + + ws.open(); + ws.close(); + + await expect(reader.read()).resolves.toEqual({ done: true, value: undefined }); + }); + + it.each([ + { + event: 'closes', + trigger: (ws: FakeWebSocket) => ws.close(), + error: 'ACP WebSocket closed before connection opened', + }, + { + event: 'errors', + trigger: (ws: FakeWebSocket) => ws.fail(), + error: 'ACP WebSocket connection failed', + }, + ])( + 'rejects a pending write when the socket $event before opening', + async ({ trigger, error }) => { + const stream = createWebSocketStream('ws://localhost/acp'); + const writer = stream.writable.getWriter(); + const write = writer.write(testRequest()); + + trigger(latestWebSocket()); + + await expect(write).rejects.toThrow(error); + } + ); + + it('rejects a write when the socket has closed', async () => { + const stream = createWebSocketStream('ws://localhost/acp'); + const writer = stream.writable.getWriter(); + const ws = latestWebSocket(); + + ws.open(); + ws.close(); + + await expect(writer.write(testRequest())).rejects.toThrow('ACP WebSocket connection lost'); + }); +}); diff --git a/ui/desktop/src/acp/acpConnection.ts b/ui/desktop/src/acp/acpConnection.ts index 1432955eb06b..8a88daff4b1e 100644 --- a/ui/desktop/src/acp/acpConnection.ts +++ b/ui/desktop/src/acp/acpConnection.ts @@ -5,6 +5,7 @@ import { } from '@aaif/goose-sdk'; import { PROTOCOL_VERSION, type InitializeResponse } from '@agentclientprotocol/sdk'; import packageJson from '../../package.json'; +import { GOOSE_SERVE_EXITED_USER_MESSAGE } from '../gooseServeLeaseRegistry'; import { handleAcpGooseSessionNotification, handleAcpSessionNotification, @@ -14,54 +15,115 @@ import { requestAcpElicitation } from './elicitationRequests'; import { requestAcpPermission } from './permissionRequests'; import { requestAcpRecipeParams } from './recipeParamRequests'; -type InitializedAcpClient = { +type AcpConnection = { client: GooseClient; + stream: ReturnType; initializeResponse: InitializeResponse; }; +type AcpRecoveryListener = (recovering: boolean) => void; + const ACP_INITIALIZE_TIMEOUT_MS = 10_000; +const ACP_RECONNECT_BASE_DELAY_MS = 500; +const ACP_RECONNECT_MAX_DELAY_MS = 30_000; -let clientPromise: Promise | null = null; -let resolvedClient: InitializedAcpClient | null = null; +let currentConnection: AcpConnection | null = null; +let pendingConnection: Promise | null = null; +let connectionGeneration = 0; +let recovering = false; +const recoveryListeners = new Set(); -function createClientCallbacks(): () => GooseClientCallbacks { - return () => ({ - requestPermission: requestAcpPermission, - unstable_createElicitation: requestAcpElicitation, - unstable_sessionRecipeRequestParams: requestAcpRecipeParams, - sessionUpdate: handleAcpSessionNotification, - unstable_sessionUpdate: handleAcpGooseSessionNotification, - }); +export async function getAcpClient(): Promise { + return (await getConnection()).client; } -function monitorConnection(client: GooseClient): void { - client.closed - .then(() => { - resolvedClient = null; - clientPromise = null; - }) - .catch(() => { - resolvedClient = null; - clientPromise = null; - }); +export async function getAcpInitializeResponse(): Promise { + return (await getConnection()).initializeResponse; } -async function withTimeout(promise: Promise, timeoutMs: number, message: string): Promise { - let timeoutId: ReturnType | null = null; - const timeout = new Promise((_, reject) => { - timeoutId = setTimeout(() => reject(new Error(message)), timeoutMs); - }); +export function reconnectAcpAfterSystemResume(): void { + recoverConnection(true); +} - try { - return await Promise.race([promise, timeout]); - } finally { - if (timeoutId !== null) { - clearTimeout(timeoutId); +export function isAcpRecovering(): boolean { + return recovering; +} + +export function subscribeToAcpRecovery(listener: AcpRecoveryListener): () => void { + recoveryListeners.add(listener); + return () => { + recoveryListeners.delete(listener); + }; +} + +function setRecovering(nextRecovering: boolean): void { + if (recovering === nextRecovering) { + return; + } + + recovering = nextRecovering; + for (const listener of recoveryListeners) { + listener(recovering); + } +} + +function recoverConnection(immediate: boolean): void { + if (!currentConnection && !pendingConnection) { + return; + } + + setRecovering(true); + const previousConnection = currentConnection; + connectionGeneration += 1; + currentConnection = null; + pendingConnection = null; + previousConnection?.stream.close(); + + const generation = connectionGeneration; + const recoveryAttempt = immediate + ? openConnection(generation).catch((error) => { + if (generation !== connectionGeneration || isGooseServeExitedError(error)) { + throw error; + } + return retryWithBackoff(generation); + }) + : retryWithBackoff(generation); + pendingConnection = recoveryAttempt; + void recoveryAttempt.then( + () => { + if (generation === connectionGeneration) { + setRecovering(false); + } + }, + () => { + if (generation === connectionGeneration) { + setRecovering(false); + } } + ); +} + +async function getConnection(): Promise { + if (currentConnection) { + return currentConnection; + } + + if (!pendingConnection) { + const generation = connectionGeneration; + let connectionAttempt: Promise; + connectionAttempt = openConnection(generation).catch((error) => { + if (pendingConnection === connectionAttempt) { + pendingConnection = null; + } + throw error; + }); + pendingConnection = connectionAttempt; } + + return pendingConnection; } -async function initializeConnection(): Promise { +async function openConnection(generation: number): Promise { const wsUrl = await window.electron.getAcpUrl(); if (!wsUrl) { throw new Error('ACP URL is not available'); @@ -96,46 +158,78 @@ async function initializeConnection(): Promise { `ACP initialize timed out after ${ACP_INITIALIZE_TIMEOUT_MS}ms` ); - monitorConnection(client); - return { client, initializeResponse }; + if (generation !== connectionGeneration) { + throw new Error('ACP connection attempt is no longer current'); + } + + const connection = { client, stream, initializeResponse }; + currentConnection = connection; + const handleClose = () => { + if (currentConnection === connection) { + recoverConnection(false); + } + }; + connection.client.closed.then(handleClose, handleClose); + return connection; } catch (error) { stream.close(); throw error; } } -export async function getAcpClient(): Promise { - return (await getInitializedAcpClient()).client; +async function retryWithBackoff(generation: number): Promise { + for (let attempt = 0; generation === connectionGeneration; attempt += 1) { + const maximumDelay = Math.min( + ACP_RECONNECT_MAX_DELAY_MS, + ACP_RECONNECT_BASE_DELAY_MS * 2 ** attempt + ); + await delay(Math.floor(Math.random() * maximumDelay)); + + if (generation !== connectionGeneration) { + break; + } + + try { + return await openConnection(generation); + } catch (error) { + if (generation !== connectionGeneration || isGooseServeExitedError(error)) { + throw error; + } + } + } + + throw new Error('ACP connection attempt is no longer current'); } -export function getAcpClientSync(): GooseClient | null { - return resolvedClient?.client ?? null; +function isGooseServeExitedError(error: unknown): boolean { + return error instanceof Error && error.message.includes(GOOSE_SERVE_EXITED_USER_MESSAGE); } -export async function getAcpInitializeResponse(): Promise { - return (await getInitializedAcpClient()).initializeResponse; +function delay(delayMs: number): Promise { + return new Promise((resolve) => setTimeout(resolve, delayMs)); } -export function isAcpClientReady(): boolean { - return resolvedClient !== null; +function createClientCallbacks(): () => GooseClientCallbacks { + return () => ({ + requestPermission: requestAcpPermission, + unstable_createElicitation: requestAcpElicitation, + unstable_sessionRecipeRequestParams: requestAcpRecipeParams, + sessionUpdate: handleAcpSessionNotification, + unstable_sessionUpdate: handleAcpGooseSessionNotification, + }); } -async function getInitializedAcpClient(): Promise { - if (resolvedClient) { - return resolvedClient; - } +async function withTimeout(promise: Promise, timeoutMs: number, message: string): Promise { + let timeoutId: ReturnType | null = null; + const timeout = new Promise((_, reject) => { + timeoutId = setTimeout(() => reject(new Error(message)), timeoutMs); + }); - if (!clientPromise) { - clientPromise = initializeConnection() - .then((clientState) => { - resolvedClient = clientState; - return clientState; - }) - .catch((error) => { - clientPromise = null; - throw error; - }); + try { + return await Promise.race([promise, timeout]); + } finally { + if (timeoutId !== null) { + clearTimeout(timeoutId); + } } - - return clientPromise; } diff --git a/ui/desktop/src/acp/chatSessionController.ts b/ui/desktop/src/acp/chatSessionController.ts index 849d6a9bbfc8..a5cc8fdb2b2d 100644 --- a/ui/desktop/src/acp/chatSessionController.ts +++ b/ui/desktop/src/acp/chatSessionController.ts @@ -5,11 +5,7 @@ import { ChatState } from '../types/chatState'; import type { Session } from '../types/session'; import { errorMessage } from '../utils/conversionUtils'; import { showExtensionLoadResults } from '../utils/extensionErrorUtils'; -import { - createUserMessage, - getPendingToolConfirmationIds, - type Message, -} from '../types/message'; +import { createUserMessage, getPendingToolConfirmationIds, type Message } from '../types/message'; import { acpChatSessionActions, acpChatSessionStore, @@ -48,6 +44,7 @@ export interface AcpChatSessionController { recipe?: AcpRecipeOptions ): Promise; loadSession(sessionId: string, options?: AcpLoadSessionOptions): Promise; + restoreSession(sessionId: string): Promise; submitMessage( sessionId: string, userMessage: Message, @@ -131,6 +128,17 @@ async function loadSession(sessionId: string, options: AcpLoadSessionOptions = { return; } + await loadSessionFromServer(sessionId, options); +} + +async function restoreSession(sessionId: string): Promise { + await loadSessionFromServer(sessionId); +} + +async function loadSessionFromServer( + sessionId: string, + options: AcpLoadSessionOptions = {} +): Promise { if (!isAcpSessionLoadInFlight(sessionId)) { acpChatSessionActions.startSessionLoad(sessionId); } @@ -316,6 +324,7 @@ async function updateMessage( export const acpChatSessionController: AcpChatSessionController = { createSession, loadSession, + restoreSession, submitMessage, stop, updateMessage, diff --git a/ui/desktop/src/acp/createWebSocketStream.ts b/ui/desktop/src/acp/createWebSocketStream.ts index 73d74481613a..9614ce928930 100644 --- a/ui/desktop/src/acp/createWebSocketStream.ts +++ b/ui/desktop/src/acp/createWebSocketStream.ts @@ -28,6 +28,11 @@ export function createWebSocketStream(wsUrl: string): ClosableAcpStream { ws.addEventListener('error', () => reject(new Error('ACP WebSocket connection failed')), { once: true, }); + ws.addEventListener( + 'close', + () => reject(new Error('ACP WebSocket closed before connection opened')), + { once: true } + ); }); ws.addEventListener('message', (event) => { @@ -67,6 +72,9 @@ export function createWebSocketStream(wsUrl: string): ClosableAcpStream { const writable = new window.WritableStream({ async write(message) { await openPromise; + if (closed || ws.readyState !== window.WebSocket.OPEN) { + throw new Error('ACP WebSocket connection lost'); + } ws.send(JSON.stringify(message)); }, close() { diff --git a/ui/desktop/src/components/BaseChat.tsx b/ui/desktop/src/components/BaseChat.tsx index d501f820c41f..5d6b09a042e9 100644 --- a/ui/desktop/src/components/BaseChat.tsx +++ b/ui/desktop/src/components/BaseChat.tsx @@ -29,6 +29,7 @@ import { useAutoSubmit } from '../hooks/useAutoSubmit'; import { Goose } from './icons'; import EnvironmentBadge from './GooseSidebar/EnvironmentBadge'; import SessionActionsHeader from './SessionActionsHeader'; +import { isAcpRecovering, subscribeToAcpRecovery } from '../acp/acpConnection'; const i18n = defineMessages({ failedToLoadSession: { @@ -39,6 +40,10 @@ const i18n = defineMessages({ id: 'baseChat.goHome', defaultMessage: 'Go home', }, + reconnecting: { + id: 'baseChat.reconnecting', + defaultMessage: 'Connection lost. Reconnecting…', + }, }); interface BaseChatProps { @@ -75,6 +80,7 @@ export default function BaseChat({ const [hasStartedUsingRecipe, setHasStartedUsingRecipe] = React.useState(false); const [hasNotAcceptedRecipe, setHasNotAcceptedRecipe] = useState(); const [hasRecipeSecurityWarnings, setHasRecipeSecurityWarnings] = useState(false); + const [acpRecovering, setAcpRecovering] = useState(isAcpRecovering); const isMobile = useIsMobile(); const navContext = useNavigationContextSafe(); const setView = useNavigation(); @@ -83,6 +89,8 @@ export default function BaseChat({ const { droppedFiles, setDroppedFiles, handleDrop, handleDragOver } = useFileDrop(); const onStreamFinish = useCallback(() => {}, []); + useEffect(() => subscribeToAcpRecovery(setAcpRecovering), []); + const { session, messages, @@ -133,6 +141,7 @@ export default function BaseChat({ // such as forks or resumes should auto-submit normally. const suppressInitialAutoSubmit = noAutoSubmit && messages.length === 0; const canAutoSubmit = + !acpRecovering && !suppressInitialAutoSubmit && (session?.session_type === 'scheduled' || !recipe || hasNotAcceptedRecipe === false); @@ -470,6 +479,12 @@ export default function BaseChat({ )} + {acpRecovering && ( +
+ {intl.formatMessage(i18n.reconnecting)} +
+ )} + ({ + useSearchParams: () => [new URLSearchParams('resumeSessionId=session-1')], +})); + +vi.mock('./BaseChat', () => ({ + default: ({ sessionId }: { sessionId: string }) =>
{sessionId}
, +})); + +vi.mock('../acp/acpConnection', () => ({ + subscribeToAcpRecovery: vi.fn(), +})); + +vi.mock('../acp/chatSessionController', () => ({ + acpChatSessionController: { + restoreSession: vi.fn().mockResolvedValue(undefined), + }, +})); + +describe('ChatSessionsContainer', () => { + beforeEach(() => { + vi.clearAllMocks(); + }); + + it('restores active chat sessions after ACP reconnects', () => { + let onRecoveryChanged: ((recovering: boolean) => void) | undefined; + vi.mocked(subscribeToAcpRecovery).mockImplementation((listener) => { + onRecoveryChanged = listener; + return () => undefined; + }); + + render( + + ); + + onRecoveryChanged?.(false); + + expect(acpChatSessionController.restoreSession).toHaveBeenCalledTimes(2); + expect(acpChatSessionController.restoreSession).toHaveBeenCalledWith('session-1'); + expect(acpChatSessionController.restoreSession).toHaveBeenCalledWith('session-2'); + }); +}); diff --git a/ui/desktop/src/components/ChatSessionsContainer.tsx b/ui/desktop/src/components/ChatSessionsContainer.tsx index b911539a62b8..0ec8b215978d 100644 --- a/ui/desktop/src/components/ChatSessionsContainer.tsx +++ b/ui/desktop/src/components/ChatSessionsContainer.tsx @@ -1,7 +1,10 @@ +import { useEffect, useRef } from 'react'; import { useSearchParams } from 'react-router-dom'; import BaseChat from './BaseChat'; import { ChatType } from '../types/chat'; import { UserInput } from '../types/message'; +import { subscribeToAcpRecovery } from '../acp/acpConnection'; +import { acpChatSessionController } from '../acp/chatSessionController'; interface ChatSessionsContainerProps { setChat: (chat: ChatType) => void; @@ -24,11 +27,6 @@ export default function ChatSessionsContainer({ const [searchParams] = useSearchParams(); const currentSessionId = searchParams.get('resumeSessionId') ?? undefined; - // Always render active sessions to keep SSE connections alive, even when not on /pair route - if (!currentSessionId && activeSessions.length === 0) { - return null; - } - // Build the list of sessions to render let sessionsToRender = activeSessions; @@ -37,6 +35,25 @@ export default function ChatSessionsContainer({ sessionsToRender = [...activeSessions, { sessionId: currentSessionId }]; } + const sessionIdsRef = useRef([]); + sessionIdsRef.current = sessionsToRender.map((session) => session.sessionId); + + useEffect(() => { + return subscribeToAcpRecovery((recovering) => { + if (recovering) { + return; + } + for (const sessionId of sessionIdsRef.current) { + void acpChatSessionController.restoreSession(sessionId); + } + }); + }, []); + + // Always render active sessions to keep SSE connections alive, even when not on /pair route + if (!currentSessionId && activeSessions.length === 0) { + return null; + } + return (
{sessionsToRender.map((session) => { diff --git a/ui/desktop/src/hooks/useAutoSubmit.test.tsx b/ui/desktop/src/hooks/useAutoSubmit.test.tsx index 324729f94ed4..555f7aee771e 100644 --- a/ui/desktop/src/hooks/useAutoSubmit.test.tsx +++ b/ui/desktop/src/hooks/useAutoSubmit.test.tsx @@ -56,7 +56,7 @@ describe('useAutoSubmit', () => { expect(dispatchEventSpy).not.toHaveBeenCalled(); }); - it('auto-submits once recipe acceptance is confirmed', () => { + it('keeps the initial message while blocked and submits it once unblocked', () => { const handleSubmit = vi.fn(); const dispatchEventSpy = vi.spyOn(window, 'dispatchEvent'); @@ -82,6 +82,7 @@ describe('useAutoSubmit', () => { ); expect(handleSubmit).not.toHaveBeenCalled(); + expect(dispatchEventSpy).not.toHaveBeenCalled(); rerender({ canAutoSubmit: true }); diff --git a/ui/desktop/src/hooks/useChatSession.ts b/ui/desktop/src/hooks/useChatSession.ts index 4b7ec02a7fff..0d32fa80bce8 100644 --- a/ui/desktop/src/hooks/useChatSession.ts +++ b/ui/desktop/src/hooks/useChatSession.ts @@ -22,6 +22,7 @@ import { useAcpChatSessionSnapshot, } from '../acp/chatSessionStore'; import { acpSteerSession } from '../acp/prompt'; +import { isAcpRecovering } from '../acp/acpConnection'; const initialTokenState: TokenState = { inputTokens: 0, @@ -150,6 +151,10 @@ export function useChatSession({ const handleSubmit = useCallback( async (input: UserInput) => { + if (isAcpRecovering()) { + return; + } + const { msg: userMessage, images } = input; const currentSnapshot = getCurrentSnapshot(); diff --git a/ui/desktop/src/i18n/messages/de.json b/ui/desktop/src/i18n/messages/de.json index 598875e21ed2..ac830aee75df 100644 --- a/ui/desktop/src/i18n/messages/de.json +++ b/ui/desktop/src/i18n/messages/de.json @@ -131,6 +131,9 @@ "baseChat.goHome": { "defaultMessage": "Zur Startseite" }, + "baseChat.reconnecting": { + "defaultMessage": "Verbindung unterbrochen. Verbindung wird wiederhergestellt…" + }, "bottomMenuExtensionSelection.extensionUpdated": { "defaultMessage": "Erweiterung aktualisiert" }, diff --git a/ui/desktop/src/i18n/messages/en.json b/ui/desktop/src/i18n/messages/en.json index 16996a89326f..b44b61ad9d99 100644 --- a/ui/desktop/src/i18n/messages/en.json +++ b/ui/desktop/src/i18n/messages/en.json @@ -131,6 +131,9 @@ "baseChat.goHome": { "defaultMessage": "Go home" }, + "baseChat.reconnecting": { + "defaultMessage": "Connection lost. Reconnecting…" + }, "bottomMenuExtensionSelection.extensionUpdated": { "defaultMessage": "Extension Updated" }, diff --git a/ui/desktop/src/i18n/messages/es.json b/ui/desktop/src/i18n/messages/es.json index 1d81c66324bb..682ced012729 100644 --- a/ui/desktop/src/i18n/messages/es.json +++ b/ui/desktop/src/i18n/messages/es.json @@ -131,6 +131,9 @@ "baseChat.goHome": { "defaultMessage": "Ir al inicio" }, + "baseChat.reconnecting": { + "defaultMessage": "Conexión perdida. Reconectando…" + }, "bottomMenuExtensionSelection.extensionUpdated": { "defaultMessage": "Extensión actualizada" }, diff --git a/ui/desktop/src/i18n/messages/fr.json b/ui/desktop/src/i18n/messages/fr.json index 86d38081e4d6..ce067a71ef4d 100644 --- a/ui/desktop/src/i18n/messages/fr.json +++ b/ui/desktop/src/i18n/messages/fr.json @@ -131,6 +131,9 @@ "baseChat.goHome": { "defaultMessage": "Aller à l'accueil" }, + "baseChat.reconnecting": { + "defaultMessage": "Connexion perdue. Reconnexion…" + }, "bottomMenuExtensionSelection.extensionUpdated": { "defaultMessage": "Extension mise à jour" }, diff --git a/ui/desktop/src/i18n/messages/hi.json b/ui/desktop/src/i18n/messages/hi.json index 17bd4a4d68a5..9efb28104fa9 100644 --- a/ui/desktop/src/i18n/messages/hi.json +++ b/ui/desktop/src/i18n/messages/hi.json @@ -131,6 +131,9 @@ "baseChat.goHome": { "defaultMessage": "घर जाओ" }, + "baseChat.reconnecting": { + "defaultMessage": "कनेक्शन टूट गया। फिर से कनेक्ट किया जा रहा है…" + }, "bottomMenuExtensionSelection.extensionUpdated": { "defaultMessage": "एक्सटेंशन अपडेट किया गया" }, diff --git a/ui/desktop/src/i18n/messages/id.json b/ui/desktop/src/i18n/messages/id.json index 576e116fa8a7..3f366765a2f2 100644 --- a/ui/desktop/src/i18n/messages/id.json +++ b/ui/desktop/src/i18n/messages/id.json @@ -131,6 +131,9 @@ "baseChat.goHome": { "defaultMessage": "Ke beranda" }, + "baseChat.reconnecting": { + "defaultMessage": "Koneksi terputus. Menghubungkan kembali…" + }, "bottomMenuExtensionSelection.extensionUpdated": { "defaultMessage": "Ekstensi Diperbarui" }, diff --git a/ui/desktop/src/i18n/messages/it.json b/ui/desktop/src/i18n/messages/it.json index 900544eedb7e..60738d9ce340 100644 --- a/ui/desktop/src/i18n/messages/it.json +++ b/ui/desktop/src/i18n/messages/it.json @@ -131,6 +131,9 @@ "baseChat.goHome": { "defaultMessage": "Vai alla home" }, + "baseChat.reconnecting": { + "defaultMessage": "Connessione persa. Riconnessione…" + }, "bottomMenuExtensionSelection.extensionUpdated": { "defaultMessage": "Estensione aggiornata" }, diff --git a/ui/desktop/src/i18n/messages/ja.json b/ui/desktop/src/i18n/messages/ja.json index d17708a5818b..3ac77bffb208 100644 --- a/ui/desktop/src/i18n/messages/ja.json +++ b/ui/desktop/src/i18n/messages/ja.json @@ -131,6 +131,9 @@ "baseChat.goHome": { "defaultMessage": "ホームへ" }, + "baseChat.reconnecting": { + "defaultMessage": "接続が失われました。再接続しています…" + }, "bottomMenuExtensionSelection.extensionUpdated": { "defaultMessage": "拡張機能を更新しました" }, diff --git a/ui/desktop/src/i18n/messages/ko.json b/ui/desktop/src/i18n/messages/ko.json index 7a1d439631bf..b35dfe0127bd 100644 --- a/ui/desktop/src/i18n/messages/ko.json +++ b/ui/desktop/src/i18n/messages/ko.json @@ -131,6 +131,9 @@ "baseChat.goHome": { "defaultMessage": "홈으로 이동" }, + "baseChat.reconnecting": { + "defaultMessage": "연결이 끊어졌습니다. 다시 연결하는 중…" + }, "bottomMenuExtensionSelection.extensionUpdated": { "defaultMessage": "익스텐션이 업데이트되었습니다." }, diff --git a/ui/desktop/src/i18n/messages/ms.json b/ui/desktop/src/i18n/messages/ms.json index 4cb5853a17a6..a68ff81e76d5 100644 --- a/ui/desktop/src/i18n/messages/ms.json +++ b/ui/desktop/src/i18n/messages/ms.json @@ -131,6 +131,9 @@ "baseChat.goHome": { "defaultMessage": "Pergi ke laman utama" }, + "baseChat.reconnecting": { + "defaultMessage": "Sambungan terputus. Menyambung semula…" + }, "bottomMenuExtensionSelection.extensionUpdated": { "defaultMessage": "Sambungan Dikemas Kini" }, diff --git a/ui/desktop/src/i18n/messages/pt.json b/ui/desktop/src/i18n/messages/pt.json index e5f5735bcb0e..f5bf0ef24a8d 100644 --- a/ui/desktop/src/i18n/messages/pt.json +++ b/ui/desktop/src/i18n/messages/pt.json @@ -131,6 +131,9 @@ "baseChat.goHome": { "defaultMessage": "Ir para o início" }, + "baseChat.reconnecting": { + "defaultMessage": "Conexão perdida. Reconectando…" + }, "bottomMenuExtensionSelection.extensionUpdated": { "defaultMessage": "Extensão atualizada" }, diff --git a/ui/desktop/src/i18n/messages/ru.json b/ui/desktop/src/i18n/messages/ru.json index 5c1fd7206895..b36fcb2744b1 100644 --- a/ui/desktop/src/i18n/messages/ru.json +++ b/ui/desktop/src/i18n/messages/ru.json @@ -131,6 +131,9 @@ "baseChat.goHome": { "defaultMessage": "На главную" }, + "baseChat.reconnecting": { + "defaultMessage": "Соединение потеряно. Повторное подключение…" + }, "bottomMenuExtensionSelection.extensionUpdated": { "defaultMessage": "Расширение обновлено" }, diff --git a/ui/desktop/src/i18n/messages/tr.json b/ui/desktop/src/i18n/messages/tr.json index e131a99e65a2..3a93602bc648 100644 --- a/ui/desktop/src/i18n/messages/tr.json +++ b/ui/desktop/src/i18n/messages/tr.json @@ -131,6 +131,9 @@ "baseChat.goHome": { "defaultMessage": "Ana sayfaya git" }, + "baseChat.reconnecting": { + "defaultMessage": "Bağlantı kesildi. Yeniden bağlanılıyor…" + }, "bottomMenuExtensionSelection.extensionUpdated": { "defaultMessage": "Uzantı Güncellendi" }, diff --git a/ui/desktop/src/i18n/messages/vi.json b/ui/desktop/src/i18n/messages/vi.json index 86e6a26b7d66..4fcdfb1cf3c3 100644 --- a/ui/desktop/src/i18n/messages/vi.json +++ b/ui/desktop/src/i18n/messages/vi.json @@ -131,6 +131,9 @@ "baseChat.goHome": { "defaultMessage": "Về trang chủ" }, + "baseChat.reconnecting": { + "defaultMessage": "Mất kết nối. Đang kết nối lại…" + }, "bottomMenuExtensionSelection.extensionUpdated": { "defaultMessage": "Đã cập nhật tiện ích mở rộng" }, diff --git a/ui/desktop/src/i18n/messages/zh-CN.json b/ui/desktop/src/i18n/messages/zh-CN.json index 2c940a376871..8b2801e8bc11 100644 --- a/ui/desktop/src/i18n/messages/zh-CN.json +++ b/ui/desktop/src/i18n/messages/zh-CN.json @@ -131,6 +131,9 @@ "baseChat.goHome": { "defaultMessage": "回到首页" }, + "baseChat.reconnecting": { + "defaultMessage": "连接已断开。正在重新连接…" + }, "bottomMenuExtensionSelection.extensionUpdated": { "defaultMessage": "扩展已更新" }, diff --git a/ui/desktop/src/i18n/messages/zh-TW.json b/ui/desktop/src/i18n/messages/zh-TW.json index 01b50e6bab89..60a1a5ed6934 100644 --- a/ui/desktop/src/i18n/messages/zh-TW.json +++ b/ui/desktop/src/i18n/messages/zh-TW.json @@ -131,6 +131,9 @@ "baseChat.goHome": { "defaultMessage": "回到首頁" }, + "baseChat.reconnecting": { + "defaultMessage": "連線已中斷。正在重新連線…" + }, "bottomMenuExtensionSelection.extensionUpdated": { "defaultMessage": "擴充功能已更新" }, diff --git a/ui/desktop/src/main.ts b/ui/desktop/src/main.ts index dd938be40c17..d15ee8013e66 100644 --- a/ui/desktop/src/main.ts +++ b/ui/desktop/src/main.ts @@ -10,6 +10,7 @@ import { MenuItem, net, Notification, + powerMonitor, powerSaveBlocker, screen, session, @@ -2408,6 +2409,14 @@ const registerGlobalShortcuts = () => { }; async function appMain() { + powerMonitor.on('resume', () => { + for (const window of BrowserWindow.getAllWindows()) { + if (!window.isDestroyed()) { + window.webContents.send('system-resume'); + } + } + }); + await configureProxy(); // Ensure Windows shims are available before any MCP processes are spawned