fix: prevent duplicate queue replay and use cached Agents
Cross-platform packages / Validate source (push) Canceled after 0s
Cross-platform packages / Authorize desktop packaging (push) Canceled after 0s
Cross-platform packages / linux arm64 (push) Canceled after 0s
Cross-platform packages / macos arm64 (push) Canceled after 0s
Cross-platform packages / windows arm64 (push) Canceled after 0s
Cross-platform packages / linux x64 (push) Canceled after 0s
Cross-platform packages / macos x64 (push) Canceled after 0s
Cross-platform packages / windows x64 (push) Canceled after 0s
Cross-platform packages / Publish GitHub and OSS release (push) Canceled after 0s
GoodBuddy Agents / Validate Agent source (push) Canceled after 0s
GoodBuddy Agents / Build Linux arm64 Agent (push) Canceled after 0s
GoodBuddy Agents / Build Linux x64 Agent (push) Canceled after 0s

An interrupted dispatch could leave the same user input both in conversation
history and in the pending queue, causing it to be sent again after restart.
Persisted user messages now retain their queue identity, startup recovery drops
only matching materialized inputs, and conversation branches discard that
ownership marker.

Explicit GoodBuddy transfer now leases a verified local Agent package when the
catalog is unavailable or the local version is at least as new, while a known
newer online package still wins. Prepare both the desktop and Agent locks for
0.11.7; the experimental LoongArch preview remains excluded.

