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) <keithpk+paseo-dev@kowal.me> Co-authored-by: keithpk <keith@kowal.me> Co-authored-by: paseo-bot[bot] <266920839+paseo-bot[bot]@users.noreply.github.com>
2026-10-14 13:22:37 +00:00 · 2026-10-11 02:19:11 +02:00
parent d5513472ed
commit ee0d901efb
2 changed files with 364 additions and 18 deletions
@@ -44,6 +44,37 @@ interface ScheduleServiceInternals {
  executeSchedule(schedule: StoredSchedule, runId: string): Promise<ScheduleExecutionResult>;
 }

+class OverlappingTickRunner {
+  private resolveFirstRunStarted: () => void = () => {};
+  private resolveFirstRunBlocked: () => void = () => {};
+  private laterScheduleRunCount = 0;
+
+  readonly firstRunStarted = new Promise<void>((resolve) => {
+    this.resolveFirstRunStarted = resolve;
+  });
+  private readonly firstRunBlocked = new Promise<void>((resolve) => {
+    this.resolveFirstRunBlocked = resolve;
+  });
+
+  async run(schedule: StoredSchedule): Promise<ScheduleExecutionResult> {
+    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<ScheduleExecutionResult> => {
+        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,
@@ -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<void> {
    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(),
        };
      }