From ee0d901efbf83a09d77edaf934c62c2e1f424d21 Mon Sep 17 00:00:00 2001 From: "paseo-bot[bot]" <266920839+paseo-bot[bot]@users.noreply.github.com> Date: Sun, 11 Oct 2026 02:19:11 +0200 Subject: [PATCH] Run each scheduled slot once without skipping the next occurrence (#6532) * fix(server): claim scheduled slots atomically * fix(server): skip elapsed schedule slots * test(server): isolate overlapping tick runner * test(schedule): reproduce duplicate cron dispatch across overlapping ticks * fix(schedule): preserve final slots and recovery after failed outcome writes Historical running records are not proof of a live execution after an outcome write fails. Keep live-run exclusion in the existing in-memory guard while the persisted cursor prevents duplicate slot claims. Final allowed runs do not need a future cron occurrence; recover interrupted final runs and preserve cadence if their limit changes during execution. --------- Co-authored-by: Keith Kowal (paseo-dev) Co-authored-by: keithpk Co-authored-by: paseo-bot[bot] <266920839+paseo-bot[bot]@users.noreply.github.com> --- .../src/server/schedule/service.test.ts | 266 ++++++++++++++++++ .../server/src/server/schedule/service.ts | 116 ++++++-- 2 files changed, 364 insertions(+), 18 deletions(-) diff --git a/packages/server/src/server/schedule/service.test.ts b/packages/server/src/server/schedule/service.test.ts index 18e23e0ed1..1a9f0f5fd4 100644 --- a/packages/server/src/server/schedule/service.test.ts +++ b/packages/server/src/server/schedule/service.test.ts @@ -44,6 +44,37 @@ interface ScheduleServiceInternals { executeSchedule(schedule: StoredSchedule, runId: string): Promise; } +class OverlappingTickRunner { + private resolveFirstRunStarted: () => void = () => {}; + private resolveFirstRunBlocked: () => void = () => {}; + private laterScheduleRunCount = 0; + + readonly firstRunStarted = new Promise((resolve) => { + this.resolveFirstRunStarted = resolve; + }); + private readonly firstRunBlocked = new Promise((resolve) => { + this.resolveFirstRunBlocked = resolve; + }); + + async run(schedule: StoredSchedule): Promise { + if (schedule.prompt === "block the first tick") { + this.resolveFirstRunStarted(); + await this.firstRunBlocked; + } else { + this.laterScheduleRunCount += 1; + } + return { agentId: null, output: "ok" }; + } + + unblockFirstRun(): void { + this.resolveFirstRunBlocked(); + } + + get laterScheduleRuns(): number { + return this.laterScheduleRunCount; + } +} + const SCHEDULE_TEST_CAPABILITIES: AgentCapabilityFlags = { supportsStreaming: true, supportsSessionPersistence: true, @@ -345,6 +376,241 @@ describe("ScheduleService", () => { expect(inspected.nextRunAt).toBe("2026-01-01T00:02:00.000Z"); }); + test.each(["new-agent", "agent", "maxRuns: 1"] as const)( + "dispatches a shared cron slot only once for %s after an earlier tick resumes", + async (targetKind) => { + const runner = new OverlappingTickRunner(); + const service = createScheduleService({ + paseoHome: tempDir, + logger: createTestLogger(), + agentManager: new AgentManager({ logger: createTestLogger() }), + agentStorage, + providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY, + now: () => now, + runner: (schedule) => runner.run(schedule), + }); + const cadence = { type: "cron" as const, expression: "0 6 * * 1", timezone: "Europe/Prague" }; + await service.create({ + prompt: "block the first tick", + cadence, + target: { type: "new-agent", config: { provider: "claude", cwd: tempDir } }, + }); + // Give the later schedule a distinct creation time so file ordering cannot affect the race. + now = new Date("2026-01-01T00:00:01.000Z"); + const agentId = "00000000-0000-0000-0000-000000000001"; + await agentStorage.upsert( + buildAgentRecord({ id: agentId, cwd: tempDir, iso: now.toISOString() }), + ); + const later = await service.create({ + prompt: "later due schedule", + cadence, + target: + targetKind === "agent" + ? { type: "agent", agentId } + : { type: "new-agent", config: { provider: "claude", cwd: tempDir } }, + maxRuns: targetKind === "maxRuns: 1" ? 1 : undefined, + }); + now = new Date("2026-01-05T05:00:00.000Z"); + const delayedTick = service.tick(); + await runner.firstRunStarted; + try { + now = new Date("2026-01-05T05:00:01.000Z"); + await service.tick(); + } finally { + runner.unblockFirstRun(); + await delayedTick; + } + const result = await service.inspect(later.id); + expect.soft(runner.laterScheduleRuns).toBe(1); + expect.soft(result.runs).toHaveLength(1); + expect.soft(result.runs.map((run) => run.scheduledFor)).toEqual(["2026-01-05T05:00:00.000Z"]); + expect(result.nextRunAt).toBe( + targetKind === "maxRuns: 1" ? null : "2026-01-12T05:00:00.000Z", + ); + expect(result.status).toBe(targetKind === "maxRuns: 1" ? "completed" : "active"); + }, + ); + + test("advances the next scheduled slot past a run that crosses a cadence boundary", async () => { + const service = createScheduleService({ + paseoHome: tempDir, + logger: createTestLogger(), + agentManager: new AgentManager({ logger: createTestLogger() }), + agentStorage, + providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY, + now: () => now, + runner: async () => { + now = new Date("2026-01-01T00:02:00.000Z"); + return { agentId: null, output: "ok" }; + }, + }); + + const created = await service.create({ + prompt: "long-running task", + cadence: { type: "every", everyMs: 60_000 }, + target: { type: "new-agent", config: { provider: "claude", cwd: tempDir } }, + }); + + now = new Date("2026-01-01T00:01:00.000Z"); + await service.tick(); + + expect((await service.inspect(created.id)).nextRunAt).toBe("2026-01-01T00:03:00.000Z"); + }); + + test("runs the final allowed cron slot without requiring another occurrence", async () => { + let dispatches = 0; + const service = createScheduleService({ + paseoHome: tempDir, + logger: createTestLogger(), + agentManager: new AgentManager({ logger: createTestLogger() }), + agentStorage, + providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY, + now: () => now, + runner: async () => { + dispatches += 1; + return { agentId: null, output: "done" }; + }, + }); + const created = await service.create({ + prompt: "Run once on January 5 when it is Monday", + cadence: { type: "cron", expression: "0 6 5 1 1" }, + target: { type: "new-agent", config: { provider: "claude", cwd: tempDir } }, + maxRuns: 1, + }); + now = new Date("2026-01-05T06:00:00.000Z"); + await service.tick(); + const result = await service.inspect(created.id); + expect(dispatches).toBe(1); + expect(result.status).toBe("completed"); + expect(result.nextRunAt).toBeNull(); + expect(result.runs).toHaveLength(1); + expect(result.runs[0].status).toBe("succeeded"); + }); + + test.skipIf(process.platform === "win32" || process.getuid?.() === 0)( + "keeps firing future slots after an outcome could not be saved", + async () => { + const schedulesDir = join(tempDir, "schedules"); + let dispatches = 0; + const failOutcomeWrite = async (): Promise => { + dispatches += 1; + if (dispatches === 1) { + await chmod(schedulesDir, 0o555); + } + return { agentId: null, output: "done" }; + }; + const service = createScheduleService({ + paseoHome: tempDir, + logger: createTestLogger(), + agentManager: new AgentManager({ logger: createTestLogger() }), + agentStorage, + providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY, + now: () => now, + runner: failOutcomeWrite, + }); + const created = await service.create({ + prompt: "Check status each minute", + runOnCreate: false, + cadence: { type: "every", everyMs: 60_000 }, + target: { type: "new-agent", config: { provider: "claude", cwd: tempDir } }, + }); + now = new Date("2026-01-01T00:01:00.000Z"); + try { + await expect(service.tick()).rejects.toThrow(); + } finally { + await chmod(schedulesDir, 0o755); + } + expect((await service.inspect(created.id)).runs[0].status).toBe("running"); + now = new Date("2026-01-01T00:02:00.000Z"); + await service.tick(); + const result = await service.inspect(created.id); + expect(dispatches).toBe(2); + expect(result.runs.map((run) => run.scheduledFor)).toEqual([ + "2026-01-01T00:01:00.000Z", + "2026-01-01T00:02:00.000Z", + ]); + expect(result.nextRunAt).toBe("2026-01-01T00:03:00.000Z"); + }, + ); + + test("completes an interrupted final slot during restart recovery", async () => { + const service = createScheduleService({ + paseoHome: tempDir, + logger: createTestLogger(), + agentManager: new AgentManager({ logger: createTestLogger() }), + agentStorage, + providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY, + now: () => now, + runner: async () => ({ agentId: null, output: "done" }), + }); + const created = await service.create({ + prompt: "Run once", + cadence: { type: "cron", expression: "0 6 5 1 1" }, + target: { type: "new-agent", config: { provider: "claude", cwd: tempDir } }, + maxRuns: 1, + }); + now = new Date("2026-01-05T06:00:00.000Z"); + const store = new ScheduleStore(join(tempDir, "schedules"), createTestLogger()); + await store.update(created.id, (schedule) => ({ + ...schedule, + nextRunAt: null, + runs: [ + { + id: "interrupted-final-slot", + scheduledFor: now.toISOString(), + startedAt: now.toISOString(), + endedAt: null, + status: "running", + agentId: null, + output: null, + error: null, + }, + ], + })); + now = new Date("2026-01-05T06:01:00.000Z"); + await service.start(); + try { + const result = await service.inspect(created.id); + expect(result.status).toBe("completed"); + expect(result.nextRunAt).toBeNull(); + expect(result.runs).toHaveLength(1); + expect(result.runs[0].status).toBe("failed"); + } finally { + await service.stop(); + } + }); + + test.each([2, null])( + "keeps a next slot when the final run's limit changes to %s", + async (maxRuns) => { + const service = createScheduleService({ + paseoHome: tempDir, + logger: createTestLogger(), + agentManager: new AgentManager({ logger: createTestLogger() }), + agentStorage, + providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY, + now: () => now, + runner: async (schedule) => { + await service.update({ id: schedule.id, maxRuns }); + return { agentId: null, output: "done" }; + }, + }); + const created = await service.create({ + prompt: "Allow more runs during execution", + runOnCreate: false, + cadence: { type: "every", everyMs: 60_000 }, + target: { type: "new-agent", config: { provider: "claude", cwd: tempDir } }, + maxRuns: 1, + }); + now = new Date("2026-01-01T00:01:30.000Z"); + await service.tick(); + const result = await service.inspect(created.id); + expect(result.status).toBe("active"); + expect(result.nextRunAt).toBe("2026-01-01T00:02:00.000Z"); + expect(result.runs).toHaveLength(1); + }, + ); + test("pause and resume update persisted schedule state", async () => { const service = createScheduleService({ paseoHome: tempDir, diff --git a/packages/server/src/server/schedule/service.ts b/packages/server/src/server/schedule/service.ts index 3608dd8636..7b5487c3f8 100644 --- a/packages/server/src/server/schedule/service.ts +++ b/packages/server/src/server/schedule/service.ts @@ -38,6 +38,18 @@ export class ScheduleTargetGoneError extends Error { } } +function advanceNextRunAtPast( + cadence: StoredSchedule["cadence"], + nextRunAt: Date, + now: Date, +): Date { + let next = nextRunAt; + while (next.getTime() <= now.getTime()) { + next = computeNextRunAt(cadence, next); + } + return next; +} + function trimOptionalName(value: string | null | undefined): string | null { if (typeof value !== "string") { return null; @@ -621,6 +633,9 @@ export class ScheduleService { error: "Daemon restarted before the scheduled run completed", }; updated = { ...updated, runs }; + if (shouldCompleteSchedule(updated, now)) { + updated = completeSchedule(updated, now); + } dirty = true; } @@ -690,20 +705,31 @@ export class ScheduleService { options?: { manual?: boolean }, ): Promise { const manual = options?.manual === true; + // Reserve before store I/O; persisted history cannot prove a run is still alive. this.runningScheduleIds.add(schedule.id); try { - const runId = randomUUID(); - const runningRun: ScheduleRun = { - id: runId, - scheduledFor: manual ? now.toISOString() : (schedule.nextRunAt ?? now.toISOString()), - startedAt: now.toISOString(), - endedAt: null, - status: "running", - agentId: null, - output: null, - error: null, - }; - const scheduleWithRun = await this.appendRunningRun(schedule.id, runningRun); + let runId: string; + let scheduleWithRun: StoredSchedule; + if (manual) { + runId = randomUUID(); + const runningRun: ScheduleRun = { + id: runId, + scheduledFor: now.toISOString(), + startedAt: now.toISOString(), + endedAt: null, + status: "running", + agentId: null, + output: null, + error: null, + }; + scheduleWithRun = await this.appendRunningRun(schedule.id, runningRun); + } else { + const claimed = await this.claimDueRun(schedule.id, now); + if (!claimed) { + return; + } + ({ schedule: scheduleWithRun, runId } = claimed); + } try { const result = await this.runner(scheduleWithRun, runId); @@ -734,6 +760,58 @@ export class ScheduleService { } } + private async claimDueRun( + scheduleId: string, + now: Date, + ): Promise<{ schedule: StoredSchedule; runId: string } | null> { + let claimed: { schedule: StoredSchedule; runId: string } | null = null; + const updatedSchedule = await this.store.update(scheduleId, (schedule) => { + if ( + schedule.status !== "active" || + !schedule.nextRunAt || + shouldCompleteSchedule(schedule, now) || + new Date(schedule.nextRunAt).getTime() > now.getTime() + ) { + return schedule; + } + + const scheduledFor = schedule.nextRunAt; + const isFinalRun = + schedule.maxRuns !== null && countCompletedRuns(schedule) + 1 >= schedule.maxRuns; + const nextRunAt = isFinalRun + ? null + : advanceNextRunAtPast( + schedule.cadence, + computeNextRunAt(schedule.cadence, new Date(scheduledFor)), + now, + ).toISOString(); + + // Claim the due slot and advance its cursor in the same serialized update. + // A tick can hold a stale list while another tick finishes this schedule. + const runId = randomUUID(); + const runningRun: ScheduleRun = { + id: runId, + scheduledFor, + startedAt: now.toISOString(), + endedAt: null, + status: "running", + agentId: null, + output: null, + error: null, + }; + const updated: StoredSchedule = { + ...schedule, + nextRunAt, + updatedAt: now.toISOString(), + runs: [...schedule.runs, runningRun], + }; + claimed = { schedule: updated, runId }; + return updated; + }); + requireSchedule(updatedSchedule, scheduleId); + return claimed; + } + private async appendRunningRun( scheduleId: string, runningRun: ScheduleRun, @@ -794,14 +872,16 @@ export class ScheduleService { nextRunAt: null, }; } else { - const after = new Date(schedule.nextRunAt ?? now.toISOString()); - let nextRunAt = computeNextRunAt(updated.cadence, after); - while (nextRunAt.getTime() <= now.getTime()) { - nextRunAt = computeNextRunAt(updated.cadence, nextRunAt); - } + // A final slot has no successor until its run limit is raised or cleared. + // Every runner invocation has a recorded run with this id. + const finishedRun = completedRuns.find((run) => run.id === params.runId)!; + const nextRunAt = updated.nextRunAt + ? new Date(updated.nextRunAt) + : computeNextRunAt(updated.cadence, new Date(finishedRun.scheduledFor)); + // Move an overdue cursor past a long run without advancing a future one twice. updated = { ...updated, - nextRunAt: nextRunAt.toISOString(), + nextRunAt: advanceNextRunAtPast(updated.cadence, nextRunAt, now).toISOString(), }; }