Release note: 修复对话中断后同一条排队输入可能重复发送的问题;GoodBuddy 传输现在可复用已验证的本地 Agent 包,同时不会阻止已知的线上新版。
This commit is contained in:
mesalogo
2026-08-29 13:31:56 +08:00
parent 459cc21e3d
commit f327341b00
17 changed files with 334 additions and 42 deletions
@@ -2690,6 +2690,92 @@ describe('AssistantDatabase', () => {
recovered.close()
})
it('does not replay a dispatched user queue item whose message was already persisted', async () => {
const directory = await mkdtemp(
join(tmpdir(), 'goodbuddy-conversation-queue-recovery-')
)
temporaryDirectories.push(directory)
const databasePath = join(directory, 'assistant.sqlite')
const initial = new AssistantDatabase(databasePath)
initial.initialize('C:\\Workspace')
const project = initial.listProjects()[0]!
const conversationId =
'00000000-0000-4000-8000-000000000911'
const userMessageId =
'00000000-0000-4000-8000-000000000912'
const assistantMessageId =
'00000000-0000-4000-8000-000000000913'
initial.replaceConversations([
{
id: conversationId,
projectId: project.id,
title: '崩溃恢复对话',
updatedAt: 1,
messages: []
}
])
const queued = initial.enqueueConversationUserInput({
conversationId,
label: '只发送一次',
payloadJson: JSON.stringify({ prompt: '只发送一次' })
})
expect(
initial.claimConversationQueueItem(conversationId, queued.id)
).toMatchObject({
source: 'user',
item: { id: queued.id }
})
initial.saveLocalConversations([
{
header: {
id: conversationId,
projectId: project.id,
title: '崩溃恢复对话',
updatedAt: 2
},
messages: [
{
id: userMessageId,
queueItemId: queued.id,
role: 'user',
content: '只发送一次',
createdAt: 2,
state: 'complete'
},
{
id: assistantMessageId,
role: 'assistant',
content: '',
createdAt: 3,
state: 'streaming'
}
]
}
])
initial.close()
const recovered = new AssistantDatabase(databasePath)
recovered.initialize('C:\\Workspace')
expect(
recovered.listConversationQueueItems(conversationId)
).toEqual([])
expect(recovered.getConversation(conversationId).messages).toEqual([
expect.objectContaining({
id: userMessageId,
queueItemId: queued.id,
role: 'user',
content: '只发送一次'
}),
expect.objectContaining({
id: assistantMessageId,
role: 'assistant',
state: 'error',
status: '上次运行意外中断,可以重新发送问题'
})
])
recovered.close()
})
it('materializes due and manual schedule runs in the conversation queue', async () => {
const database = await createDatabase()
const schedule = database.createSchedule({
@@ -3408,6 +3494,8 @@ describe('AssistantDatabase', () => {
messages: [
{
id: sourceMessageIds[0]!,
queueItemId:
'00000000-0000-4000-8000-000000000551',
role: 'user',
content: '比较两个发布方案',
createdAt: 1_775_000_000_000,
@@ -3516,6 +3604,7 @@ describe('AssistantDatabase', () => {
expect(branch.messages[0]?.attachments).toEqual(
sourceBefore.messages[0]?.attachments
)
expect(branch.messages[0]?.queueItemId).toBeUndefined()
expect(branch.messages[1]?.task).toBeUndefined()
expect(branch.messages[1]?.artifactIds).toBeUndefined()
expect(database.getConversation(sourceConversationId)).toEqual(
@@ -3549,6 +3638,16 @@ describe('AssistantDatabase', () => {
expect(branchMetadata).not.toHaveProperty('artifactIds')
expect(branchMetadata).not.toHaveProperty('subagents')
expect(branchMetadata).not.toHaveProperty('task')
const branchUserMetadataRow = durable
.prepare(
`SELECT metadata_json
FROM messages
WHERE conversation_id = ? AND sequence = 0`
)
.get(branch.id) as { metadata_json: string }
expect(
JSON.parse(branchUserMetadataRow.metadata_json)
).not.toHaveProperty('queueItemId')
durable.close()
database.saveLocalConversations([
+22 -1
View File
@@ -208,6 +208,7 @@ type MagicTodoRow = {
type MessageMetadata = {
createdAt?: number
queueItemId?: string
status?: string
reasoning?: ConversationSnapshot['messages'][number]['reasoning']
blocks?: ConversationSnapshot['messages'][number]['blocks']
@@ -998,6 +999,7 @@ function toConversationSnapshot(
const interrupted = message.state === 'streaming'
return {
id: message.id,
queueItemId: metadata.queueItemId,
role: message.role,
content: message.content,
reasoning: metadata.reasoning,
@@ -1031,6 +1033,7 @@ function serializeConversationMessageMetadata(
): string {
return JSON.stringify({
createdAt: message.createdAt,
queueItemId: message.queueItemId,
status: message.status,
reasoning: message.reasoning,
blocks: message.blocks,
@@ -1340,6 +1343,23 @@ export class AssistantDatabase {
WHERE status = 'dispatching'`
)
.run()
database
.prepare(
`DELETE FROM conversation_queue_items
WHERE source = 'user'
AND EXISTS (
SELECT 1
FROM messages
WHERE messages.conversation_id =
conversation_queue_items.conversation_id
AND messages.role = 'user'
AND json_extract(
messages.metadata_json,
'$.queueItemId'
) = conversation_queue_items.id
)`
)
.run()
database.exec(`
INSERT OR IGNORE INTO conversation_queue_items
(id, conversation_id, source, label, payload_json,
@@ -2496,7 +2516,8 @@ export class AssistantDatabase {
metadata_json,
'$.artifactIds',
'$.subagents',
'$.task'
'$.task',
'$.queueItemId'
),
created_at
FROM messages
@@ -73,9 +73,12 @@ vi.mock('./agent-package-verifier', async (importOriginal) => {
const { readFile: readFixture } =
await import('node:fs/promises')
const archive = await readFixture(options.archivePath)
const version = archive.toString('utf8') === '2.1.0'
? '2.1.0'
: '2.0.0'
const archiveVersion = archive.toString('utf8')
const version =
archiveVersion === '1.9.0' ||
archiveVersion === '2.1.0'
? archiveVersion
: '2.0.0'
await mkdir(join(options.destinationDirectory, 'agent'), {
recursive: true
})
@@ -492,6 +495,107 @@ describe('AgentPackageManager remote install candidates', () => {
lease.release()
})
it('prefers a verified local package over an older available online catalog for GoodBuddy transfer', async () => {
const fixture = await createFixture('mirror')
const transport = vi.fn<typeof fetch>(async (input) => {
const url = input.toString()
if (url.endsWith('latest.json')) {
return response(fixture.pointer)
}
if (url.endsWith('agent-catalog.json')) {
return response(fixture.catalogBytes)
}
if (url.endsWith('agent-catalog.sig')) {
return response(fixture.signature)
}
throw new Error(`Archive must not be requested: ${url}`)
})
const importDirectory = await mkdtemp(
join(tmpdir(), 'goodbuddy-agent-local-newer-')
)
temporaryDirectories.push(importDirectory)
const localArchive = join(
importDirectory,
'goodbuddy-agent-2.1.0-linux-x64.gbagent'
)
await writeFile(localArchive, '2.1.0')
const manager = fixture.manager(transport)
await manager.importArchive(localArchive)
await manager.getSnapshot()
const lease = await manager.acquireGoodBuddyInstallArchive('x64')
expect(lease.candidate).toMatchObject({
architecture: 'x64',
version: '2.1.0',
urls: []
})
expect(
transport.mock.calls.some(([input]) =>
input.toString().endsWith(fixture.archive)
)
).toBe(false)
await expect(readFile(lease.path)).resolves.toEqual(
Buffer.from('2.1.0')
)
lease.release()
})
it('downloads a newer online package instead of transferring a stale local package', async () => {
const onlineArchive = Buffer.from('online-2.0.0')
const fixture = await createFixture(
'mirror',
undefined,
onlineArchive
)
const entry = fixture.catalog.entries[0]!
let archiveRequests = 0
const transport = vi.fn<typeof fetch>(async (input) => {
const url = input.toString()
if (url.endsWith('latest.json')) {
return response(fixture.pointer)
}
if (url.endsWith('agent-catalog.json')) {
return response(fixture.catalogBytes)
}
if (url.endsWith('agent-catalog.sig')) {
return response(fixture.signature)
}
if (url.endsWith(entry.archive)) {
archiveRequests += 1
return response(onlineArchive, {
headers: {
'content-length': String(onlineArchive.byteLength)
}
})
}
throw new Error(`Unexpected URL ${url}`)
})
const importDirectory = await mkdtemp(
join(tmpdir(), 'goodbuddy-agent-local-stale-')
)
temporaryDirectories.push(importDirectory)
const localArchive = join(
importDirectory,
'goodbuddy-agent-1.9.0-linux-x64.gbagent'
)
await writeFile(localArchive, '1.9.0')
const manager = fixture.manager(transport)
await manager.importArchive(localArchive)
await manager.getSnapshot()
const lease = await manager.acquireGoodBuddyInstallArchive('x64')
expect(lease.candidate).toMatchObject({
architecture: 'x64',
version: '2.0.0',
sha256: entry.sha256
})
expect(archiveRequests).toBe(1)
await expect(readFile(lease.path)).resolves.toEqual(onlineArchive)
lease.release()
})
it('propagates cancellation into an active archive download', async () => {
const archiveBytes = Buffer.from('package')
const fixture = await createFixture(
+32 -15
View File
@@ -540,26 +540,43 @@ export class AgentPackageManager {
options.signal?.throwIfAborted()
const source = await this.#getUpdateSource()
options.signal?.throwIfAborted()
let entry: AgentPackageCatalogEntry | undefined
const record = await this.#loadInstalled(architecture)
const leaseRecord = async (
installed: InstalledRecord
): Promise<AgentPackageArchiveLease> => {
const status = await lstat(installed.archivePath)
return this.#archiveLease(
installed,
candidateFromVerifiedRecord(
installed,
source,
status.size
)
)
}
if (
record &&
this.#catalogState?.state !== 'available'
) {
return leaseRecord(record)
}
let entry: AgentPackageCatalogEntry
if (this.#catalogState?.state === 'available') {
entry = selectLatestCompatibleEntry(
this.#catalogState.catalog,
architecture,
this.#desktopVersion
)
} else {
const record = await this.#loadInstalled(architecture)
if (record) {
const status = await lstat(record.archivePath)
return this.#archiveLease(
record,
candidateFromVerifiedRecord(
record,
source,
status.size
)
)
if (
record &&
compareSemanticVersions(
record.verified.descriptor.version,
entry.version
) >= 0
) {
return leaseRecord(record)
}
} else {
if (this.#catalogState?.state === 'unavailable') {
throw new Error(this.#catalogState.error)
}
@@ -577,14 +594,14 @@ export class AgentPackageManager {
entry,
source
)
const record = await this.#ensureInstalledCandidate(
const installed = await this.#ensureInstalledCandidate(
architecture,
entry,
source,
options.onProgress,
options.signal
)
return this.#archiveLease(record, candidate)
return this.#archiveLease(installed, candidate)
}
)
}
+4 -4
View File
@@ -62,7 +62,7 @@ describe('sandboxed preload', () => {
)
const embeddings =
source.match(
/embeddings: \{(?<body>[\s\S]*?)\n {2}\},\n {2}documentParsing:/u
/embeddings: \{(?<body>[\s\S]*?)\r?\n {2}\},\r?\n {2}documentParsing:/u
)?.groups?.body ?? ''
expect(embeddings).toContain('getSnapshot:')
expect(embeddings).toContain('getModelProgress:')
@@ -82,8 +82,8 @@ describe('sandboxed preload', () => {
expect(source).toContain(
'test: (purpose: DocumentParsingTestPurpose)'
)
expect(source).toContain(
'ipcChannels.documentParsingTest,\n { purpose }'
expect(source).toMatch(
/ipcChannels\.documentParsingTest,\r?\n {8}\{ purpose \}/u
)
})
@@ -125,7 +125,7 @@ describe('sandboxed preload', () => {
)
const feedback =
source.match(
/feedback: \{(?<body>[\s\S]*?)\n {2}\},\n {2}shortcuts:/u
/feedback: \{(?<body>[\s\S]*?)\r?\n {2}\},\r?\n {2}shortcuts:/u
)?.groups?.body ?? ''
expect(feedback).toContain('submit: (input: FeedbackSubmitInput)')
expect(feedback).toContain('ipcChannels.feedbackSubmit')
+39
View File
@@ -4802,6 +4802,45 @@ describe('App', () => {
).not.toBeInTheDocument()
})
it('persists the claimed queue identity with a dispatched user message', async () => {
render(<App />)
fireEvent.change(screen.getByLabelText('向 GoodBuddy 提问'), {
target: { value: '只发送一次' }
})
fireEvent.click(await screen.findByLabelText('发送'))
await waitFor(() => expect(run).toHaveBeenCalledOnce())
const request = run.mock.calls[0]?.[0]
if (!request?.queueItemId) {
throw new Error('Missing queued request')
}
act(() => {
agentListener?.({
requestId: request.requestId,
type: 'error',
status: 'failed',
message: 'Remote Runtime failed'
})
})
await waitFor(() =>
expect(
vi.mocked(api.conversations.saveLocal).mock.calls.some(
([batch]) =>
batch.some((conversation) =>
conversation.messages.some(
(message) =>
message.role === 'user' &&
message.content === '只发送一次' &&
message.queueItemId === request.queueItemId
)
)
)
).toBe(true)
)
})
it('keeps sent documents and images in conversation history', async () => {
const documentAttachment = {
id: '00000000-0000-4000-8000-000000000301',
+5
View File
@@ -120,6 +120,7 @@ import type {
WorkspaceChanges
} from '../../shared/assistant-contracts'
import {
assistantIdSchema,
conversationAttachmentSchema,
conversationBranchSchema,
conversationContextCompressionMarkerSchema,
@@ -1182,6 +1183,8 @@ function isConversation(value: unknown): value is Conversation {
const entry = message as Record<string, unknown>
return (
typeof entry.id === 'string' &&
(entry.queueItemId === undefined ||
assistantIdSchema.safeParse(entry.queueItemId).success) &&
(entry.role === 'user' || entry.role === 'assistant') &&
typeof entry.content === 'string' &&
entry.content.length <= 1_000_000 &&
@@ -1262,6 +1265,7 @@ function toConversationSnapshots(
function toConversationMessage(message: Message): ConversationMessage {
return {
id: message.id,
queueItemId: message.queueItemId,
role: message.role,
content: message.content,
reasoning: message.reasoning,
@@ -6519,6 +6523,7 @@ function App(): React.JSX.Element {
setConversationActivity(conversationId, true)
const userMessage: Message = {
id: crypto.randomUUID(),
queueItemId: queuedDispatch.item.id,
role: 'user',
content: prompt,
createdAt: Date.now(),
+1
View File
@@ -41,6 +41,7 @@ export type KnowledgeRetrievalStatus = Omit<
export type Message = {
id: string
queueItemId?: ConversationMessage['queueItemId']
role: 'user' | 'assistant'
content: string
reasoning?: string
+1
View File
@@ -226,6 +226,7 @@ export type ConversationContextCompressionMarker = z.infer<
export const conversationMessageSchema = z
.object({
id: assistantIdSchema,
queueItemId: assistantIdSchema.optional(),
role: z.enum(['user', 'assistant']),
content: z.string().max(1_000_000),
reasoning: z.string().optional(),