From d859cbfdeb1494808c0fdc1b0b9c07113f0660fc Mon Sep 17 00:00:00 2001 From: mesalogo Date: Fri, 2 Oct 2026 23:11:11 +0800 Subject: [PATCH] perf: persist remote runtime events in one transaction per checkpoint Remote semantic events were written with one or two synchronous SQLite transactions each before being forwarded; recoverable conversations also rewrote the whole assistant message per event. Events now commit together at each remote checkpoint (before the Agent can acknowledge), and before non-remote, question and terminal events, on a size cap or a 50 ms timer. Only newly stored events are forwarded, in order; a failed batch rolls back and blocks later commits. Applies to interactive, background and recovery runs. --- src/main/assistant/assistant-database.test.ts | 198 ++++++++ src/main/assistant/assistant-database.ts | 457 ++++++++++++------ src/main/ipc.test.ts | 205 +++++++- src/main/ipc.ts | 287 +++++++---- src/main/remote-event-batcher.test.ts | 66 +++ src/main/remote-event-batcher.ts | 149 ++++++ 6 files changed, 1087 insertions(+), 275 deletions(-) create mode 100644 src/main/remote-event-batcher.test.ts create mode 100644 src/main/remote-event-batcher.ts diff --git a/src/main/assistant/assistant-database.test.ts b/src/main/assistant/assistant-database.test.ts index 7764293f..ec009150 100644 --- a/src/main/assistant/assistant-database.test.ts +++ b/src/main/assistant/assistant-database.test.ts @@ -4103,6 +4103,204 @@ describe('AssistantDatabase', () => { raw.close() }) + describe('remote conversation event batches', () => { + const conversationId = '00000000-0000-4000-8000-000000000711' + const taskId = '00000000-0000-4000-8000-000000000712' + const assistantMessageId = '00000000-0000-4000-8000-000000000714' + type BatchEvent = Parameters< + AssistantDatabase['appendRemoteConversationTaskEventsBatch'] + >[0]['events'][number] + + const openRecoverableTask = (): AssistantDatabase => { + const database = new AssistantDatabase(':memory:') + database.initialize('C:\\Workspace') + const project = database.createSshProject(validatedSshProjectWrite()) + database.saveLocalConversations([{ + header: { id: conversationId, projectId: project.id, title: '批量', updatedAt: 0 }, + messages: [] + }]) + database.createTask({ + id: taskId, projectId: project.id, conversationId, + title: '批量', instructions: '回复', workMode: 'execute', + remoteRecovery: { + recoverable: true, + currentUserMessageId: '00000000-0000-4000-8000-000000000713', + currentAssistantMessageId: assistantMessageId + } + }) + return database + } + const entry = ( + semanticSequence: string, + eventIndex: number, + event: BatchEvent['event'] + ): BatchEvent => ({ + bindingId: 'batch-binding', operationId: taskId, + semanticSequence, eventIndex, event + }) + const taskEvents = (database: AssistantDatabase, id = taskId) => + ((database as unknown as { requireDatabase(): DatabaseSync }).requireDatabase() + .prepare('SELECT kind, payload_json FROM task_events WHERE task_id = ? AND remote_semantic_sequence IS NOT NULL ORDER BY id') + .all(id) as Array<{ kind: string; payload_json: string }>) + .map(({ kind, payload_json }) => ({ kind, payload: JSON.parse(payload_json) as unknown })) + const page: BatchEvent[] = [ + entry('1', 0, { requestId: taskId, type: 'reasoning', delta: 'plan ' }), + entry('1', 1, { requestId: taskId, type: 'text', delta: 'Hello' }), + entry('1', 2, { requestId: taskId, type: 'remote-semantic-checkpoint' }), + entry('2', 0, { + requestId: taskId, type: 'tool', callId: 'read', name: 'Read', + state: 'running', summary: 'Read file', input: '{"path":"a"}' + }), + entry('2', 1, { requestId: taskId, type: 'status', message: 'working' }), + entry('2', 2, { + requestId: taskId, type: 'context-metrics', contextTokens: 77, + effectiveTriggerTokens: 1000, compressionEnabled: false, source: 'provider' + }), + entry('2', 3, { requestId: taskId, type: 'remote-semantic-checkpoint' }), + entry('3', 0, { + requestId: taskId, type: 'tool', callId: 'read', name: 'Read', + state: 'completed', summary: 'Read file', output: 'contents' + }), + entry('3', 1, { requestId: taskId, type: 'text', delta: ' world' }), + entry('3', 2, { requestId: taskId, type: 'reasoning', delta: 'done' }), + entry('3', 3, { requestId: taskId, type: 'done', sessionId: 'session' }), + entry('3', 4, { requestId: taskId, type: 'remote-semantic-checkpoint' }) + ] + const owner = { taskId, conversationId, assistantMessageId } + // Block ids are random per reduction; compare everything else. + const comparable = (database: AssistantDatabase) => { + const conversation = database.getConversation(conversationId) + const message = conversation.messages.find( + (candidate) => candidate.id === assistantMessageId + )! + return { + message: { + ...message, + createdAt: undefined, + blocks: message.blocks?.map((block) => ({ ...block, id: undefined })) + }, + contextMetrics: conversation.contextMetrics, + status: database.getTask(taskId).status, + error: database.getTask(taskId).error, + events: taskEvents(database), + cursor: database.getHighestCommittedRemoteTaskEventSequenceForTask(taskId) + } + } + + it('matches sequential per-event persistence with one message rewrite and terminalizes at its checkpoint', () => { + const sequential = openRecoverableTask() + const batched = openRecoverableTask() + try { + for (const item of page) { + expect(sequential.appendRemoteConversationTaskEventOnce({ ...owner, ...item })).toBe(true) + } + const prepare = vi.spyOn( + (batched as unknown as { requireDatabase(): DatabaseSync }).requireDatabase(), + 'prepare' + ) + expect(batched.appendRemoteConversationTaskEventsBatch({ ...owner, events: page })) + .toEqual(page.map(() => true)) + const statements = prepare.mock.calls.map(([sql]) => sql.replace(/\s+/gu, ' ').trim()) + prepare.mockRestore() + expect(statements.filter((sql) => sql.startsWith('UPDATE messages'))).toHaveLength(1) + expect(statements.filter((sql) => sql.startsWith('SELECT id, role, content'))).toHaveLength(1) + expect(statements.filter((sql) => sql.startsWith('UPDATE conversations'))).toHaveLength(1) + + expect(comparable(batched)).toEqual(comparable(sequential)) + expect(comparable(batched)).toMatchObject({ + message: { + content: 'Hello world', reasoning: 'plan done', state: 'complete', + tools: [expect.objectContaining({ callId: 'read', state: 'completed' })] + }, + contextMetrics: expect.objectContaining({ contextTokens: 77 }), + status: 'completed', + cursor: '3' + }) + } finally { + sequential.close() + batched.close() + } + }) + + it('deduplicates replayed events within and across batches and keeps their order', () => { + const database = openRecoverableTask() + try { + expect(database.appendRemoteConversationTaskEventsBatch({ + ...owner, events: page.slice(0, 3) + })).toEqual([true, true, true]) + // A reconnect replays page 1 before delivering page 2. + expect(database.appendRemoteConversationTaskEventsBatch({ + ...owner, events: [...page.slice(0, 3), ...page.slice(3, 7)] + })).toEqual([false, false, false, true, true, true, true]) + expect(database.appendRemoteConversationTaskEventsBatch({ + ...owner, events: page.slice(0, 7) + })).toEqual(page.slice(0, 7).map(() => false)) + const message = database.getConversation(conversationId).messages + .find((candidate) => candidate.id === assistantMessageId)! + expect(message).toMatchObject({ content: 'Hello', reasoning: 'plan ', status: 'working' }) + expect(message.blocks?.map((block) => block.type)).toEqual(['reasoning', 'text', 'tool']) + expect(taskEvents(database).map(({ kind }) => kind)).toEqual([ + 'reasoning', 'text', 'remote-semantic-checkpoint', + 'tool', 'status', 'context-metrics', 'remote-semantic-checkpoint' + ]) + expect(database.getTask(taskId).status).toBe('running') + } finally { + database.close() + } + }) + + it('rolls back the whole batch when one event conflicts', () => { + const database = openRecoverableTask() + try { + database.appendRemoteConversationTaskEventsBatch({ ...owner, events: page.slice(0, 3) }) + const before = comparable(database) + expect(() => database.appendRemoteConversationTaskEventsBatch({ + ...owner, + events: [ + page[3]!, + entry('1', 1, { requestId: taskId, type: 'text', delta: 'conflict' }) + ] + })).toThrow('conflicts') + expect(comparable(database)).toEqual(before) + } finally { + database.close() + } + }) + }) + + it('appends plain remote task event batches in one transaction with replay dedupe', () => { + const database = new AssistantDatabase(':memory:') + database.initialize('C:\\Workspace') + try { + const taskId = '00000000-0000-4000-8000-000000000721' + database.createTask({ id: taskId, title: '批量', instructions: '回复', workMode: 'ask' }) + const event = (semanticSequence: string, eventIndex: number, delta: string) => ({ + taskId, bindingId: 'plain-batch', operationId: taskId, semanticSequence, eventIndex, + kind: 'text', payload: { requestId: taskId, type: 'text', delta } + }) + const exec = vi.spyOn( + (database as unknown as { requireDatabase(): DatabaseSync }).requireDatabase(), + 'exec' + ) + expect(database.appendRemoteTaskEventsOnce([event('1', 0, 'a'), event('1', 1, 'b')])) + .toEqual([true, true]) + expect(exec.mock.calls.map(([sql]) => sql)).toEqual(['BEGIN IMMEDIATE', 'COMMIT']) + exec.mockRestore() + expect(database.appendRemoteTaskEventsOnce([event('1', 1, 'b'), event('2', 0, 'c')])) + .toEqual([false, true]) + expect(() => database.appendRemoteTaskEventsOnce([event('3', 0, 'd'), event('2', 0, 'x')])) + .toThrow('conflicts') + expect(((database as unknown as { requireDatabase(): DatabaseSync }).requireDatabase() + .prepare('SELECT payload_json FROM task_events WHERE task_id = ? AND remote_semantic_sequence IS NOT NULL ORDER BY id') + .all(taskId) as Array<{ payload_json: string }>) + .map(({ payload_json }) => (JSON.parse(payload_json) as { delta: string }).delta)) + .toEqual(['a', 'b', 'c']) + expect(database.appendRemoteTaskEventsOnce([])).toEqual([]) + } finally { + database.close() + } + }) + it.each(['failed', 'cancelled'] as const)('creates and recovers a remote-authoritative conversation task, then ends it as %s', async (terminalStatus) => { const directory = await mkdtemp( join(tmpdir(), 'goodbuddy-remote-conversation-recovery-') diff --git a/src/main/assistant/assistant-database.ts b/src/main/assistant/assistant-database.ts index 5f629ec1..ee4b4e92 100644 --- a/src/main/assistant/assistant-database.ts +++ b/src/main/assistant/assistant-database.ts @@ -161,6 +161,21 @@ export type RemoteConversationTaskEventInput = Omit< } } +export type RemoteConversationTaskEventsBatchInput = Pick< + RemoteConversationTaskEventInput, + 'taskId' | 'conversationId' | 'assistantMessageId' | 'runtimeSelection' +> & { + events: ReadonlyArray< + Pick< + RemoteConversationTaskEventInput, + 'bindingId' | 'operationId' | 'semanticSequence' | 'eventIndex' | 'event' + > + > +} + +/** Upper bound for one live remote persistence transaction. */ +export const REMOTE_TASK_EVENT_BATCH_MAXIMUM_EVENTS = 512 + export type RecoverableRemoteTask = { taskId: string projectId: string @@ -1728,6 +1743,42 @@ function reduceRecoveredAgentEvent( return conversationMessageSchema.parse(next) } +/** + * Projects the assistant-message fields that remote event persistence reads + * back from storage before reducing the next remote event. + */ +function storedRemoteAssistantMessage( + id: string, + content: string, + state: ConversationMessage['state'], + createdAt: number, + metadata: MessageMetadata +): ConversationMessage { + return { + id, + role: 'assistant', + content, + state, + createdAt, + status: metadata.status, + terminalStatus: metadata.terminalStatus, + reasoning: metadata.reasoning, + runtimeChecklist: metadata.runtimeChecklist, + blocks: metadata.blocks, + displayCaptureTruncated: metadata.displayCaptureTruncated, + contextCompression: metadata.contextCompression, + contextCompressions: metadata.contextCompressions, + tools: metadata.tools, + subagents: metadata.subagents, + sources: metadata.sources, + sourceReferences: metadata.sourceReferences, + knowledgeRetrieval: metadata.knowledgeRetrieval, + artifactIds: metadata.artifactIds, + task: metadata.task, + attachments: metadata.attachments, + answeredQuestions: metadata.answeredQuestions + } +} function stripRemoteEventProvenance( event: RemoteConversationTaskEventInput['event'] ): RemoteConversationTaskEventInput['event'] { @@ -6069,28 +6120,92 @@ export class AssistantDatabase { ) } + /** + * Appends many remote semantic events in ONE task-event transaction. + * Returns one flag per input event: true when the event was newly stored, + * false when the same provenance was already stored (replay dedupe). + */ + appendRemoteTaskEventsOnce( + events: readonly RemoteTaskEventInput[] + ): boolean[] { + if (events.length === 0) { + return [] + } + if (events.length > REMOTE_TASK_EVENT_BATCH_MAXIMUM_EVENTS) { + throw new RangeError( + `Remote task event batch must contain at most ${REMOTE_TASK_EVENT_BATCH_MAXIMUM_EVENTS} events` + ) + } + const validated = events.map(validateRemoteTaskEvent) + const database = this.requireDatabase() + database.exec('BEGIN IMMEDIATE') + try { + const inserted = validated.map((event) => + this.insertRemoteTaskEventOnce(database, event) + ) + database.exec('COMMIT') + return inserted + } catch (error) { + database.exec('ROLLBACK') + throw error + } + } + appendRemoteConversationTaskEventOnce( input: RemoteConversationTaskEventInput ): boolean { + return this.appendRemoteConversationTaskEventsBatch({ + taskId: input.taskId, + conversationId: input.conversationId, + assistantMessageId: input.assistantMessageId, + runtimeSelection: input.runtimeSelection, + events: [input] + })[0]! + } + + /** + * Persists remote semantic events of one recoverable task in ONE + * transaction. The assistant message and conversation row are read at most + * once and written at most once per batch, while events are still reduced + * one by one in their original order. Already stored events (same remote + * provenance) are skipped and reported as false. + */ + appendRemoteConversationTaskEventsBatch( + input: RemoteConversationTaskEventsBatchInput + ): boolean[] { + if (input.events.length === 0) { + return [] + } + if (input.events.length > REMOTE_TASK_EVENT_BATCH_MAXIMUM_EVENTS) { + throw new RangeError( + `Remote task event batch must contain at most ${REMOTE_TASK_EVENT_BATCH_MAXIMUM_EVENTS} events` + ) + } const conversationId = assistantIdSchema.parse(input.conversationId) const assistantMessageId = assistantIdSchema.parse( input.assistantMessageId ) const requestRuntimeSelection = optionalAgentRuntimeSelectionSchema.parse(input.runtimeSelection) - const publicEvent = stripRemoteEventProvenance(input.event) - if (publicEvent.requestId !== input.taskId) { - throw new Error('远程事件的请求 ID 与任务不匹配') - } - const event = validateRemoteTaskEvent({ - taskId: input.taskId, - bindingId: input.bindingId, - operationId: input.operationId, - semanticSequence: input.semanticSequence, - eventIndex: input.eventIndex, - kind: publicEvent.type, - payload: publicEvent + const prepared = input.events.map((item) => { + const publicEvent = stripRemoteEventProvenance(item.event) + if (publicEvent.requestId !== input.taskId) { + throw new Error('远程事件的请求 ID 与任务不匹配') + } + return { + publicEvent, + event: validateRemoteTaskEvent({ + taskId: input.taskId, + bindingId: item.bindingId, + operationId: item.operationId, + semanticSequence: item.semanticSequence, + eventIndex: item.eventIndex, + kind: publicEvent.type, + payload: publicEvent + }) + } }) + const taskId = prepared[0]!.event.taskId const database = this.requireDatabase() database.exec('BEGIN IMMEDIATE') try { @@ -6101,7 +6216,7 @@ export class AssistantDatabase { FROM tasks WHERE id = ? AND remote_recoverable = 1` ) - .get(event.taskId) as + .get(taskId) as | { project_id: string | null conversation_id: string | null @@ -6115,162 +6230,190 @@ export class AssistantDatabase { ) { throw new Error('远程事件与可恢复任务的消息归属不匹配') } - const inserted = this.insertRemoteTaskEventOnce(database, event) - if (!inserted) { - database.exec('COMMIT') - return false - } - if (publicEvent.type === 'remote-semantic-checkpoint') { - this.terminalizeRemoteTaskAtCheckpoint(database, event) - database.exec('COMMIT') - return true - } - const row = database - .prepare( - `SELECT id, role, content, state, metadata_json, created_at - FROM messages - WHERE id = ? AND conversation_id = ? AND request_id = ?` - ) - .get( - assistantMessageId, - conversationId, - event.taskId - ) as + const inserted: boolean[] = [] + let message: ConversationMessage | undefined + let lastReduced: ConversationMessage | undefined + let conversationState: | { - id: string - role: MessageRow['role'] - content: string - state: ConversationMessage['state'] - metadata_json: string - created_at: string + contextState: ReturnType + runtimeSelectionKey: ReturnType } | undefined - if (!row || row.role !== 'assistant') { - throw new Error('远程事件的助理消息不存在或不受任务管理') - } - const metadata = JSON.parse(row.metadata_json) as MessageMetadata - const message = conversationMessageSchema.parse({ - id: row.id, - role: 'assistant', - content: row.content, - state: row.state, - createdAt: - metadata.createdAt ?? Date.parse(row.created_at), - status: metadata.status, - terminalStatus: metadata.terminalStatus, - reasoning: metadata.reasoning, - runtimeChecklist: metadata.runtimeChecklist, - blocks: metadata.blocks, - displayCaptureTruncated: - metadata.displayCaptureTruncated, - contextCompression: metadata.contextCompression, - contextCompressions: metadata.contextCompressions, - tools: metadata.tools, - subagents: metadata.subagents, - sources: metadata.sources, - sourceReferences: metadata.sourceReferences, - knowledgeRetrieval: metadata.knowledgeRetrieval, - artifactIds: metadata.artifactIds, - task: metadata.task, - attachments: metadata.attachments, - answeredQuestions: metadata.answeredQuestions - }) - const reduced = reduceRecoveredAgentEvent(message, publicEvent) - const messageUpdate = database - .prepare( - `UPDATE messages - SET content = ?, state = ?, metadata_json = ? - WHERE id = ? AND conversation_id = ? AND request_id = ?` + const loadMessage = (): ConversationMessage => { + if (message) { + return message + } + const row = database + .prepare( + `SELECT id, role, content, state, metadata_json, created_at + FROM messages + WHERE id = ? AND conversation_id = ? AND request_id = ?` + ) + .get( + assistantMessageId, + conversationId, + taskId + ) as + | { + id: string + role: MessageRow['role'] + content: string + state: ConversationMessage['state'] + metadata_json: string + created_at: string + } + | undefined + if (!row || row.role !== 'assistant') { + throw new Error('远程事件的助理消息不存在或不受任务管理') + } + const metadata = JSON.parse(row.metadata_json) as MessageMetadata + message = conversationMessageSchema.parse( + storedRemoteAssistantMessage( + row.id, + row.content, + row.state, + metadata.createdAt ?? Date.parse(row.created_at), + metadata + ) ) - .run( + return message + } + const loadConversationState = (): NonNullable< + typeof conversationState + > => { + if (conversationState) { + return conversationState + } + const conversation = database + .prepare( + `SELECT c.runtime_selection_json, c.context_state_json, + p.runtime_selection_json AS project_runtime_selection_json + FROM conversations c + LEFT JOIN projects p ON p.id = c.project_id + WHERE c.id = ?` + ) + .get(conversationId) as + | { + runtime_selection_json: string | null + project_runtime_selection_json: string | null + context_state_json: string | null + } + | undefined + if (!conversation) { + throw new Error('远程事件的对话不存在') + } + conversationState = { + contextState: parseConversationContextState( + conversation.context_state_json + ), + runtimeSelectionKey: remoteEventSelectionKey( + requestRuntimeSelection, + parseRuntimeSelection(conversation.runtime_selection_json), + parseRuntimeSelection( + conversation.project_runtime_selection_json + ) + ) + } + return conversationState + } + for (const { event, publicEvent } of prepared) { + if (!this.insertRemoteTaskEventOnce(database, event)) { + inserted.push(false) + continue + } + inserted.push(true) + if (publicEvent.type === 'remote-semantic-checkpoint') { + this.terminalizeRemoteTaskAtCheckpoint(database, event) + continue + } + const reduced = reduceRecoveredAgentEvent(loadMessage(), publicEvent) + lastReduced = reduced + // Match sequential single-event writes exactly: each of them re-read + // only the stored metadata fields before reducing the next event. + message = storedRemoteAssistantMessage( + reduced.id, reduced.content, reduced.state, - serializeConversationMessageMetadata(reduced), - assistantMessageId, - conversationId, - event.taskId + reduced.createdAt, + reduced ) - if (messageUpdate.changes !== 1) { - throw new Error('远程事件的助理消息写入失败') - } - const conversation = database - .prepare( - `SELECT c.runtime_selection_json, c.context_state_json, - p.runtime_selection_json AS project_runtime_selection_json - FROM conversations c - LEFT JOIN projects p ON p.id = c.project_id - WHERE c.id = ?` - ) - .get(conversationId) as - | { - runtime_selection_json: string | null - project_runtime_selection_json: string | null - context_state_json: string | null + const state = loadConversationState() + if (publicEvent.type === 'context-metrics') { + state.contextState = { + ...state.contextState, + contextMetrics: { + runtimeSelectionKey: state.runtimeSelectionKey, + contextTokens: publicEvent.contextTokens, + source: publicEvent.source, + basis: 'model-call' + } } - | undefined - if (!conversation) { - throw new Error('远程事件的对话不存在') - } - let contextState = parseConversationContextState( - conversation.context_state_json - ) - const runtimeSelectionKey = remoteEventSelectionKey( - requestRuntimeSelection, - parseRuntimeSelection(conversation.runtime_selection_json), - parseRuntimeSelection(conversation.project_runtime_selection_json) - ) - if (publicEvent.type === 'context-metrics') { - contextState = { - ...contextState, - contextMetrics: { - runtimeSelectionKey, - contextTokens: publicEvent.contextTokens, - source: publicEvent.source, - basis: 'model-call' - } - } - } else if ( - publicEvent.type === 'context-compression' && - publicEvent.scope !== 'agent-run' - ) { - const estimatedAfterTokens = - publicEvent.estimatedAfterTokens - contextState = { - ...contextState, - ...(publicEvent.conversationState - ? { - contextCompressionState: - publicEvent.conversationState - } - : {}), - ...(publicEvent.state === 'completed' && - estimatedAfterTokens !== undefined - ? { - contextMetrics: { - runtimeSelectionKey, - contextTokens: estimatedAfterTokens, - source: 'estimated' as const, - basis: 'conversation' as const + } else if ( + publicEvent.type === 'context-compression' && + publicEvent.scope !== 'agent-run' + ) { + const estimatedAfterTokens = + publicEvent.estimatedAfterTokens + state.contextState = { + ...state.contextState, + ...(publicEvent.conversationState + ? { + contextCompressionState: + publicEvent.conversationState } - } - : {}) + : {}), + ...(publicEvent.state === 'completed' && + estimatedAfterTokens !== undefined + ? { + contextMetrics: { + runtimeSelectionKey: state.runtimeSelectionKey, + contextTokens: estimatedAfterTokens, + source: 'estimated' as const, + basis: 'conversation' as const + } + } + : {}) + } } } - database - .prepare( - `UPDATE conversations - SET context_state_json = ?, updated_at = ? - WHERE id = ?` - ) - .run( - serializeConversationContextState( - conversationContextStateSchema.parse(contextState) - ), - new Date().toISOString(), - conversationId - ) + if (lastReduced) { + const messageUpdate = database + .prepare( + `UPDATE messages + SET content = ?, state = ?, metadata_json = ? + WHERE id = ? AND conversation_id = ? AND request_id = ?` + ) + .run( + lastReduced.content, + lastReduced.state, + serializeConversationMessageMetadata(lastReduced), + assistantMessageId, + conversationId, + taskId + ) + if (messageUpdate.changes !== 1) { + throw new Error('远程事件的助理消息写入失败') + } + } + if (conversationState) { + database + .prepare( + `UPDATE conversations + SET context_state_json = ?, updated_at = ? + WHERE id = ?` + ) + .run( + serializeConversationContextState( + conversationContextStateSchema.parse( + conversationState.contextState + ) + ), + new Date().toISOString(), + conversationId + ) + } database.exec('COMMIT') - return true + return inserted } catch (error) { database.exec('ROLLBACK') throw error diff --git a/src/main/ipc.test.ts b/src/main/ipc.test.ts index ae25648e..50557c75 100644 --- a/src/main/ipc.test.ts +++ b/src/main/ipc.test.ts @@ -5383,13 +5383,25 @@ describe('registerIpcHandlers agent terminal state', () => { nativeTerminalManager?: Parameters[41], applicationSettingsStore?: ApplicationSettingsStore ) { + const appendRemoteTaskEventOnce = vi.fn< + (input: RemoteTaskEventMockInput) => boolean + >(() => true) + const appendRemoteConversationTaskEventOnce = + vi.fn(() => true) const assistantDatabase = { createTask: vi.fn(), appendTaskEvent: vi.fn(), - appendRemoteTaskEventOnce: vi.fn< - (input: RemoteTaskEventMockInput) => boolean - >(() => true), - appendRemoteConversationTaskEventOnce: vi.fn(() => true), + appendRemoteTaskEventOnce, + appendRemoteConversationTaskEventOnce, + // Batch writes delegate to the per-event mocks by default so tests can + // assert both per-event payloads and the number of transactions. + appendRemoteTaskEventsOnce: vi.fn< + (inputs: RemoteTaskEventMockInput[]) => boolean[] + >((inputs): boolean[] => inputs.map((input): boolean => appendRemoteTaskEventOnce(input))), + appendRemoteConversationTaskEventsBatch: vi.fn( + ({ events, ...owner }): boolean[] => events.map((event): boolean => + appendRemoteConversationTaskEventOnce({ ...owner, ...event })) + ), listConversations: vi.fn<() => ConversationSnapshot[]>(() => []), listConversationSummaries: vi.fn<(ids?: string[]) => ConversationSnapshot[]>(() => []), listRecoverableRemoteTasks: vi.fn< @@ -7342,7 +7354,7 @@ describe('registerIpcHandlers agent terminal state', () => { await harness.dispose() }) - it('durably commits remote text before resume, deduplicates it, and preserves its semantic payload', async () => { + it('durably commits remote text at its checkpoint before resume, deduplicates it, and preserves its semantic payload', async () => { const requestId = '00000000-0000-4000-8000-000000000763' const provenance = { source: 'remote-semantic-transcript' as const, @@ -7364,24 +7376,43 @@ describe('registerIpcHandlers agent terminal state', () => { delta: taggedDelta, remoteProvenance: provenance } as const + const checkpoint = { + requestId: request.requestId, + type: 'remote-semantic-checkpoint', + remoteProvenance: { ...provenance, eventIndex: 1 } + } as const yield remoteText + yield checkpoint + // The Agent may ACK the entry here: text and checkpoint must already + // be committed, together, in one batch. expect( - harness.assistantDatabase.appendRemoteTaskEventOnce - ).toHaveBeenCalledWith({ - taskId: request.requestId, - bindingId: provenance.bindingId, - operationId: provenance.operationId, - semanticSequence: provenance.semanticSequence, - eventIndex: provenance.eventIndex, - kind: 'text', - payload: { - requestId: request.requestId, - type: 'text', - delta: taggedDelta - } - }) + harness.assistantDatabase.appendRemoteTaskEventsOnce + ).toHaveBeenCalledOnce() + expect( + harness.assistantDatabase.appendRemoteTaskEventsOnce + ).toHaveBeenLastCalledWith([ + { + taskId: request.requestId, + bindingId: provenance.bindingId, + operationId: provenance.operationId, + semanticSequence: provenance.semanticSequence, + eventIndex: provenance.eventIndex, + kind: 'text', + payload: { + requestId: request.requestId, + type: 'text', + delta: taggedDelta + } + }, + expect.objectContaining({ + eventIndex: 1, + kind: 'remote-semantic-checkpoint' + }) + ]) resumed('first') + // Reconnect replay of the same transcript entry. yield remoteText + yield checkpoint resumed('duplicate') yield { requestId: request.requestId, @@ -7397,6 +7428,8 @@ describe('registerIpcHandlers agent terminal state', () => { const { harness } = fixture harness.assistantDatabase.appendRemoteTaskEventOnce .mockReturnValueOnce(true) + .mockReturnValueOnce(true) + .mockReturnValueOnce(false) .mockReturnValueOnce(false) .mockReturnValueOnce(true) @@ -7415,6 +7448,10 @@ describe('registerIpcHandlers agent terminal state', () => { ) expect(resumed.mock.calls).toEqual([['first'], ['duplicate']]) + // One transaction per checkpoint plus one for the terminal event. + expect( + harness.assistantDatabase.appendRemoteTaskEventsOnce + ).toHaveBeenCalledTimes(3) const persistedTextCalls = harness.assistantDatabase.appendRemoteTaskEventOnce.mock.calls.filter( ([input]) => input.kind === 'text' @@ -7444,6 +7481,136 @@ describe('registerIpcHandlers agent terminal state', () => { await harness.dispose() }) + it('batches a remote transcript page into one transaction and keeps order with local events', async () => { + const requestId = '00000000-0000-4000-8000-000000000767' + const conversationId = 'managed-ssh-remote-batch' + const remote = (semanticSequence: string, eventIndex: number) => ({ + source: 'remote-semantic-transcript' as const, + bindingId: 'binding-remote-batch', + operationId: requestId, + semanticSequence, + eventIndex + }) + const lifecycle: string[] = [] + const selectedRuntime = { + runtimeId: 'opencode', + capability: 'chat', + supportsToolExecution: true, + async *run(request: { requestId: string }) { + const id = request.requestId + yield { requestId: id, type: 'reasoning', delta: 'think', remoteProvenance: remote('1', 0) } as const + yield { requestId: id, type: 'text', delta: 'A', remoteProvenance: remote('1', 1) } as const + yield { requestId: id, type: 'text', delta: 'B', remoteProvenance: remote('1', 2) } as const + // Replayed event from an earlier page: already stored. + yield { requestId: id, type: 'text', delta: 'old', remoteProvenance: remote('1', 3) } as const + lifecycle.push('before-checkpoint') + yield { requestId: id, type: 'remote-semantic-checkpoint', remoteProvenance: remote('1', 4) } as const + lifecycle.push('after-checkpoint') + yield { requestId: id, type: 'text', delta: 'C', remoteProvenance: remote('2', 0) } as const + // A local event forces the pending remote batch to commit first. + yield { requestId: id, type: 'status', message: 'local status' } as const + lifecycle.push('after-local') + yield { requestId: id, type: 'remote-semantic-checkpoint', remoteProvenance: remote('2', 1) } as const + yield { requestId: id, type: 'done', remoteProvenance: remote('3', 0) } as const + yield { requestId: id, type: 'remote-semantic-checkpoint', remoteProvenance: remote('3', 1) } as const + } + } + const { harness, projectId } = createManagedSshHarness(selectedRuntime) + harness.assistantDatabase.getConversation.mockReturnValue({ + id: conversationId, projectId, messages: [] + }) + harness.assistantDatabase.appendRemoteConversationTaskEventsBatch.mockImplementation( + ({ events }) => { + lifecycle.push(`batch:${events.map(({ event }) => event.type).join(',')}`) + return events.map(({ event }) => !(event.type === 'text' && event.delta === 'old')) + } + ) + await harness.handler?.(trustedEvent(harness.webContents), { + ...managedSshRequest(projectId, requestId, conversationId), + currentUserMessageId: '00000000-0000-4000-8000-000000000768', + currentAssistantMessageId: '00000000-0000-4000-8000-000000000769' + }) + await vi.waitFor(() => expect(lifecycle).toContain('batch:remote-semantic-checkpoint')) + + expect(lifecycle).toEqual([ + 'before-checkpoint', + 'batch:reasoning,text,text,text,remote-semantic-checkpoint', + 'after-checkpoint', + 'batch:text', + 'after-local', + 'batch:remote-semantic-checkpoint', + 'batch:done', + 'batch:remote-semantic-checkpoint' + ]) + expect(harness.assistantDatabase.appendRemoteConversationTaskEventOnce).not.toHaveBeenCalled() + expect(harness.assistantDatabase.appendRemoteTaskEventOnce).not.toHaveBeenCalled() + const publicEvents = harness.webContents.send.mock.calls + .filter(([channel]) => channel === ipcChannels.agentEvent) + .map(([, event]) => event) + .filter((event) => event.requestId === requestId) + // Renderer coalescing may merge adjacent text deltas; compare the stream. + const stream: string[] = [] + for (const event of publicEvents) { + const previous = stream.at(-1) + if ((event.type === 'text' || event.type === 'reasoning') && previous?.startsWith(`${event.type}:`)) { + stream[stream.length - 1] = `${previous}${event.delta}` + } else { + stream.push(event.type === 'text' || event.type === 'reasoning' + ? `${event.type}:${event.delta}` + : event.type === 'status' ? `status:${event.message}` : event.type) + } + } + expect(stream).toEqual(['reasoning:think', 'text:ABC', 'status:local status', 'done']) + expect(publicEvents.some((event) => event.type === 'text' && event.delta.includes('old'))).toBe(false) + await harness.dispose() + }) + + it('fails the run without forwarding or acknowledging when a remote batch commit fails', async () => { + const requestId = '00000000-0000-4000-8000-000000000770' + const remote = (eventIndex: number) => ({ + source: 'remote-semantic-transcript' as const, + bindingId: 'binding-remote-batch-failure', + operationId: requestId, + semanticSequence: '1', + eventIndex + }) + const resumedAfterCheckpoint = vi.fn() + const selectedRuntime = { + runtimeId: 'opencode', + capability: 'chat', + supportsToolExecution: true, + async *run(request: { requestId: string }) { + yield { requestId: request.requestId, type: 'text', delta: 'lost', remoteProvenance: remote(0) } as const + yield { requestId: request.requestId, type: 'remote-semantic-checkpoint', remoteProvenance: remote(1) } as const + resumedAfterCheckpoint() + } + } + const fixture = createManagedSshHarness(selectedRuntime) + const { harness } = fixture + harness.assistantDatabase.appendRemoteTaskEventsOnce.mockImplementation(() => { + throw new Error('disk full') + }) + await harness.handler?.( + trustedEvent(harness.webContents), + managedSshRequest(fixture.projectId, requestId, 'managed-ssh-batch-failure') + ) + await vi.waitFor(() => + expect(harness.assistantDatabase.updateTaskStatus).toHaveBeenCalledWith( + requestId, + expect.stringMatching(/^(failed|interrupted)$/u), + expect.any(String) + ) + ) + expect(resumedAfterCheckpoint).not.toHaveBeenCalled() + const publicEvents = harness.webContents.send.mock.calls + .filter(([channel]) => channel === ipcChannels.agentEvent) + .map(([, event]) => event) + .filter((event) => event.requestId === requestId) + expect(publicEvents.some((event) => event.type === 'text')).toBe(false) + expect(publicEvents.at(-1)).toMatchObject({ type: 'error' }) + await harness.dispose() + }) + it('persists a remote semantic checkpoint without publishing it', async () => { const requestId = '00000000-0000-4000-8000-000000000764' const provenance = { diff --git a/src/main/ipc.ts b/src/main/ipc.ts index 23c19525..30764108 100644 --- a/src/main/ipc.ts +++ b/src/main/ipc.ts @@ -245,7 +245,8 @@ import type { ToolApprovalBroker } from './tool-approval-broker' import { registerClipboardIpcHandlers, registerWindowIpcHandlers } from './window-ipc' import type { AssistantDatabase, - RecoverableRemoteTask + RecoverableRemoteTask, + RemoteConversationTaskEventInput } from './assistant/assistant-database' import { RemoteDelegationService } from './assistant/remote-delegation-service' import { @@ -316,6 +317,10 @@ import type { GoodBuddyConfigService } from './goodbuddy-config-service' import { weixinVerificationInputSchema } from '../shared/weixin-channel-contracts' import type { RemoteChannelActivity } from '../shared/remote-channel-contracts' import { AgentEventBuffer } from './agent-event-buffer' +import { + RemoteEventBatcher, + type RemoteEventBatchEntry +} from './remote-event-batcher' import { withImageConversationContext } from './agent/image-conversation-context' import type { ImageGenerationService } from './agent/image-generation-service' import { imageOperationTargetSchema } from '../shared/image-operation-ipc' @@ -611,6 +616,47 @@ function stripRemoteSemanticProvenance( return publicEvent } +type RemoteBatchEvent = Exclude< + RemoteConversationTaskEventInput['event'], + { type: 'checkpoint' } +> + +/** + * Remote events are committed in one transaction per remote semantic + * checkpoint. The owned-prompt poller yields a checkpoint after each transcript + * entry and may acknowledge the page as soon as it is resumed after one, so a + * checkpoint is the latest point at which buffered events can still be + * committed before the Agent ACK. + */ +function remoteTaskEventBatchInputs( + taskId: string, + entries: readonly RemoteEventBatchEntry[] +): Parameters[0] { + return entries.map(({ provenance, event }) => ({ + taskId, + bindingId: provenance.bindingId, + operationId: provenance.operationId, + semanticSequence: provenance.semanticSequence, + eventIndex: provenance.eventIndex, + kind: event.type, + payload: event + })) +} + +function remoteConversationBatchEvents( + entries: readonly RemoteEventBatchEntry[] +): Parameters< + AssistantDatabase['appendRemoteConversationTaskEventsBatch'] +>[0]['events'] { + return entries.map(({ provenance, event }) => ({ + bindingId: provenance.bindingId, + operationId: provenance.operationId, + semanticSequence: provenance.semanticSequence, + eventIndex: provenance.eventIndex, + event + })) +} + function createPromiseTracker(): { track(operation: Promise): Promise drain(): Promise @@ -1580,6 +1626,18 @@ export function registerIpcHandlers( lease.detachOnApplicationExit = true lease.recoveredMessageId = task.currentAssistantMessageId let sawTerminal = false + // Replayed transcript entries commit in one transaction per checkpoint. + const remoteEventBatcher = new RemoteEventBatcher({ + onError: (error) => controller.abort(error), + persist: (entries) => + assistantDatabase.appendRemoteConversationTaskEventsBatch({ + taskId: task.taskId, + conversationId: task.conversationId, + runtimeSelection, + assistantMessageId: task.currentAssistantMessageId, + events: remoteConversationBatchEvents(entries) + }) + }) try { publishRemoteProjectRecovery({ projectId: task.projectId, @@ -1631,6 +1689,7 @@ export function registerIpcHandlers( controller.signal )) { if (rawEvent.type === 'question' || rawEvent.type === 'question-resolved') { + remoteEventBatcher.flush() if (rawEvent.type === 'question') { assistantDatabase.recordRemoteTaskQuestionArrival(task.taskId, rawEvent.questionId) pendingAgentQuestions.set(rawEvent.questionId, { @@ -1710,17 +1769,15 @@ export function registerIpcHandlers( } } } - assistantDatabase.appendRemoteConversationTaskEventOnce({ - taskId: task.taskId, - conversationId: task.conversationId, - runtimeSelection, - assistantMessageId: task.currentAssistantMessageId, - bindingId: provenance.bindingId, - operationId: provenance.operationId, - semanticSequence: provenance.semanticSequence, - eventIndex: provenance.eventIndex, - event - }) + remoteEventBatcher.add(provenance, event) + if ( + event.type === 'remote-semantic-checkpoint' || + event.type === 'done' || + event.type === 'error' + ) { + // Commit before the Agent ACK (checkpoint) or terminal handling. + remoteEventBatcher.flush() + } if (event.type === 'remote-semantic-checkpoint') { publishRemoteProjectRecovery({ projectId: task.projectId, @@ -1734,10 +1791,16 @@ export function registerIpcHandlers( sawTerminal = true } } + remoteEventBatcher.flush() if (!sawTerminal) { throw new Error('远端 Agent 恢复流未提供任务终态') } } catch (error) { + try { + remoteEventBatcher.flush() + } catch { + // The original error decides the outcome; nothing was acknowledged. + } if (sawTerminal) { return } @@ -1757,6 +1820,7 @@ export function registerIpcHandlers( } throw error } finally { + remoteEventBatcher.dispose() for (const [id, pending] of pendingAgentQuestions) { if (pending.requestId === task.taskId) pendingAgentQuestions.delete(id) } @@ -2127,6 +2191,15 @@ export function registerIpcHandlers( ) } }) + // Remote semantic events commit in one transaction per checkpoint (see + // remoteTaskEventBatchInputs) instead of one transaction per event. + const remoteEventBatcher = new RemoteEventBatcher({ + onError: (error) => controller.abort(error), + persist: (entries) => + assistantDatabase.appendRemoteTaskEventsOnce( + remoteTaskEventBatchInputs(taskId, entries) + ) + }) try { const requestRuntime = remoteContext?.runtime ?? @@ -2292,20 +2365,16 @@ export function registerIpcHandlers( const provenance = remoteSemanticProvenance(agentEvent) if (provenance !== undefined) { activeRequestLease.detachOnApplicationExit = true + } else { + remoteEventBatcher.flush() } if (agentEvent.type === 'remote-semantic-checkpoint') { - assistantDatabase.appendRemoteTaskEventOnce({ - taskId, - bindingId: provenance!.bindingId, - operationId: provenance!.operationId, - semanticSequence: provenance!.semanticSequence, - eventIndex: provenance!.eventIndex, - kind: agentEvent.type, - payload: { - requestId: agentEvent.requestId, - type: agentEvent.type - } + // The Agent may acknowledge this entry once the generator resumes. + remoteEventBatcher.add(provenance!, { + requestId: agentEvent.requestId, + type: agentEvent.type }) + remoteEventBatcher.flush() continue } if (agentEvent.type === 'model-usage') { @@ -2315,17 +2384,9 @@ export function registerIpcHandlers( callId: agentEvent.callId }) if (provenance !== undefined) { - assistantDatabase.appendRemoteTaskEventOnce({ - taskId, - bindingId: provenance.bindingId, - operationId: provenance.operationId, - semanticSequence: provenance.semanticSequence, - eventIndex: provenance.eventIndex, - kind: 'remote-semantic-checkpoint', - payload: { - requestId: agentEvent.requestId, - type: 'remote-semantic-checkpoint' - } + remoteEventBatcher.add(provenance, { + requestId: agentEvent.requestId, + type: 'remote-semantic-checkpoint' }) } continue @@ -2378,6 +2439,7 @@ export function registerIpcHandlers( artifactIds.push(taskEvent.artifactId) } if (taskEvent.type === 'question') { + remoteEventBatcher.flush() const error = new Error( '后台任务无法回答 Runtime 交互提问。请改为在 GoodBuddy 对话中运行，或调整提示词和工具配置以避免交互提问。' ) @@ -2411,15 +2473,10 @@ export function registerIpcHandlers( if (provenance === undefined) { eventBuffer.push(taskEvent) } else { - assistantDatabase.appendRemoteTaskEventOnce({ - taskId, - bindingId: provenance.bindingId, - operationId: provenance.operationId, - semanticSequence: provenance.semanticSequence, - eventIndex: provenance.eventIndex, - kind: taskEvent.type, - payload: taskEvent - }) + remoteEventBatcher.add(provenance, taskEvent) + if (taskEvent.type === 'done' || taskEvent.type === 'error') { + remoteEventBatcher.flush() + } } if (taskEvent.type === 'tool' && remoteContext) { publishRemoteActivity({ @@ -2456,6 +2513,7 @@ export function registerIpcHandlers( completed = true } } + remoteEventBatcher.flush() if (!completed) { throw new Error('Agent Runtime 未报告任务完成，定时任务已失败') } @@ -2511,6 +2569,11 @@ export function registerIpcHandlers( ...(artifactIds.length > 0 ? { artifactIds } : {}) } } catch (error) { + try { + remoteEventBatcher.flush() + } catch { + // The original error already fails the task; nothing was acknowledged. + } eventBuffer.flush() const message = backgroundQuestionError ? backgroundQuestionError.message @@ -2539,6 +2602,7 @@ export function registerIpcHandlers( error: message } } finally { + remoteEventBatcher.dispose() eventBuffer.close() externalSignal?.removeEventListener( 'abort', @@ -3919,76 +3983,81 @@ export function registerIpcHandlers( publicEventBuffer.flush() } } + // Remote semantic events are committed in batches. A batch is closed + // (one SQLite transaction) at the next remote semantic checkpoint, before + // any non-batched event, at the size cap, or by a short safety timer. + let remoteEventBatchClosed = false + const remoteEventBatcher = new RemoteEventBatcher({ + onError: (error) => controller.abort(error), + persist: (entries) => + remoteConversationRecovery + ? assistantDatabase.appendRemoteConversationTaskEventsBatch({ + taskId: request.requestId, + conversationId: request.conversationId, + runtimeSelection: remoteEventSelection, + assistantMessageId: + remoteConversationRecovery.currentAssistantMessageId, + events: remoteConversationBatchEvents(entries) + }) + : assistantDatabase.appendRemoteTaskEventsOnce( + remoteTaskEventBatchInputs(request.requestId, entries) + ) + }) + const flushRemoteEvents = (): void => { + if (!remoteEventBatchClosed) { + remoteEventBatcher.flush() + } + } const eventBuffer = { push: (event: AgentEvent): void => { + flushRemoteEvents() pushPublicEvent(event) persistedEventBuffer.push(event) }, pushPublic: pushPublicEvent, flush: (): void => { + flushRemoteEvents() publicStreamType = undefined publicEventBuffer.flush() persistedEventBuffer.flush() }, close: (): void => { + remoteEventBatchClosed = true + remoteEventBatcher.dispose() publicEventBuffer.close() persistedEventBuffer.close() } } + /** + * Queues a remote public event. It is forwarded to the renderer only + * after its batch commits, and only when it was newly stored. With + * `immediate`, the batch (including this event) is committed before + * returning so callers can rely on its durability right away. + */ const persistRemotePublicEvent = ( provenance: RemoteSemanticEventProvenance, - publicEvent: AgentEvent - ): boolean => - remoteConversationRecovery - ? assistantDatabase.appendRemoteConversationTaskEventOnce({ - taskId: request.requestId, - conversationId: request.conversationId, - runtimeSelection: remoteEventSelection, - assistantMessageId: - remoteConversationRecovery.currentAssistantMessageId, - bindingId: provenance.bindingId, - operationId: provenance.operationId, - semanticSequence: provenance.semanticSequence, - eventIndex: provenance.eventIndex, - event: publicEvent - }) - : assistantDatabase.appendRemoteTaskEventOnce({ - taskId: request.requestId, - bindingId: provenance.bindingId, - operationId: provenance.operationId, - semanticSequence: provenance.semanticSequence, - eventIndex: provenance.eventIndex, - kind: publicEvent.type, - payload: publicEvent - }) + publicEvent: AgentEvent, + immediate: boolean + ): void => { + remoteEventBatcher.add(provenance, publicEvent, (event, inserted) => { + if (inserted) { + eventBuffer.pushPublic(event as AgentEvent) + } + }) + if (immediate) { + flushRemoteEvents() + } + } + // The Agent may acknowledge the transcript entry once the generator is + // resumed after its checkpoint, so the checkpoint always commits the + // whole pending batch together with itself. const persistRemoteCheckpoint = ( provenance: RemoteSemanticEventProvenance, requestId: string, type: 'remote-semantic-checkpoint' - ): boolean => { - const checkpoint = { requestId, type } as const - return remoteConversationRecovery - ? assistantDatabase.appendRemoteConversationTaskEventOnce({ - taskId: request.requestId, - conversationId: request.conversationId, - runtimeSelection: remoteEventSelection, - assistantMessageId: - remoteConversationRecovery.currentAssistantMessageId, - bindingId: provenance.bindingId, - operationId: provenance.operationId, - semanticSequence: provenance.semanticSequence, - eventIndex: provenance.eventIndex, - event: checkpoint - }) - : assistantDatabase.appendRemoteTaskEventOnce({ - taskId: request.requestId, - bindingId: provenance.bindingId, - operationId: provenance.operationId, - semanticSequence: provenance.semanticSequence, - eventIndex: provenance.eventIndex, - kind: type, - payload: checkpoint - }) + ): void => { + remoteEventBatcher.add(provenance, { requestId, type }) + flushRemoteEvents() } activeEventBuffers.set(request.requestId, eventBuffer) const toolStates = new Map< @@ -4263,6 +4332,11 @@ export function registerIpcHandlers( : runSmartRoute() for await (const agentEvent of splitTaggedReasoning(eventStream)) { const provenance = remoteSemanticProvenance(agentEvent) + if (provenance === undefined) { + // Keep renderer and storage order: earlier remote events commit + // and publish before any event that bypasses the remote batch. + flushRemoteEvents() + } if (agentEvent.type === 'remote-semantic-checkpoint') { persistRemoteCheckpoint( agentEvent.remoteProvenance, @@ -4290,13 +4364,12 @@ export function registerIpcHandlers( )! if (provenance === undefined) { eventBuffer.push(contextMetricsEvent) - } else if ( + } else { persistRemotePublicEvent( provenance, - contextMetricsEvent + contextMetricsEvent, + false ) - ) { - eventBuffer.pushPublic(contextMetricsEvent) } } else if (provenance !== undefined) { persistRemoteCheckpoint( @@ -4332,6 +4405,9 @@ export function registerIpcHandlers( toolStates.delete(publicEvent.runtimeCallId) } if (publicEvent.type === 'question') { + // The arrival rewrites the assistant message; commit earlier + // remote events first so message blocks keep their order. + flushRemoteEvents() assistantDatabase.recordRemoteTaskQuestionArrival(request.requestId, publicEvent.questionId) pendingAgentQuestions.set(publicEvent.questionId, { requestId: request.requestId, @@ -4378,10 +4454,14 @@ export function registerIpcHandlers( } else { eventBuffer.push(publicEvent) } - } else if ( - persistRemotePublicEvent(provenance, publicEvent) - ) { - eventBuffer.pushPublic(publicEvent) + } else { + // Terminal events commit immediately: task status, notification + // and failure handling below rely on them being durable. + persistRemotePublicEvent( + provenance, + publicEvent, + publicEvent.type === 'done' || publicEvent.type === 'error' + ) } if ( provenance !== undefined && @@ -4415,10 +4495,19 @@ export function registerIpcHandlers( } } } + flushRemoteEvents() if (!completed) { throw new Error('Agent Runtime 未报告任务完成，任务已标记为失败') } } catch (error) { + try { + // Commit what the remote Agent already produced. Nothing of a + // failed batch was acknowledged, so recovery can replay it. + flushRemoteEvents() + } catch { + remoteEventBatchClosed = true + remoteEventBatcher.dispose() + } publishReferences() eventBuffer.flush() const cancelled = diff --git a/src/main/remote-event-batcher.test.ts b/src/main/remote-event-batcher.test.ts new file mode 100644 index 00000000..c705c205 --- /dev/null +++ b/src/main/remote-event-batcher.test.ts @@ -0,0 +1,66 @@ +import { afterEach, describe, expect, it, vi } from 'vitest' +import { RemoteEventBatcher } from './remote-event-batcher' + +const provenance = (eventIndex: number) => ({ + source: 'remote-semantic-transcript' as const, + bindingId: 'binding', + operationId: 'operation', + semanticSequence: '1', + eventIndex +}) + +describe('RemoteEventBatcher', () => { + afterEach(() => { + vi.useRealTimers() + }) + + it('commits buffered events once and reports them in order', () => { + const persist = vi.fn((entries: readonly { event: string }[]) => + entries.map(({ event }) => event !== 'replayed')) + const committed: Array<[string, boolean]> = [] + const batcher = new RemoteEventBatcher({ persist }) + for (const [index, event] of ['a', 'replayed', 'b'].entries()) { + batcher.add(provenance(index), event, (value, inserted) => committed.push([value, inserted])) + } + expect(persist).not.toHaveBeenCalled() + batcher.flush() + batcher.flush() + expect(persist).toHaveBeenCalledOnce() + expect(committed).toEqual([['a', true], ['replayed', false], ['b', true]]) + batcher.dispose() + }) + + it('flushes at the size cap and on the safety timer', () => { + vi.useFakeTimers() + const persist = vi.fn((entries: readonly unknown[]) => entries.map(() => true)) + const batcher = new RemoteEventBatcher({ persist, maximumEvents: 2, flushIntervalMs: 50 }) + batcher.add(provenance(0), 0) + batcher.add(provenance(1), 1) + expect(persist).toHaveBeenCalledTimes(1) + batcher.add(provenance(2), 2) + vi.advanceTimersByTime(49) + expect(persist).toHaveBeenCalledTimes(1) + vi.advanceTimersByTime(1) + expect(persist).toHaveBeenCalledTimes(2) + expect(persist.mock.calls[1]![0]).toHaveLength(1) + batcher.dispose() + }) + + it('rethrows a failed commit and stays failed so later checkpoints are never committed', () => { + vi.useFakeTimers() + const onError = vi.fn() + const persist = vi.fn((): boolean[] => { + throw new Error('disk full') + }) + const onCommitted = vi.fn() + const batcher = new RemoteEventBatcher({ persist, onError }) + batcher.add(provenance(0), 'a', onCommitted) + vi.advanceTimersByTime(50) + expect(onError).toHaveBeenCalledWith(new Error('disk full')) + expect(() => batcher.add(provenance(1), 'checkpoint')).toThrow('disk full') + expect(() => batcher.flush()).toThrow('disk full') + expect(persist).toHaveBeenCalledOnce() + expect(onCommitted).not.toHaveBeenCalled() + batcher.dispose() + }) +}) diff --git a/src/main/remote-event-batcher.ts b/src/main/remote-event-batcher.ts new file mode 100644 index 00000000..b1a936d0 --- /dev/null +++ b/src/main/remote-event-batcher.ts @@ -0,0 +1,149 @@ +import type { RemoteSemanticEventProvenance } from './agent/runtime' + +export type RemoteEventBatchEntry = { + provenance: RemoteSemanticEventProvenance + event: TEvent +} + +export type RemoteEventCommitted = ( + event: TEvent, + inserted: boolean +) => void + +export type RemoteEventBatcherOptions = { + /** + * Durably stores every entry in ONE transaction and returns one flag per + * entry: true when it was newly stored, false for an already stored replay. + */ + persist(entries: readonly RemoteEventBatchEntry[]): readonly boolean[] + /** Called when a timer-driven flush fails (for example to abort the run). */ + onError?(error: unknown): void + /** Safety flush for buffered events that are not followed by a boundary. */ + flushIntervalMs?: number + maximumEvents?: number +} + +type PendingEntry = RemoteEventBatchEntry & { + onCommitted?: RemoteEventCommitted +} + +export const REMOTE_EVENT_BATCH_FLUSH_INTERVAL_MS = 50 +export const REMOTE_EVENT_BATCH_MAXIMUM_EVENTS = 256 + +/** + * Buffers remote semantic events so one SQLite transaction covers a whole + * remote transcript sequence instead of one transaction per event. + * + * Callers must flush before any point where the producer may acknowledge the + * buffered events (the remote semantic checkpoint), and before publishing any + * unrelated event so renderer order is preserved. `onCommitted` callbacks run + * after the commit, in the original order, with `inserted = false` for + * replayed events that were already stored. + * + * A failed commit stores nothing from that batch and poisons the batcher: + * every later `add` or `flush` rethrows the same error, so a later checkpoint + * can never be committed (and acknowledged) after an earlier batch was lost. + */ +export class RemoteEventBatcher { + private readonly persist: RemoteEventBatcherOptions['persist'] + private readonly onError: RemoteEventBatcherOptions['onError'] + private readonly flushIntervalMs: number + private readonly maximumEvents: number + private pending: PendingEntry[] = [] + private timer: ReturnType | undefined + private failure: { error: unknown } | undefined + + constructor(options: RemoteEventBatcherOptions) { + this.persist = options.persist + this.onError = options.onError + this.flushIntervalMs = + options.flushIntervalMs ?? REMOTE_EVENT_BATCH_FLUSH_INTERVAL_MS + this.maximumEvents = + options.maximumEvents ?? REMOTE_EVENT_BATCH_MAXIMUM_EVENTS + if (this.flushIntervalMs <= 0) { + throw new Error('flushIntervalMs must be positive') + } + if (!Number.isSafeInteger(this.maximumEvents) || this.maximumEvents < 1) { + throw new Error('maximumEvents must be a positive integer') + } + } + + get size(): number { + return this.pending.length + } + + add( + provenance: RemoteSemanticEventProvenance, + event: TEvent, + onCommitted?: RemoteEventCommitted + ): void { + this.throwIfFailed() + this.pending.push({ provenance, event, onCommitted }) + if (this.pending.length >= this.maximumEvents) { + this.flush() + return + } + this.scheduleFlush() + } + + flush(): void { + this.clearTimer() + this.throwIfFailed() + if (this.pending.length === 0) { + return + } + const entries = this.pending + this.pending = [] + let inserted: readonly boolean[] + try { + inserted = this.persist( + entries.map(({ provenance, event }) => ({ provenance, event })) + ) + if (inserted.length !== entries.length) { + throw new Error('远程事件批量写入结果数量不匹配') + } + } catch (error) { + this.failure = { error } + throw error + } + for (let index = 0; index < entries.length; index += 1) { + const entry = entries[index]! + entry.onCommitted?.(entry.event, inserted[index] === true) + } + } + + /** Stops the safety timer and drops uncommitted events. */ + dispose(): void { + this.clearTimer() + this.pending = [] + } + + private throwIfFailed(): void { + if (this.failure) { + throw this.failure.error + } + } + + private scheduleFlush(): void { + if (this.timer) { + return + } + this.timer = setTimeout(() => { + this.timer = undefined + try { + this.flush() + } catch (error) { + this.onError?.(error) + } + }, this.flushIntervalMs) + this.timer.unref?.() + } + + private clearTimer(): void { + if (!this.timer) { + return + } + clearTimeout(this.timer) + this.timer = undefined + } +}