diff --git a/apps/desktop/src/app/desktop-controller.tsx b/apps/desktop/src/app/desktop-controller.tsx index 8aedb6b4608f..0f3365f66046 100644 --- a/apps/desktop/src/app/desktop-controller.tsx +++ b/apps/desktop/src/app/desktop-controller.tsx @@ -66,6 +66,7 @@ import { PersistentTerminal, TerminalSlot } from './right-sidebar/terminal/persi import { NEW_CHAT_ROUTE, routeSessionId, sessionRoute, SETTINGS_ROUTE } from './routes' import { useContextSuggestions } from './session/hooks/use-context-suggestions' import { useCwdActions } from './session/hooks/use-cwd-actions' +import { useExternalSessionRefresh } from './session/hooks/use-external-session-refresh' import { useHermesConfig } from './session/hooks/use-hermes-config' import { useMessageStream } from './session/hooks/use-message-stream' import { useModelControls } from './session/hooks/use-model-controls' @@ -491,6 +492,15 @@ export function DesktopController() { } }, [gatewayState, refreshCurrentModel, refreshSessions]) + useExternalSessionRefresh({ + activeSessionId, + gatewayState, + getRuntimeState: runtimeSessionId => sessionStateByRuntimeIdRef.current.get(runtimeSessionId), + refreshSessions, + selectedStoredSessionId, + updateSessionState + }) + useRouteResume({ activeSessionId, activeSessionIdRef, diff --git a/apps/desktop/src/app/session/hooks/use-external-session-refresh.test.ts b/apps/desktop/src/app/session/hooks/use-external-session-refresh.test.ts new file mode 100644 index 000000000000..8f8c7227113e --- /dev/null +++ b/apps/desktop/src/app/session/hooks/use-external-session-refresh.test.ts @@ -0,0 +1,212 @@ +import { describe, expect, it, vi } from 'vitest' + +import { type ChatMessage, textPart } from '@/lib/chat-messages' +import type { SessionMessage, SessionMessagesResponse } from '@/types/hermes' + +import type { ClientSessionState } from '../../types' + +import { refreshExternalSessionSnapshot } from './use-external-session-refresh' + +const remoteMessages = (messages: SessionMessage[]): SessionMessagesResponse => ({ + messages, + session_id: 'stored-1' +}) + +const clientState = (messages: ChatMessage[], extra: Partial = {}): ClientSessionState => + ({ + awaitingResponse: false, + branch: '', + busy: false, + cwd: '', + interrupted: false, + messages, + pendingBranchGroup: null, + sawAssistantPayload: false, + storedSessionId: 'stored-1', + streamId: null, + ...extra + }) as ClientSessionState + +describe('refreshExternalSessionSnapshot', () => { + it('hydrates the active desktop session when stored messages change outside the app', async () => { + const local = clientState([ + { + id: 'local-user-1', + parts: [textPart('old')], + role: 'user', + timestamp: 1 + } + ]) + + const getRuntimeState = vi.fn(() => local) + + const getSessionMessages = vi.fn(async () => + remoteMessages([ + { content: 'old', role: 'user', timestamp: 1 }, + { content: 'phone update', role: 'user', timestamp: 2 } + ]) + ) + + const refreshSessions = vi.fn(async () => undefined) + + const updateSessionState = vi.fn((sessionId, updater, storedSessionId) => { + const next = updater(local) + + return { sessionId, state: next, storedSessionId } + }) + + const refreshed = await refreshExternalSessionSnapshot({ + getRuntimeState, + getSessionMessages, + lastRemoteRevision: new Map(), + refreshSessions, + runtimeSessionId: 'runtime-1', + storedSessionId: 'stored-1', + updateSessionState + }) + + expect(refreshed).toBe(true) + expect(updateSessionState).toHaveBeenCalledTimes(1) + + const [runtimeId, updater, storedId] = updateSessionState.mock.calls[0] as [ + string, + (state: ClientSessionState) => ClientSessionState, + string + ] + + const hydrated = updater(local) + expect(runtimeId).toBe('runtime-1') + expect(storedId).toBe('stored-1') + expect(hydrated.messages.map((message: ChatMessage) => message.role)).toEqual(['user', 'user']) + expect(hydrated.messages.at(-1)?.parts).toEqual([textPart('phone update')]) + expect(refreshSessions).toHaveBeenCalledTimes(1) + }) + + it('does not rehydrate unchanged revisions on each poll tick', async () => { + const local = clientState([ + { + id: 'local-user-1', + parts: [textPart('same')], + role: 'user', + timestamp: 1 + } + ]) + + const lastRemoteRevision = new Map([['stored-1', '1:user:1:same']]) + const getSessionMessages = vi.fn(async () => remoteMessages([{ content: 'same', role: 'user', timestamp: 1 }])) + + const updateSessionState = vi.fn((sessionId, updater, storedSessionId) => ({ + sessionId, + state: updater(local), + storedSessionId + })) + + const refreshed = await refreshExternalSessionSnapshot({ + getRuntimeState: () => local, + getSessionMessages, + lastRemoteRevision, + refreshSessions: vi.fn(async () => undefined), + runtimeSessionId: 'runtime-1', + storedSessionId: 'stored-1', + updateSessionState + }) + + expect(refreshed).toBe(false) + expect(getSessionMessages).toHaveBeenCalled() + expect(updateSessionState).not.toHaveBeenCalled() + }) + + it('does not repeatedly rehydrate when the remote revision is unchanged after an initial repair', async () => { + const local = clientState([]) + const lastRemoteRevision = new Map() + const getSessionMessages = vi.fn(async () => remoteMessages([{ content: 'phone update', role: 'user', timestamp: 2 }])) + const refreshSessions = vi.fn(async () => undefined) + + const updateSessionState = vi.fn((sessionId, updater, storedSessionId) => ({ + sessionId, + state: updater(local), + storedSessionId + })) + + const options = { + getRuntimeState: () => local, + getSessionMessages, + lastRemoteRevision, + refreshSessions, + runtimeSessionId: 'runtime-1', + storedSessionId: 'stored-1', + updateSessionState + } + + await expect(refreshExternalSessionSnapshot(options)).resolves.toBe(true) + await expect(refreshExternalSessionSnapshot(options)).resolves.toBe(false) + + expect(updateSessionState).toHaveBeenCalledTimes(1) + expect(refreshSessions).toHaveBeenCalledTimes(1) + }) + + it('does not overwrite a session that becomes busy while refresh is in flight', async () => { + const idle = clientState([ + { + id: 'local-user-1', + parts: [textPart('old')], + role: 'user', + timestamp: 1 + } + ]) + + const busy = clientState([ + ...idle.messages, + { + id: 'optimistic-user-2', + parts: [textPart('new prompt')], + role: 'user', + timestamp: 2 + } + ], { awaitingResponse: true }) + + let runtimeState = idle + + const getSessionMessages = vi.fn(async () => { + runtimeState = busy + + return remoteMessages([ + { content: 'old', role: 'user', timestamp: 1 }, + { content: 'phone update', role: 'user', timestamp: 2 } + ]) + }) + + const updateSessionState = vi.fn() + + const refreshed = await refreshExternalSessionSnapshot({ + getRuntimeState: () => runtimeState, + getSessionMessages, + lastRemoteRevision: new Map(), + refreshSessions: vi.fn(async () => undefined), + runtimeSessionId: 'runtime-1', + storedSessionId: 'stored-1', + updateSessionState + }) + + expect(refreshed).toBe(false) + expect(updateSessionState).not.toHaveBeenCalled() + }) + + it('skips polling while the active session is busy', async () => { + const local = clientState([], { busy: true }) + const getSessionMessages = vi.fn(async () => remoteMessages([{ content: 'busy update', role: 'user', timestamp: 1 }])) + + const refreshed = await refreshExternalSessionSnapshot({ + getRuntimeState: () => local, + getSessionMessages, + lastRemoteRevision: new Map(), + refreshSessions: vi.fn(async () => undefined), + runtimeSessionId: 'runtime-1', + storedSessionId: 'stored-1', + updateSessionState: vi.fn() + }) + + expect(refreshed).toBe(false) + expect(getSessionMessages).not.toHaveBeenCalled() + }) +}) diff --git a/apps/desktop/src/app/session/hooks/use-external-session-refresh.ts b/apps/desktop/src/app/session/hooks/use-external-session-refresh.ts new file mode 100644 index 000000000000..ae6eac16d8f5 --- /dev/null +++ b/apps/desktop/src/app/session/hooks/use-external-session-refresh.ts @@ -0,0 +1,193 @@ +import { useEffect, useRef } from 'react' + +import { getSessionMessages as fetchSessionMessages } from '@/hermes' +import { + type ChatMessage, + chatMessageText, + preserveLocalAssistantErrors, + toChatMessages +} from '@/lib/chat-messages' +import type { SessionMessage, SessionMessagesResponse } from '@/types/hermes' + +import type { ClientSessionState } from '../../types' + +const DEFAULT_EXTERNAL_SESSION_REFRESH_MS = 4_000 +const MIN_EXTERNAL_SESSION_REFRESH_MS = 1_000 + +interface ExternalSessionRefreshCallbacks { + getRuntimeState: (runtimeSessionId: string) => ClientSessionState | undefined + getSessionMessages: (storedSessionId: string) => Promise + refreshSessions: () => Promise + updateSessionState: ( + sessionId: string, + updater: (state: ClientSessionState) => ClientSessionState, + storedSessionId?: null | string + ) => ClientSessionState | unknown +} + +interface ExternalSessionRefreshOptions extends Omit { + activeSessionId: null | string + gatewayState: string + getSessionMessages?: (storedSessionId: string) => Promise + pollMs?: number + selectedStoredSessionId: null | string +} + +interface RefreshExternalSessionSnapshotOptions extends ExternalSessionRefreshCallbacks { + lastRemoteRevision: Map + runtimeSessionId: string + storedSessionId: string +} + +function safeContent(value: unknown): string { + if (typeof value === 'string') { + return value + } + + if (value === null || value === undefined) { + return '' + } + + try { + return JSON.stringify(value) + } catch { + return String(value) + } +} + +export function sessionMessageRevision(messages: readonly SessionMessage[]): string { + const last = messages.at(-1) + + if (!last) { + return '0' + } + + return [messages.length, last.role, last.timestamp ?? '', safeContent(last.content ?? last.text)].join(':') +} + +export function chatMessageRevision(messages: readonly ChatMessage[]): string { + const last = messages.at(-1) + + if (!last) { + return '0' + } + + return [messages.length, last.role, last.timestamp ?? '', chatMessageText(last)].join(':') +} + +export async function refreshExternalSessionSnapshot({ + getRuntimeState, + getSessionMessages, + lastRemoteRevision, + refreshSessions, + runtimeSessionId, + storedSessionId, + updateSessionState +}: RefreshExternalSessionSnapshotOptions): Promise { + const state = getRuntimeState(runtimeSessionId) + + // Local sends already drive their own live stream. Do not fight the active + // run by replacing messages with the persisted snapshot mid-turn. + if (state?.busy || state?.awaitingResponse) { + return false + } + + const latest = await getSessionMessages(storedSessionId) + const stateAfterRead = getRuntimeState(runtimeSessionId) + + if (stateAfterRead?.busy || stateAfterRead?.awaitingResponse) { + return false + } + + const remoteRevision = sessionMessageRevision(latest.messages) + const localRevision = stateAfterRead ? chatMessageRevision(stateAfterRead.messages) : null + const previousRemoteRevision = lastRemoteRevision.get(storedSessionId) + const firstSeen = previousRemoteRevision === undefined + + lastRemoteRevision.set(storedSessionId, remoteRevision) + + if (firstSeen ? localRevision === remoteRevision : previousRemoteRevision === remoteRevision) { + return false + } + + updateSessionState( + runtimeSessionId, + current => ({ + ...current, + messages: preserveLocalAssistantErrors(toChatMessages(latest.messages), current.messages) + }), + storedSessionId + ) + await refreshSessions().catch(() => undefined) + + return true +} + +export function useExternalSessionRefresh({ + activeSessionId, + gatewayState, + getRuntimeState, + getSessionMessages = fetchSessionMessages, + pollMs = DEFAULT_EXTERNAL_SESSION_REFRESH_MS, + refreshSessions, + selectedStoredSessionId, + updateSessionState +}: ExternalSessionRefreshOptions) { + const callbacksRef = useRef({ getRuntimeState, getSessionMessages, refreshSessions, updateSessionState }) + const inFlightRef = useRef(false) + const lastRemoteRevisionRef = useRef(new Map()) + + callbacksRef.current = { getRuntimeState, getSessionMessages, refreshSessions, updateSessionState } + + useEffect(() => { + if (gatewayState !== 'open' || !activeSessionId || !selectedStoredSessionId) { + return undefined + } + + let cancelled = false + const intervalMs = Math.max(MIN_EXTERNAL_SESSION_REFRESH_MS, pollMs) + + const poll = async () => { + if (inFlightRef.current) { + return + } + + inFlightRef.current = true + + try { + await refreshExternalSessionSnapshot({ + ...callbacksRef.current, + getSessionMessages: async storedSessionId => { + const latest = await callbacksRef.current.getSessionMessages(storedSessionId) + + if (cancelled) { + throw new Error('external session refresh cancelled') + } + + return latest + }, + lastRemoteRevision: lastRemoteRevisionRef.current, + runtimeSessionId: activeSessionId, + storedSessionId: selectedStoredSessionId + }) + + if (cancelled) { + return + } + } catch { + // Best-effort read repair. The normal gateway stream and manual Cmd+R + // remain available if a transient API read fails. + } finally { + inFlightRef.current = false + } + } + + void poll() + const handle = window.setInterval(() => void poll(), intervalMs) + + return () => { + cancelled = true + window.clearInterval(handle) + } + }, [activeSessionId, gatewayState, pollMs, selectedStoredSessionId]) +}