From 67cb69f07d235714c3a3711e700ef2d4bcbb5651 Mon Sep 17 00:00:00 2001 From: lofyer Date: Thu, 13 Aug 2026 01:41:52 +0800 Subject: [PATCH] fix: recover interrupted schedules --- src/main/assistant/assistant-database.test.ts | 76 ++++++- src/main/assistant/assistant-database.ts | 211 ++++++++++++++---- src/main/ipc.ts | 26 ++- 3 files changed, 264 insertions(+), 49 deletions(-) diff --git a/src/main/assistant/assistant-database.test.ts b/src/main/assistant/assistant-database.test.ts index 4cb9162..718c5a9 100644 --- a/src/main/assistant/assistant-database.test.ts +++ b/src/main/assistant/assistant-database.test.ts @@ -955,9 +955,23 @@ describe('AssistantDatabase', () => { recurrence: 'daily', nextRunAt: '2026-07-31T00:00:00.000Z' }) - expect( - database.claimDueSchedules(new Date('2026-07-31T00:01:00.000Z')) - ).toEqual([expect.objectContaining({ id: schedule.id })]) + const [claim] = database.claimDueSchedules( + new Date('2026-07-31T00:01:00.000Z') + ) + expect(claim?.schedule).toEqual( + expect.objectContaining({ id: schedule.id }) + ) + expect(database.listSchedules(project.id)[0]).toMatchObject({ + id: schedule.id, + nextRunAt: '2026-07-31T00:00:00.000Z', + lastRunAt: undefined + }) + database.completeScheduleRun( + claim!.runId, + 'completed', + undefined, + new Date('2026-07-31T00:01:00.000Z') + ) expect(database.listSchedules(project.id)[0]).toMatchObject({ id: schedule.id, nextRunAt: '2026-08-01T00:00:00.000Z', @@ -971,7 +985,13 @@ describe('AssistantDatabase', () => { recurrence: 'daily', nextRunAt: '2025-07-31T00:00:00.000Z' }) - database.claimDueSchedules( + const [overdueClaim] = database.claimDueSchedules( + new Date('2026-07-31T00:01:00.000Z') + ) + database.completeScheduleRun( + overdueClaim!.runId, + 'completed', + undefined, new Date('2026-07-31T00:01:00.000Z') ) expect( @@ -984,6 +1004,54 @@ describe('AssistantDatabase', () => { database.close() }) + it('recovers a claimed schedule without swallowing its occurrence', async () => { + const directory = await mkdtemp( + join(tmpdir(), 'goodbuddy-schedule-recovery-') + ) + temporaryDirectories.push(directory) + const databasePath = join(directory, 'assistant.sqlite') + const initial = new AssistantDatabase(databasePath) + initial.initialize('C:\\Workspace') + const schedule = initial.createSchedule({ + title: '一次提醒', + prompt: '提醒我检查结果', + workMode: 'ask', + recurrence: 'once', + nextRunAt: '2026-08-13T00:00:00.000Z' + }) + const [claimed] = initial.claimDueSchedules( + new Date('2026-08-13T00:01:00.000Z') + ) + expect(claimed?.schedule.id).toBe(schedule.id) + initial.close() + + const recovered = new AssistantDatabase(databasePath) + recovered.initialize('C:\\Workspace') + const [reclaimed] = recovered.claimDueSchedules( + new Date('2026-08-13T00:02:00.000Z') + ) + expect(reclaimed).toMatchObject({ + runId: claimed!.runId, + schedule: { + id: schedule.id, + enabled: true, + nextRunAt: '2026-08-13T00:00:00.000Z' + } + }) + recovered.completeScheduleRun( + reclaimed!.runId, + 'completed', + undefined, + new Date('2026-08-13T00:02:00.000Z') + ) + expect(recovered.listSchedules()[0]).toMatchObject({ + id: schedule.id, + enabled: false, + lastRunAt: '2026-08-13T00:02:00.000Z' + }) + recovered.close() + }) + it('durably interrupts active tasks with completion times and audit events on startup', async () => { const directory = await mkdtemp( join(tmpdir(), 'goodbuddy-assistant-recovery-') diff --git a/src/main/assistant/assistant-database.ts b/src/main/assistant/assistant-database.ts index fa3dbd6..fcb8389 100644 --- a/src/main/assistant/assistant-database.ts +++ b/src/main/assistant/assistant-database.ts @@ -237,6 +237,11 @@ type ScheduleRow = { updated_at: string } +export type ClaimedSchedule = { + schedule: AssistantSchedule + runId: string +} + type ExpertRow = { id: string name: string @@ -842,6 +847,13 @@ export class AssistantDatabase { const recoveredAt = new Date().toISOString() database.exec('BEGIN IMMEDIATE') try { + database + .prepare( + `UPDATE schedule_runs + SET status = 'pending' + WHERE status = 'running'` + ) + .run() const interruptedTasks = database .prepare( `SELECT id, error @@ -3109,65 +3121,182 @@ export class AssistantDatabase { } } - claimDueSchedules(now = new Date()): AssistantSchedule[] { + claimDueSchedules(now = new Date()): ClaimedSchedule[] { const database = this.requireDatabase() - const due = ( - database + const nowIso = now.toISOString() + database.exec('BEGIN IMMEDIATE') + try { + const pending = database + .prepare( + `SELECT sr.id AS run_id, s.* + FROM schedule_runs sr + INNER JOIN schedules s ON s.id = sr.schedule_id + WHERE sr.status = 'pending' + ORDER BY sr.scheduled_for + LIMIT 1` + ) + .get() as (ScheduleRow & { run_id: string }) | undefined + if (pending) { + database + .prepare( + `UPDATE schedule_runs + SET status = 'running' + WHERE id = ? AND status = 'pending'` + ) + .run(pending.run_id) + database.exec('COMMIT') + return [{ + schedule: toSchedule(pending), + runId: pending.run_id + }] + } + + const row = database .prepare( `SELECT * FROM schedules WHERE enabled = 1 AND next_run_at <= ? ORDER BY next_run_at LIMIT 1` ) - .all(now.toISOString()) as ScheduleRow[] - ).map(toSchedule) - for (const schedule of due) { - const next = new Date(schedule.nextRunAt) - if (schedule.recurrence === 'daily') { - const intervals = - Math.floor( - (now.getTime() - next.getTime()) / (24 * 60 * 60 * 1_000) - ) + 1 - next.setUTCDate(next.getUTCDate() + intervals) - } else if (schedule.recurrence === 'weekly') { - const intervals = - Math.floor( - (now.getTime() - next.getTime()) / - (7 * 24 * 60 * 60 * 1_000) - ) + 1 - next.setUTCDate(next.getUTCDate() + intervals * 7) + .get(nowIso) as ScheduleRow | undefined + if (!row) { + database.exec('COMMIT') + return [] + } + const schedule = toSchedule(row) + const runId = randomUUID() + const inserted = database + .prepare( + `INSERT OR IGNORE INTO schedule_runs + (id, schedule_id, scheduled_for, task_id, status) + VALUES (?, ?, ?, NULL, 'running')` + ) + .run(runId, schedule.id, schedule.nextRunAt) + if (inserted.changes !== 1) { + database.exec('COMMIT') + return [] + } + database.exec('COMMIT') + return [{ schedule, runId }] + } catch (error) { + database.exec('ROLLBACK') + throw error + } + } + + claimScheduleNow(scheduleId: string): ClaimedSchedule { + const schedule = this.getSchedule(scheduleId) + const database = this.requireDatabase() + const runId = randomUUID() + database + .prepare( + `INSERT INTO schedule_runs + (id, schedule_id, scheduled_for, task_id, status) + VALUES (?, ?, ?, NULL, 'running')` + ) + .run(runId, scheduleId, new Date().toISOString()) + return { schedule, runId } + } + + completeScheduleRun( + runId: string, + status: 'completed' | 'failed', + taskId: string | undefined, + now = new Date() + ): void { + const database = this.requireDatabase() + const nowIso = now.toISOString() + database.exec('BEGIN IMMEDIATE') + try { + const row = database + .prepare( + `SELECT s.*, sr.scheduled_for + FROM schedule_runs sr + INNER JOIN schedules s ON s.id = sr.schedule_id + WHERE sr.id = ? AND sr.status = 'running'` + ) + .get(runId) as + | (ScheduleRow & { scheduled_for: string }) + | undefined + if (!row) { + throw new Error('定时任务运行记录不存在或已完成') } database .prepare( - `UPDATE schedules - SET enabled = ?, next_run_at = ?, last_run_at = ?, updated_at = ? - WHERE id = ? AND next_run_at = ?` - ) - .run( - schedule.recurrence === 'once' ? 0 : 1, - schedule.recurrence === 'once' - ? schedule.nextRunAt - : next.toISOString(), - now.toISOString(), - now.toISOString(), - schedule.id, - schedule.nextRunAt + `UPDATE schedule_runs + SET task_id = ?, status = ? + WHERE id = ?` ) + .run(taskId ?? null, status, runId) + const schedule = toSchedule(row) + if (row.scheduled_for === row.next_run_at) { + const next = new Date(row.scheduled_for) + if (schedule.recurrence === 'daily') { + const intervals = + Math.floor( + (now.getTime() - next.getTime()) / + (24 * 60 * 60 * 1_000) + ) + 1 + next.setUTCDate(next.getUTCDate() + intervals) + } else if (schedule.recurrence === 'weekly') { + const intervals = + Math.floor( + (now.getTime() - next.getTime()) / + (7 * 24 * 60 * 60 * 1_000) + ) + 1 + next.setUTCDate(next.getUTCDate() + intervals * 7) + } + database + .prepare( + `UPDATE schedules + SET enabled = ?, next_run_at = ?, last_run_at = ?, updated_at = ? + WHERE id = ? AND next_run_at = ?` + ) + .run( + schedule.recurrence === 'once' ? 0 : 1, + schedule.recurrence === 'once' + ? schedule.nextRunAt + : next.toISOString(), + 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) + } + database.exec('COMMIT') + } catch (error) { + database.exec('ROLLBACK') + throw error } - return due } - claimScheduleNow(scheduleId: string): AssistantSchedule { - const schedule = this.getSchedule(scheduleId) - const now = new Date() + bindScheduleRunTask(scheduleId: string, taskId: string): void { this.requireDatabase() .prepare( - `UPDATE schedules - SET last_run_at = ?, updated_at = ? + `UPDATE schedule_runs + SET task_id = ? + WHERE schedule_id = ? AND status = 'running' AND task_id IS NULL` + ) + .run(taskId, scheduleId) + } + + getScheduleRunTaskId(runId: string): string | undefined { + const row = this.requireDatabase() + .prepare( + `SELECT task_id + FROM schedule_runs WHERE id = ?` ) - .run(now.toISOString(), now.toISOString(), scheduleId) - return schedule + .get(runId) as { task_id: string | null } | undefined + return row?.task_id ?? undefined } listHeartbeatConfigs(projectId?: string): AssistantHeartbeatConfig[] { diff --git a/src/main/ipc.ts b/src/main/ipc.ts index 4e53439..2344dce 100644 --- a/src/main/ipc.ts +++ b/src/main/ipc.ts @@ -1033,6 +1033,9 @@ export function registerIpcHandlers( origin: origin === 'channel' ? 'delegation' : origin }) } + if (origin === 'schedule') { + assistantDatabase.bindScheduleRunTask(schedule.id, requestId) + } const modeInstruction = schedule.workMode === 'execute' ? 'Work mode: Execute. Follow the request using the selected backend. Tool actions must remain within the configured workspace, sandbox, enabled capabilities, and security policy.' @@ -1400,8 +1403,15 @@ export function registerIpcHandlers( } scheduleTickRunning = true try { - for (const schedule of assistantDatabase.claimDueSchedules()) { - await trackExecution(executeSchedule(schedule)) + for (const claim of assistantDatabase.claimDueSchedules()) { + const result = await trackExecution( + executeSchedule(claim.schedule) + ) + assistantDatabase.completeScheduleRun( + claim.runId, + result.status, + assistantDatabase.getScheduleRunTaskId(claim.runId) + ) } if (!shuttingDown && !executionPaused) { await trackExecution(heartbeatService.processDue()) @@ -3531,10 +3541,18 @@ export function registerIpcHandlers( if (executionPaused || shuttingDown) { throw new Error('本地数据维护期间暂不接受新任务') } - const schedule = assistantDatabase.claimScheduleNow( + const claim = assistantDatabase.claimScheduleNow( assistantIdSchema.parse(input) ) - void trackExecution(executeSchedule(schedule)).catch(() => undefined) + void trackExecution(executeSchedule(claim.schedule)) + .then((result) => { + assistantDatabase.completeScheduleRun( + claim.runId, + result.status, + assistantDatabase.getScheduleRunTaskId(claim.runId) + ) + }) + .catch(() => undefined) }) ipcMain.handle(ipcChannels.heartbeatsList, (event, input: unknown) => {