diff --git a/docs-web/content/docs/user-dashboard-scheduler.mdx b/docs-web/content/docs/user-dashboard-scheduler.mdx index a5037eb1fb..3899aeccc2 100644 --- a/docs-web/content/docs/user-dashboard-scheduler.mdx +++ b/docs-web/content/docs/user-dashboard-scheduler.mdx @@ -46,9 +46,11 @@ weekly). The page previews the next occurrences so you can confirm the cadence b For node-flow schedules, recurrence uses the same model as other targets. Due runs call the node-flow runtime with `triggerType = "scheduler"` and trigger metadata containing the scheduler entry id, -scheduled occurrence time, target type, and stored flow version when present. Node-flow schedules -advance only when the returned run status is `succeeded`; returned `failed` or `cancelled` runs move -the entry to `failed`, record the attempted occurrence, and show the run error in the list. +scheduled occurrence time, target type, and stored flow version when present. The due occurrence is +claimed in SQLite before the runtime is awaited so a restart does not dispatch the same occurrence +again. Node-flow schedules are then finalized from the returned run status: `succeeded` advances or +completes the schedule, while returned `failed` or `cancelled` runs move the entry to `failed`, +record the attempted occurrence, and show the run error in the list. ## Managing entries diff --git a/docs-web/user/dashboard/scheduler.md b/docs-web/user/dashboard/scheduler.md index 190d5d58cf..7113f3ea7b 100644 --- a/docs-web/user/dashboard/scheduler.md +++ b/docs-web/user/dashboard/scheduler.md @@ -46,9 +46,11 @@ weekly). The page previews the next occurrences so you can confirm the cadence b For node-flow schedules, recurrence uses the same model as other targets. Due runs call the node-flow runtime with `triggerType = "scheduler"` and trigger metadata containing the scheduler entry id, -scheduled occurrence time, target type, and stored flow version when present. Node-flow schedules -advance only when the returned run status is `succeeded`; returned `failed` or `cancelled` runs move -the entry to `failed`, record the attempted occurrence, and show the run error in the list. +scheduled occurrence time, target type, and stored flow version when present. The due occurrence is +claimed in SQLite before the runtime is awaited so a restart does not dispatch the same occurrence +again. Node-flow schedules are then finalized from the returned run status: `succeeded` advances or +completes the schedule, while returned `failed` or `cancelled` runs move the entry to `failed`, +record the attempted occurrence, and show the run error in the list. ## Managing entries diff --git a/docs/dashboard/scheduler.md b/docs/dashboard/scheduler.md index 632f4358c5..bc584ecb11 100644 --- a/docs/dashboard/scheduler.md +++ b/docs/dashboard/scheduler.md @@ -141,7 +141,7 @@ Due entries execute through existing production paths: AI memory remediation entries create a `remediation` invocation record even when no cleanup candidates are found; in that case the invocation is completed with a skipped reason instead of dispatching an empty provider request. -After a successful run, the service advances `nextRunAt` from the scheduled occurrence time. One-time entries move to `completed`; recurring entries stay `scheduled` until their count or end date/time is exhausted. Failed entries move to `failed` with `lastError` for operator visibility. Node-flow entries advance only after `runFlow` returns, so a runtime startup failure does not mark the schedule succeeded. +After a successful run, the service advances `nextRunAt` from the scheduled occurrence time. One-time entries move to `completed`; recurring entries stay `scheduled` until their count or end date/time is exhausted. Failed entries move to `failed` with `lastError` for operator visibility. Node-flow entries are durably claimed before `runFlow` is awaited so the same due occurrence is not dispatched again after a restart, then the scheduler entry is finalized from the returned node-flow run status. ### Node-Flow Schedules @@ -156,8 +156,9 @@ Behavior: - Resume recomputes `nextRunAt` to the first future occurrence so missed runs are not replayed immediately. - Due execution calls `NodeFlowRuntimeService.runFlow(projectId, flowId, input, { triggerType: "scheduler", triggerPayload })`. - The trigger payload includes scheduler entry id, scheduled occurrence time, target type, and `flowVersion` when the schedule stored one. +- Before the runtime is awaited, the due occurrence is claimed in SQLite. Absolute entries move `nextRunAt` off the claimed occurrence, and anchored entries record the claimed anchor occurrence so restart due checks skip it. - On success, one-time entries complete and recurring entries advance to the next occurrence. -- On failure, the schedule moves to `failed` with `lastError`; for node flows, `nextRunAt` is not advanced before runtime startup succeeds. +- On returned `failed` or `cancelled` runtime status, the schedule moves to `failed` with `lastError`, `lastRunAt`, and `runCount` recording the attempted occurrence. Anchored entries are evaluated separately from absolute `nextRunAt` polling: - An `after_sprint_end` entry is due only after the source sprint reaches `completed`, `failed`, or `cancelled`. diff --git a/src/repositories/scheduler-repository.ts b/src/repositories/scheduler-repository.ts index 45d79e931b..f3de02937b 100644 --- a/src/repositories/scheduler-repository.ts +++ b/src/repositories/scheduler-repository.ts @@ -254,6 +254,38 @@ export class SchedulerRepository { return updated; } + claimDueOccurrence(entryId: string, occurrenceIso: string, nextRunAt: string | null): SchedulerEntryRecord | null { + const current = this.getEntry(entryId); + if (!current || current.status !== "scheduled") { + return null; + } + + const now = new Date().toISOString(); + const result = current.scheduleAnchor + ? this.db.prepare(` + UPDATE scheduler_entries + SET next_run_at = ?, last_run_at = ?, last_error = NULL, updated_at = ? + WHERE id = ? + AND status = 'scheduled' + AND (last_run_at IS NULL OR last_run_at != ?) + `).run(nextRunAt, occurrenceIso, now, entryId, occurrenceIso) + : this.db.prepare(` + UPDATE scheduler_entries + SET next_run_at = ?, last_error = NULL, updated_at = ? + WHERE id = ? + AND status = 'scheduled' + AND next_run_at = ? + `).run(nextRunAt, now, entryId, occurrenceIso); + + if (result.changes === 0) { + return null; + } + + const updated = this.requireEntry(entryId); + this.publishProjectStructureRefresh(updated.projectId); + return updated; + } + markRunFailed(entryId: string, error: string, occurrenceIso?: string): SchedulerEntryRecord { const current = this.requireEntry(entryId); const now = new Date().toISOString(); diff --git a/src/services/scheduler-service.ts b/src/services/scheduler-service.ts index 0d89d6a95e..92b09c9b06 100644 --- a/src/services/scheduler-service.ts +++ b/src/services/scheduler-service.ts @@ -214,18 +214,24 @@ export class SchedulerService { : computeNextRunAfterOccurrence(occurrenceIso, freshEntry.recurrence, freshEntry.runCount + 1); if (freshEntry.targetType === "node_flow") { - this.executeNodeFlowEntry(freshEntry, occurrenceIso).then((result) => { + const claimedEntry = this.claimDueOccurrence(freshEntry, occurrenceIso, nextRunAt); + if (!claimedEntry) { + this.inFlightEntryIds.delete(entry.id); + continue; + } + + this.executeNodeFlowEntry(claimedEntry, occurrenceIso).then((result) => { if (result.run.status === "succeeded") { - this.deps.schedulerRepository.markRunSucceeded(freshEntry.id, occurrenceIso, nextRunAt); + this.deps.schedulerRepository.markRunSucceeded(claimedEntry.id, occurrenceIso, nextRunAt); return; } this.deps.schedulerRepository.markRunFailed( - freshEntry.id, + claimedEntry.id, result.run.errorMessage ?? `Node flow run ${result.run.status}.`, occurrenceIso, ); }).catch((error) => { - this.handleExecutionFailure(freshEntry, error); + this.handleExecutionFailure(claimedEntry, error); }).finally(() => { this.inFlightEntryIds.delete(entry.id); }); @@ -254,6 +260,18 @@ export class SchedulerService { this.deps.schedulerRepository.markRunFailed(entry.id, message); } + private claimDueOccurrence( + entry: SchedulerEntryRecord, + occurrenceIso: string, + nextRunAt: string | null, + ): SchedulerEntryRecord | null { + const claimDueOccurrence = this.deps.schedulerRepository.claimDueOccurrence; + if (typeof claimDueOccurrence !== "function") { + return entry; + } + return claimDueOccurrence.call(this.deps.schedulerRepository, entry.id, occurrenceIso, nextRunAt); + } + private async executeEntry(entry: SchedulerEntryRecord, occurrenceIso: string): Promise { if (entry.targetType === "sprint") { const sprintId = entry.sprintTarget?.sprintId; @@ -495,7 +513,11 @@ export class SchedulerService { return null; } const dueAt = new Date(anchorTime.getTime() + ((entry.scheduleAnchor.offsetMinutes ?? 0) * 60_000)); - return dueAt.getTime() <= now.getTime() ? dueAt.toISOString() : null; + const dueIso = dueAt.toISOString(); + if (entry.lastRunAt === dueIso) { + return null; + } + return dueAt.getTime() <= now.getTime() ? dueIso : null; } private resolveAnchorOccurrenceStart(entry: SchedulerEntryRecord): string | null { diff --git a/tests/backend/repositories/scheduler-repository.test.ts b/tests/backend/repositories/scheduler-repository.test.ts index fd749c4566..1bac634aa3 100644 --- a/tests/backend/repositories/scheduler-repository.test.ts +++ b/tests/backend/repositories/scheduler-repository.test.ts @@ -310,6 +310,38 @@ describe("SchedulerRepository", () => { expect(updated.runCount).toBe(1); }); + it("claims due occurrences without incrementing run accounting", async () => { + const { dir, projectRepository, schedulerRepository } = await createRepositories(); + const project = projectRepository.createProject({ + name: "Scheduler Project", + sourceType: "local", + sourceRef: dir, + }); + + const entry = schedulerRepository.createEntry(project.id, { + targetType: "node_flow", + scheduledFor: "2026-05-18T09:00:00.000Z", + recurrence: { frequency: "daily", interval: 1, endMode: "after_count", count: 2 }, + nodeFlowTarget: { flowId: "flow-1" }, + }); + + const claimed = schedulerRepository.claimDueOccurrence( + entry.id, + "2026-05-18T09:00:00.000Z", + "2026-05-19T09:00:00.000Z", + ); + const duplicateClaim = schedulerRepository.claimDueOccurrence( + entry.id, + "2026-05-18T09:00:00.000Z", + "2026-05-19T09:00:00.000Z", + ); + + expect(claimed?.nextRunAt).toBe("2026-05-19T09:00:00.000Z"); + expect(claimed?.lastRunAt).toBeNull(); + expect(claimed?.runCount).toBe(0); + expect(duplicateClaim).toBeNull(); + }); + it("persists settings-managed memory remediation targets", async () => { const { dir, projectRepository, schedulerRepository } = await createRepositories(); const project = projectRepository.createProject({ diff --git a/tests/backend/services/scheduler-service.test.ts b/tests/backend/services/scheduler-service.test.ts index 35160ed774..da8ec3d3d3 100644 --- a/tests/backend/services/scheduler-service.test.ts +++ b/tests/backend/services/scheduler-service.test.ts @@ -1,7 +1,19 @@ -import { describe, expect, it, vi } from "vitest"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import * as fs from "fs/promises"; +import * as os from "os"; +import * as path from "path"; import { SchedulerService } from "../../../src/services/scheduler-service.js"; import { normalizeRecurrenceRule } from "../../../src/domain/scheduler/schedule-time.js"; import type { SchedulerEntryRecord } from "../../../src/contracts/scheduler-types.js"; +import { AppDbStorage } from "../../../src/repositories/app-db-storage.js"; +import { ProjectManagementRepository } from "../../../src/repositories/project-management-repository.js"; +import { SchedulerRepository } from "../../../src/repositories/scheduler-repository.js"; + +const tempDirs: string[] = []; + +afterEach(async () => { + await Promise.all(tempDirs.splice(0).map((dir) => fs.rm(dir, { recursive: true, force: true }))); +}); const createLogger = () => ({ error: vi.fn(), @@ -337,6 +349,61 @@ describe("SchedulerService", () => { expect(repo.markRunSucceeded).toHaveBeenCalledWith("entry-1", "2026-05-18T09:00:00.000Z", null); }); + it("durably claims a due node flow occurrence before a long-running runtime resolves", async () => { + const dir = await fs.mkdtemp(path.join(os.tmpdir(), "scheduler-service-")); + tempDirs.push(dir); + const storage = new AppDbStorage(path.join(dir, "app.db")); + const projectManagementRepository = new ProjectManagementRepository(storage); + const schedulerRepository = new SchedulerRepository(storage); + const project = projectManagementRepository.createProject({ + name: "Scheduler Project", + sourceType: "local", + sourceRef: dir, + }); + const entry = schedulerRepository.createEntry(project.id, { + targetType: "node_flow", + scheduledFor: "2026-05-18T09:00:00.000Z", + nodeFlowTarget: { flowId: "flow-1" }, + }); + let resolveRun: (value: { run: { status: "succeeded" }; nodeRuns: []; output: Record }) => void = () => {}; + const nodeFlowRuntimeService = { + runFlow: vi.fn(() => new Promise((resolve) => { + resolveRun = resolve; + })), + }; + const nodeFlowRepository = { + getFlow: vi.fn(() => ({ id: "flow-1", projectId: project.id })), + }; + const firstScheduler = buildService(schedulerRepository, { + projectManagementRepository, + nodeFlowRuntimeService, + nodeFlowRepository, + }); + const restartedScheduler = buildService(schedulerRepository, { + projectManagementRepository, + nodeFlowRuntimeService, + nodeFlowRepository, + }); + const now = new Date("2026-05-18T09:00:01.000Z"); + + await firstScheduler.runDueEntries(now); + + expect(nodeFlowRuntimeService.runFlow).toHaveBeenCalledTimes(1); + expect(schedulerRepository.listDueEntries(now.toISOString()).find((due) => due.id === entry.id)).toBeUndefined(); + + await restartedScheduler.runDueEntries(now); + + expect(nodeFlowRuntimeService.runFlow).toHaveBeenCalledTimes(1); + + resolveRun({ run: { status: "succeeded" }, nodeRuns: [], output: {} }); + await flush(); + + const finalized = schedulerRepository.getEntry(entry.id); + expect(finalized?.status).toBe("completed"); + expect(finalized?.lastRunAt).toBe("2026-05-18T09:00:00.000Z"); + expect(finalized?.runCount).toBe(1); + }); + it("marks node flow schedules failed when the runtime resolves a failed run", async () => { const entry = createEntry({ targetType: "node_flow",