diff --git a/src/main/ipc.test.ts b/src/main/ipc.test.ts index 786fa1b8..c15c3628 100644 --- a/src/main/ipc.test.ts +++ b/src/main/ipc.test.ts @@ -6229,6 +6229,20 @@ describe('registerIpcHandlers agent terminal state', () => { }) ) ) + // The renderer converges on the persisted terminal state by re-reading + // conversations, so the change notice must land before `completed`. + const sendCalls = harness.webContents.send.mock.calls + const completedIndex = sendCalls.findIndex( + ([channel, payload]) => + channel === ipcChannels.remoteProjectRecoveryProgress && + (payload as { stage?: string }).stage === 'completed' + ) + expect(completedIndex).toBeGreaterThan(0) + expect( + sendCalls + .slice(0, completedIndex) + .some(([channel]) => channel === ipcChannels.conversationsChanged) + ).toBe(true) await harness.dispose() }) diff --git a/src/main/ipc.ts b/src/main/ipc.ts index 46f618d5..6708a2bc 100644 --- a/src/main/ipc.ts +++ b/src/main/ipc.ts @@ -1935,12 +1935,16 @@ export function registerIpcHandlers( for (const task of tasks) { await recoverRemoteTask(task, requestId) } + // Recovery may terminalize the task without a trailing checkpoint; + // let the renderer converge on the persisted terminal state. + publishConversationChange() publishRemoteProjectRecovery({ projectId, requestId, stage: 'completed' }) } catch (error) { + publishConversationChange() publishRemoteProjectRecovery({ projectId, requestId, diff --git a/src/renderer/src/App.test.tsx b/src/renderer/src/App.test.tsx index b9ee99bf..14e2ce87 100644 --- a/src/renderer/src/App.test.tsx +++ b/src/renderer/src/App.test.tsx @@ -6466,6 +6466,236 @@ describe('App', () => { ) }) + describe('persisted terminal state convergence', () => { + let conversationsChangedListener: (() => void) | undefined + + const startStreamingRun = async (): Promise<{ + requestId: string + conversationId: string + assistantMessageId: string + userMessageId: string + }> => { + conversationsChangedListener = undefined + vi.mocked(api.conversations.onChanged).mockImplementation( + (listener) => { + conversationsChangedListener = listener + return () => { + if (conversationsChangedListener === listener) { + conversationsChangedListener = undefined + } + } + } + ) + render() + fireEvent.change(screen.getByLabelText('向 GoodBuddy 提问'), { + target: { value: '断网期间继续执行' } + }) + await waitFor(() => + expect(screen.getByLabelText('发送')).toBeEnabled() + ) + fireEvent.click(screen.getByLabelText('发送')) + await waitFor(() => expect(run).toHaveBeenCalledOnce()) + const request = run.mock.calls[0]?.[0] + if ( + !request?.currentAssistantMessageId || + !request.currentUserMessageId + ) { + throw new Error('Missing request') + } + expect( + screen + .getByText('正在连接 Agent Runtime') + .querySelector('.message__status-dot') + ).toHaveClass('message__status-dot--active') + expect(screen.getByLabelText('加入待发送队列')).toBeInTheDocument() + vi.mocked(api.conversationQueue.ready).mockClear() + return { + requestId: request.requestId, + conversationId: request.conversationId, + assistantMessageId: request.currentAssistantMessageId, + userMessageId: request.currentUserMessageId + } + } + + const persistedSnapshot = ( + input: { + conversationId: string + userMessageId: string + assistantMessageId: string + }, + assistant: { + state: 'streaming' | 'complete' | 'error' + content: string + status?: string + } + ): ConversationSnapshot => ({ + id: input.conversationId, + projectId, + runtimeSelection: { provider: 'model', profileId: modelProfileId }, + title: '断网期间继续执行', + updatedAt: 1_000, + messages: [ + { + id: input.userMessageId, + role: 'user', + content: '断网期间继续执行', + createdAt: 900, + state: 'complete' + }, + { + id: input.assistantMessageId, + role: 'assistant', + content: assistant.content, + createdAt: 950, + state: assistant.state, + status: assistant.status + } + ] + }) + + it('converges a streaming message onto the persisted completed state after reconnecting', async () => { + const started = await startStreamingRun() + expect(conversationsChangedListener).toBeDefined() + + vi.mocked(api.conversations.list).mockResolvedValue([ + persistedSnapshot(started, { + state: 'complete', + content: '远端在断线期间已经完成的回复', + status: '任务已完成' + }) + ]) + act(() => { + conversationsChangedListener?.() + }) + + expect( + await screen.findByText('远端在断线期间已经完成的回复') + ).toBeInTheDocument() + await waitFor(() => + expect( + screen.queryByText('正在连接 Agent Runtime') + ).not.toBeInTheDocument() + ) + expect( + document.querySelector('.message__status-dot--active') + ).toBeNull() + expect(screen.getByLabelText('发送')).toBeInTheDocument() + await waitFor(() => + expect(api.conversationQueue.ready).toHaveBeenCalledWith( + started.conversationId + ) + ) + + // A late terminal event for the settled run must be ignored rather + // than resurrecting the message or double-releasing the queue. + vi.mocked(api.conversationQueue.ready).mockClear() + act(() => { + agentListener?.({ + requestId: started.requestId, + type: 'error', + status: 'failed', + message: '迟到的连接中断错误' + }) + }) + expect( + screen.queryByText('迟到的连接中断错误') + ).not.toBeInTheDocument() + expect( + screen.getByText('远端在断线期间已经完成的回复') + ).toBeInTheDocument() + expect(api.conversationQueue.ready).not.toHaveBeenCalled() + }) + + it('converges a streaming message onto the persisted failed state after reconnecting', async () => { + const started = await startStreamingRun() + + vi.mocked(api.conversations.list).mockResolvedValue([ + persistedSnapshot(started, { + state: 'error', + content: '', + status: '远端 Agent 执行失败' + }) + ]) + act(() => { + conversationsChangedListener?.() + }) + + expect( + await screen.findByText('远端 Agent 执行失败') + ).toBeInTheDocument() + expect( + screen.queryByText('正在连接 Agent Runtime') + ).not.toBeInTheDocument() + expect( + document.querySelector('.message__status-dot--active') + ).toBeNull() + await waitFor(() => + expect(screen.getByLabelText('发送')).toBeInTheDocument() + ) + await waitFor(() => + expect(api.conversationQueue.ready).toHaveBeenCalledWith( + started.conversationId + ) + ) + }) + + it('keeps the run in flight while the persisted message is still streaming and refreshes on focus', async () => { + const started = await startStreamingRun() + + vi.mocked(api.conversations.list).mockResolvedValue([ + persistedSnapshot(started, { + state: 'streaming', + content: '', + status: '远端 Agent 正在处理请求' + }) + ]) + act(() => { + conversationsChangedListener?.() + }) + await waitFor(() => + expect(api.conversations.list).toHaveBeenCalledTimes(2) + ) + // The local streaming copy still wins over a non-terminal snapshot. + expect( + screen.getByText('正在连接 Agent Runtime') + ).toBeInTheDocument() + expect(screen.getByLabelText('加入待发送队列')).toBeInTheDocument() + expect(api.conversationQueue.ready).not.toHaveBeenCalled() + + // Regaining focus re-reads the store and converges once the remote + // terminal state has landed, even without another change event. + vi.mocked(api.conversations.list).mockResolvedValue([ + persistedSnapshot(started, { + state: 'complete', + content: '聚焦后读到的最终回复', + status: '任务已完成' + }) + ]) + act(() => { + window.dispatchEvent(new Event('focus')) + }) + expect( + await screen.findByText('聚焦后读到的最终回复') + ).toBeInTheDocument() + expect( + screen.queryByText('正在连接 Agent Runtime') + ).not.toBeInTheDocument() + await waitFor(() => + expect(api.conversationQueue.ready).toHaveBeenCalledWith( + started.conversationId + ) + ) + + // The live done event now targets an already-settled run: no-op. + act(() => { + agentListener?.({ requestId: started.requestId, type: 'done' }) + }) + expect( + screen.getByText('聚焦后读到的最终回复') + ).toBeInTheDocument() + }) + }) + it('retries a failed recovery with a new request ID and cleans up its listener', async () => { installRemoteProjectsSetting(true) const remoteProject = { diff --git a/src/renderer/src/App.tsx b/src/renderer/src/App.tsx index 2a764de5..2112cb57 100644 --- a/src/renderer/src/App.tsx +++ b/src/renderer/src/App.tsx @@ -1482,6 +1482,19 @@ function mergeArtifacts( ) } +/** + * The persisted store is the authority for a message's terminal state. + * A locally streaming message (or a locally interrupted one that the + * remote recovery later completed) must never shadow a persisted + * terminal state, even when the local conversation was touched later. + */ +function persistedTerminalStateOverridesLocal( + local: Message, + persisted: Message +): boolean { + return persisted.state !== 'streaming' && local.state !== 'complete' +} + function mergePersistedConversations( current: readonly Conversation[], incoming: readonly ConversationSnapshot[], @@ -1510,12 +1523,19 @@ function mergePersistedConversations( conversation.messages.map((message) => message.id) ) const messages = [ - ...conversation.messages.map( - (message) => - localIsNewer - ? localMessageById.get(message.id) ?? message - : message - ), + ...conversation.messages.map((message) => { + if (!localIsNewer) { + return message + } + const localMessage = localMessageById.get(message.id) + if ( + !localMessage || + persistedTerminalStateOverridesLocal(localMessage, message) + ) { + return message + } + return localMessage + }), ...local.messages.filter( (message) => !serverMessageIds.has(message.id) ) @@ -4140,6 +4160,39 @@ function App(): React.JSX.Element { [] ) + const releaseConversationQueueAfterRun = useCallback( + (run: Pick): void => { + requestAnimationFrame(() => { + if ( + run.projectId && + ((!projectRecoverySnapshotReadyRef.current && + isManagedSshProject( + projectsRef.current.find( + (project) => project.id === run.projectId + ) + )) || + isProjectRecoveryUnsettled( + projectRecoveryByProjectIdRef.current[run.projectId] + )) + ) { + return + } + void window.goodbuddy.conversationQueue + .ready(run.conversationId) + .catch(() => { + notify({ + tone: 'error', + message: tRef.current( + 'notices.conversationQueueResumeFailed' + ), + dedupeKey: 'conversation-queue-resume' + }) + }) + }) + }, + [notify] + ) + const handleAgentEvent = useCallback( (event: AgentEvent): void => { const run = activeRuns.current.get(event.requestId) @@ -4959,33 +5012,7 @@ function App(): React.JSX.Element { }) activeRuns.current.delete(event.requestId) setConversationActivity(run.conversationId, false) - requestAnimationFrame(() => { - if ( - run.projectId && - ((!projectRecoverySnapshotReadyRef.current && - isManagedSshProject( - projectsRef.current.find( - (project) => project.id === run.projectId - ) - )) || - isProjectRecoveryUnsettled( - projectRecoveryByProjectIdRef.current[run.projectId] - )) - ) { - return - } - void window.goodbuddy.conversationQueue - .ready(run.conversationId) - .catch(() => { - notify({ - tone: 'error', - message: tRef.current( - 'notices.conversationQueueResumeFailed' - ), - dedupeKey: 'conversation-queue-resume' - }) - }) - }) + releaseConversationQueueAfterRun(run) flushConversationPersistenceAfterRenderRef.current = true } }, @@ -4993,6 +5020,7 @@ function App(): React.JSX.Element { recordActivity, loadWorkspaceChanges, refreshTokenUsage, + releaseConversationQueueAfterRun, setConversationActivity, updateMessage, updateRequestActivity @@ -5115,6 +5143,36 @@ function App(): React.JSX.Element { refresh() }, 50) } + // A run whose assistant message has already reached a persisted + // terminal state (for example a remote task that finished while the + // desktop was disconnected) must stop counting as in-flight, otherwise + // the renderer keeps showing "processing" forever. + const settleActiveRunsFromPersistedMessages = ( + persisted: readonly ConversationSnapshot[] + ): void => { + if (activeRuns.current.size === 0) { + return + } + const persistedMessageStates = new Map() + for (const conversation of persisted) { + for (const message of conversation.messages) { + persistedMessageStates.set( + `${conversation.id}\0${message.id}`, + message.state + ) + } + } + for (const [requestId, run] of activeRuns.current) { + const state = persistedMessageStates.get( + `${run.conversationId}\0${run.messageId}` + ) + if (state !== undefined && state !== 'streaming') { + activeRuns.current.delete(requestId) + setConversationActivity(run.conversationId, false) + releaseConversationQueueAfterRun(run) + } + } + } const refresh = (): void => { if (refreshInFlight) { refreshQueued = true @@ -5215,6 +5273,7 @@ function App(): React.JSX.Element { persistedLocalConversationsRef.current ) ) + settleActiveRunsFromPersistedMessages(persisted) }) .catch(() => { if (active && sequence === refreshSequence) { @@ -5238,15 +5297,28 @@ function App(): React.JSX.Element { } }) } + const refreshWhenVisible = (): void => { + if (document.visibilityState !== 'hidden') { + queueRefresh() + } + } const remove = window.goodbuddy.conversations.onChanged(queueRefresh) + window.addEventListener('focus', refreshWhenVisible) + document.addEventListener('visibilitychange', refreshWhenVisible) return () => { active = false if (refreshTimer !== undefined) { window.clearTimeout(refreshTimer) } remove() + window.removeEventListener('focus', refreshWhenVisible) + document.removeEventListener('visibilitychange', refreshWhenVisible) } - }, [conversationStoreReady]) + }, [ + conversationStoreReady, + releaseConversationQueueAfterRun, + setConversationActivity + ]) useEffect(() => { let active = true