feat: improve application responsiveness
This commit is contained in:
mesalogo
@@ -0,0 +1,177 @@
|
||||
import { afterEach, describe, expect, it, vi } from 'vitest'
|
||||
import type { AgentEvent } from '../shared/contracts'
|
||||
import { AgentEventBuffer } from './agent-event-buffer'
|
||||
|
||||
const requestId = '00000000-0000-4000-8000-000000000001'
|
||||
|
||||
describe('AgentEventBuffer', () => {
|
||||
afterEach(() => {
|
||||
vi.useRealTimers()
|
||||
})
|
||||
|
||||
it('combines 100 adjacent text deltas into one event', () => {
|
||||
vi.useFakeTimers()
|
||||
const events: AgentEvent[] = []
|
||||
const buffer = new AgentEventBuffer({
|
||||
onEvent: (event) => events.push(event)
|
||||
})
|
||||
|
||||
for (let index = 0; index < 100; index += 1) {
|
||||
buffer.push({ requestId, type: 'text', delta: `${index},` })
|
||||
}
|
||||
expect(events).toEqual([])
|
||||
|
||||
buffer.close()
|
||||
expect(events).toEqual([
|
||||
{
|
||||
requestId,
|
||||
type: 'text',
|
||||
delta: Array.from({ length: 100 }, (_, index) => `${index},`).join(
|
||||
''
|
||||
)
|
||||
}
|
||||
])
|
||||
})
|
||||
|
||||
it('preserves text, reasoning, and immediate tool order', () => {
|
||||
vi.useFakeTimers()
|
||||
const events: AgentEvent[] = []
|
||||
const buffer = new AgentEventBuffer({
|
||||
onEvent: (event) => events.push(event)
|
||||
})
|
||||
|
||||
buffer.push({ requestId, type: 'text', delta: 'answer' })
|
||||
buffer.push({ requestId, type: 'reasoning', delta: 'thought' })
|
||||
buffer.push({
|
||||
requestId,
|
||||
type: 'tool',
|
||||
callId: 'call-1',
|
||||
name: 'read',
|
||||
state: 'completed',
|
||||
summary: 'read completed'
|
||||
})
|
||||
|
||||
expect(events.map((event) => event.type)).toEqual([
|
||||
'text',
|
||||
'reasoning',
|
||||
'tool'
|
||||
])
|
||||
})
|
||||
|
||||
it('flushes before a combined delta exceeds the size bound', () => {
|
||||
vi.useFakeTimers()
|
||||
const events: AgentEvent[] = []
|
||||
const buffer = new AgentEventBuffer({
|
||||
maximumBufferedBytes: 5,
|
||||
onEvent: (event) => events.push(event)
|
||||
})
|
||||
|
||||
buffer.push({ requestId, type: 'text', delta: '123' })
|
||||
buffer.push({ requestId, type: 'text', delta: '456' })
|
||||
expect(events).toEqual([
|
||||
{ requestId, type: 'text', delta: '123' }
|
||||
])
|
||||
|
||||
buffer.close()
|
||||
expect(events).toEqual([
|
||||
{ requestId, type: 'text', delta: '123' },
|
||||
{ requestId, type: 'text', delta: '456' }
|
||||
])
|
||||
})
|
||||
|
||||
it('flushes buffered deltas when its timer expires', () => {
|
||||
vi.useFakeTimers()
|
||||
const events: AgentEvent[] = []
|
||||
const buffer = new AgentEventBuffer({
|
||||
flushIntervalMs: 32,
|
||||
onEvent: (event) => events.push(event)
|
||||
})
|
||||
|
||||
buffer.push({ requestId, type: 'reasoning', delta: 'thinking' })
|
||||
vi.advanceTimersByTime(31)
|
||||
expect(events).toEqual([])
|
||||
vi.advanceTimersByTime(1)
|
||||
expect(events).toEqual([
|
||||
{ requestId, type: 'reasoning', delta: 'thinking' }
|
||||
])
|
||||
})
|
||||
|
||||
it('supports frame-paced UI updates with coarser durable writes', () => {
|
||||
vi.useFakeTimers()
|
||||
const publicEvents: AgentEvent[] = []
|
||||
const persistedEvents: AgentEvent[] = []
|
||||
const publicBuffer = new AgentEventBuffer({
|
||||
flushIntervalMs: 16,
|
||||
onEvent: (event) => publicEvents.push(event)
|
||||
})
|
||||
const persistedBuffer = new AgentEventBuffer({
|
||||
flushIntervalMs: 32,
|
||||
onEvent: (event) => persistedEvents.push(event)
|
||||
})
|
||||
const first: AgentEvent = {
|
||||
requestId,
|
||||
type: 'text',
|
||||
delta: 'first'
|
||||
}
|
||||
publicBuffer.push(first)
|
||||
persistedBuffer.push(first)
|
||||
|
||||
vi.advanceTimersByTime(16)
|
||||
expect(publicEvents).toEqual([first])
|
||||
expect(persistedEvents).toEqual([])
|
||||
|
||||
const second: AgentEvent = {
|
||||
requestId,
|
||||
type: 'text',
|
||||
delta: 'second'
|
||||
}
|
||||
publicBuffer.push(second)
|
||||
persistedBuffer.push(second)
|
||||
publicBuffer.close()
|
||||
persistedBuffer.close()
|
||||
|
||||
expect(publicEvents).toEqual([first, second])
|
||||
expect(persistedEvents).toEqual([
|
||||
{
|
||||
requestId,
|
||||
type: 'text',
|
||||
delta: 'firstsecond'
|
||||
}
|
||||
])
|
||||
})
|
||||
|
||||
it('closes idempotently and ignores events after close', () => {
|
||||
vi.useFakeTimers()
|
||||
const events: AgentEvent[] = []
|
||||
const buffer = new AgentEventBuffer({
|
||||
onEvent: (event) => events.push(event)
|
||||
})
|
||||
|
||||
buffer.push({ requestId, type: 'text', delta: 'once' })
|
||||
buffer.close()
|
||||
buffer.close()
|
||||
buffer.push({ requestId, type: 'text', delta: 'late' })
|
||||
vi.runAllTimers()
|
||||
|
||||
expect(events).toEqual([
|
||||
{ requestId, type: 'text', delta: 'once' }
|
||||
])
|
||||
})
|
||||
|
||||
it('reports timer publication errors instead of throwing asynchronously', () => {
|
||||
vi.useFakeTimers()
|
||||
const error = new Error('database unavailable')
|
||||
const onError = vi.fn()
|
||||
const buffer = new AgentEventBuffer({
|
||||
flushIntervalMs: 32,
|
||||
onError,
|
||||
onEvent: () => {
|
||||
throw error
|
||||
}
|
||||
})
|
||||
|
||||
buffer.push({ requestId, type: 'text', delta: 'pending' })
|
||||
expect(() => vi.advanceTimersByTime(32)).not.toThrow()
|
||||
expect(onError).toHaveBeenCalledWith(error)
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,125 @@
|
||||
import type { AgentEvent } from '../shared/contracts'
|
||||
|
||||
type BufferedAgentEvent = Extract<
|
||||
AgentEvent,
|
||||
{ type: 'text' | 'reasoning' }
|
||||
>
|
||||
|
||||
export type AgentEventBufferOptions = {
|
||||
onEvent(event: AgentEvent): void
|
||||
onError?(error: unknown): void
|
||||
flushIntervalMs?: number
|
||||
maximumBufferedBytes?: number
|
||||
}
|
||||
|
||||
const DEFAULT_FLUSH_INTERVAL_MS = 32
|
||||
const DEFAULT_MAXIMUM_BUFFERED_BYTES = 64 * 1024
|
||||
|
||||
export class AgentEventBuffer {
|
||||
private readonly onEvent: (event: AgentEvent) => void
|
||||
private readonly onError: ((error: unknown) => void) | undefined
|
||||
private readonly flushIntervalMs: number
|
||||
private readonly maximumBufferedBytes: number
|
||||
private pending: BufferedAgentEvent | undefined
|
||||
private pendingBytes = 0
|
||||
private timer: ReturnType<typeof setTimeout> | undefined
|
||||
private closed = false
|
||||
|
||||
constructor(options: AgentEventBufferOptions) {
|
||||
this.onEvent = options.onEvent
|
||||
this.onError = options.onError
|
||||
this.flushIntervalMs =
|
||||
options.flushIntervalMs ?? DEFAULT_FLUSH_INTERVAL_MS
|
||||
this.maximumBufferedBytes =
|
||||
options.maximumBufferedBytes ??
|
||||
DEFAULT_MAXIMUM_BUFFERED_BYTES
|
||||
if (this.flushIntervalMs <= 0) {
|
||||
throw new Error('flushIntervalMs must be positive')
|
||||
}
|
||||
if (this.maximumBufferedBytes <= 0) {
|
||||
throw new Error('maximumBufferedBytes must be positive')
|
||||
}
|
||||
}
|
||||
|
||||
push(event: AgentEvent): void {
|
||||
if (this.closed) {
|
||||
return
|
||||
}
|
||||
if (event.type !== 'text' && event.type !== 'reasoning') {
|
||||
this.flush()
|
||||
this.onEvent(event)
|
||||
return
|
||||
}
|
||||
|
||||
const eventBytes = Buffer.byteLength(event.delta)
|
||||
const matchesPending =
|
||||
this.pending?.requestId === event.requestId &&
|
||||
this.pending.type === event.type
|
||||
if (!matchesPending) {
|
||||
this.flush()
|
||||
} else if (
|
||||
this.pendingBytes + eventBytes >
|
||||
this.maximumBufferedBytes
|
||||
) {
|
||||
this.flush()
|
||||
}
|
||||
|
||||
if (eventBytes > this.maximumBufferedBytes) {
|
||||
this.onEvent(event)
|
||||
return
|
||||
}
|
||||
if (this.pending) {
|
||||
this.pending = {
|
||||
...this.pending,
|
||||
delta: this.pending.delta + event.delta
|
||||
}
|
||||
this.pendingBytes += eventBytes
|
||||
return
|
||||
}
|
||||
|
||||
this.pending = { ...event }
|
||||
this.pendingBytes = eventBytes
|
||||
this.scheduleFlush()
|
||||
}
|
||||
|
||||
flush(): void {
|
||||
this.clearTimer()
|
||||
const event = this.pending
|
||||
this.pending = undefined
|
||||
this.pendingBytes = 0
|
||||
if (event) {
|
||||
this.onEvent(event)
|
||||
}
|
||||
}
|
||||
|
||||
close(): void {
|
||||
if (this.closed) {
|
||||
return
|
||||
}
|
||||
this.closed = true
|
||||
this.flush()
|
||||
}
|
||||
|
||||
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
|
||||
}
|
||||
}
|
||||
@@ -1386,6 +1386,429 @@ describe('AssistantDatabase', () => {
|
||||
database.close()
|
||||
})
|
||||
|
||||
it('incrementally saves local changes without replacing unrelated or remote data', async () => {
|
||||
const directory = await mkdtemp(
|
||||
join(tmpdir(), 'goodbuddy-incremental-conversations-')
|
||||
)
|
||||
temporaryDirectories.push(directory)
|
||||
const databasePath = join(directory, 'assistant.sqlite')
|
||||
const database = new AssistantDatabase(databasePath)
|
||||
database.initialize('C:\\Workspace')
|
||||
const project = database.listProjects()[0]!
|
||||
const conversationId =
|
||||
'00000000-0000-4000-8000-000000000501'
|
||||
const unrelatedId =
|
||||
'00000000-0000-4000-8000-000000000502'
|
||||
const streamingMessageId =
|
||||
'00000000-0000-4000-8000-000000000503'
|
||||
const newMessageId =
|
||||
'00000000-0000-4000-8000-000000000504'
|
||||
database.replaceConversations([
|
||||
{
|
||||
id: conversationId,
|
||||
projectId: project.id,
|
||||
title: '增量对话',
|
||||
updatedAt: 1_775_000_000_000,
|
||||
messages: [
|
||||
{
|
||||
id: streamingMessageId,
|
||||
role: 'assistant',
|
||||
content: '生成中',
|
||||
createdAt: 1_775_000_000_001,
|
||||
state: 'streaming',
|
||||
status: '正在生成'
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
id: unrelatedId,
|
||||
title: '不相关本地对话',
|
||||
updatedAt: 1_775_000_000_002,
|
||||
messages: []
|
||||
}
|
||||
])
|
||||
const channelProject = database.ensureChannelProjects(
|
||||
'C:\\Users\\test',
|
||||
channelDefaultProfileId
|
||||
)[0]!
|
||||
const remote = database.getOrCreateRemoteConversation({
|
||||
projectId: channelProject.id,
|
||||
channel: 'weixin',
|
||||
accountId: 'default',
|
||||
externalConversationId: 'incremental-preserved',
|
||||
conversationType: 'direct',
|
||||
title: '保留的远程对话',
|
||||
accountDisplay: '发送者 ****0501'
|
||||
})
|
||||
const raw = new DatabaseSync(databasePath)
|
||||
raw
|
||||
.prepare('UPDATE messages SET request_id = ? WHERE id = ?')
|
||||
.run('preserved-request-id', streamingMessageId)
|
||||
raw.close()
|
||||
|
||||
const save = [
|
||||
{
|
||||
header: {
|
||||
id: conversationId,
|
||||
projectId: project.id,
|
||||
title: '增量对话（已完成）',
|
||||
updatedAt: 1_775_000_001_000
|
||||
},
|
||||
messages: [
|
||||
{
|
||||
id: streamingMessageId,
|
||||
role: 'assistant' as const,
|
||||
content: '生成完成',
|
||||
createdAt: 1_775_000_000_001,
|
||||
state: 'complete' as const,
|
||||
status: '已完成'
|
||||
},
|
||||
{
|
||||
id: newMessageId,
|
||||
role: 'user' as const,
|
||||
content: '继续',
|
||||
createdAt: 1_775_000_001_000,
|
||||
state: 'complete' as const
|
||||
}
|
||||
]
|
||||
}
|
||||
]
|
||||
database.saveLocalConversations(save)
|
||||
database.saveLocalConversations(save)
|
||||
|
||||
expect(database.getConversation(conversationId)).toMatchObject({
|
||||
title: '增量对话（已完成）',
|
||||
messages: [
|
||||
{
|
||||
id: streamingMessageId,
|
||||
content: '生成完成',
|
||||
state: 'complete',
|
||||
status: '已完成'
|
||||
},
|
||||
{
|
||||
id: newMessageId,
|
||||
content: '继续',
|
||||
state: 'complete'
|
||||
}
|
||||
]
|
||||
})
|
||||
expect(database.getConversation(unrelatedId).title).toBe(
|
||||
'不相关本地对话'
|
||||
)
|
||||
expect(database.getConversation(remote.id).remote?.channel).toBe(
|
||||
'weixin'
|
||||
)
|
||||
|
||||
const durable = new DatabaseSync(databasePath)
|
||||
expect(
|
||||
durable
|
||||
.prepare(
|
||||
`SELECT id, sequence, request_id
|
||||
FROM messages
|
||||
WHERE conversation_id = ?
|
||||
ORDER BY sequence`
|
||||
)
|
||||
.all(conversationId)
|
||||
).toEqual([
|
||||
{
|
||||
id: streamingMessageId,
|
||||
sequence: 0,
|
||||
request_id: 'preserved-request-id'
|
||||
},
|
||||
{
|
||||
id: newMessageId,
|
||||
sequence: 1,
|
||||
request_id: null
|
||||
}
|
||||
])
|
||||
durable.close()
|
||||
database.close()
|
||||
})
|
||||
|
||||
it('keeps incremental local conversation storage bounded to 500 messages', async () => {
|
||||
const directory = await mkdtemp(
|
||||
join(tmpdir(), 'goodbuddy-bounded-local-conversation-')
|
||||
)
|
||||
temporaryDirectories.push(directory)
|
||||
const databasePath = join(directory, 'assistant.sqlite')
|
||||
const database = new AssistantDatabase(databasePath)
|
||||
database.initialize('C:\\Workspace')
|
||||
const project = database.listProjects()[0]!
|
||||
const conversationId =
|
||||
'00000000-0000-4000-8000-000000000521'
|
||||
const messageId = (index: number): string =>
|
||||
`00000000-0000-4000-8001-${String(index).padStart(12, '0')}`
|
||||
database.replaceConversations([
|
||||
{
|
||||
id: conversationId,
|
||||
projectId: project.id,
|
||||
title: '有界增量对话',
|
||||
updatedAt: 1_775_000_000_000,
|
||||
messages: Array.from({ length: 500 }, (_, index) => ({
|
||||
id: messageId(index),
|
||||
role: index % 2 === 0 ? 'user' as const : 'assistant' as const,
|
||||
content: `消息 ${index}`,
|
||||
createdAt: 1_775_000_000_000 + index,
|
||||
state: 'complete' as const
|
||||
}))
|
||||
}
|
||||
])
|
||||
|
||||
const newestMessageId = messageId(500)
|
||||
database.saveLocalConversations([
|
||||
{
|
||||
header: {
|
||||
id: conversationId,
|
||||
projectId: project.id,
|
||||
title: '有界增量对话',
|
||||
updatedAt: 1_775_000_001_000
|
||||
},
|
||||
messages: [
|
||||
{
|
||||
id: newestMessageId,
|
||||
role: 'user',
|
||||
content: '最新消息',
|
||||
createdAt: 1_775_000_001_000,
|
||||
state: 'complete'
|
||||
}
|
||||
]
|
||||
}
|
||||
])
|
||||
|
||||
const restored = database.getConversation(conversationId)
|
||||
expect(restored.messages).toHaveLength(500)
|
||||
expect(restored.messages[0]?.id).toBe(messageId(1))
|
||||
expect(restored.messages.at(-1)?.id).toBe(newestMessageId)
|
||||
const raw = new DatabaseSync(databasePath)
|
||||
expect(
|
||||
raw
|
||||
.prepare(
|
||||
`SELECT COUNT(*) AS count, MIN(sequence) AS minimum,
|
||||
MAX(sequence) AS maximum
|
||||
FROM messages
|
||||
WHERE conversation_id = ?`
|
||||
)
|
||||
.get(conversationId)
|
||||
).toEqual({
|
||||
count: 500,
|
||||
minimum: 1,
|
||||
maximum: 500
|
||||
})
|
||||
raw.close()
|
||||
database.close()
|
||||
})
|
||||
|
||||
it('rolls back incremental saves when message ownership or role changes', async () => {
|
||||
const database = await createDatabase()
|
||||
const firstConversationId =
|
||||
'00000000-0000-4000-8000-000000000511'
|
||||
const secondConversationId =
|
||||
'00000000-0000-4000-8000-000000000512'
|
||||
const firstMessageId =
|
||||
'00000000-0000-4000-8000-000000000513'
|
||||
const secondMessageId =
|
||||
'00000000-0000-4000-8000-000000000514'
|
||||
const rolledBackMessageId =
|
||||
'00000000-0000-4000-8000-000000000515'
|
||||
database.replaceConversations([
|
||||
{
|
||||
id: firstConversationId,
|
||||
title: '第一对话',
|
||||
updatedAt: 1,
|
||||
messages: [
|
||||
{
|
||||
id: firstMessageId,
|
||||
role: 'user',
|
||||
content: '第一条',
|
||||
createdAt: 1,
|
||||
state: 'complete'
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
id: secondConversationId,
|
||||
title: '第二对话',
|
||||
updatedAt: 2,
|
||||
messages: [
|
||||
{
|
||||
id: secondMessageId,
|
||||
role: 'assistant',
|
||||
content: '第二条',
|
||||
createdAt: 2,
|
||||
state: 'complete'
|
||||
}
|
||||
]
|
||||
}
|
||||
])
|
||||
|
||||
expect(() =>
|
||||
database.saveLocalConversations([
|
||||
{
|
||||
header: {
|
||||
id: firstConversationId,
|
||||
title: '不应提交的标题',
|
||||
updatedAt: 3
|
||||
},
|
||||
messages: [
|
||||
{
|
||||
id: rolledBackMessageId,
|
||||
role: 'user',
|
||||
content: '不应提交',
|
||||
createdAt: 3,
|
||||
state: 'complete'
|
||||
},
|
||||
{
|
||||
id: secondMessageId,
|
||||
role: 'assistant',
|
||||
content: '错误归属',
|
||||
createdAt: 2,
|
||||
state: 'complete'
|
||||
}
|
||||
]
|
||||
}
|
||||
])
|
||||
).toThrow('消息 ID 已属于其他对话')
|
||||
expect(database.getConversation(firstConversationId)).toMatchObject({
|
||||
title: '第一对话',
|
||||
messages: [{ id: firstMessageId }]
|
||||
})
|
||||
|
||||
expect(() =>
|
||||
database.saveLocalConversations([
|
||||
{
|
||||
header: {
|
||||
id: firstConversationId,
|
||||
title: '仍不应提交的标题',
|
||||
updatedAt: 4
|
||||
},
|
||||
messages: [
|
||||
{
|
||||
id: firstMessageId,
|
||||
role: 'assistant',
|
||||
content: '错误角色',
|
||||
createdAt: 1,
|
||||
state: 'complete'
|
||||
}
|
||||
]
|
||||
}
|
||||
])
|
||||
).toThrow('消息角色不能更改')
|
||||
expect(database.getConversation(firstConversationId)).toMatchObject({
|
||||
title: '第一对话',
|
||||
messages: [
|
||||
{
|
||||
id: firstMessageId,
|
||||
role: 'user',
|
||||
content: '第一条'
|
||||
}
|
||||
]
|
||||
})
|
||||
database.close()
|
||||
})
|
||||
|
||||
it('explicitly deletes only local conversations and cascades messages', async () => {
|
||||
const database = await createDatabase()
|
||||
const localId = '00000000-0000-4000-8000-000000000521'
|
||||
database.replaceConversations([
|
||||
{
|
||||
id: localId,
|
||||
title: '待删除本地对话',
|
||||
updatedAt: 1,
|
||||
messages: [
|
||||
{
|
||||
id: '00000000-0000-4000-8000-000000000522',
|
||||
role: 'user',
|
||||
content: '待删除消息',
|
||||
createdAt: 1,
|
||||
state: 'complete'
|
||||
}
|
||||
]
|
||||
}
|
||||
])
|
||||
const channelProject = database.ensureChannelProjects(
|
||||
'C:\\Users\\test',
|
||||
channelDefaultProfileId
|
||||
)[0]!
|
||||
const remote = database.getOrCreateRemoteConversation({
|
||||
projectId: channelProject.id,
|
||||
channel: 'weixin',
|
||||
accountId: 'default',
|
||||
externalConversationId: 'protected-delete',
|
||||
conversationType: 'direct',
|
||||
title: '受保护远程对话',
|
||||
accountDisplay: '发送者 ****0521'
|
||||
})
|
||||
database.appendRemoteConversationMessage({
|
||||
conversationId: remote.id,
|
||||
role: 'user',
|
||||
content: '远程消息'
|
||||
})
|
||||
|
||||
expect(database.deleteLocalConversation(localId)).toBe(true)
|
||||
expect(database.deleteLocalConversation(localId)).toBe(false)
|
||||
expect(() =>
|
||||
database.getConversation(localId)
|
||||
).toThrow('对话不存在')
|
||||
expect(() =>
|
||||
database.deleteLocalConversation(remote.id)
|
||||
).toThrow('远程对话不能作为本地对话删除')
|
||||
expect(() =>
|
||||
database.saveLocalConversations([
|
||||
{
|
||||
header: {
|
||||
id: remote.id,
|
||||
title: '冲突本地标题',
|
||||
updatedAt: 2
|
||||
},
|
||||
messages: []
|
||||
}
|
||||
])
|
||||
).toThrow('本地对话 ID 与远程对话冲突')
|
||||
expect(database.getConversation(remote.id)).toMatchObject({
|
||||
title: '受保护远程对话',
|
||||
messages: [{ content: '远程消息' }]
|
||||
})
|
||||
database.close()
|
||||
})
|
||||
|
||||
it('gets a targeted conversation outside the latest 100', async () => {
|
||||
const database = await createDatabase()
|
||||
database.saveLocalConversations(
|
||||
Array.from({ length: 100 }, (_, index) => ({
|
||||
header: {
|
||||
id: `00000000-0000-4000-8000-${String(index + 1).padStart(12, '0')}`,
|
||||
title: `较新对话 ${index}`,
|
||||
updatedAt: index + 2
|
||||
},
|
||||
messages: []
|
||||
}))
|
||||
)
|
||||
const oldestId = '00000000-0000-4000-8000-000000000999'
|
||||
database.saveLocalConversations([
|
||||
{
|
||||
header: {
|
||||
id: oldestId,
|
||||
title: '第 101 个对话',
|
||||
updatedAt: 1
|
||||
},
|
||||
messages: []
|
||||
}
|
||||
])
|
||||
|
||||
expect(database.listConversations()).toHaveLength(100)
|
||||
expect(
|
||||
database.listConversations().some(
|
||||
(conversation) => conversation.id === oldestId
|
||||
)
|
||||
).toBe(false)
|
||||
expect(database.getConversation(oldestId)).toMatchObject({
|
||||
id: oldestId,
|
||||
title: '第 101 个对话',
|
||||
messages: []
|
||||
})
|
||||
database.close()
|
||||
})
|
||||
|
||||
it('repairs unattended channel selections without rebinding ordinary conversations', async () => {
|
||||
const database = await createDatabase()
|
||||
const removedProfileId =
|
||||
|
||||
@@ -14,6 +14,7 @@ import type {
|
||||
AssistantProject,
|
||||
AssistantSchedule,
|
||||
AssistantTask,
|
||||
ConversationMessage,
|
||||
ConversationSnapshot,
|
||||
ExpertCreateInput,
|
||||
ExpertUpdateInput,
|
||||
@@ -21,6 +22,7 @@ import type {
|
||||
HeartbeatSummaryOutput,
|
||||
HeartbeatUpdateInput,
|
||||
LegacyWorkMode,
|
||||
LocalConversationSaveBatch,
|
||||
MemoryCreateInput,
|
||||
ModelUsageCallInput,
|
||||
ProjectChannel,
|
||||
@@ -749,6 +751,80 @@ function interruptActiveToolBlocks(
|
||||
)
|
||||
}
|
||||
|
||||
function toConversationSnapshot(
|
||||
conversation: ConversationRow,
|
||||
messages: MessageRow[]
|
||||
): ConversationSnapshot {
|
||||
return {
|
||||
id: conversation.id,
|
||||
projectId: conversation.project_id ?? undefined,
|
||||
runtimeSelection: parseRuntimeSelection(
|
||||
conversation.runtime_selection_json
|
||||
),
|
||||
knowledgeRetrievalMode:
|
||||
conversation.knowledge_retrieval_mode ?? undefined,
|
||||
...(conversation.channel &&
|
||||
conversation.conversation_type &&
|
||||
conversation.account_display
|
||||
? {
|
||||
remote: {
|
||||
channel: conversation.channel,
|
||||
accountDisplay: conversation.account_display,
|
||||
conversationType: conversation.conversation_type
|
||||
}
|
||||
}
|
||||
: {}),
|
||||
title: conversation.title,
|
||||
updatedAt: Date.parse(conversation.updated_at),
|
||||
messages: messages.map((message) => {
|
||||
const metadata = JSON.parse(
|
||||
message.metadata_json
|
||||
) as MessageMetadata
|
||||
const interrupted = message.state === 'streaming'
|
||||
return {
|
||||
id: message.id,
|
||||
role: message.role,
|
||||
content: message.content,
|
||||
reasoning: metadata.reasoning,
|
||||
blocks: interrupted
|
||||
? interruptActiveToolBlocks(metadata.blocks)
|
||||
: metadata.blocks,
|
||||
createdAt:
|
||||
metadata.createdAt ?? Date.parse(message.created_at),
|
||||
state: interrupted ? ('error' as const) : message.state,
|
||||
status: interrupted
|
||||
? interruptedMessageStatus
|
||||
: metadata.status,
|
||||
tools: interrupted
|
||||
? interruptActiveTools(metadata.tools)
|
||||
: metadata.tools,
|
||||
sources: metadata.sources,
|
||||
sourceReferences: metadata.sourceReferences,
|
||||
knowledgeRetrieval: metadata.knowledgeRetrieval,
|
||||
artifactIds: metadata.artifactIds,
|
||||
attachments: metadata.attachments
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
function serializeConversationMessageMetadata(
|
||||
message: ConversationMessage
|
||||
): string {
|
||||
return JSON.stringify({
|
||||
createdAt: message.createdAt,
|
||||
status: message.status,
|
||||
reasoning: message.reasoning,
|
||||
blocks: message.blocks,
|
||||
tools: message.tools,
|
||||
sources: message.sources,
|
||||
sourceReferences: message.sourceReferences,
|
||||
knowledgeRetrieval: message.knowledgeRetrieval,
|
||||
artifactIds: message.artifactIds,
|
||||
attachments: message.attachments
|
||||
})
|
||||
}
|
||||
|
||||
export class AssistantDatabase {
|
||||
private database?: DatabaseSync
|
||||
private channelEventWrites = 0
|
||||
@@ -1279,69 +1355,45 @@ export class AssistantDatabase {
|
||||
)
|
||||
ORDER BY sequence ASC`
|
||||
)
|
||||
return conversations.map((conversation) => ({
|
||||
id: conversation.id,
|
||||
projectId: conversation.project_id ?? undefined,
|
||||
runtimeSelection: parseRuntimeSelection(
|
||||
conversation.runtime_selection_json
|
||||
),
|
||||
knowledgeRetrievalMode:
|
||||
conversation.knowledge_retrieval_mode ?? undefined,
|
||||
...(conversation.channel &&
|
||||
conversation.conversation_type &&
|
||||
conversation.account_display
|
||||
? {
|
||||
remote: {
|
||||
channel: conversation.channel,
|
||||
accountDisplay: conversation.account_display,
|
||||
conversationType: conversation.conversation_type
|
||||
}
|
||||
}
|
||||
: {}),
|
||||
title: conversation.title,
|
||||
updatedAt: Date.parse(conversation.updated_at),
|
||||
messages: (
|
||||
return conversations.map((conversation) =>
|
||||
toConversationSnapshot(
|
||||
conversation,
|
||||
messageStatement.all(conversation.id) as MessageRow[]
|
||||
).map((message) => {
|
||||
const metadata = JSON.parse(
|
||||
message.metadata_json
|
||||
) as MessageMetadata
|
||||
const interrupted = message.state === 'streaming'
|
||||
return {
|
||||
id: message.id,
|
||||
role: message.role,
|
||||
content: message.content,
|
||||
reasoning: metadata.reasoning,
|
||||
blocks: interrupted
|
||||
? interruptActiveToolBlocks(metadata.blocks)
|
||||
: metadata.blocks,
|
||||
createdAt:
|
||||
metadata.createdAt ?? Date.parse(message.created_at),
|
||||
state: interrupted ? ('error' as const) : message.state,
|
||||
status: interrupted
|
||||
? interruptedMessageStatus
|
||||
: metadata.status,
|
||||
tools: interrupted
|
||||
? interruptActiveTools(metadata.tools)
|
||||
: metadata.tools,
|
||||
sources: metadata.sources,
|
||||
sourceReferences: metadata.sourceReferences,
|
||||
knowledgeRetrieval: metadata.knowledgeRetrieval,
|
||||
artifactIds: metadata.artifactIds,
|
||||
attachments: metadata.attachments
|
||||
}
|
||||
})
|
||||
}))
|
||||
)
|
||||
)
|
||||
}
|
||||
|
||||
getConversation(conversationId: string): ConversationSnapshot {
|
||||
const conversation = this.listConversations().find(
|
||||
(candidate) => candidate.id === conversationId
|
||||
)
|
||||
const database = this.requireDatabase()
|
||||
const conversation = database
|
||||
.prepare(
|
||||
`SELECT id, project_id, runtime_selection_json,
|
||||
knowledge_retrieval_mode, title, channel,
|
||||
external_account_id, external_conversation_id,
|
||||
conversation_type, account_display, updated_at
|
||||
FROM conversations
|
||||
WHERE id = ? AND status = 'active'`
|
||||
)
|
||||
.get(conversationId) as ConversationRow | undefined
|
||||
if (!conversation) {
|
||||
throw new Error('对话不存在')
|
||||
}
|
||||
return conversation
|
||||
const messages = database
|
||||
.prepare(
|
||||
`SELECT id, conversation_id, role, content, state, metadata_json,
|
||||
created_at
|
||||
FROM (
|
||||
SELECT id, conversation_id, role, content, state,
|
||||
metadata_json, created_at, sequence
|
||||
FROM messages
|
||||
WHERE conversation_id = ?
|
||||
ORDER BY sequence DESC
|
||||
LIMIT 500
|
||||
)
|
||||
ORDER BY sequence ASC`
|
||||
)
|
||||
.all(conversationId) as MessageRow[]
|
||||
return toConversationSnapshot(conversation, messages)
|
||||
}
|
||||
|
||||
repairConversationRuntimeSelections(
|
||||
@@ -1502,6 +1554,167 @@ export class AssistantDatabase {
|
||||
}
|
||||
}
|
||||
|
||||
saveLocalConversations(batch: LocalConversationSaveBatch): void {
|
||||
const database = this.requireDatabase()
|
||||
const findConversation = database.prepare(
|
||||
'SELECT channel FROM conversations WHERE id = ?'
|
||||
)
|
||||
const insertConversation = database.prepare(
|
||||
`INSERT INTO conversations
|
||||
(id, project_id, runtime_selection_json, knowledge_retrieval_mode,
|
||||
work_mode, title, status, created_at, updated_at)
|
||||
VALUES (?, ?, ?, ?, 'ask', ?, 'active', ?, ?)`
|
||||
)
|
||||
const updateConversation = database.prepare(
|
||||
`UPDATE conversations
|
||||
SET project_id = ?, runtime_selection_json = ?,
|
||||
knowledge_retrieval_mode = ?, title = ?, status = 'active',
|
||||
updated_at = ?
|
||||
WHERE id = ? AND channel IS NULL`
|
||||
)
|
||||
const findMessage = database.prepare(
|
||||
`SELECT conversation_id, role
|
||||
FROM messages
|
||||
WHERE id = ?`
|
||||
)
|
||||
const nextSequence = database.prepare(
|
||||
`SELECT COALESCE(MAX(sequence), -1) + 1 AS sequence
|
||||
FROM messages
|
||||
WHERE conversation_id = ?`
|
||||
)
|
||||
const insertMessage = database.prepare(
|
||||
`INSERT INTO messages
|
||||
(id, conversation_id, request_id, role, content, state, sequence,
|
||||
metadata_json, created_at)
|
||||
VALUES (?, ?, NULL, ?, ?, ?, ?, ?, ?)`
|
||||
)
|
||||
const updateMessage = database.prepare(
|
||||
`UPDATE messages
|
||||
SET content = ?, state = ?, metadata_json = ?
|
||||
WHERE id = ?`
|
||||
)
|
||||
const trimMessages = database.prepare(
|
||||
`DELETE FROM messages
|
||||
WHERE id IN (
|
||||
SELECT id
|
||||
FROM messages
|
||||
WHERE conversation_id = ?
|
||||
ORDER BY sequence DESC
|
||||
LIMIT -1 OFFSET 500
|
||||
)`
|
||||
)
|
||||
|
||||
database.exec('BEGIN IMMEDIATE')
|
||||
try {
|
||||
for (const save of batch) {
|
||||
const { header } = save
|
||||
const existingConversation = findConversation.get(
|
||||
header.id
|
||||
) as { channel: ProjectChannel | null } | undefined
|
||||
const updatedAt = new Date(header.updatedAt).toISOString()
|
||||
if (existingConversation?.channel) {
|
||||
throw new Error('本地对话 ID 与远程对话冲突')
|
||||
}
|
||||
if (existingConversation) {
|
||||
const result = updateConversation.run(
|
||||
header.projectId ?? null,
|
||||
header.runtimeSelection
|
||||
? JSON.stringify(header.runtimeSelection)
|
||||
: null,
|
||||
header.knowledgeRetrievalMode ?? null,
|
||||
header.title,
|
||||
updatedAt,
|
||||
header.id
|
||||
)
|
||||
if (result.changes !== 1) {
|
||||
throw new Error('无法更新本地对话')
|
||||
}
|
||||
} else {
|
||||
insertConversation.run(
|
||||
header.id,
|
||||
header.projectId ?? null,
|
||||
header.runtimeSelection
|
||||
? JSON.stringify(header.runtimeSelection)
|
||||
: null,
|
||||
header.knowledgeRetrievalMode ?? null,
|
||||
header.title,
|
||||
updatedAt,
|
||||
updatedAt
|
||||
)
|
||||
}
|
||||
|
||||
let sequence = (
|
||||
nextSequence.get(header.id) as { sequence: number }
|
||||
).sequence
|
||||
let insertedMessage = false
|
||||
for (const message of save.messages) {
|
||||
const existingMessage = findMessage.get(message.id) as
|
||||
| {
|
||||
conversation_id: string
|
||||
role: MessageRow['role']
|
||||
}
|
||||
| undefined
|
||||
if (existingMessage) {
|
||||
if (existingMessage.conversation_id !== header.id) {
|
||||
throw new Error('消息 ID 已属于其他对话')
|
||||
}
|
||||
if (existingMessage.role !== message.role) {
|
||||
throw new Error('消息角色不能更改')
|
||||
}
|
||||
updateMessage.run(
|
||||
message.content,
|
||||
message.state,
|
||||
serializeConversationMessageMetadata(message),
|
||||
message.id
|
||||
)
|
||||
continue
|
||||
}
|
||||
insertMessage.run(
|
||||
message.id,
|
||||
header.id,
|
||||
message.role,
|
||||
message.content,
|
||||
message.state,
|
||||
sequence,
|
||||
serializeConversationMessageMetadata(message),
|
||||
new Date(message.createdAt).toISOString()
|
||||
)
|
||||
sequence += 1
|
||||
insertedMessage = true
|
||||
}
|
||||
if (insertedMessage) {
|
||||
trimMessages.run(header.id)
|
||||
}
|
||||
}
|
||||
database.exec('COMMIT')
|
||||
} catch (error) {
|
||||
database.exec('ROLLBACK')
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
deleteLocalConversation(conversationId: string): boolean {
|
||||
const database = this.requireDatabase()
|
||||
const conversation = database
|
||||
.prepare('SELECT channel FROM conversations WHERE id = ?')
|
||||
.get(conversationId) as
|
||||
| { channel: ProjectChannel | null }
|
||||
| undefined
|
||||
if (!conversation) {
|
||||
return false
|
||||
}
|
||||
if (conversation.channel) {
|
||||
throw new Error('远程对话不能作为本地对话删除')
|
||||
}
|
||||
return (
|
||||
database
|
||||
.prepare(
|
||||
'DELETE FROM conversations WHERE id = ? AND channel IS NULL'
|
||||
)
|
||||
.run(conversationId).changes === 1
|
||||
)
|
||||
}
|
||||
|
||||
getOrCreateRemoteConversation(input: {
|
||||
projectId: string
|
||||
channel: ProjectChannel
|
||||
|
||||
+271
-1
@@ -4,7 +4,7 @@ import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { ipcChannels } from '../shared/ipc-channels'
|
||||
import type { AssistantProject } from '../shared/assistant-contracts'
|
||||
import type { BrowserLiveState } from '../shared/contracts'
|
||||
import type { AgentEvent, BrowserLiveState } from '../shared/contracts'
|
||||
import { defaultKnowledgeOntologySettings } from '../shared/knowledge-ontology'
|
||||
import { AssistantDatabase } from './assistant/assistant-database'
|
||||
import { registerIpcHandlers } from './ipc'
|
||||
@@ -1715,6 +1715,177 @@ describe('registerIpcHandlers token usage', () => {
|
||||
})
|
||||
})
|
||||
|
||||
describe('registerIpcHandlers local conversation persistence', () => {
|
||||
afterEach(() => {
|
||||
electronMocks.handlers.clear()
|
||||
vi.clearAllMocks()
|
||||
})
|
||||
|
||||
it('validates and forwards incremental saves and explicit deletions', async () => {
|
||||
const assistantDatabase = {
|
||||
claimDueSchedules: vi.fn(() => []),
|
||||
saveLocalConversations: vi.fn(),
|
||||
deleteLocalConversation: vi.fn(() => true)
|
||||
}
|
||||
const webContents = {
|
||||
mainFrame: {
|
||||
url: 'file:///goodbuddy/index.html'
|
||||
},
|
||||
getURL: vi.fn(() => 'file:///goodbuddy/index.html')
|
||||
}
|
||||
const window = {
|
||||
webContents,
|
||||
isDestroyed: vi.fn(() => false),
|
||||
on: vi.fn(),
|
||||
removeListener: vi.fn()
|
||||
}
|
||||
const dispose = registerIpcHandlers(
|
||||
window as never,
|
||||
{ capability: 'text' } as never,
|
||||
'CommandOrControl+Shift+Space',
|
||||
{} as never,
|
||||
{} as never,
|
||||
{ clear: vi.fn() } as never,
|
||||
{} as never,
|
||||
assistantDatabase as never,
|
||||
{ clear: vi.fn() } as never,
|
||||
{} as never,
|
||||
vi.fn(async () => {})
|
||||
)
|
||||
const event = {
|
||||
sender: webContents,
|
||||
senderFrame: webContents.mainFrame
|
||||
}
|
||||
const conversationId =
|
||||
'00000000-0000-4000-8000-000000000301'
|
||||
const messageId = '00000000-0000-4000-8000-000000000302'
|
||||
const batch = [
|
||||
{
|
||||
header: {
|
||||
id: conversationId,
|
||||
title: '增量会话',
|
||||
updatedAt: 1
|
||||
},
|
||||
messages: [
|
||||
{
|
||||
id: messageId,
|
||||
role: 'assistant' as const,
|
||||
content: '增量内容',
|
||||
createdAt: 1,
|
||||
state: 'streaming' as const
|
||||
}
|
||||
]
|
||||
}
|
||||
]
|
||||
|
||||
expect(
|
||||
electronMocks.handlers.get(
|
||||
ipcChannels.conversationsSaveLocal
|
||||
)?.(event, batch)
|
||||
).toBeUndefined()
|
||||
expect(
|
||||
assistantDatabase.saveLocalConversations
|
||||
).toHaveBeenCalledWith(batch)
|
||||
expect(
|
||||
electronMocks.handlers.get(
|
||||
ipcChannels.conversationsDeleteLocal
|
||||
)?.(event, conversationId)
|
||||
).toBe(true)
|
||||
expect(
|
||||
assistantDatabase.deleteLocalConversation
|
||||
).toHaveBeenCalledWith(conversationId)
|
||||
|
||||
expect(() =>
|
||||
electronMocks.handlers.get(
|
||||
ipcChannels.conversationsSaveLocal
|
||||
)?.(event, [
|
||||
{
|
||||
...batch[0],
|
||||
header: {
|
||||
...batch[0]!.header,
|
||||
remote: {
|
||||
channel: 'weixin',
|
||||
accountDisplay: 'remote',
|
||||
conversationType: 'direct'
|
||||
}
|
||||
}
|
||||
}
|
||||
])
|
||||
).toThrow()
|
||||
expect(() =>
|
||||
electronMocks.handlers.get(
|
||||
ipcChannels.conversationsDeleteLocal
|
||||
)?.(event, 'not-a-uuid')
|
||||
).toThrow()
|
||||
|
||||
await dispose()
|
||||
})
|
||||
|
||||
it('waits for the renderer persistence acknowledgement before removing handlers', async () => {
|
||||
const assistantDatabase = {
|
||||
claimDueSchedules: vi.fn(() => [])
|
||||
}
|
||||
const webContents = {
|
||||
mainFrame: {
|
||||
url: 'file:///goodbuddy/index.html'
|
||||
},
|
||||
getURL: vi.fn(() => 'file:///goodbuddy/index.html'),
|
||||
isDestroyed: vi.fn(() => false),
|
||||
send: vi.fn()
|
||||
}
|
||||
const window = {
|
||||
webContents,
|
||||
isDestroyed: vi.fn(() => false),
|
||||
on: vi.fn(),
|
||||
removeListener: vi.fn()
|
||||
}
|
||||
const dispose = registerIpcHandlers(
|
||||
window as never,
|
||||
{ capability: 'text' } as never,
|
||||
'CommandOrControl+Shift+Space',
|
||||
{} as never,
|
||||
{} as never,
|
||||
{ clear: vi.fn() } as never,
|
||||
{} as never,
|
||||
assistantDatabase as never,
|
||||
{ clear: vi.fn() } as never,
|
||||
{} as never,
|
||||
vi.fn(async () => {})
|
||||
)
|
||||
const event = {
|
||||
sender: webContents,
|
||||
senderFrame: webContents.mainFrame
|
||||
}
|
||||
electronMocks.handlers.get(
|
||||
ipcChannels.appRendererPersistenceReady
|
||||
)?.(event)
|
||||
|
||||
const disposal = dispose()
|
||||
await vi.waitFor(() =>
|
||||
expect(webContents.send).toHaveBeenCalledWith(
|
||||
ipcChannels.appRendererPersistenceRequest,
|
||||
expect.any(String)
|
||||
)
|
||||
)
|
||||
expect(
|
||||
electronMocks.handlers.has(ipcChannels.conversationsSaveLocal)
|
||||
).toBe(true)
|
||||
const requestId = webContents.send.mock.calls.find(
|
||||
([channel]) =>
|
||||
channel === ipcChannels.appRendererPersistenceRequest
|
||||
)?.[1]
|
||||
expect(requestId).toEqual(expect.any(String))
|
||||
electronMocks.handlers.get(
|
||||
ipcChannels.appRendererPersistenceComplete
|
||||
)?.(event, requestId)
|
||||
|
||||
await disposal
|
||||
expect(
|
||||
electronMocks.handlers.has(ipcChannels.conversationsSaveLocal)
|
||||
).toBe(false)
|
||||
})
|
||||
})
|
||||
|
||||
describe('registerIpcHandlers agent terminal state', () => {
|
||||
afterEach(() => {
|
||||
electronMocks.handlers.clear()
|
||||
@@ -2258,6 +2429,105 @@ describe('registerIpcHandlers agent terminal state', () => {
|
||||
await harness.dispose()
|
||||
})
|
||||
|
||||
it('coalesces burst deltas while preserving output and terminal order', async () => {
|
||||
const deltas = Array.from(
|
||||
{ length: 100 },
|
||||
(_, index) => `chunk-${index};`
|
||||
)
|
||||
const expectedOutput = deltas.join('')
|
||||
const runtime = {
|
||||
runtimeId: 'model',
|
||||
capability: 'chat',
|
||||
supportsToolExecution: true,
|
||||
async *run(request: { requestId: string }) {
|
||||
for (const delta of deltas) {
|
||||
yield {
|
||||
requestId: request.requestId,
|
||||
type: 'text',
|
||||
delta
|
||||
}
|
||||
}
|
||||
yield {
|
||||
requestId: request.requestId,
|
||||
type: 'tool',
|
||||
callId: 'call-burst',
|
||||
name: 'read',
|
||||
state: 'completed',
|
||||
summary: 'read completed'
|
||||
}
|
||||
yield { requestId: request.requestId, type: 'done' }
|
||||
}
|
||||
}
|
||||
const harness = createHarness(runtime)
|
||||
const requestId = '00000000-0000-4000-8000-000000000025'
|
||||
|
||||
await harness.handler?.(trustedEvent(harness.webContents), {
|
||||
requestId,
|
||||
conversationId: 'burst-stream',
|
||||
prompt: 'stream',
|
||||
workMode: 'ask',
|
||||
knowledgeLibraryIds: []
|
||||
})
|
||||
await vi.waitFor(() =>
|
||||
expect(harness.assistantDatabase.updateTaskStatus).toHaveBeenCalledWith(
|
||||
requestId,
|
||||
'completed'
|
||||
)
|
||||
)
|
||||
|
||||
const persistedEvents =
|
||||
harness.assistantDatabase.appendTaskEvent.mock.calls
|
||||
.filter(([taskId]) => taskId === requestId)
|
||||
.map(([, , payload]) => payload)
|
||||
const publicEvents = harness.webContents.send.mock.calls
|
||||
.filter(([channel]) => channel === ipcChannels.agentEvent)
|
||||
.map(([, payload]) => payload)
|
||||
.filter((payload) => payload.requestId === requestId)
|
||||
expect(persistedEvents).toHaveLength(3)
|
||||
expect(publicEvents).toHaveLength(4)
|
||||
expect(persistedEvents.map((event) => event.type)).toEqual([
|
||||
'text',
|
||||
'tool',
|
||||
'done'
|
||||
])
|
||||
expect(publicEvents.map((event) => event.type)).toEqual([
|
||||
'text',
|
||||
'text',
|
||||
'tool',
|
||||
'done'
|
||||
])
|
||||
expect(publicEvents[0]).toEqual({
|
||||
requestId,
|
||||
type: 'text',
|
||||
delta: deltas[0]
|
||||
})
|
||||
expect(persistedEvents[0]).toEqual({
|
||||
requestId,
|
||||
type: 'text',
|
||||
delta: expectedOutput
|
||||
})
|
||||
expect(
|
||||
publicEvents
|
||||
.filter(
|
||||
(
|
||||
event
|
||||
): event is Extract<AgentEvent, { type: 'text' }> =>
|
||||
event.type === 'text'
|
||||
)
|
||||
.map((event) => event.delta)
|
||||
.join('')
|
||||
).toBe(expectedOutput)
|
||||
expect(
|
||||
harness.assistantDatabase.createTextArtifact
|
||||
).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
taskId: requestId,
|
||||
content: expectedOutput
|
||||
})
|
||||
)
|
||||
await harness.dispose()
|
||||
})
|
||||
|
||||
it('preflights always-retrieve mode and injects bounded untrusted evidence', async () => {
|
||||
const libraryId = '11111111-1111-4111-8111-111111111111'
|
||||
const documentId = '33333333-3333-4333-8333-333333333333'
|
||||
|
||||
+189
-58
@@ -131,6 +131,7 @@ import {
|
||||
import {
|
||||
assistantIdSchema,
|
||||
conversationSnapshotsSchema,
|
||||
localConversationSaveBatchSchema,
|
||||
memoryCreateSchema,
|
||||
normalizeInteractiveWorkMode,
|
||||
projectChannelLabels,
|
||||
@@ -241,6 +242,7 @@ import {
|
||||
analyzeMagicNoteEntry,
|
||||
analyzeMagicTodo
|
||||
} from './magic-notes/magic-note-analyzer'
|
||||
import { AgentEventBuffer } from './agent-event-buffer'
|
||||
|
||||
const requestIdSchema = z.string().uuid()
|
||||
const GOODBUDDY_RELEASES_URL =
|
||||
@@ -832,6 +834,7 @@ export function registerIpcHandlers(
|
||||
goodbuddyConfigService?: GoodBuddyConfigService
|
||||
): () => Promise<void> {
|
||||
const activeRequests = new Map<string, AbortController>()
|
||||
const activeEventBuffers = new Map<string, { flush(): void }>()
|
||||
const pendingAgentQuestions = new Map<
|
||||
string,
|
||||
{ requestId: string; runtime: AgentRuntime }
|
||||
@@ -840,6 +843,8 @@ export function registerIpcHandlers(
|
||||
let shuttingDown = false
|
||||
let executionPaused = false
|
||||
let clearLocalDataOperation: Promise<void> | undefined
|
||||
let rendererPersistenceReady = false
|
||||
const pendingRendererPersistence = new Map<string, () => void>()
|
||||
let pendingGoodBuddyConfigReload = false
|
||||
let goodBuddyConfigReloadQueue: Promise<void> = Promise.resolve()
|
||||
const executionTracker = createPromiseTracker()
|
||||
@@ -902,6 +907,40 @@ export function registerIpcHandlers(
|
||||
ipcMain.removeHandler(channel)
|
||||
}
|
||||
|
||||
const requestRendererPersistence = async (): Promise<void> => {
|
||||
if (
|
||||
!rendererPersistenceReady ||
|
||||
window.isDestroyed() ||
|
||||
(typeof window.webContents.isDestroyed === 'function' &&
|
||||
window.webContents.isDestroyed())
|
||||
) {
|
||||
return
|
||||
}
|
||||
const requestId = randomUUID()
|
||||
const completion = new Promise<void>((resolve) => {
|
||||
const finish = (): void => {
|
||||
clearTimeout(timeout)
|
||||
pendingRendererPersistence.delete(requestId)
|
||||
resolve()
|
||||
}
|
||||
pendingRendererPersistence.set(requestId, finish)
|
||||
const timeout = setTimeout(finish, 1_500)
|
||||
timeout.unref?.()
|
||||
})
|
||||
window.webContents.send(
|
||||
ipcChannels.appRendererPersistenceRequest,
|
||||
requestId
|
||||
)
|
||||
await completion
|
||||
}
|
||||
|
||||
const waitForRendererQuiescence = async (): Promise<void> => {
|
||||
await Promise.allSettled([
|
||||
executionTracker.drain(),
|
||||
maintenanceTracker.drain()
|
||||
])
|
||||
}
|
||||
|
||||
const notifyMaximizedChanged = (): void => {
|
||||
if (!window.isDestroyed()) {
|
||||
window.webContents.send(
|
||||
@@ -967,6 +1006,7 @@ export function registerIpcHandlers(
|
||||
},
|
||||
signal,
|
||||
(approvalEvent) => {
|
||||
activeEventBuffers.get(event.requestId)?.flush()
|
||||
if (!window.isDestroyed()) {
|
||||
window.webContents.send(ipcChannels.agentEvent, approvalEvent)
|
||||
}
|
||||
@@ -1029,6 +1069,7 @@ export function registerIpcHandlers(
|
||||
parentTaskId: string,
|
||||
event: Extract<AgentEvent, { type: 'subagent' }>
|
||||
): void => {
|
||||
activeEventBuffers.get(parentTaskId)?.flush()
|
||||
assistantDatabase.appendTaskEvent(
|
||||
parentTaskId,
|
||||
event.type,
|
||||
@@ -1218,6 +1259,16 @@ export function registerIpcHandlers(
|
||||
let knowledgeCapabilityToken: string | undefined
|
||||
const resultAttachments: ChannelMediaAttachment[] = []
|
||||
const artifactIds: string[] = []
|
||||
const eventBuffer = new AgentEventBuffer({
|
||||
onError: (error) => controller.abort(error),
|
||||
onEvent: (event) => {
|
||||
assistantDatabase.appendTaskEvent(
|
||||
requestId,
|
||||
event.type,
|
||||
event
|
||||
)
|
||||
}
|
||||
})
|
||||
try {
|
||||
const requestRuntime =
|
||||
remoteContext?.runtime ??
|
||||
@@ -1296,6 +1347,7 @@ export function registerIpcHandlers(
|
||||
},
|
||||
controller.signal,
|
||||
(approvalEvent) => {
|
||||
eventBuffer.flush()
|
||||
if (!window.isDestroyed()) {
|
||||
window.webContents.send(
|
||||
ipcChannels.agentEvent,
|
||||
@@ -1380,11 +1432,7 @@ export function registerIpcHandlers(
|
||||
if (taskEvent.type === 'artifact') {
|
||||
artifactIds.push(taskEvent.artifactId)
|
||||
}
|
||||
assistantDatabase.appendTaskEvent(
|
||||
requestId,
|
||||
taskEvent.type,
|
||||
taskEvent
|
||||
)
|
||||
eventBuffer.push(taskEvent)
|
||||
if (taskEvent.type === 'tool' && remoteContext) {
|
||||
publishRemoteActivity({
|
||||
requestId,
|
||||
@@ -1474,6 +1522,7 @@ export function registerIpcHandlers(
|
||||
...(artifactIds.length > 0 ? { artifactIds } : {})
|
||||
}
|
||||
} catch (error) {
|
||||
eventBuffer.flush()
|
||||
const message = safeRuntimeError(error, '定时任务执行失败')
|
||||
assistantDatabase.updateTaskStatus(
|
||||
requestId,
|
||||
@@ -1492,6 +1541,7 @@ export function registerIpcHandlers(
|
||||
})
|
||||
return { status: 'failed', error: message }
|
||||
} finally {
|
||||
eventBuffer.close()
|
||||
externalSignal?.removeEventListener(
|
||||
'abort',
|
||||
abortFromExternal
|
||||
@@ -2040,6 +2090,25 @@ export function registerIpcHandlers(
|
||||
}
|
||||
})
|
||||
|
||||
registerHandler(
|
||||
ipcChannels.appRendererPersistenceReady,
|
||||
(event) => {
|
||||
assertTrustedSender(event, window)
|
||||
rendererPersistenceReady = true
|
||||
},
|
||||
false
|
||||
)
|
||||
|
||||
registerHandler(
|
||||
ipcChannels.appRendererPersistenceComplete,
|
||||
(event, input: unknown) => {
|
||||
assertTrustedSender(event, window)
|
||||
const requestId = requestIdSchema.parse(input)
|
||||
pendingRendererPersistence.get(requestId)?.()
|
||||
},
|
||||
false
|
||||
)
|
||||
|
||||
registerHandler(ipcChannels.appShow, (event) => {
|
||||
assertTrustedSender(event, window)
|
||||
showWindow(window)
|
||||
@@ -2280,10 +2349,58 @@ export function registerIpcHandlers(
|
||||
const execution = (async () => {
|
||||
let outputText = ''
|
||||
let completed = false
|
||||
let persistedRuntimeError = false
|
||||
let runtimeErrorEvent:
|
||||
| Extract<AgentEvent, { type: 'error' }>
|
||||
| undefined
|
||||
let executionRequest = request
|
||||
let preflightReferences: KnowledgeSearchReference[] = []
|
||||
let referencesPublished = false
|
||||
const persistedEventBuffer = new AgentEventBuffer({
|
||||
onError: (error) => controller.abort(error),
|
||||
onEvent: (event) => {
|
||||
assistantDatabase.appendTaskEvent(
|
||||
request.requestId,
|
||||
event.type,
|
||||
event
|
||||
)
|
||||
}
|
||||
})
|
||||
const publicEventBuffer = new AgentEventBuffer({
|
||||
flushIntervalMs: 16,
|
||||
onError: (error) => controller.abort(error),
|
||||
onEvent: (event) => {
|
||||
if (!window.isDestroyed()) {
|
||||
window.webContents.send(ipcChannels.agentEvent, event)
|
||||
}
|
||||
}
|
||||
})
|
||||
let publicStreamType: 'text' | 'reasoning' | undefined
|
||||
const eventBuffer = {
|
||||
push: (event: AgentEvent): void => {
|
||||
const streamType =
|
||||
event.type === 'text' || event.type === 'reasoning'
|
||||
? event.type
|
||||
: undefined
|
||||
const startsStreamSegment =
|
||||
streamType !== undefined && streamType !== publicStreamType
|
||||
publicStreamType = streamType
|
||||
publicEventBuffer.push(event)
|
||||
if (startsStreamSegment) {
|
||||
publicEventBuffer.flush()
|
||||
}
|
||||
persistedEventBuffer.push(event)
|
||||
},
|
||||
flush: (): void => {
|
||||
publicStreamType = undefined
|
||||
publicEventBuffer.flush()
|
||||
persistedEventBuffer.flush()
|
||||
},
|
||||
close: (): void => {
|
||||
publicEventBuffer.close()
|
||||
persistedEventBuffer.close()
|
||||
}
|
||||
}
|
||||
activeEventBuffers.set(request.requestId, eventBuffer)
|
||||
const toolStates = new Map<
|
||||
string,
|
||||
Extract<AgentEvent, { type: 'tool' }>
|
||||
@@ -2294,17 +2411,7 @@ export function registerIpcHandlers(
|
||||
{ type: 'knowledge-retrieval' }
|
||||
>
|
||||
): void => {
|
||||
assistantDatabase.appendTaskEvent(
|
||||
request.requestId,
|
||||
retrievalEvent.type,
|
||||
retrievalEvent
|
||||
)
|
||||
if (!window.isDestroyed()) {
|
||||
window.webContents.send(
|
||||
ipcChannels.agentEvent,
|
||||
retrievalEvent
|
||||
)
|
||||
}
|
||||
eventBuffer.push(retrievalEvent)
|
||||
}
|
||||
const publishReferences = (): void => {
|
||||
if (referencesPublished) {
|
||||
@@ -2337,17 +2444,7 @@ export function registerIpcHandlers(
|
||||
type: 'source-references',
|
||||
references
|
||||
}
|
||||
assistantDatabase.appendTaskEvent(
|
||||
request.requestId,
|
||||
referenceEvent.type,
|
||||
referenceEvent
|
||||
)
|
||||
if (!window.isDestroyed()) {
|
||||
window.webContents.send(
|
||||
ipcChannels.agentEvent,
|
||||
referenceEvent
|
||||
)
|
||||
}
|
||||
eventBuffer.push(referenceEvent)
|
||||
}
|
||||
try {
|
||||
controller.signal.throwIfAborted()
|
||||
@@ -2593,12 +2690,7 @@ export function registerIpcHandlers(
|
||||
})
|
||||
}
|
||||
if (publicEvent.type === 'error') {
|
||||
assistantDatabase.appendTaskEvent(
|
||||
request.requestId,
|
||||
publicEvent.type,
|
||||
publicEvent
|
||||
)
|
||||
persistedRuntimeError = true
|
||||
runtimeErrorEvent = publicEvent
|
||||
throw new Error(publicEvent.message)
|
||||
}
|
||||
if (publicEvent.type === 'done') {
|
||||
@@ -2615,12 +2707,15 @@ export function registerIpcHandlers(
|
||||
)
|
||||
}
|
||||
publishReferences()
|
||||
eventBuffer.flush()
|
||||
assistantDatabase.appendTaskEvent(
|
||||
request.requestId,
|
||||
publicEvent.type,
|
||||
publicEvent
|
||||
)
|
||||
} else {
|
||||
eventBuffer.push(publicEvent)
|
||||
}
|
||||
assistantDatabase.appendTaskEvent(
|
||||
request.requestId,
|
||||
publicEvent.type,
|
||||
publicEvent
|
||||
)
|
||||
if (publicEvent.type === 'done') {
|
||||
completed = true
|
||||
if (outputText.trim()) {
|
||||
@@ -2641,9 +2736,12 @@ export function registerIpcHandlers(
|
||||
title: 'GoodBuddy 任务已完成',
|
||||
body: '任务结果已保存到成果工作栏。'
|
||||
})
|
||||
}
|
||||
if (!window.isDestroyed()) {
|
||||
window.webContents.send(ipcChannels.agentEvent, publicEvent)
|
||||
if (!window.isDestroyed()) {
|
||||
window.webContents.send(
|
||||
ipcChannels.agentEvent,
|
||||
publicEvent
|
||||
)
|
||||
}
|
||||
}
|
||||
if (completed) {
|
||||
break
|
||||
@@ -2654,27 +2752,31 @@ export function registerIpcHandlers(
|
||||
}
|
||||
} catch (error) {
|
||||
publishReferences()
|
||||
eventBuffer.flush()
|
||||
const errorMessage = controller.signal.aborted
|
||||
? '请求已取消'
|
||||
: safeRuntimeError(error, 'Agent Runtime 执行失败')
|
||||
const agentEvent: AgentEvent =
|
||||
runtimeErrorEvent && !controller.signal.aborted
|
||||
? runtimeErrorEvent
|
||||
: {
|
||||
requestId: request.requestId,
|
||||
type: 'error',
|
||||
status: controller.signal.aborted
|
||||
? 'cancelled'
|
||||
: 'failed',
|
||||
message: errorMessage
|
||||
}
|
||||
assistantDatabase.updateTaskStatus(
|
||||
request.requestId,
|
||||
controller.signal.aborted ? 'cancelled' : 'failed',
|
||||
errorMessage
|
||||
)
|
||||
const agentEvent: AgentEvent = {
|
||||
requestId: request.requestId,
|
||||
type: 'error',
|
||||
status: controller.signal.aborted ? 'cancelled' : 'failed',
|
||||
message: errorMessage
|
||||
}
|
||||
if (!persistedRuntimeError) {
|
||||
assistantDatabase.appendTaskEvent(
|
||||
request.requestId,
|
||||
agentEvent.type,
|
||||
agentEvent
|
||||
)
|
||||
}
|
||||
assistantDatabase.appendTaskEvent(
|
||||
request.requestId,
|
||||
agentEvent.type,
|
||||
agentEvent
|
||||
)
|
||||
showDesktopNotificationWhenUnfocused(window, {
|
||||
title: controller.signal.aborted
|
||||
? 'GoodBuddy 任务已取消'
|
||||
@@ -2685,6 +2787,8 @@ export function registerIpcHandlers(
|
||||
window.webContents.send(ipcChannels.agentEvent, agentEvent)
|
||||
}
|
||||
} finally {
|
||||
eventBuffer.close()
|
||||
activeEventBuffers.delete(request.requestId)
|
||||
for (const [questionId, pending] of pendingAgentQuestions) {
|
||||
if (pending.requestId === request.requestId) {
|
||||
pendingAgentQuestions.delete(questionId)
|
||||
@@ -3573,6 +3677,26 @@ export function registerIpcHandlers(
|
||||
}
|
||||
)
|
||||
|
||||
registerHandler(
|
||||
ipcChannels.conversationsSaveLocal,
|
||||
(event, input: unknown) => {
|
||||
assertTrustedSender(event, window)
|
||||
assistantDatabase.saveLocalConversations(
|
||||
localConversationSaveBatchSchema.parse(input)
|
||||
)
|
||||
}
|
||||
)
|
||||
|
||||
registerHandler(
|
||||
ipcChannels.conversationsDeleteLocal,
|
||||
(event, input: unknown) => {
|
||||
assertTrustedSender(event, window)
|
||||
return assistantDatabase.deleteLocalConversation(
|
||||
assistantIdSchema.parse(input)
|
||||
)
|
||||
}
|
||||
)
|
||||
|
||||
registerHandler(
|
||||
ipcChannels.workspaceChangesGet,
|
||||
async (event, input: unknown) => {
|
||||
@@ -5000,9 +5124,6 @@ export function registerIpcHandlers(
|
||||
clearInterval(scheduleInterval)
|
||||
window.removeListener('maximize', notifyMaximizedChanged)
|
||||
window.removeListener('unmaximize', notifyMaximizedChanged)
|
||||
for (const channel of channels) {
|
||||
ipcMain.removeHandler(channel)
|
||||
}
|
||||
abortActiveRequests('应用正在退出')
|
||||
for (const controller of heartbeatControllers) {
|
||||
controller.abort(new Error('应用正在退出'))
|
||||
@@ -5019,6 +5140,16 @@ export function registerIpcHandlers(
|
||||
approvalBroker.clear()
|
||||
goodbuddyConfigService?.clear()
|
||||
pendingGoodBuddyConfigReload = false
|
||||
await waitForRendererQuiescence()
|
||||
await requestRendererPersistence()
|
||||
for (const channel of channels) {
|
||||
ipcMain.removeHandler(channel)
|
||||
}
|
||||
rendererPersistenceReady = false
|
||||
for (const complete of pendingRendererPersistence.values()) {
|
||||
complete()
|
||||
}
|
||||
pendingRendererPersistence.clear()
|
||||
await goodBuddyConfigReloadQueue
|
||||
const channelCleanup = Promise.allSettled([
|
||||
...channelServices.map((service) => service.stop()),
|
||||
|
||||
Reference in New Issue
Block a user