diff --git a/docs/features/remote-host/technical-design.md b/docs/features/remote-host/technical-design.md index 5c831800..12fdd6e3 100644 --- a/docs/features/remote-host/technical-design.md +++ b/docs/features/remote-host/technical-design.md @@ -340,6 +340,12 @@ OC `todowrite` / CN `Checklist` 顶部展示属于待实施功能，权威需求 `toolCallId` 继续路由，不等待新的父 Task metadata，也不重放模型请求。 远端实时过程需要包含该插件的 Agent；旧 Agent 的最终结果仍可查看，历史未采集过程 不会被补造。 +- 远端 ACP 普通工具失败后，只有最后一次工具失败之后收到非空白助手正文、Prompt 正常 + 完成且没有 `pending` / `in_progress` 工具时，才在 `done` 前将仍失败的工具更新为 + `recoverable`，保留调用 ID、名称、输入、输出和错误。失败前的正文、推理和纯空白 + 不作为恢复依据；后续再次失败会清除该依据。原生子代理失败不参与普通工具恢复， + Prompt 失败、取消或 `outcome-unknown` 也不触发转换。Agent-owned 与直接 ACP 路径 + 共用此规则；转换表示后续响应已处理失败，不表示工具重新执行成功。 - 本机 OpenCode SDK 路径允许不同会话并行，同一会话仍按请求顺序执行。每个请求独立拥有 SSE 事件订阅，并在正常完成、错误、取消或消费方结束迭代时主动关闭自己的响应流， 不关闭其他会话的订阅或共享 Server。聊天与原生 Compact 共用 `consumeEventSubscription`， @@ -376,6 +382,14 @@ OC `todowrite` / CN `Checklist` 顶部展示属于待实施功能，权威需求 时提交 Task 终态；事务成功后生成器才继续并向 Agent ACK。断电发生在同一 sequence 中间时会重读该 sequence，已写事件按 `(binding, operation, sequence, eventIndex)` 去重， 未写事件继续归并，不会重复 Provider 或工具执行。 +- 工具失败恢复所需的正文依据从 Desktop 已提交的 `task_events` 推导：按 + `remote_semantic_sequence` 数值及 `remote_event_index` 顺序判断最后一次失败后是否有 + 非空白 `text`，重连时与已有工具状态一起交给 Runtime，保留断线前的有效依据。 + 不从会原地更新或被截断的展示块推断顺序，不新增表、字段或持久快照。终态转换按 + `callId` 排序，并重发已为 `recoverable` 的工具，使部分终态落库后的重读仍使用稳定的 + `eventIndex`，沿用既有 provenance 去重和 checkpoint 提交。 + 重连时，工具及原生子代理的状态也以已提交的远端活动事件为准，覆盖消息终态处理派生的 + 失败状态，避免将未完成活动误判为可恢复错误，保证终态重放内容一致。 - 远程 context metrics 与对话级压缩后的估算值按本次请求实际使用的 Runtime 选择写入 `context_state_json`。实时接收与恢复入口都传入请求的选择；数据库在未传入时按对话、 项目、auto 顺序解析。继承项目设置的对话保持 `runtime_selection_json = NULL`， diff --git a/src/main/agent/acp-remote-runtime.test.ts b/src/main/agent/acp-remote-runtime.test.ts index 1d7109f9..a540efba 100644 --- a/src/main/agent/acp-remote-runtime.test.ts +++ b/src/main/agent/acp-remote-runtime.test.ts @@ -6,7 +6,8 @@ import { type Agent, type AgentCapabilities, type PromptRequest, - type PromptResponse + type PromptResponse, + type SessionUpdate } from '@agentclientprotocol/sdk' import { describe, expect, it, vi } from 'vitest' import { canonicalJson } from '../../shared/agent-protocol/canonical' @@ -17,7 +18,7 @@ import type { import { REMOTE_RUNTIME_LAUNCH_LIMITS } from '../../shared/remote-runtime-launch-contracts' -import type { RuntimeEvent } from './runtime' +import type { RuntimeEvent, RemoteSemanticEventProvenance } from './runtime' import { AcpRemoteRuntime, RemotePromptCancelledError, @@ -35,6 +36,7 @@ import type { RemoteRuntimeChannel } from './remote-runtime-channel' import { MemoryRuntimeSessionBindingStore } from './runtime-session-binding-store' +import { AssistantDatabase } from '../assistant/assistant-database' type FakeServer = { channel: RemoteRuntimeChannel @@ -2589,6 +2591,7 @@ describe('AcpRemoteRuntime Agent-owned prompts', () => { } function ownedChannel(options?: { + updates?: SessionUpdate[] state?: 'completed' | 'failed' | 'cancelled' | 'outcome-unknown' terminalPayload?: unknown includePermissionCheckpoint?: boolean @@ -2619,7 +2622,7 @@ describe('AcpRemoteRuntime Agent-owned prompts', () => { state: 'running' as const, latestSemanticSequence: '0' })) - const events = [ + const defaultEvents = [ { sequence: '1', kind: 'session-update' as const, @@ -2708,6 +2711,13 @@ describe('AcpRemoteRuntime Agent-owned prompts', () => { createdAt: 5 } ] + const events = options?.updates ? [ + ...options.updates.map((update, index) => ({ + sequence: String(index + 1), kind: 'session-update' as const, + payload: { sessionId: 'owned-session', update }, createdAt: index + 1 + })), + { ...defaultEvents.at(-1)!, sequence: String(options.updates.length + 1) } + ] : defaultEvents const pageOwnedPromptTranscript = vi.fn(async ({ bindingId, operationId, @@ -3132,6 +3142,186 @@ describe('AcpRemoteRuntime Agent-owned prompts', () => { ) }) + const failedRead: SessionUpdate = { + sessionUpdate: 'tool_call', toolCallId: 'read-1', title: 'Read', + status: 'failed', rawInput: { filePath: '/missing' }, + content: [{ type: 'content', content: { type: 'text', text: 'File not found' } }] + } + const responseText: SessionUpdate = { + sessionUpdate: 'agent_message_chunk', content: { type: 'text', text: 'Handled the missing file.' } + } + + describe.each(['owned', 'direct'] as const)('%s tool failure recovery', (path) => { + it.each([ + { name: 'response after failure', updates: [failedRead, responseText], recovered: true }, + { name: 'no response', updates: [failedRead], recovered: false }, + { name: 'response before failure', updates: [responseText, failedRead], recovered: false }, + { name: 'reasoning only', updates: [failedRead, { ...responseText, sessionUpdate: 'agent_thought_chunk' }], recovered: false }, + { name: 'whitespace only', updates: [failedRead, { ...responseText, content: { type: 'text', text: ' \n\t' } }], recovered: false }, + { name: 'another failure after response', updates: [failedRead, responseText, { ...failedRead, toolCallId: 'read-2' }], recovered: false }, + { name: 'unfinished tool', updates: [failedRead, responseText, { ...failedRead, toolCallId: 'read-2', status: 'in_progress' }], recovered: false }, + { name: 'native subagent failure', updates: [{ ...failedRead, rawInput: { subagent_type: 'explorer', prompt: 'Read a file' } }, responseText], recovered: false } + ] satisfies Array<{ name: string; updates: SessionUpdate[]; recovered: boolean }>)('$name', async ({ updates, recovered }) => { + const fixture = path === 'owned' ? ownedChannel({ updates }) : undefined + const server = fixture ? undefined : fakeServer({ + prompt: async ({ sessionId }) => { + for (const update of updates) await server!.client.sessionUpdate({ sessionId, update }) + return { stopReason: 'end_turn' } + } + }) + const instance = fixture?.instance ?? runtime(server!) + try { + const events = await collect(instance.run(request, new AbortController().signal)) + const recovery = events.filter(event => event.type === 'tool' && event.state === 'recoverable') + expect(recovery).toHaveLength(recovered ? 1 : 0) + if (recovered) expect(recovery[0]).toMatchObject({ + callId: 'read-1', name: 'Read', input: '{"filePath":"/missing"}', output: '"File not found"' + }) + expect(events.some(event => event.type === 'done')).toBe(true) + } finally { + await instance.dispose() + } + }) + + it.each(['failed', 'cancelled', 'outcome-unknown'] as const)('does not recover a %s prompt', async (state) => { + const updates = [failedRead, responseText] + const fixture = path === 'owned' ? ownedChannel({ updates, state }) : undefined + const server = fixture ? undefined : fakeServer({ + prompt: async ({ sessionId }) => { + for (const update of updates) await server!.client.sessionUpdate({ sessionId, update }) + if (state === 'cancelled') return { stopReason: 'cancelled' } + throw new Error(state) + } + }) + const instance = fixture?.instance ?? runtime(server!) + const events: RuntimeEvent[] = [] + try { + await expect((async () => { + for await (const event of instance.run(request, new AbortController().signal)) events.push(event) + })()).rejects.toThrow() + expect(events.some(event => event.type === 'tool' && event.state === 'recoverable')).toBe(false) + expect(events.some(event => event.type === 'done')).toBe(false) + } finally { + await instance.dispose() + } + }) + }) + + it.each(['failed', 'cancelled', 'interrupted', 'running'] as const)('does not promote an ineligible recovered %s tool', async (state) => { + const fixture = ownedChannel({ updates: [] }) + try { + const events = await collect(fixture.instance.run({ + ...request, remoteRecoveredTools: [{ callId: 'read', name: 'Read', state, summary: 'Read' }], + remoteHasResponseTextAfterToolFailure: state !== 'failed' + }, new AbortController().signal)) + expect(events.some(event => event.type === 'tool')).toBe(false) + } finally { + await fixture.instance.dispose() + } + }) + + it.each(['recoverable', 'done', 'unfinished-tool', 'unfinished-subagent'] as const)( + 'replays a partial %s terminal after real message reduction without provenance conflicts', async (cut) => { + const failedTerminal = cut.startsWith('unfinished-') + const fixture = ownedChannel({ + terminalPayload: { status: 'completed', response: { stopReason: 'end_turn' } }, + updates: failedTerminal ? [failedRead, responseText, { + ...failedRead, toolCallId: 'running-call', status: 'in_progress', + ...(cut === 'unfinished-subagent' + ? { rawInput: { subagent_type: 'explorer', prompt: 'Read a file' } } : {}) + }] : [ + { ...failedRead, toolCallId: 'z-read' }, + { ...failedRead, toolCallId: 'a-read' }, responseText + ] + }) + const database = new AssistantDatabase(':memory:') + database.initialize('C:\\Workspace') + const project = database.createSshProject({ + project: { name: 'Recovery', description: '', rootPath: '/srv/project', defaultWorkMode: 'execute', runtimeSelection: { provider: 'opencode' } }, + executionSpace: { kind: 'ssh', hostId: '00000000-0000-4000-8000-000000000850', remoteRootPath: '/srv/project' }, + assertCurrent: () => {} + }) + const assistantMessageId = '00000000-0000-4000-8000-000000000851' + const conversationId = '00000000-0000-4000-8000-000000000853' + database.saveLocalConversations([{ + header: { id: conversationId, projectId: project.id, title: 'Recovery', updatedAt: 0 }, messages: [] + }]) + database.createTask({ + id: request.requestId, projectId: project.id, conversationId, + title: 'Recovery', instructions: 'Read', workMode: 'execute', + remoteRecovery: { recoverable: true, currentUserMessageId: '00000000-0000-4000-8000-000000000852', currentAssistantMessageId: assistantMessageId } + }) + const persist = (event: RuntimeEvent & { remoteProvenance?: RemoteSemanticEventProvenance }) => { + if (!event.remoteProvenance) return undefined + if (event.type === 'model-usage' || event.type === 'generated-image') throw new Error('Unexpected fixture event') + const { remoteProvenance, ...publicEvent } = event + // IPC rejects done while the original failed Read is still unsuccessful. + const payload = failedTerminal && publicEvent.type === 'done' ? { + requestId: request.requestId, type: 'error' as const, status: 'failed' as const, + message: 'Read 工具执行失败' + } : publicEvent + return database.appendRemoteConversationTaskEventOnce({ + taskId: request.requestId, conversationId, assistantMessageId, + ...remoteProvenance, event: payload + }) + } + const stream = fixture.instance.run({ ...request, conversationId }, new AbortController().signal) + let partial: RuntimeEvent | undefined + for (;;) { + const next = await stream.next() + if (next.done) throw new Error('Expected partial terminal') + persist(next.value) + if (cut === 'recoverable' + ? next.value.type === 'tool' && next.value.state === 'recoverable' + : next.value.type === 'done') { + partial = next.value + break + } + } + await stream.return() + const replacement = factoryRuntime(async () => fixture.channel, fixture.store, { + modelBridgePolicy, modelProfile: ownedModelProfile + }) + try { + const cursor = database.getHighestCommittedRemoteTaskEventSequenceForTask(request.requestId) + expect(cursor).toBe('3') + expect(database.getTask(request.requestId).status).toBe('running') + expect(database.hasRemoteResponseTextAfterToolFailure(request.requestId)).toBe(true) + const message = database.getConversation(conversationId).messages[1]! + const states = database.getRemoteTaskActivityStates(request.requestId) + if (cut === 'unfinished-tool') { + expect(message.tools?.map(tool => tool.state)).toEqual(['failed', 'failed']) + expect(states.tools.get('running-call')).toBe('running') + } else if (cut === 'unfinished-subagent') { + expect(message.subagents?.[0]).toMatchObject({ state: 'failed', error: 'Read 工具执行失败' }) + expect(states.subagents.get(message.subagents![0]!.childTaskId)).toEqual({ state: 'running', error: undefined }) + } + const recovered = await collect(replacement.run({ + ...request, conversationId, remoteRecoveryOnly: true, remoteSemanticAfterSequence: cursor, + remoteHasResponseTextAfterToolFailure: database.hasRemoteResponseTextAfterToolFailure(request.requestId), + remoteRecoveredTools: [...(message.tools ?? [])].reverse().map(tool => ({ + ...tool, callId: tool.callId!, state: states.tools.get(tool.callId!) ?? tool.state + })), + remoteRecoveredSubagents: message.subagents?.map(subagent => ({ + ...subagent, ...states.subagents.get(subagent.childTaskId) + })) + }, new AbortController().signal)) + expect(recovered).toContainEqual(partial) + const replayed = recovered.filter(event => event.type === 'tool') + expect(replayed.map(event => event.callId)).toEqual(failedTerminal ? [] : ['a-read', 'z-read']) + expect(persist(partial!)).toBe(false) + for (const event of recovered) persist(event) + expect(database.getHighestCommittedRemoteTaskEventSequenceForTask(request.requestId)).toBe('4') + expect(recovered.some(event => event.type === 'done')).toBe(true) + expect(database.getTask(request.requestId).status).toBe(failedTerminal ? 'failed' : 'completed') + expect(fixture.startOwnedPrompt).toHaveBeenCalledOnce() + expect(fixture.attachOwnedPrompt).toHaveBeenCalledOnce() + } finally { + database.close() + await Promise.all([replacement.dispose(), fixture.instance.dispose()]) + } + }) + it('never starts a replacement prompt for attach-only recovery', async () => { const fixture = ownedChannel() const recoveryRequest = { diff --git a/src/main/agent/acp-remote-runtime.ts b/src/main/agent/acp-remote-runtime.ts index 08c71e3b..559dd89a 100644 --- a/src/main/agent/acp-remote-runtime.ts +++ b/src/main/agent/acp-remote-runtime.ts @@ -147,6 +147,8 @@ export class RemotePromptCancelledError extends Error { } type ActivePrompt = { + failedTools?: Map> + hasResponseTextAfterToolFailure?: boolean reportedQuestions?: Set subagentProgress?: OpenCodeSubagentProgress requestId: string @@ -2110,6 +2112,46 @@ export class AcpRemoteRuntime implements AgentRuntime { private mapUpdate( prompt: ActivePrompt, update: SessionUpdate + ): RuntimePublicEvent | undefined { + const mapped = this.mapSessionUpdate(prompt, update) + if (!mapped) return undefined + const event = this.limitEvent(mapped) + if (event.type === 'text' && event.delta.trim()) { + prompt.hasResponseTextAfterToolFailure = true + } else if (event.type === 'tool') { + if (event.state === 'failed') { + prompt.failedTools ??= new Map() + prompt.failedTools.set(event.callId, event) + prompt.hasResponseTextAfterToolFailure = false + } else { + prompt.failedTools?.delete(event.callId) + } + } else if (event.type === 'subagent' && event.runtimeCallId) { + prompt.failedTools?.delete(event.runtimeCallId) + } + return event + } + + private recoveredToolEvents(prompt: ActivePrompt): RuntimePublicEvent[] { + if (!prompt.hasResponseTextAfterToolFailure || + [...prompt.toolCalls.values()].some(tool => + tool.status === 'pending' || tool.status === 'in_progress')) { + return [] + } + // Re-emit already recovered tools too: a terminal may be only partly + // committed. Sorting keeps every event index stable across reattachment. + return [...(prompt.failedTools ?? [])] + .sort(([left], [right]) => left < right ? -1 : left > right ? 1 : 0) + .map(([, tool]) => ({ + ...tool, + state: 'recoverable' as const, + summary: `远端 Runtime 已在后续响应中处理工具失败：${tool.name}` + })) + } + + private mapSessionUpdate( + prompt: ActivePrompt, + update: SessionUpdate ): RuntimePublicEvent | undefined { const extension = update._meta?.goodbuddyQuestion if (this.options.runtimeId === 'opencode' && extension && typeof extension === 'object' && @@ -2432,11 +2474,26 @@ export class AcpRemoteRuntime implements AgentRuntime { pendingUpdateBytes: 0, inboundPaused: false, toolCalls: new Map(), + hasResponseTextAfterToolFailure: request.remoteHasResponseTextAfterToolFailure, + failedTools: new Map(), open: true, completed: false, context } for (const tool of request.remoteRecoveredTools ?? []) { + if (tool.state === 'failed' || tool.state === 'recoverable') { + prompt.failedTools!.set(tool.callId, { + requestId: request.requestId, + type: 'tool', + callId: tool.callId, + name: tool.name, + state: tool.state, + summary: tool.summary, + ...(tool.input ? { input: tool.input } : {}), + ...(tool.output ? { output: tool.output } : {}), + ...(tool.error ? { error: tool.error } : {}) + }) + } prompt.toolCalls.set(tool.callId.slice(0, 256), { title: tool.name.slice(0, 200), status: @@ -2457,6 +2514,7 @@ export class AcpRemoteRuntime implements AgentRuntime { if (!subagent.runtimeCallId) { continue } + prompt.failedTools?.delete(subagent.runtimeCallId) const subagentTitle = 'actor' in subagent ? subagent.actor.label @@ -2631,7 +2689,7 @@ export class AcpRemoteRuntime implements AgentRuntime { const mapped = page.pendingQuestions !== undefined && notification.update._meta?.goodbuddyQuestion ? undefined : this.mapUpdate(prompt, notification.update) if (mapped !== undefined) { - publicEvents.push(this.limitEvent(mapped)) + publicEvents.push(mapped) } } else if (transcriptEvent.kind === 'prompt-terminal') { const terminal = transcriptTerminal( @@ -2650,6 +2708,7 @@ export class AcpRemoteRuntime implements AgentRuntime { } } if (terminal.state === 'completed') { + publicEvents.push(...this.recoveredToolEvents(prompt)) publicEvents.push({ requestId: request.requestId, type: 'done', @@ -2979,7 +3038,7 @@ export class AcpRemoteRuntime implements AgentRuntime { } const event = this.mapUpdate(prompt, update) if (event) { - yield this.limitEvent(event) + yield event } } await this.awaitOperation( @@ -3025,6 +3084,7 @@ export class AcpRemoteRuntime implements AgentRuntime { signal ) session.binding = binding + yield* this.recoveredToolEvents(prompt) yield { requestId: request.requestId, type: 'done', diff --git a/src/main/agent/runtime.ts b/src/main/agent/runtime.ts index 26e51032..0dcf81f9 100644 --- a/src/main/agent/runtime.ts +++ b/src/main/agent/runtime.ts @@ -158,6 +158,8 @@ export type AgentExecutionRequest = Omit & { remoteRecoveryOnly?: boolean /** Main-only durable tool state used to merge partial recovery updates. */ remoteRecoveredTools?: readonly RemoteRecoveredTool[] + /** Derived from committed task events, including text before disconnect. */ + remoteHasResponseTextAfterToolFailure?: boolean /** Main-only durable native subagent state for partial recovery updates. */ remoteRecoveredSubagents?: readonly RemoteRecoveredSubagent[] } diff --git a/src/main/assistant/assistant-database.test.ts b/src/main/assistant/assistant-database.test.ts index d386f80c..36744cda 100644 --- a/src/main/assistant/assistant-database.test.ts +++ b/src/main/assistant/assistant-database.test.ts @@ -3544,6 +3544,40 @@ describe('AssistantDatabase', () => { inspected.close() }) + it('derives remote recovery evidence from semantic order across a restart', async () => { + const directory = await mkdtemp(join(tmpdir(), 'goodbuddy-tool-recovery-')) + temporaryDirectories.push(directory) + const path = join(directory, 'assistant.sqlite') + let database = new AssistantDatabase(path) + database.initialize('C:\\Workspace') + const taskId = '00000000-0000-4000-8000-000000000702' + database.createTask({ id: taskId, title: 'Recovery', instructions: 'Read', workMode: 'execute' }) + const append = (sequence: string, kind: string, payload: object) => database.appendRemoteTaskEventOnce({ + taskId, bindingId: 'tool-recovery', operationId: taskId, + semanticSequence: sequence, eventIndex: 0, kind, payload + }) + expect(database.hasRemoteResponseTextAfterToolFailure(taskId)).toBe(false) + append('1', 'text', { delta: 'Before failure' }) + append('2', 'tool', { state: 'failed', callId: 'read' }) + append('3', 'reasoning', { delta: 'Still thinking' }) + append('4', 'text', { delta: '\n\t\u3000' }) + expect(database.hasRemoteResponseTextAfterToolFailure(taskId)).toBe(false) + append('10', 'text', { delta: 'Handled the failure' }) + // Late insertion of an older event must not change semantic ordering. + append('9', 'tool', { state: 'failed', callId: 'read-again' }) + expect(database.hasRemoteResponseTextAfterToolFailure(taskId)).toBe(true) + database.close() + database = new AssistantDatabase(path) + database.initialize('C:\\Workspace') + expect(database.hasRemoteResponseTextAfterToolFailure(taskId)).toBe(true) + append('11', 'tool', { state: 'recoverable', callId: 'read' }) + expect(database.hasRemoteResponseTextAfterToolFailure(taskId)).toBe(true) + append('12', 'tool', { state: 'failed', callId: 'third-read' }) + expect(database.hasRemoteResponseTextAfterToolFailure(taskId)).toBe(false) + expect(database.hasRemoteResponseTextAfterToolFailure('00000000-0000-4000-8000-000000000703')).toBe(false) + database.close() + }) + it('persists inherited remote metrics and replay without pinning the conversation', async () => { const directory = await mkdtemp( join(tmpdir(), 'goodbuddy-inherited-remote-metrics-') diff --git a/src/main/assistant/assistant-database.ts b/src/main/assistant/assistant-database.ts index 28009a56..9b9c51f9 100644 --- a/src/main/assistant/assistant-database.ts +++ b/src/main/assistant/assistant-database.ts @@ -5405,6 +5405,57 @@ export class AssistantDatabase { return row.sequence === null ? '0' : String(row.sequence) } + hasRemoteResponseTextAfterToolFailure(taskIdInput: string): boolean { + const taskId = assistantIdSchema.parse(taskIdInput) + // Use event order, not display blocks: tool blocks are updated in place, + // and display capture can be truncated before the final response arrives. + const rows = this.requireDatabase().prepare( + `SELECT kind, payload_json FROM task_events + WHERE task_id = ? AND remote_semantic_sequence IS NOT NULL + AND (kind = 'text' OR + (kind = 'tool' AND json_extract(payload_json, '$.state') = 'failed')) + ORDER BY CAST(remote_semantic_sequence AS INTEGER) DESC, remote_event_index DESC` + ).iterate(taskId) + for (const row of rows) { + if (row.kind === 'tool') return false + const event = JSON.parse(row.payload_json as string) as { delta?: string } + if (event.delta?.trim()) return true + } + return false + } + + getRemoteTaskActivityStates(taskIdInput: string): { + tools: Map + subagents: Map> + } { + const taskId = assistantIdSchema.parse(taskIdInput) + const tools = new Map() + const subagents = new Map>() + // Terminal message projection changes unfinished activities. Only their + // original remote events can supply states for deterministic terminal replay. + const rows = this.requireDatabase().prepare( + `SELECT kind, json_extract(payload_json, '$.callId') AS call_id, + json_extract(payload_json, '$.childTaskId') AS child_task_id, + json_extract(payload_json, '$.state') AS state, + json_extract(payload_json, '$.error') AS error + FROM task_events + WHERE task_id = ? AND remote_semantic_sequence IS NOT NULL + AND kind IN ('tool', 'subagent') + ORDER BY CAST(remote_semantic_sequence AS INTEGER), remote_event_index` + ).iterate(taskId) + for (const row of rows) { + if (row.kind === 'tool') { + tools.set(row.call_id as string, row.state as ConversationToolActivity['state']) + } else { + subagents.set(row.child_task_id as string, { + state: row.state as ConversationSubagentActivity['state'], + error: row.error === null ? undefined : row.error as string + }) + } + } + return { tools, subagents } + } + private queryHighestCommittedRemoteTaskEventSequence( database: DatabaseSync, bindingId: string, diff --git a/src/main/ipc.test.ts b/src/main/ipc.test.ts index cc4e17e4..5baed4e6 100644 --- a/src/main/ipc.test.ts +++ b/src/main/ipc.test.ts @@ -5086,9 +5086,11 @@ describe('registerIpcHandlers agent terminal state', () => { status: 'running' | 'waiting_approval' | 'interrupted' }> >(() => []), - getHighestCommittedRemoteTaskEventSequenceForTask: vi.fn( + getHighestCommittedRemoteTaskEventSequenceForTask: vi.fn( () => '0' ), + hasRemoteResponseTextAfterToolFailure: vi.fn(() => false), + getRemoteTaskActivityStates: vi.fn(() => ({ tools: new Map(), subagents: new Map() })), endRecoverableRemoteTask: vi.fn(), recordRemoteTaskQuestionAnswer: vi.fn(), getTask: vi.fn(), @@ -6519,7 +6521,7 @@ describe('registerIpcHandlers agent terminal state', () => { await harness.dispose() }) - it('reattaches a recoverable task from its committed cursor and completes project synchronization', async () => { + it.each([false, true])('reattaches from its committed cursor with tool recovery evidence %s', async (hasRecoveredResponse) => { const projectId = '00000000-0000-4000-8000-000000000781' const taskId = '00000000-0000-4000-8000-000000000782' const conversationId = @@ -6543,6 +6545,7 @@ describe('registerIpcHandlers agent terminal state', () => { requestId: taskId, remoteRecoveryOnly: true, remoteSemanticAfterSequence: '2', + remoteHasResponseTextAfterToolFailure: hasRecoveredResponse, remoteRecoveredTools: [ expect.objectContaining({ callId: 'tool-before-restart', @@ -6575,6 +6578,14 @@ describe('registerIpcHandlers agent terminal state', () => { eventIndex: 1 } } as const + if (hasRecoveredResponse) yield { + requestId: taskId, type: 'tool', callId: 'tool-before-restart', + name: 'Read', state: 'recoverable', summary: 'Handled missing file', error: 'missing file', + remoteProvenance: { + source: 'remote-semantic-transcript', bindingId: 'binding-recovery', + operationId: taskId, semanticSequence: '4', eventIndex: 0 + } + } as const yield { requestId: taskId, type: 'done', @@ -6584,7 +6595,7 @@ describe('registerIpcHandlers agent terminal state', () => { bindingId: 'binding-recovery', operationId: taskId, semanticSequence: '4', - eventIndex: 0 + eventIndex: hasRecoveredResponse ? 1 : 0 } } as const consumed.push('4') @@ -6596,7 +6607,7 @@ describe('registerIpcHandlers agent terminal state', () => { bindingId: 'binding-recovery', operationId: taskId, semanticSequence: '4', - eventIndex: 1 + eventIndex: hasRecoveredResponse ? 2 : 1 } } as const } @@ -6653,6 +6664,7 @@ describe('registerIpcHandlers agent terminal state', () => { harness.assistantDatabase.getHighestCommittedRemoteTaskEventSequenceForTask.mockReturnValue( '2' ) + harness.assistantDatabase.hasRemoteResponseTextAfterToolFailure.mockReturnValue(hasRecoveredResponse) harness.recoveryGetHandler?.( trustedEvent(harness.webContents) @@ -6662,7 +6674,7 @@ describe('registerIpcHandlers agent terminal state', () => { await vi.waitFor(() => expect( harness.assistantDatabase.appendRemoteConversationTaskEventOnce - ).toHaveBeenCalledTimes(4) + ).toHaveBeenCalledTimes(hasRecoveredResponse ? 5 : 4) ) expect( harness.assistantDatabase.appendRemoteConversationTaskEventOnce @@ -6683,11 +6695,13 @@ describe('registerIpcHandlers agent terminal state', () => { expect( harness.assistantDatabase.appendRemoteConversationTaskEventOnce ).toHaveBeenNthCalledWith( - 3, + hasRecoveredResponse ? 4 : 3, expect.objectContaining({ taskId, semanticSequence: '4', - event: { + event: hasRecoveredResponse ? { + requestId: taskId, type: 'done', sessionId: 'session-recovery' + } : { requestId: taskId, type: 'error', status: 'failed', @@ -6722,6 +6736,83 @@ describe('registerIpcHandlers agent terminal state', () => { await harness.dispose() }) + it('replays an error committed before its checkpoint using remote tool states instead of terminal message states', async () => { + const database = new AssistantDatabase(':memory:') + database.initialize('C:\\Workspace') + const project = database.createSshProject({ + project: { name: 'Partial terminal', description: '', rootPath: '/srv/project', defaultWorkMode: 'execute', runtimeSelection: { provider: 'opencode' } }, + executionSpace: { kind: 'ssh', hostId: '00000000-0000-4000-8000-000000000850', remoteRootPath: '/srv/project' }, + assertCurrent: () => {} + }) + const taskId = '00000000-0000-4000-8000-000000000851' + const conversationId = '00000000-0000-4000-8000-000000000852' + const assistantMessageId = '00000000-0000-4000-8000-000000000853' + database.saveLocalConversations([{ + header: { id: conversationId, projectId: project.id, title: 'Partial terminal', updatedAt: 0 }, messages: [] + }]) + database.createTask({ + id: taskId, projectId: project.id, conversationId, title: 'Partial terminal', instructions: 'Read', workMode: 'execute', + remoteRecovery: { recoverable: true, currentUserMessageId: '00000000-0000-4000-8000-000000000854', currentAssistantMessageId: assistantMessageId } + }) + const append = (semanticSequence: string, eventIndex: number, event: Parameters[0]['event']) => + database.appendRemoteConversationTaskEventOnce({ + taskId, conversationId, assistantMessageId, bindingId: 'partial-terminal', operationId: taskId, + semanticSequence, eventIndex, event + }) + const events: AgentEvent[] = [ + { requestId: taskId, type: 'tool', callId: 'A', name: 'Read', state: 'failed', summary: 'Read', error: 'missing file' }, + { requestId: taskId, type: 'text', delta: 'Handled missing file' }, + { requestId: taskId, type: 'tool', callId: 'B', name: 'Write', state: 'running', summary: 'Write', input: 'preserve input' } + ] + events.forEach((event, index) => { + append(String(index + 1), 0, event) + append(String(index + 1), 1, { requestId: taskId, type: 'remote-semantic-checkpoint' }) + }) + const error: AgentEvent = { requestId: taskId, type: 'error', status: 'failed', message: 'Read 工具执行失败：missing file' } + append('4', 0, error) + expect(database.getConversation(conversationId).messages[1]?.tools?.map(tool => tool.state)).toEqual(['failed', 'failed']) + expect(database.getTask(taskId).status).toBe('running') + const recoveredRuntime = { + runtimeId: 'opencode', capability: 'chat', supportsToolExecution: true, + async *run(request: AgentExecutionRequest) { + expect(request).toMatchObject({ + requestId: taskId, remoteRecoveryOnly: true, remoteSemanticAfterSequence: '3', + remoteHasResponseTextAfterToolFailure: true, + remoteRecoveredTools: [ + expect.objectContaining({ callId: 'A', state: 'failed', error: 'missing file' }), + expect.objectContaining({ callId: 'B', state: 'running', input: 'preserve input' }) + ] + }) + const remoteProvenance = { + source: 'remote-semantic-transcript' as const, bindingId: 'partial-terminal', + operationId: taskId, semanticSequence: '4', eventIndex: 0 + } + yield { requestId: taskId, type: 'done' as const, remoteProvenance } + yield { requestId: taskId, type: 'remote-semantic-checkpoint' as const, remoteProvenance: { ...remoteProvenance, eventIndex: 1 } } + } + } + const { harness } = createManagedSshHarness(recoveredRuntime) + harness.assistantDatabase.getProject.mockImplementation(database.getProject.bind(database)) + harness.assistantDatabase.getConversation.mockImplementation(database.getConversation.bind(database)) + harness.assistantDatabase.listRecoverableRemoteTasks.mockImplementation(database.listRecoverableRemoteTasks.bind(database)) + harness.assistantDatabase.getHighestCommittedRemoteTaskEventSequenceForTask.mockImplementation(database.getHighestCommittedRemoteTaskEventSequenceForTask.bind(database)) + harness.assistantDatabase.hasRemoteResponseTextAfterToolFailure.mockImplementation(database.hasRemoteResponseTextAfterToolFailure.bind(database)) + harness.assistantDatabase.getRemoteTaskActivityStates.mockImplementation(database.getRemoteTaskActivityStates.bind(database)) + harness.assistantDatabase.appendRemoteConversationTaskEventOnce.mockImplementation(database.appendRemoteConversationTaskEventOnce.bind(database)) + try { + harness.recoveryGetHandler?.(trustedEvent(harness.webContents)) + await vi.waitFor(() => expect(database.getTask(taskId).status).toBe('failed')) + expect(database.getHighestCommittedRemoteTaskEventSequenceForTask(taskId)).toBe('4') + expect(harness.assistantDatabase.appendRemoteConversationTaskEventOnce).toHaveBeenNthCalledWith(1, + expect.objectContaining({ semanticSequence: '4', eventIndex: 0, event: error })) + expect(harness.assistantDatabase.appendRemoteConversationTaskEventOnce.mock.results.map(result => result.value)).toEqual([false, true]) + expect(database.getRemoteTaskActivityStates(taskId).tools.get('B')).toBe('running') + } finally { + await harness.dispose() + database.close() + } + }) + it.each(['runtime', 'customization', 'capabilities'] as const)('restores concurrent live questions and their answer history across %s settings reload', async (settingsKind) => { const directory = await mkdtemp(join(tmpdir(), 'goodbuddy-recovered-question-history-')) const databasePath = join(directory, 'assistant.sqlite') diff --git a/src/main/ipc.ts b/src/main/ipc.ts index e607e800..26f606cf 100644 --- a/src/main/ipc.ts +++ b/src/main/ipc.ts @@ -1919,17 +1919,24 @@ export function registerIpcHandlers( const recoveredAssistantMessage = conversation.messages.find( (message) => message.id === task.currentAssistantMessageId ) + const activityStates = assistantDatabase.getRemoteTaskActivityStates(task.taskId) const recoveredTools = recoveredAssistantMessage?.tools?.filter( (tool): tool is typeof tool & { callId: string } => Boolean(tool.callId) - ) ?? [] + ).map(tool => ({ + ...tool, + state: activityStates.tools.get(tool.callId) ?? tool.state + })) ?? [] const recoveredSubagents = recoveredAssistantMessage?.subagents?.filter( (subagent) => subagent.routingMode === 'native' && subagent.runtimeCallId - ) ?? [] + ).map(subagent => ({ + ...subagent, + ...activityStates.subagents.get(subagent.childTaskId) + })) ?? [] const toolStates = new Map( recoveredTools.map((tool) => [tool.callId, tool]) ) @@ -1982,6 +1989,8 @@ export function registerIpcHandlers( remoteSemanticAfterSequence: highestCommitted, remoteRecoveryOnly: true, remoteRecoveredTools: recoveredTools, + remoteHasResponseTextAfterToolFailure: + assistantDatabase.hasRemoteResponseTextAfterToolFailure(task.taskId), remoteRecoveredSubagents: recoveredSubagents } let recoveryMetricSettings: