feat: add persistent conversation input queue
Messages and Scheduled Task occurrences previously could not share one ordered path while a Conversation was active. They now enter a durable FIFO queue with frozen execution settings and bounded attachments, recover safely after restart, and never write concurrently to the same timeline. The compact queue above Composer supports removal and explicit interrupt-and-promote actions. SQLite schema v23 preserves pending schedule work, while Conversation Task children reuse the shared status-dot semantics for running, completed, failed, approval, paused, and cancelled states. Release note: 回复生成期间仍可继续发送普通消息；消息与 Scheduled Task 会按 Conversation 顺序排队，并支持删除待发送项或立即中断后优先执行。
This commit is contained in:
mesalogo
@@ -26,6 +26,64 @@ async function createDatabase(): Promise<AssistantDatabase> {
|
||||
return database
|
||||
}
|
||||
|
||||
function claimQueuedSchedules(
|
||||
database: AssistantDatabase,
|
||||
now: Date,
|
||||
limit = 4
|
||||
): Array<{
|
||||
schedule: ReturnType<AssistantDatabase['listSchedules']>[number]
|
||||
runId: string
|
||||
}> {
|
||||
database.queueDueSchedules(now, limit)
|
||||
const claims: Array<{
|
||||
schedule: ReturnType<AssistantDatabase['listSchedules']>[number]
|
||||
runId: string
|
||||
}> = []
|
||||
const seenConversations = new Set<string>()
|
||||
for (const item of database.listConversationQueueItems()) {
|
||||
if (
|
||||
item.source !== 'schedule' ||
|
||||
seenConversations.has(item.conversationId) ||
|
||||
claims.length >= limit
|
||||
) {
|
||||
continue
|
||||
}
|
||||
const claimed = database.claimConversationQueueItem(
|
||||
item.conversationId,
|
||||
item.id
|
||||
)
|
||||
if (claimed?.source === 'schedule') {
|
||||
claims.push({
|
||||
schedule: claimed.schedule,
|
||||
runId: claimed.runId
|
||||
})
|
||||
seenConversations.add(item.conversationId)
|
||||
}
|
||||
}
|
||||
return claims
|
||||
}
|
||||
|
||||
function claimManualScheduleQueueItem(
|
||||
database: AssistantDatabase,
|
||||
scheduleId: string
|
||||
): {
|
||||
schedule: ReturnType<AssistantDatabase['listSchedules']>[number]
|
||||
runId: string
|
||||
} {
|
||||
const item = database.queueScheduleNow(scheduleId)
|
||||
const claimed = database.claimConversationQueueItem(
|
||||
item.conversationId,
|
||||
item.id
|
||||
)
|
||||
if (claimed?.source !== 'schedule') {
|
||||
throw new Error('Expected a claimed schedule queue item')
|
||||
}
|
||||
return {
|
||||
schedule: claimed.schedule,
|
||||
runId: claimed.runId
|
||||
}
|
||||
}
|
||||
|
||||
describe('AssistantDatabase', () => {
|
||||
it('rejects a newer unsupported schema without changing its version', async () => {
|
||||
const directory = await mkdtemp(
|
||||
@@ -98,7 +156,7 @@ describe('AssistantDatabase', () => {
|
||||
database.close()
|
||||
})
|
||||
|
||||
it('migrates existing databases to schema version 22', async () => {
|
||||
it('migrates existing databases to schema version 23', async () => {
|
||||
const directory = await mkdtemp(
|
||||
join(tmpdir(), 'goodbuddy-assistant-migration-')
|
||||
)
|
||||
@@ -127,7 +185,7 @@ describe('AssistantDatabase', () => {
|
||||
user_version: number
|
||||
}
|
||||
).user_version
|
||||
).toBe(22)
|
||||
).toBe(23)
|
||||
expect(
|
||||
current
|
||||
.prepare(
|
||||
@@ -198,6 +256,15 @@ describe('AssistantDatabase', () => {
|
||||
)
|
||||
.get()
|
||||
).toEqual({ name: 'magic_todos' })
|
||||
expect(
|
||||
current
|
||||
.prepare(
|
||||
`SELECT name FROM sqlite_master
|
||||
WHERE type = 'table'
|
||||
AND name = 'conversation_queue_items'`
|
||||
)
|
||||
.get()
|
||||
).toEqual({ name: 'conversation_queue_items' })
|
||||
current.close()
|
||||
})
|
||||
|
||||
@@ -231,7 +298,7 @@ describe('AssistantDatabase', () => {
|
||||
user_version: number
|
||||
}
|
||||
).user_version
|
||||
).toBe(22)
|
||||
).toBe(23)
|
||||
expect(
|
||||
current
|
||||
.prepare(
|
||||
@@ -369,7 +436,7 @@ describe('AssistantDatabase', () => {
|
||||
const inspected = new DatabaseSync(databasePath)
|
||||
expect(
|
||||
inspected.prepare('PRAGMA user_version').get()
|
||||
).toEqual({ user_version: 22 })
|
||||
).toEqual({ user_version: 23 })
|
||||
expect(
|
||||
inspected
|
||||
.prepare(
|
||||
@@ -1142,7 +1209,8 @@ describe('AssistantDatabase', () => {
|
||||
title: '每日摘要',
|
||||
messages: []
|
||||
})
|
||||
const [claim] = database.claimDueSchedules(
|
||||
const [claim] = claimQueuedSchedules(
|
||||
database,
|
||||
new Date('2026-07-31T00:01:00.000Z')
|
||||
)
|
||||
expect(claim?.schedule).toEqual(
|
||||
@@ -1169,12 +1237,17 @@ describe('AssistantDatabase', () => {
|
||||
status: 'idle',
|
||||
completedAt: undefined
|
||||
})
|
||||
const manualClaim = database.claimScheduleNow(schedule.id)
|
||||
const manualClaim = claimManualScheduleQueueItem(
|
||||
database,
|
||||
schedule.id
|
||||
)
|
||||
expect(manualClaim.schedule).toMatchObject({
|
||||
taskId: schedule.taskId,
|
||||
conversationId: schedule.conversationId
|
||||
})
|
||||
expect(() => database.claimScheduleNow(schedule.id)).toThrow(
|
||||
expect(() =>
|
||||
claimManualScheduleQueueItem(database, schedule.id)
|
||||
).toThrow(
|
||||
'已有一次运行正在进行'
|
||||
)
|
||||
database.completeScheduleRun(
|
||||
@@ -1254,7 +1327,8 @@ describe('AssistantDatabase', () => {
|
||||
recurrence: 'daily',
|
||||
nextRunAt: '2025-07-31T00:00:00.000Z'
|
||||
})
|
||||
const [overdueClaim] = database.claimDueSchedules(
|
||||
const [overdueClaim] = claimQueuedSchedules(
|
||||
database,
|
||||
new Date('2026-07-31T00:01:00.000Z')
|
||||
)
|
||||
database.completeScheduleRun(
|
||||
@@ -1288,6 +1362,137 @@ describe('AssistantDatabase', () => {
|
||||
database.close()
|
||||
})
|
||||
|
||||
it('persists and arbitrates a FIFO conversation input queue', async () => {
|
||||
const directory = await mkdtemp(
|
||||
join(tmpdir(), 'goodbuddy-conversation-queue-')
|
||||
)
|
||||
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-000000000901'
|
||||
initial.replaceConversations([
|
||||
{
|
||||
id: conversationId,
|
||||
projectId: project.id,
|
||||
title: '排队对话',
|
||||
updatedAt: Date.now(),
|
||||
messages: []
|
||||
}
|
||||
])
|
||||
|
||||
const first = initial.enqueueConversationUserInput({
|
||||
conversationId,
|
||||
label: '第一条消息',
|
||||
payloadJson: JSON.stringify({ prompt: '第一条消息' })
|
||||
})
|
||||
const second = initial.enqueueConversationUserInput({
|
||||
conversationId,
|
||||
label: '第二条消息',
|
||||
payloadJson: JSON.stringify({ prompt: '第二条消息' })
|
||||
})
|
||||
expect(
|
||||
initial
|
||||
.listConversationQueueItems(conversationId)
|
||||
.map((item) => item.id)
|
||||
).toEqual([first.id, second.id])
|
||||
|
||||
const preferred = initial.claimConversationQueueItem(
|
||||
conversationId,
|
||||
second.id
|
||||
)
|
||||
expect(preferred).toMatchObject({
|
||||
source: 'user',
|
||||
item: { id: second.id, source: 'user' },
|
||||
payloadJson: JSON.stringify({ prompt: '第二条消息' })
|
||||
})
|
||||
expect(initial.listConversationQueueItems(conversationId)).toEqual([
|
||||
expect.objectContaining({ id: first.id })
|
||||
])
|
||||
initial.completeConversationUserQueueItem(second.id)
|
||||
|
||||
const firstClaim =
|
||||
initial.claimConversationQueueItem(conversationId)
|
||||
expect(firstClaim).toMatchObject({
|
||||
source: 'user',
|
||||
item: { id: first.id }
|
||||
})
|
||||
initial.close()
|
||||
|
||||
const recovered = new AssistantDatabase(databasePath)
|
||||
recovered.initialize('C:\\Workspace')
|
||||
expect(
|
||||
recovered.listConversationQueueItems(conversationId)
|
||||
).toEqual([
|
||||
expect.objectContaining({
|
||||
id: first.id,
|
||||
source: 'user',
|
||||
label: '第一条消息'
|
||||
})
|
||||
])
|
||||
recovered.cancelConversationQueueItem(first.id)
|
||||
expect(
|
||||
recovered.listConversationQueueItems(conversationId)
|
||||
).toEqual([])
|
||||
recovered.close()
|
||||
})
|
||||
|
||||
it('materializes due and manual schedule runs in the conversation queue', async () => {
|
||||
const database = await createDatabase()
|
||||
const schedule = database.createSchedule({
|
||||
title: '排队提醒',
|
||||
prompt: '检查排队结果',
|
||||
workMode: 'execute',
|
||||
recurrence: 'daily',
|
||||
nextRunAt: '2026-08-20T09:00:00.000Z'
|
||||
})
|
||||
|
||||
const [dueItem] = database.queueDueSchedules(
|
||||
new Date('2026-08-20T09:01:00.000Z')
|
||||
)
|
||||
expect(dueItem).toMatchObject({
|
||||
conversationId: schedule.conversationId,
|
||||
source: 'schedule',
|
||||
scheduleId: schedule.id,
|
||||
taskId: schedule.taskId
|
||||
})
|
||||
expect(database.listConversationQueueItems()).toEqual([
|
||||
expect.objectContaining({ id: dueItem!.id })
|
||||
])
|
||||
expect(
|
||||
database.listPendingScheduleQueueConversationIds()
|
||||
).toEqual([schedule.conversationId])
|
||||
expect(database.listPendingConversationQueueIds()).toEqual([
|
||||
schedule.conversationId
|
||||
])
|
||||
|
||||
const claimed = database.claimConversationQueueItem(
|
||||
schedule.conversationId
|
||||
)
|
||||
expect(claimed).toMatchObject({
|
||||
source: 'schedule',
|
||||
item: { id: dueItem!.id },
|
||||
schedule: { id: schedule.id },
|
||||
runId: dueItem!.id
|
||||
})
|
||||
database.completeScheduleRun(
|
||||
dueItem!.id,
|
||||
'completed',
|
||||
new Date('2026-08-20T09:02:00.000Z')
|
||||
)
|
||||
|
||||
const manualItem = database.queueScheduleNow(schedule.id)
|
||||
expect(manualItem).toMatchObject({
|
||||
source: 'schedule',
|
||||
scheduleId: schedule.id
|
||||
})
|
||||
database.cancelConversationQueueItem(manualItem.id)
|
||||
expect(database.listConversationQueueItems()).toEqual([])
|
||||
database.close()
|
||||
})
|
||||
|
||||
it('recovers a claimed schedule without swallowing its occurrence', async () => {
|
||||
const directory = await mkdtemp(
|
||||
join(tmpdir(), 'goodbuddy-schedule-recovery-')
|
||||
@@ -1303,7 +1508,8 @@ describe('AssistantDatabase', () => {
|
||||
recurrence: 'once',
|
||||
nextRunAt: '2026-08-13T00:00:00.000Z'
|
||||
})
|
||||
const [claimed] = initial.claimDueSchedules(
|
||||
const [claimed] = claimQueuedSchedules(
|
||||
initial,
|
||||
new Date('2026-08-13T00:01:00.000Z')
|
||||
)
|
||||
expect(claimed?.schedule.id).toBe(schedule.id)
|
||||
@@ -1311,7 +1517,8 @@ describe('AssistantDatabase', () => {
|
||||
|
||||
const recovered = new AssistantDatabase(databasePath)
|
||||
recovered.initialize('C:\\Workspace')
|
||||
const [reclaimed] = recovered.claimDueSchedules(
|
||||
const [reclaimed] = claimQueuedSchedules(
|
||||
recovered,
|
||||
new Date('2026-08-13T00:02:00.000Z')
|
||||
)
|
||||
expect(reclaimed).toMatchObject({
|
||||
@@ -1350,7 +1557,8 @@ describe('AssistantDatabase', () => {
|
||||
}).id
|
||||
)
|
||||
|
||||
const firstBatch = database.claimDueSchedules(
|
||||
const firstBatch = claimQueuedSchedules(
|
||||
database,
|
||||
new Date('2026-08-13T00:01:00.000Z'),
|
||||
2
|
||||
)
|
||||
@@ -1359,7 +1567,8 @@ describe('AssistantDatabase', () => {
|
||||
new Set(firstBatch.map((claim) => claim.schedule.id)).size
|
||||
).toBe(2)
|
||||
|
||||
const secondBatch = database.claimDueSchedules(
|
||||
const secondBatch = claimQueuedSchedules(
|
||||
database,
|
||||
new Date('2026-08-13T00:01:00.000Z'),
|
||||
2
|
||||
)
|
||||
|
||||
@@ -16,6 +16,7 @@ import type {
|
||||
AssistantProject,
|
||||
AssistantSchedule,
|
||||
AssistantTask,
|
||||
ConversationQueueItem,
|
||||
ConversationMessage,
|
||||
ConversationSnapshot,
|
||||
ExpertCreateInput,
|
||||
@@ -251,11 +252,32 @@ type ScheduleWithTaskRow = ScheduleRow & {
|
||||
conversation_id: string
|
||||
}
|
||||
|
||||
export type ClaimedSchedule = {
|
||||
schedule: AssistantSchedule
|
||||
runId: string
|
||||
type ConversationQueueRow = {
|
||||
id: string
|
||||
conversation_id: string
|
||||
source: ConversationQueueItem['source']
|
||||
label: string
|
||||
payload_json: string
|
||||
schedule_run_id: string | null
|
||||
schedule_id: string | null
|
||||
task_id: string | null
|
||||
status: 'pending' | 'dispatching'
|
||||
created_at: string
|
||||
}
|
||||
|
||||
export type ClaimedConversationQueueItem =
|
||||
| {
|
||||
source: 'user'
|
||||
item: ConversationQueueItem & { source: 'user' }
|
||||
payloadJson: string
|
||||
}
|
||||
| {
|
||||
source: 'schedule'
|
||||
item: ConversationQueueItem & { source: 'schedule' }
|
||||
schedule: AssistantSchedule
|
||||
runId: string
|
||||
}
|
||||
|
||||
type ExpertRow = {
|
||||
id: string
|
||||
name: string
|
||||
@@ -554,6 +576,21 @@ function toSchedule(row: ScheduleWithTaskRow): AssistantSchedule {
|
||||
}
|
||||
}
|
||||
|
||||
function toConversationQueueItem(
|
||||
row: ConversationQueueRow
|
||||
): ConversationQueueItem {
|
||||
return {
|
||||
id: row.id,
|
||||
conversationId: row.conversation_id,
|
||||
source: row.source,
|
||||
label: row.label,
|
||||
createdAt: row.created_at,
|
||||
scheduleRunId: row.schedule_run_id ?? undefined,
|
||||
scheduleId: row.schedule_id ?? undefined,
|
||||
taskId: row.task_id ?? undefined
|
||||
}
|
||||
}
|
||||
|
||||
function toExpert(row: ExpertRow): AssistantExpert {
|
||||
let routingKeywords: string[]
|
||||
let modelProfileId: string | undefined
|
||||
@@ -1012,6 +1049,24 @@ export class AssistantDatabase {
|
||||
WHERE status = 'running'`
|
||||
)
|
||||
.run()
|
||||
database
|
||||
.prepare(
|
||||
`UPDATE conversation_queue_items
|
||||
SET status = 'pending'
|
||||
WHERE status = 'dispatching'`
|
||||
)
|
||||
.run()
|
||||
database.exec(`
|
||||
INSERT OR IGNORE INTO conversation_queue_items
|
||||
(id, conversation_id, source, label, payload_json,
|
||||
schedule_run_id, schedule_id, task_id, status, created_at)
|
||||
SELECT sr.id, t.conversation_id, 'schedule', t.title, '{}',
|
||||
sr.id, sr.schedule_id, t.id, 'pending', sr.scheduled_for
|
||||
FROM schedule_runs sr
|
||||
INNER JOIN tasks t ON t.schedule_id = sr.schedule_id
|
||||
WHERE sr.status = 'pending'
|
||||
AND t.conversation_id IS NOT NULL;
|
||||
`)
|
||||
const interruptedTasks = database
|
||||
.prepare(
|
||||
`SELECT id, error
|
||||
@@ -1115,6 +1170,7 @@ export class AssistantDatabase {
|
||||
'delegation_outbox',
|
||||
'delegations',
|
||||
'notifications',
|
||||
'conversation_queue_items',
|
||||
'schedule_runs',
|
||||
'schedules',
|
||||
'memory_items',
|
||||
@@ -3517,6 +3573,317 @@ export class AssistantDatabase {
|
||||
}
|
||||
}
|
||||
|
||||
listConversationQueueItems(
|
||||
conversationId?: string
|
||||
): ConversationQueueItem[] {
|
||||
const database = this.requireDatabase()
|
||||
const rows = conversationId
|
||||
? database
|
||||
.prepare(
|
||||
`SELECT *
|
||||
FROM conversation_queue_items
|
||||
WHERE conversation_id = ? AND status = 'pending'
|
||||
ORDER BY created_at ASC, rowid ASC
|
||||
LIMIT 100`
|
||||
)
|
||||
.all(conversationId)
|
||||
: database
|
||||
.prepare(
|
||||
`SELECT *
|
||||
FROM conversation_queue_items
|
||||
WHERE status = 'pending'
|
||||
ORDER BY created_at ASC, rowid ASC
|
||||
LIMIT 500`
|
||||
)
|
||||
.all()
|
||||
return (rows as ConversationQueueRow[]).map(
|
||||
toConversationQueueItem
|
||||
)
|
||||
}
|
||||
|
||||
listPendingScheduleQueueConversationIds(limit = 4): string[] {
|
||||
const safeLimit = Math.max(1, Math.min(20, Math.trunc(limit)))
|
||||
const rows = this.requireDatabase()
|
||||
.prepare(
|
||||
`SELECT conversation_id
|
||||
FROM conversation_queue_items
|
||||
WHERE source = 'schedule' AND status = 'pending'
|
||||
GROUP BY conversation_id
|
||||
ORDER BY MIN(created_at) ASC, MIN(rowid) ASC
|
||||
LIMIT ?`
|
||||
)
|
||||
.all(safeLimit) as Array<{ conversation_id: string }>
|
||||
return rows.map((row) => row.conversation_id)
|
||||
}
|
||||
|
||||
listPendingConversationQueueIds(): string[] {
|
||||
const rows = this.requireDatabase()
|
||||
.prepare(
|
||||
`SELECT conversation_id
|
||||
FROM conversation_queue_items
|
||||
WHERE status = 'pending'
|
||||
GROUP BY conversation_id
|
||||
ORDER BY MIN(created_at) ASC, MIN(rowid) ASC`
|
||||
)
|
||||
.all() as Array<{ conversation_id: string }>
|
||||
return rows.map((row) => row.conversation_id)
|
||||
}
|
||||
|
||||
getConversationQueueItem(
|
||||
itemId: string
|
||||
): ConversationQueueItem | undefined {
|
||||
const row = this.requireDatabase()
|
||||
.prepare(
|
||||
`SELECT *
|
||||
FROM conversation_queue_items
|
||||
WHERE id = ?`
|
||||
)
|
||||
.get(itemId) as ConversationQueueRow | undefined
|
||||
return row ? toConversationQueueItem(row) : undefined
|
||||
}
|
||||
|
||||
getConversationUserQueuePayloadJson(itemId: string): string | undefined {
|
||||
const row = this.requireDatabase()
|
||||
.prepare(
|
||||
`SELECT payload_json
|
||||
FROM conversation_queue_items
|
||||
WHERE id = ? AND source = 'user'`
|
||||
)
|
||||
.get(itemId) as { payload_json: string } | undefined
|
||||
return row?.payload_json
|
||||
}
|
||||
|
||||
isConversationUserQueueItemDispatching(itemId: string): boolean {
|
||||
return Boolean(
|
||||
this.requireDatabase()
|
||||
.prepare(
|
||||
`SELECT 1
|
||||
FROM conversation_queue_items
|
||||
WHERE id = ? AND source = 'user' AND status = 'dispatching'`
|
||||
)
|
||||
.get(itemId)
|
||||
)
|
||||
}
|
||||
|
||||
enqueueConversationUserInput(input: {
|
||||
conversationId: string
|
||||
label: string
|
||||
payloadJson: string
|
||||
}): ConversationQueueItem {
|
||||
const database = this.requireDatabase()
|
||||
const id = randomUUID()
|
||||
const now = new Date().toISOString()
|
||||
const label = input.label.trim().slice(0, 200)
|
||||
if (!label) {
|
||||
throw new Error('待发送消息标题不能为空')
|
||||
}
|
||||
database.exec('BEGIN IMMEDIATE')
|
||||
try {
|
||||
const conversation = database
|
||||
.prepare(
|
||||
`SELECT 1
|
||||
FROM conversations
|
||||
WHERE id = ? AND status = 'active' AND channel IS NULL`
|
||||
)
|
||||
.get(input.conversationId)
|
||||
if (!conversation) {
|
||||
throw new Error('对话不存在或不能加入发送队列')
|
||||
}
|
||||
const count = database
|
||||
.prepare(
|
||||
`SELECT COUNT(*) AS count
|
||||
FROM conversation_queue_items
|
||||
WHERE conversation_id = ?
|
||||
AND status IN ('pending', 'dispatching')`
|
||||
)
|
||||
.get(input.conversationId) as { count: number }
|
||||
if (count.count >= 20) {
|
||||
throw new Error('当前对话最多保留 20 条待执行项')
|
||||
}
|
||||
database
|
||||
.prepare(
|
||||
`INSERT INTO conversation_queue_items
|
||||
(id, conversation_id, source, label, payload_json,
|
||||
schedule_run_id, schedule_id, task_id, status, created_at)
|
||||
VALUES (?, ?, 'user', ?, ?, NULL, NULL, NULL, 'pending', ?)`
|
||||
)
|
||||
.run(
|
||||
id,
|
||||
input.conversationId,
|
||||
label,
|
||||
input.payloadJson,
|
||||
now
|
||||
)
|
||||
database.exec('COMMIT')
|
||||
} catch (error) {
|
||||
database.exec('ROLLBACK')
|
||||
throw error
|
||||
}
|
||||
return this.getConversationQueueItem(id)!
|
||||
}
|
||||
|
||||
claimConversationQueueItem(
|
||||
conversationId: string,
|
||||
preferredItemId?: string
|
||||
): ClaimedConversationQueueItem | undefined {
|
||||
const database = this.requireDatabase()
|
||||
database.exec('BEGIN IMMEDIATE')
|
||||
try {
|
||||
const row = database
|
||||
.prepare(
|
||||
preferredItemId
|
||||
? `SELECT *
|
||||
FROM conversation_queue_items
|
||||
WHERE id = ? AND conversation_id = ?
|
||||
AND status = 'pending'`
|
||||
: `SELECT *
|
||||
FROM conversation_queue_items
|
||||
WHERE conversation_id = ? AND status = 'pending'
|
||||
ORDER BY created_at ASC, rowid ASC
|
||||
LIMIT 1`
|
||||
)
|
||||
.get(
|
||||
...(preferredItemId
|
||||
? [preferredItemId, conversationId]
|
||||
: [conversationId])
|
||||
) as ConversationQueueRow | undefined
|
||||
if (!row) {
|
||||
database.exec('COMMIT')
|
||||
return undefined
|
||||
}
|
||||
const claimed = database
|
||||
.prepare(
|
||||
`UPDATE conversation_queue_items
|
||||
SET status = 'dispatching'
|
||||
WHERE id = ? AND status = 'pending'`
|
||||
)
|
||||
.run(row.id)
|
||||
if (claimed.changes !== 1) {
|
||||
database.exec('COMMIT')
|
||||
return undefined
|
||||
}
|
||||
const item = toConversationQueueItem({
|
||||
...row,
|
||||
status: 'dispatching'
|
||||
})
|
||||
if (row.source === 'user') {
|
||||
database.exec('COMMIT')
|
||||
return {
|
||||
source: 'user',
|
||||
item: { ...item, source: 'user' },
|
||||
payloadJson: row.payload_json
|
||||
}
|
||||
}
|
||||
if (!row.schedule_run_id) {
|
||||
throw new Error('定时任务队列项缺少运行记录')
|
||||
}
|
||||
const scheduleRow = database
|
||||
.prepare(
|
||||
`SELECT s.*, t.id AS task_id,
|
||||
t.conversation_id AS conversation_id
|
||||
FROM schedule_runs sr
|
||||
INNER JOIN schedules s ON s.id = sr.schedule_id
|
||||
INNER JOIN tasks t ON t.schedule_id = s.id
|
||||
WHERE sr.id = ? AND sr.status = 'pending'`
|
||||
)
|
||||
.get(row.schedule_run_id) as ScheduleWithTaskRow | undefined
|
||||
if (!scheduleRow) {
|
||||
throw new Error('定时任务队列项已失效')
|
||||
}
|
||||
const scheduleClaim = database
|
||||
.prepare(
|
||||
`UPDATE schedule_runs
|
||||
SET status = 'running'
|
||||
WHERE id = ? AND status = 'pending'`
|
||||
)
|
||||
.run(row.schedule_run_id)
|
||||
if (scheduleClaim.changes !== 1) {
|
||||
throw new Error('定时任务运行记录无法认领')
|
||||
}
|
||||
database.exec('COMMIT')
|
||||
return {
|
||||
source: 'schedule',
|
||||
item: { ...item, source: 'schedule' },
|
||||
schedule: toSchedule(scheduleRow),
|
||||
runId: row.schedule_run_id
|
||||
}
|
||||
} catch (error) {
|
||||
database.exec('ROLLBACK')
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
completeConversationUserQueueItem(itemId: string): void {
|
||||
const result = this.requireDatabase()
|
||||
.prepare(
|
||||
`DELETE FROM conversation_queue_items
|
||||
WHERE id = ? AND source = 'user' AND status = 'dispatching'`
|
||||
)
|
||||
.run(itemId)
|
||||
if (result.changes !== 1) {
|
||||
throw new Error('待发送消息不存在或状态已变化')
|
||||
}
|
||||
}
|
||||
|
||||
releaseConversationUserQueueItem(itemId: string): void {
|
||||
const result = this.requireDatabase()
|
||||
.prepare(
|
||||
`UPDATE conversation_queue_items
|
||||
SET status = 'pending'
|
||||
WHERE id = ? AND source = 'user' AND status = 'dispatching'`
|
||||
)
|
||||
.run(itemId)
|
||||
if (
|
||||
result.changes !== 1 &&
|
||||
!this.requireDatabase()
|
||||
.prepare(
|
||||
`SELECT 1
|
||||
FROM conversation_queue_items
|
||||
WHERE id = ? AND source = 'user' AND status = 'pending'`
|
||||
)
|
||||
.get(itemId)
|
||||
) {
|
||||
throw new Error('待发送消息不存在或状态已变化')
|
||||
}
|
||||
}
|
||||
|
||||
removeConversationUserQueueItem(itemId: string): void {
|
||||
const result = this.requireDatabase()
|
||||
.prepare(
|
||||
`DELETE FROM conversation_queue_items
|
||||
WHERE id = ? AND source = 'user' AND status = 'pending'`
|
||||
)
|
||||
.run(itemId)
|
||||
if (result.changes !== 1) {
|
||||
throw new Error('待发送消息不存在或状态已变化')
|
||||
}
|
||||
}
|
||||
|
||||
cancelConversationQueueItem(
|
||||
itemId: string
|
||||
): ConversationQueueItem {
|
||||
const row = this.requireDatabase()
|
||||
.prepare(
|
||||
`SELECT *
|
||||
FROM conversation_queue_items
|
||||
WHERE id = ? AND status = 'pending'`
|
||||
)
|
||||
.get(itemId) as ConversationQueueRow | undefined
|
||||
if (!row) {
|
||||
throw new Error('待执行项不存在或状态已变化')
|
||||
}
|
||||
const item = toConversationQueueItem(row)
|
||||
if (row.source === 'user') {
|
||||
this.removeConversationUserQueueItem(itemId)
|
||||
return item
|
||||
}
|
||||
if (!row.schedule_run_id) {
|
||||
throw new Error('定时任务队列项缺少运行记录')
|
||||
}
|
||||
this.completeScheduleRun(row.schedule_run_id, 'cancelled')
|
||||
return item
|
||||
}
|
||||
|
||||
listSchedules(projectId?: string): AssistantSchedule[] {
|
||||
const rows = projectId
|
||||
? this.requireDatabase()
|
||||
@@ -3772,21 +4139,22 @@ export class AssistantDatabase {
|
||||
}
|
||||
}
|
||||
|
||||
claimDueSchedules(now = new Date(), limit = 4): ClaimedSchedule[] {
|
||||
queueDueSchedules(
|
||||
now = new Date(),
|
||||
limit = 100
|
||||
): ConversationQueueItem[] {
|
||||
const database = this.requireDatabase()
|
||||
const nowIso = now.toISOString()
|
||||
const safeLimit = Math.max(1, Math.min(16, Math.trunc(limit)))
|
||||
const safeLimit = Math.max(1, Math.min(100, Math.trunc(limit)))
|
||||
database.exec('BEGIN IMMEDIATE')
|
||||
try {
|
||||
const claims: ClaimedSchedule[] = []
|
||||
const pending = database
|
||||
const due = database
|
||||
.prepare(
|
||||
`SELECT sr.id AS run_id, s.*, t.id AS task_id,
|
||||
`SELECT s.*, t.id AS task_id,
|
||||
t.conversation_id AS conversation_id
|
||||
FROM schedule_runs sr
|
||||
INNER JOIN schedules s ON s.id = sr.schedule_id
|
||||
FROM schedules s
|
||||
INNER JOIN tasks t ON t.schedule_id = s.id
|
||||
WHERE sr.status = 'pending'
|
||||
WHERE s.enabled = 1 AND s.next_run_at <= ?
|
||||
AND (
|
||||
s.project_id IS NULL OR EXISTS (
|
||||
SELECT 1 FROM projects p
|
||||
@@ -3795,83 +4163,69 @@ export class AssistantDatabase {
|
||||
AND p.kind = 'user'
|
||||
)
|
||||
)
|
||||
ORDER BY sr.scheduled_for
|
||||
AND NOT EXISTS (
|
||||
SELECT 1
|
||||
FROM schedule_runs active
|
||||
WHERE active.schedule_id = s.id
|
||||
AND active.status IN ('pending', 'running')
|
||||
)
|
||||
ORDER BY s.next_run_at
|
||||
LIMIT ?`
|
||||
)
|
||||
.all(safeLimit) as Array<
|
||||
ScheduleWithTaskRow & { run_id: string }
|
||||
>
|
||||
const claimPending = database
|
||||
.prepare(
|
||||
`UPDATE schedule_runs
|
||||
SET status = 'running'
|
||||
WHERE id = ? AND status = 'pending'`
|
||||
)
|
||||
for (const row of pending) {
|
||||
if (claimPending.run(row.run_id).changes === 1) {
|
||||
claims.push({
|
||||
schedule: toSchedule(row),
|
||||
runId: row.run_id
|
||||
.all(nowIso, safeLimit) as ScheduleWithTaskRow[]
|
||||
const insertRun = database.prepare(
|
||||
`INSERT OR IGNORE INTO schedule_runs
|
||||
(id, schedule_id, scheduled_for, task_id, status)
|
||||
VALUES (?, ?, ?, ?, 'pending')`
|
||||
)
|
||||
const insertQueueItem = database.prepare(
|
||||
`INSERT INTO conversation_queue_items
|
||||
(id, conversation_id, source, label, payload_json,
|
||||
schedule_run_id, schedule_id, task_id, status, created_at)
|
||||
VALUES (?, ?, 'schedule', ?, '{}', ?, ?, ?, 'pending', ?)`
|
||||
)
|
||||
const queued: ConversationQueueItem[] = []
|
||||
for (const row of due) {
|
||||
const schedule = toSchedule(row)
|
||||
const runId = randomUUID()
|
||||
if (
|
||||
insertRun.run(
|
||||
runId,
|
||||
schedule.id,
|
||||
schedule.nextRunAt,
|
||||
schedule.taskId
|
||||
).changes === 1
|
||||
) {
|
||||
insertQueueItem.run(
|
||||
runId,
|
||||
schedule.conversationId,
|
||||
schedule.title,
|
||||
runId,
|
||||
schedule.id,
|
||||
schedule.taskId,
|
||||
schedule.nextRunAt
|
||||
)
|
||||
queued.push({
|
||||
id: runId,
|
||||
conversationId: schedule.conversationId,
|
||||
source: 'schedule',
|
||||
label: schedule.title,
|
||||
createdAt: schedule.nextRunAt,
|
||||
scheduleRunId: runId,
|
||||
scheduleId: schedule.id,
|
||||
taskId: schedule.taskId
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
const remaining = safeLimit - claims.length
|
||||
if (remaining > 0) {
|
||||
const due = database
|
||||
.prepare(
|
||||
`SELECT s.*, t.id AS task_id,
|
||||
t.conversation_id AS conversation_id
|
||||
FROM schedules s
|
||||
INNER JOIN tasks t ON t.schedule_id = s.id
|
||||
WHERE s.enabled = 1 AND s.next_run_at <= ?
|
||||
AND (
|
||||
s.project_id IS NULL OR EXISTS (
|
||||
SELECT 1 FROM projects p
|
||||
WHERE p.id = s.project_id
|
||||
AND p.status = 'active'
|
||||
AND p.kind = 'user'
|
||||
)
|
||||
)
|
||||
AND NOT EXISTS (
|
||||
SELECT 1
|
||||
FROM schedule_runs active
|
||||
WHERE active.schedule_id = s.id
|
||||
AND active.status IN ('pending', 'running')
|
||||
)
|
||||
ORDER BY s.next_run_at
|
||||
LIMIT ?`
|
||||
)
|
||||
.all(nowIso, remaining) as ScheduleWithTaskRow[]
|
||||
const insertRun = database.prepare(
|
||||
`INSERT OR IGNORE INTO schedule_runs
|
||||
(id, schedule_id, scheduled_for, task_id, status)
|
||||
VALUES (?, ?, ?, ?, 'running')`
|
||||
)
|
||||
for (const row of due) {
|
||||
const schedule = toSchedule(row)
|
||||
const runId = randomUUID()
|
||||
if (
|
||||
insertRun.run(
|
||||
runId,
|
||||
schedule.id,
|
||||
schedule.nextRunAt,
|
||||
schedule.taskId
|
||||
).changes === 1
|
||||
) {
|
||||
claims.push({ schedule, runId })
|
||||
}
|
||||
}
|
||||
}
|
||||
database.exec('COMMIT')
|
||||
return claims
|
||||
return queued
|
||||
} catch (error) {
|
||||
database.exec('ROLLBACK')
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
claimScheduleNow(scheduleId: string): ClaimedSchedule {
|
||||
queueScheduleNow(scheduleId: string): ConversationQueueItem {
|
||||
const schedule = this.getSchedule(scheduleId)
|
||||
const database = this.requireDatabase()
|
||||
const runId = randomUUID()
|
||||
@@ -3897,7 +4251,7 @@ export class AssistantDatabase {
|
||||
.prepare(
|
||||
`INSERT INTO schedule_runs
|
||||
(id, schedule_id, scheduled_for, task_id, status)
|
||||
VALUES (?, ?, ?, ?, 'running')
|
||||
VALUES (?, ?, ?, ?, 'pending')
|
||||
ON CONFLICT(schedule_id, scheduled_for) DO NOTHING`
|
||||
)
|
||||
.run(
|
||||
@@ -3910,17 +4264,34 @@ export class AssistantDatabase {
|
||||
if (!inserted) {
|
||||
throw new Error('定时任务正在启动，请稍后重试')
|
||||
}
|
||||
const createdAt = new Date(baseTime).toISOString()
|
||||
database
|
||||
.prepare(
|
||||
`INSERT INTO conversation_queue_items
|
||||
(id, conversation_id, source, label, payload_json,
|
||||
schedule_run_id, schedule_id, task_id, status, created_at)
|
||||
VALUES (?, ?, 'schedule', ?, '{}', ?, ?, ?, 'pending', ?)`
|
||||
)
|
||||
.run(
|
||||
runId,
|
||||
schedule.conversationId,
|
||||
schedule.title,
|
||||
runId,
|
||||
schedule.id,
|
||||
schedule.taskId,
|
||||
createdAt
|
||||
)
|
||||
database.exec('COMMIT')
|
||||
} catch (error) {
|
||||
database.exec('ROLLBACK')
|
||||
throw error
|
||||
}
|
||||
return { schedule, runId }
|
||||
return this.getConversationQueueItem(runId)!
|
||||
}
|
||||
|
||||
completeScheduleRun(
|
||||
runId: string,
|
||||
status: 'completed' | 'failed',
|
||||
status: 'completed' | 'failed' | 'cancelled',
|
||||
now = new Date()
|
||||
): void {
|
||||
const database = this.requireDatabase()
|
||||
@@ -3935,7 +4306,8 @@ export class AssistantDatabase {
|
||||
FROM schedule_runs sr
|
||||
INNER JOIN schedules s ON s.id = sr.schedule_id
|
||||
INNER JOIN tasks t ON t.schedule_id = s.id
|
||||
WHERE sr.id = ? AND sr.status = 'running'`
|
||||
WHERE sr.id = ?
|
||||
AND sr.status IN ('pending', 'running')`
|
||||
)
|
||||
.get(runId) as
|
||||
| (ScheduleWithTaskRow & { scheduled_for: string })
|
||||
@@ -3950,6 +4322,12 @@ export class AssistantDatabase {
|
||||
WHERE id = ?`
|
||||
)
|
||||
.run(status, runId)
|
||||
database
|
||||
.prepare(
|
||||
`DELETE FROM conversation_queue_items
|
||||
WHERE schedule_run_id = ?`
|
||||
)
|
||||
.run(runId)
|
||||
const schedule = toSchedule(row)
|
||||
if (row.scheduled_for === row.next_run_at) {
|
||||
const next = new Date(row.scheduled_for)
|
||||
@@ -3972,7 +4350,12 @@ export class AssistantDatabase {
|
||||
.prepare(
|
||||
`UPDATE schedules
|
||||
SET enabled = CASE WHEN ? = 1 THEN 0 ELSE enabled END,
|
||||
next_run_at = ?, last_run_at = ?, updated_at = ?
|
||||
next_run_at = ?,
|
||||
last_run_at = CASE
|
||||
WHEN ? = 'cancelled' THEN last_run_at
|
||||
ELSE ?
|
||||
END,
|
||||
updated_at = ?
|
||||
WHERE id = ? AND next_run_at = ?`
|
||||
)
|
||||
.run(
|
||||
@@ -3980,19 +4363,22 @@ export class AssistantDatabase {
|
||||
schedule.recurrence === 'once'
|
||||
? schedule.nextRunAt
|
||||
: next.toISOString(),
|
||||
status,
|
||||
nowIso,
|
||||
nowIso,
|
||||
schedule.id,
|
||||
row.scheduled_for
|
||||
)
|
||||
} else {
|
||||
database
|
||||
.prepare(
|
||||
`UPDATE schedules
|
||||
SET last_run_at = ?, updated_at = ?
|
||||
WHERE id = ?`
|
||||
)
|
||||
.run(nowIso, nowIso, schedule.id)
|
||||
if (status !== 'cancelled') {
|
||||
database
|
||||
.prepare(
|
||||
`UPDATE schedules
|
||||
SET last_run_at = ?, updated_at = ?
|
||||
WHERE id = ?`
|
||||
)
|
||||
.run(nowIso, nowIso, schedule.id)
|
||||
}
|
||||
}
|
||||
const updatedSchedule = database
|
||||
.prepare('SELECT enabled FROM schedules WHERE id = ?')
|
||||
@@ -4000,6 +4386,9 @@ export class AssistantDatabase {
|
||||
const finalTaskStatus =
|
||||
status === 'failed'
|
||||
? 'failed'
|
||||
: status === 'cancelled' &&
|
||||
schedule.recurrence === 'once'
|
||||
? 'cancelled'
|
||||
: updatedSchedule.enabled === 1
|
||||
? 'queued'
|
||||
: schedule.recurrence === 'once'
|
||||
@@ -4010,7 +4399,7 @@ export class AssistantDatabase {
|
||||
`UPDATE tasks
|
||||
SET status = ?,
|
||||
completed_at = CASE
|
||||
WHEN ? IN ('completed', 'failed') THEN ?
|
||||
WHEN ? IN ('completed', 'failed', 'cancelled') THEN ?
|
||||
ELSE NULL
|
||||
END
|
||||
WHERE id = ?`
|
||||
@@ -5352,12 +5741,12 @@ export class AssistantDatabase {
|
||||
const version = database
|
||||
.prepare('PRAGMA user_version')
|
||||
.get() as { user_version: number }
|
||||
if (version.user_version > 22) {
|
||||
if (version.user_version > 23) {
|
||||
throw new Error(
|
||||
`当前 GoodBuddy 不支持助理数据库版本 ${version.user_version}，请升级应用后重试`
|
||||
)
|
||||
}
|
||||
if (version.user_version === 22) {
|
||||
if (version.user_version === 23) {
|
||||
return
|
||||
}
|
||||
if (version.user_version < 1) {
|
||||
@@ -6562,6 +6951,56 @@ export class AssistantDatabase {
|
||||
throw error
|
||||
}
|
||||
}
|
||||
if (version.user_version < 23) {
|
||||
database.exec('BEGIN IMMEDIATE')
|
||||
try {
|
||||
database.exec(`
|
||||
CREATE TABLE IF NOT EXISTS conversation_queue_items (
|
||||
id TEXT PRIMARY KEY,
|
||||
conversation_id TEXT NOT NULL
|
||||
REFERENCES conversations(id) ON DELETE CASCADE,
|
||||
source TEXT NOT NULL CHECK(source IN ('user', 'schedule')),
|
||||
label TEXT NOT NULL,
|
||||
payload_json TEXT NOT NULL DEFAULT '{}',
|
||||
schedule_run_id TEXT UNIQUE
|
||||
REFERENCES schedule_runs(id) ON DELETE CASCADE,
|
||||
schedule_id TEXT
|
||||
REFERENCES schedules(id) ON DELETE CASCADE,
|
||||
task_id TEXT
|
||||
REFERENCES tasks(id) ON DELETE SET NULL,
|
||||
status TEXT NOT NULL
|
||||
CHECK(status IN ('pending', 'dispatching')),
|
||||
created_at TEXT NOT NULL
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS conversation_queue_pending_idx
|
||||
ON conversation_queue_items(
|
||||
conversation_id, status, created_at, id
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS conversation_queue_schedule_pending_idx
|
||||
ON conversation_queue_items(
|
||||
source, status, created_at, conversation_id
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS conversation_queue_status_created_idx
|
||||
ON conversation_queue_items(
|
||||
status, created_at, conversation_id
|
||||
);
|
||||
INSERT OR IGNORE INTO conversation_queue_items
|
||||
(id, conversation_id, source, label, payload_json,
|
||||
schedule_run_id, schedule_id, task_id, status, created_at)
|
||||
SELECT sr.id, t.conversation_id, 'schedule', t.title, '{}',
|
||||
sr.id, sr.schedule_id, t.id, 'pending', sr.scheduled_for
|
||||
FROM schedule_runs sr
|
||||
INNER JOIN tasks t ON t.schedule_id = sr.schedule_id
|
||||
WHERE sr.status IN ('pending', 'running')
|
||||
AND t.conversation_id IS NOT NULL;
|
||||
PRAGMA user_version = 23;
|
||||
COMMIT;
|
||||
`)
|
||||
} catch (error) {
|
||||
database.exec('ROLLBACK')
|
||||
throw error
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private requireDatabase(): DatabaseSync {
|
||||
|
||||
@@ -102,7 +102,7 @@ describe('AssistantDatabase heartbeat persistence', () => {
|
||||
).count
|
||||
check.close()
|
||||
migrated.close()
|
||||
expect(version).toBe(22)
|
||||
expect(version).toBe(23)
|
||||
expect(heartbeatTableCount).toBe(4)
|
||||
})
|
||||
|
||||
|
||||
Reference in New Issue
Block a user