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.
This commit is contained in:
mesalogo
@@ -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-')
|
||||
|
||||
@@ -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<typeof parseConversationContextState>
|
||||
runtimeSelectionKey: ReturnType<typeof remoteEventSelectionKey>
|
||||
}
|
||||
| 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
|
||||
|
||||
+186
-19
@@ -5383,13 +5383,25 @@ describe('registerIpcHandlers agent terminal state', () => {
|
||||
nativeTerminalManager?: Parameters<typeof registerIpcHandlers>[41],
|
||||
applicationSettingsStore?: ApplicationSettingsStore
|
||||
) {
|
||||
const appendRemoteTaskEventOnce = vi.fn<
|
||||
(input: RemoteTaskEventMockInput) => boolean
|
||||
>(() => true)
|
||||
const appendRemoteConversationTaskEventOnce =
|
||||
vi.fn<AssistantDatabase['appendRemoteConversationTaskEventOnce']>(() => true)
|
||||
const assistantDatabase = {
|
||||
createTask: vi.fn(),
|
||||
appendTaskEvent: vi.fn(),
|
||||
appendRemoteTaskEventOnce: vi.fn<
|
||||
(input: RemoteTaskEventMockInput) => boolean
|
||||
>(() => true),
|
||||
appendRemoteConversationTaskEventOnce: vi.fn<AssistantDatabase['appendRemoteConversationTaskEventOnce']>(() => 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<AssistantDatabase['appendRemoteConversationTaskEventsBatch']>(
|
||||
({ 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 = {
|
||||
|
||||
+188
-99
@@ -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<RemoteBatchEvent>[]
|
||||
): Parameters<AssistantDatabase['appendRemoteTaskEventsOnce']>[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<RemoteBatchEvent>[]
|
||||
): 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<T>(operation: Promise<T>): Promise<T>
|
||||
drain(): Promise<void>
|
||||
@@ -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<RemoteBatchEvent>({
|
||||
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<RemoteBatchEvent>({
|
||||
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<RemoteBatchEvent>({
|
||||
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 =
|
||||
|
||||
@@ -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<string>({ 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<number>({ 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<string>({ 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()
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,149 @@
|
||||
import type { RemoteSemanticEventProvenance } from './agent/runtime'
|
||||
|
||||
export type RemoteEventBatchEntry<TEvent> = {
|
||||
provenance: RemoteSemanticEventProvenance
|
||||
event: TEvent
|
||||
}
|
||||
|
||||
export type RemoteEventCommitted<TEvent> = (
|
||||
event: TEvent,
|
||||
inserted: boolean
|
||||
) => void
|
||||
|
||||
export type RemoteEventBatcherOptions<TEvent> = {
|
||||
/**
|
||||
* 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<TEvent>[]): 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<TEvent> = RemoteEventBatchEntry<TEvent> & {
|
||||
onCommitted?: RemoteEventCommitted<TEvent>
|
||||
}
|
||||
|
||||
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<TEvent> {
|
||||
private readonly persist: RemoteEventBatcherOptions<TEvent>['persist']
|
||||
private readonly onError: RemoteEventBatcherOptions<TEvent>['onError']
|
||||
private readonly flushIntervalMs: number
|
||||
private readonly maximumEvents: number
|
||||
private pending: PendingEntry<TEvent>[] = []
|
||||
private timer: ReturnType<typeof setTimeout> | undefined
|
||||
private failure: { error: unknown } | undefined
|
||||
|
||||
constructor(options: RemoteEventBatcherOptions<TEvent>) {
|
||||
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<TEvent>
|
||||
): 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
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user