root/goodbuddy
1
0
Fork 0
Code Issues Pull Requests Actions 3 Packages Projects Releases Wiki Activity

fix: converge remote run state on persisted terminal messages

Browse Source 
After a managed SSH disconnect the renderer kept showing the remote
Agent as processing even though the recovered task had already been
persisted as completed. Local streaming copies shadowed the persisted
terminal state in mergePersistedConversations, and recovery completion
never asked the renderer to re-read conversations.

- Prefer the persisted terminal state over a local streaming message
- Settle active runs whose messages already reached a persisted terminal
  state and release the conversation queue once
- Refresh conversations on window focus and visibility changes
- Publish a conversation change before recovery reports completed/failed
This commit is contained in:
mesalogo
2026-09-04 16:54:35 +08:00
parent 670171ce5d
commit d180f63abd
4 changed files with 354 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/ipc.test.ts
+14
View File
@@ -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()
})
src/main/ipc.ts
+4
View File
@@ -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,
src/renderer/src/App.test.tsx
+230
View File
@@ -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(<App />)
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 = {
src/renderer/src/App.tsx
+106 -34
View File
@@ -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<ActiveRun, 'conversationId' | 'projectId'>): 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<string, Message['state']>()
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