diff --git a/packages/opencode/src/automation/index.ts b/packages/opencode/src/automation/index.ts index 4b70ebc15..74fab2967 100644 --- a/packages/opencode/src/automation/index.ts +++ b/packages/opencode/src/automation/index.ts @@ -234,7 +234,7 @@ export namespace Automation { definitions: Map runs: Map activeWriters: Set - activeRuns: Map + activeRuns: Map } const state = Instance.state(() => ({ definitions: new Map(), @@ -503,11 +503,15 @@ export namespace Automation { return replaceRun(Run.parse(next)) } - function stopRun(run: Run, stopReason: Extract["stopReason"]): Run { + function stopRun( + run: Run, + stopReason: Extract["stopReason"], + options?: { now?: number }, + ): Run { if (run.state === "stopped" || run.state === "succeeded" || run.state === "failed") return run return reviseRun(run, { state: "stopped", - completedAt: Date.now(), + completedAt: options?.now ?? Date.now(), result: null, error: null, stopReason, @@ -518,12 +522,26 @@ export namespace Automation { const active = state().activeRuns.get(automationID) if (!active) return undefined active.controller.abort() - const current = state().runs.get(automationID)?.find((run) => ( - run.state === "scheduled" || run.state === "running" || run.state === "awaiting_input" - )) + const current = state().runs.get(automationID)?.find((run) => run.id === active.runID) return current ? stopRun(current, "cancelled") : undefined } + export function stopRunByID( + runID: string, + stopReason: Extract["stopReason"], + options?: { now?: number }, + ): Run | undefined { + for (const runs of state().runs.values()) { + const run = runs.find((item) => item.id === runID) + if (!run) continue + const active = state().activeRuns.get(run.automationID) + if (active?.runID === runID) active.controller.abort() + const stopped = stopRun(run, stopReason, options) + return stopped === run ? undefined : stopped + } + return undefined + } + export function markRunStarted(run: Run, sessionID: SessionID, options?: { now?: number }): Run { return reviseRun(run, { state: "running", @@ -584,6 +602,32 @@ export namespace Automation { return run } + export function hasActiveRun(automationID: string): boolean { + if (state().activeRuns.has(automationID)) return true + return (state().runs.get(automationID) ?? []).some( + (run) => run.state === "scheduled" || run.state === "running" || run.state === "awaiting_input", + ) + } + + export function hasRunTriggeredAtOrAfter(automationID: string, triggeredAt: number): boolean { + return (state().runs.get(automationID) ?? []).some((run) => run.triggeredAt >= triggeredAt) + } + + export function completedRunCount(automationID: string): number { + get(automationID) + const runs = state().runs.get(automationID) ?? [] + return runs.filter((run) => run.state === "succeeded" || run.state === "failed").length + } + + export function recordStoppedRun( + automationID: string, + stopReason: Extract["stopReason"], + options?: { now?: number; triggeredAt?: number }, + ): Run { + const run = runNow(automationID, { now: options?.triggeredAt ?? options?.now }) + return stopRun(run, stopReason, options) + } + export function runNowExecuting( id: string, options: { executor: RunExecutor; attendance?: AutomationRunAttendance; now?: number }, @@ -608,7 +652,7 @@ export namespace Automation { } data.activeWriters.add(writerKey) const controller = new AbortController() - data.activeRuns.set(initial.automationID, { writerKey, controller }) + data.activeRuns.set(initial.automationID, { writerKey, controller, runID: initial.id }) let current = initial try { const prepared = await executor({ definition, run: initial, attendance, signal: controller.signal }) @@ -663,8 +707,11 @@ export namespace Automation { }) await publishRunUpdated(failed) } finally { - data.activeRuns.delete(initial.automationID) - data.activeWriters.delete(writerKey) + const active = data.activeRuns.get(initial.automationID) + if (active?.runID === initial.id) { + data.activeRuns.delete(initial.automationID) + data.activeWriters.delete(writerKey) + } } } diff --git a/packages/opencode/src/automation/scheduler.ts b/packages/opencode/src/automation/scheduler.ts new file mode 100644 index 000000000..c3e7b96e5 --- /dev/null +++ b/packages/opencode/src/automation/scheduler.ts @@ -0,0 +1,333 @@ +import { Context, Effect, Fiber, Layer } from "effect" +import { Automation } from "." +import { Bus } from "@/bus" +import { Instance, type InstanceContext } from "@/project/instance" +import { NotFoundError } from "@/storage/db" +import { sessionPromptExecutor } from "./runner" + +export namespace AutomationScheduler { + const MAX_TIMER_DELAY_MS = 2_147_483_647 + const MISSED_SCHEDULE_GRACE_MS = 60_000 + + export interface Clock { + now(): number + sleep(delayMs: number, signal: AbortSignal): Promise + } + + export interface Task { + interrupt(): void + } + + export interface TaskRuntime { + fork(run: (signal: AbortSignal) => Effect.Effect): Task + } + + export interface Interface { + stop(): void + stopOwnedRuns(): void + reschedule(definition: Automation.Definition): void + cancel(automationID: string): void + computeNextFireAt(definition: Automation.Definition, from?: number): number | null + } + + export interface Options { + clock?: Clock + executor?: Automation.RunExecutor + runtime?: TaskRuntime + } + + type ScheduledTask = { + task: Task + fireAt: number + token: symbol + definition: Automation.Definition + } + + export const liveClock: Clock = { + now: () => Date.now(), + sleep: (delayMs, signal) => + new Promise((resolve) => { + if (signal.aborted) { + resolve() + return + } + const id = setTimeout(resolve, delayMs) + id.unref?.() + signal.addEventListener( + "abort", + () => { + clearTimeout(id) + resolve() + }, + { once: true }, + ) + }), + } + + export const liveRuntime: TaskRuntime = { + fork(run) { + const controller = new AbortController() + const fiber = Effect.runFork(run(controller.signal)) + return { + interrupt() { + controller.abort() + Effect.runFork(Fiber.interrupt(fiber)) + }, + } + }, + } + + export class Service extends Context.Service()("@pawwork/AutomationScheduler") {} + + export const layer = (options?: Options) => + Layer.effect( + Service, + Effect.gen(function* () { + const scheduler = make(options) + yield* Effect.addFinalizer(() => Effect.sync(() => scheduler.stop())) + return Service.of(scheduler) + }), + ) + export const defaultLayer = layer() + + export function computeNextFireAt(definition: Automation.Definition, from: number): number | null { + if (definition.paused) return null + if (definition.kind === "oneshot") { + if (Automation.hasRunTriggeredAtOrAfter(definition.id, definition.fireAt)) return null + return definition.fireAt + } + if (!canScheduleRecurring(definition)) return null + if (definition.rhythm.kind !== "interval") return null + return from + definition.rhythm.everyMs + } + + function canScheduleRecurring(definition: Extract) { + if (definition.stop.kind === "never") return true + if (definition.stop.kind === "count") return Automation.completedRunCount(definition.id) < definition.stop.count + return false + } + + function isSameSchedule(left: Automation.Definition, right: Automation.Definition) { + if (left.kind !== right.kind || left.paused !== right.paused) return false + if (left.kind === "oneshot" && right.kind === "oneshot") return left.fireAt === right.fireAt + if (left.kind === "recurring" && right.kind === "recurring") { + return JSON.stringify(left.rhythm) === JSON.stringify(right.rhythm) && JSON.stringify(left.stop) === JSON.stringify(right.stop) + } + return false + } + + export function make(options: Options = {}): Interface { + const clock = options.clock ?? liveClock + const executor = options.executor ?? sessionPromptExecutor + const runtime = options.runtime ?? liveRuntime + const tasks = new Map() + const ownedRuns = new Map() + const schedulerStoppedRuns = new Set() + let running = true + + const cancel = (automationID: string) => { + const entry = tasks.get(automationID) + if (!entry) return + tasks.delete(automationID) + entry.task.interrupt() + } + + const scheduleNextInterval = (automationID: string) => { + try { + const latest = Automation.get(automationID) + if (latest.kind === "recurring" && latest.rhythm.kind === "interval" && !latest.paused && canScheduleRecurring(latest)) { + schedule(latest, clock.now() + latest.rhythm.everyMs) + } else { + cancel(automationID) + } + } catch (error) { + cancel(automationID) + if (!NotFoundError.isInstance(error)) throw error + } + } + + const fire = (automationID: string, triggeredAt: number) => { + tasks.delete(automationID) + const firedAt = clock.now() + try { + const latest = Automation.get(automationID) + if (latest.paused) return + if (latest.kind === "oneshot" && latest.fireAt !== triggeredAt) return + if (latest.kind === "recurring" && (latest.rhythm.kind !== "interval" || !canScheduleRecurring(latest))) return + if (firedAt - triggeredAt > MISSED_SCHEDULE_GRACE_MS) { + const stopped = Automation.recordStoppedRun(automationID, "missed_schedule", { now: firedAt, triggeredAt }) + schedulerStoppedRuns.add(stopped.id) + void Automation.publishRunUpdated(stopped) + if (latest.kind === "recurring") scheduleNextInterval(automationID) + return + } + } catch (error) { + if (!NotFoundError.isInstance(error)) throw error + return + } + if (Automation.hasActiveRun(automationID)) { + const stopped = Automation.recordStoppedRun(automationID, "previous_run_awaiting_input", { now: triggeredAt }) + schedulerStoppedRuns.add(stopped.id) + void Automation.publishRunUpdated(stopped) + scheduleNextInterval(automationID) + return + } + try { + const run = Automation.runNowExecuting(automationID, { + executor, + attendance: "unattended", + now: triggeredAt, + }) + ownedRuns.set(run.id, automationID) + } catch (error) { + if (!NotFoundError.isInstance(error)) throw error + } + } + + const isCurrentTask = (automationID: string, fireAt: number, token: symbol, signal: AbortSignal) => { + const current = tasks.get(automationID) + return !signal.aborted && running && current?.token === token && current.fireAt === fireAt + } + + const waitUntil = (automationID: string, fireAt: number, token: symbol, signal: AbortSignal): Effect.Effect => + Effect.gen(function* () { + while (clock.now() < fireAt) { + const delayMs = Math.max(0, fireAt - clock.now()) + yield* Effect.promise(() => clock.sleep(Math.min(delayMs, MAX_TIMER_DELAY_MS), signal)) + if (!isCurrentTask(automationID, fireAt, token, signal)) return + } + if (!isCurrentTask(automationID, fireAt, token, signal)) return + fire(automationID, fireAt) + }) + + const schedule = (definition: Automation.Definition, fireAt: number) => { + cancel(definition.id) + if (!running || definition.paused) return + const token = Symbol(definition.id) + tasks.set(definition.id, { + task: runtime.fork((signal) => waitUntil(definition.id, fireAt, token, signal)), + fireAt, + token, + definition, + }) + } + + const preservePendingSchedule = (definition: Automation.Definition) => { + const current = tasks.get(definition.id) + if (!current || current.fireAt <= clock.now() || !isSameSchedule(current.definition, definition)) return false + current.definition = definition + return true + } + + const hasSchedulerOwnedActiveRun = (automationID: string) => { + for (const ownedAutomationID of ownedRuns.values()) { + if (ownedAutomationID === automationID) return true + } + return false + } + + const reschedule = (definition: Automation.Definition) => { + const next = computeNextFireAt(definition, clock.now()) + if (next === null) { + cancel(definition.id) + return + } + if (preservePendingSchedule(definition)) return + if (!tasks.has(definition.id) && definition.kind === "recurring" && hasSchedulerOwnedActiveRun(definition.id)) return + schedule(definition, next) + } + + const unsubscribeRunUpdates = Bus.subscribe(Automation.Event.RunUpdated, (event) => { + if (!running) return + const run = event.properties + if (run.state === "scheduled" || run.state === "running" || run.state === "awaiting_input") return + const wasOwned = ownedRuns.delete(run.id) + const wasSchedulerStopped = schedulerStoppedRuns.delete(run.id) + if (run.state === "stopped" && !wasOwned && !wasSchedulerStopped) return + scheduleNextInterval(run.automationID) + }) + const unsubscribeDefinitionUpdates = Bus.subscribe(Automation.Event.DefinitionUpdated, (event) => { + if (!running) return + reschedule(event.properties) + }) + const unsubscribeDefinitionDeletes = Bus.subscribe(Automation.Event.DefinitionDeleted, (event) => { + if (!running) return + cancel(event.properties.id) + }) + + for (const definition of Automation.list()) reschedule(definition) + + const stopOwnedRuns = () => { + for (const runID of [...ownedRuns.keys()]) { + const stopped = Automation.stopRunByID(runID, "cancelled", { now: clock.now() }) + ownedRuns.delete(runID) + if (stopped) void Automation.publishRunUpdated(stopped) + } + } + + return { + stop() { + running = false + unsubscribeRunUpdates() + unsubscribeDefinitionUpdates() + unsubscribeDefinitionDeletes() + stopOwnedRuns() + for (const automationID of [...tasks.keys()]) cancel(automationID) + }, + stopOwnedRuns, + reschedule, + cancel, + computeNextFireAt(definition, from = clock.now()) { + return computeNextFireAt(definition, from) + }, + } + } + + type OwnerState = { + context: InstanceContext + scheduler: Interface + } + + const owners = new Map() + const owner = Instance.state( + () => { + const context = Instance.current + const state = { context, scheduler: make() } + owners.set(context.directory, state) + return state + }, + async (state) => { + owners.delete(state.context.directory) + Instance.restore(state.context, () => state.scheduler.stop()) + }, + ) + + export function current(): Interface { + return owner().scheduler + } + + export function install(scheduler: Interface): Interface { + const state = owner() + const previous = state.scheduler + previous.stop() + state.scheduler = scheduler + return previous + } + + export function stopCurrentOwnedRuns(): void { + const state = owner() + Instance.restore(state.context, () => state.scheduler.stopOwnedRuns()) + } + + export function stopDirectoryOwnedRuns(directory: string): void { + const state = owners.get(directory) + if (!state) return + Instance.restore(state.context, () => state.scheduler.stopOwnedRuns()) + } + + export function stopAllOwnedRuns(): void { + for (const state of owners.values()) { + Instance.restore(state.context, () => state.scheduler.stopOwnedRuns()) + } + } +} diff --git a/packages/opencode/src/project/instance.ts b/packages/opencode/src/project/instance.ts index 44b29e32a..f8ac65b99 100644 --- a/packages/opencode/src/project/instance.ts +++ b/packages/opencode/src/project/instance.ts @@ -1,3 +1,4 @@ +import { Log } from "@opencode-ai/core/util/log" import { Filesystem } from "@/util/filesystem" import { context, containsPath as containsPathInContext, type InstanceContext } from "./instance-context" import { Project } from "./project" @@ -6,11 +7,27 @@ import { State } from "./state" export type { InstanceContext } from "./instance-context" const directories = new Set() +const log = Log.create({ service: "instance" }) async function runtime() { return (await import("./instance-runtime")).InstanceRuntime } +async function scheduler() { + return (await import("@/automation/scheduler")).AutomationScheduler +} + +async function stopSchedulerOwnedRuns(kind: "current" | "directory" | "all", directory?: string) { + try { + const automationScheduler = await scheduler() + if (kind === "current") automationScheduler.stopCurrentOwnedRuns() + else if (kind === "directory" && directory) automationScheduler.stopDirectoryOwnedRuns(directory) + else automationScheduler.stopAllOwnedRuns() + } catch (error) { + log.error("failed to stop scheduler-owned automation runs before instance disposal", { error, kind, directory }) + } +} + export const Instance = { async provide(input: { directory: string @@ -123,6 +140,7 @@ export const Instance = { }, async dispose(input?: { mode?: "maintenance" | "force"; onCompleted?: () => void | Promise }) { const ctx = Instance.current + if ((input?.mode ?? "maintenance") === "maintenance") await stopSchedulerOwnedRuns("current") const instanceRuntime = await runtime() const onCompleted = async () => { directories.delete(ctx.directory) @@ -136,6 +154,7 @@ export const Instance = { input?: { mode?: "maintenance" | "force"; onCompleted?: () => void | Promise }, ) { const directory = Filesystem.resolve(inputDirectory) + if ((input?.mode ?? "maintenance") === "maintenance") await stopSchedulerOwnedRuns("directory", directory) const instanceRuntime = await runtime() const onCompleted = async () => { directories.delete(directory) @@ -146,6 +165,7 @@ export const Instance = { }, async disposeAll(input?: { mode?: "maintenance" | "force"; onCompleted?: () => void | Promise }) { const { disposeAllLoadedInstances } = await import("./instance-store") + if ((input?.mode ?? "maintenance") === "maintenance") await stopSchedulerOwnedRuns("all") const onCompleted = async () => { directories.clear() await input?.onCompleted?.() diff --git a/packages/opencode/src/server/instance/automation.ts b/packages/opencode/src/server/instance/automation.ts index 07a66786b..96f562854 100644 --- a/packages/opencode/src/server/instance/automation.ts +++ b/packages/opencode/src/server/instance/automation.ts @@ -4,6 +4,7 @@ import { describeRoute, resolver, validator } from "hono-openapi" import z from "zod" import { Automation, AutomationID, ValidationError } from "@/automation" import { sessionPromptExecutor } from "@/automation/runner" +import { AutomationScheduler } from "@/automation/scheduler" import { errors } from "../error" function validationError(error: ValidationError) { @@ -100,6 +101,7 @@ export const AutomationRoutes = (): Hono => validator("json", Automation.CreateInput, automationBodyValidationHook), async (c) => { try { + AutomationScheduler.current() const definition = Automation.create(c.req.valid("json")) await Automation.publishDefinitionUpdated(definition) return c.json(definition) @@ -149,6 +151,7 @@ export const AutomationRoutes = (): Hono => async (c) => { try { const automationID = c.req.valid("param").automationID + AutomationScheduler.current() const previous = Automation.get(automationID) const definition = Automation.update(automationID, c.req.valid("json")) await publishIfChanged(previous, definition) @@ -176,6 +179,7 @@ export const AutomationRoutes = (): Hono => validator("param", z.object({ automationID: AutomationID.Definition.zod })), async (c) => { const automationID = c.req.valid("param").automationID + AutomationScheduler.current() const previous = Automation.get(automationID) const definition = Automation.update(automationID, { paused: true }) await publishIfChanged(previous, definition) @@ -199,6 +203,7 @@ export const AutomationRoutes = (): Hono => validator("param", z.object({ automationID: AutomationID.Definition.zod })), async (c) => { const automationID = c.req.valid("param").automationID + AutomationScheduler.current() const previous = Automation.get(automationID) const definition = Automation.update(automationID, { paused: false }) await publishIfChanged(previous, definition) @@ -223,6 +228,7 @@ export const AutomationRoutes = (): Hono => validator("param", z.object({ automationID: AutomationID.Definition.zod })), async (c) => { const removed = Automation.remove(c.req.valid("param").automationID) + AutomationScheduler.current().cancel(removed.tombstone.id) if (removed.stoppedRun) await Automation.publishRunUpdated(removed.stoppedRun) await Automation.publishDefinitionDeleted(removed.tombstone) return c.json(removed.tombstone) diff --git a/packages/opencode/src/tool/automate.ts b/packages/opencode/src/tool/automate.ts index 5ec65afce..3d8a674a9 100644 --- a/packages/opencode/src/tool/automate.ts +++ b/packages/opencode/src/tool/automate.ts @@ -1,5 +1,6 @@ import { Effect, Schema } from "effect" import { Automation, ValidationError } from "@/automation" +import { AutomationScheduler } from "@/automation/scheduler" import * as Tool from "./tool" const Where = Schema.Struct({ @@ -89,6 +90,7 @@ export function createAutomateDefinition(): Tool.DefWithoutID= count) return items + await Bun.sleep(5) + } + throw new Error(`Timed out waiting for automation run count: ${count}`) +} + type RecurringCreateInput = Extract type OneshotCreateInput = Extract @@ -83,6 +94,39 @@ function run(overrides: Record = {}) { } describe("automation routes", () => { + test("route deletion cancels timers before publishing the tombstone", async () => { + await withAutomationApp(async ({ app, projectID }) => { + const cancelled: string[] = [] + AutomationScheduler.install({ + stop: () => undefined, + stopOwnedRuns: () => undefined, + reschedule: () => undefined, + cancel: (automationID) => cancelled.push(automationID), + computeNextFireAt: () => null, + }) + + const created = await json(app, "/automation", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify(recurringInput(projectID)), + }) + await json(app, `/automation/${created.id}`, { + method: "PUT", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ title: "Updated brief" }), + }) + await json(app, `/automation/${created.id}`, { + method: "PUT", + headers: { "content-type": "application/json" }, + body: JSON.stringify({}), + }) + await json(app, `/automation/${created.id}/pause`, { method: "POST" }) + await json(app, `/automation/${created.id}`, { method: "DELETE" }) + + expect(cancelled).toEqual([created.id]) + }) + }) + test("create echoes the resolved definition with revision and normalization warnings", async () => { await withAutomationApp(async ({ app, projectID }) => { const response = await app.request("/automation", { @@ -114,6 +158,20 @@ describe("automation routes", () => { }) }) + test("create lazily starts the scheduler before publishing definition updates", async () => { + await withAutomationApp(async ({ app, projectID }) => { + const created = await json(app, "/automation", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify(oneshotInput(projectID, { fireAt: Date.now() + 20 })), + }) + + const runs = await waitForRunCount(created.id, 1) + + expect(runs[0]?.automationID).toBe(created.id) + }) + }) + test("rejects a definition scoped to a different project", async () => { await withAutomationApp(async ({ app }) => { const response = await app.request("/automation", { diff --git a/packages/opencode/test/server/automation-scheduler.test.ts b/packages/opencode/test/server/automation-scheduler.test.ts new file mode 100644 index 000000000..3fccbccc9 --- /dev/null +++ b/packages/opencode/test/server/automation-scheduler.test.ts @@ -0,0 +1,953 @@ +import { afterEach, describe, expect, test } from "bun:test" +import { Effect } from "effect" +import { Automation } from "../../src/automation" +import { AutomationScheduler } from "../../src/automation/scheduler" +import { Instance } from "../../src/project/instance" +import { ProjectID } from "../../src/project/schema" +import { trackActiveRun } from "../../src/session/lifecycle-provenance" +import { MessageID, SessionID } from "../../src/session/schema" +import { createAutomateDefinition } from "../../src/tool/automate" +import { tmpdir } from "../fixture/fixture" + +afterEach(async () => { + await Instance.disposeAll() +}) + +class FakeClock implements AutomationScheduler.Clock { + private sleepers = new Map void }>() + private nextID = 1 + + constructor(private current: number) {} + + now() { + return this.current + } + + sleep(delayMs: number, signal: AbortSignal) { + return new Promise((resolve) => { + if (signal.aborted) { + resolve() + return + } + const id = this.nextID++ + this.sleepers.set(id, { at: this.current + Math.max(0, delayMs), resolve }) + signal.addEventListener( + "abort", + () => { + this.sleepers.delete(id) + resolve() + }, + { once: true }, + ) + }) + } + + async flush() { + await Bun.sleep(0) + } + + async advance(ms: number) { + await this.flush() + const target = this.current + ms + while (true) { + const next = [...this.sleepers.entries()] + .filter(([, sleeper]) => sleeper.at <= target) + .sort((left, right) => left[1].at - right[1].at || left[0] - right[0])[0] + if (!next) break + const [id, sleeper] = next + this.sleepers.delete(id) + this.current = sleeper.at + sleeper.resolve() + await this.flush() + } + this.current = target + } +} + +class OversleepClock implements AutomationScheduler.Clock { + private current: number + + constructor( + start: number, + private readonly oversleptAt: number, + ) { + this.current = start + } + + now() { + return this.current + } + + sleep(_delayMs: number, signal: AbortSignal) { + return new Promise((resolve) => { + if (signal.aborted) { + resolve() + return + } + queueMicrotask(() => { + this.current = this.oversleptAt + resolve() + }) + }) + } +} + +class ManualRuntime implements AutomationScheduler.TaskRuntime { + private queued: Array<{ run: (signal: AbortSignal) => Effect.Effect; controller: AbortController }> = [] + + fork(run: (signal: AbortSignal) => Effect.Effect): AutomationScheduler.Task { + const controller = new AbortController() + this.queued.push({ run, controller }) + return { + interrupt() { + controller.abort() + }, + } + } + + start(index: number) { + const queued = this.queued[index] + if (!queued) throw new Error(`Missing queued task: ${index}`) + Effect.runFork(queued.run(queued.controller.signal)) + } +} + +async function withAutomation(fn: (projectID: ProjectID) => Promise) { + await using tmp = await tmpdir({ git: true }) + return Instance.provide({ + directory: tmp.path, + fn: () => fn(Instance.project.id), + }) +} + +function oneshotInput(projectID: ProjectID, fireAt: number): Automation.CreateInput { + return { + kind: "oneshot", + title: "One-time repo brief", + prompt: "Summarize repo changes once.", + context: "fresh", + where: { projectID }, + timezone: "Asia/Shanghai", + fireAt, + } +} + +type RecurringInput = Extract + +function recurringInput(projectID: ProjectID, everyMs: number, overrides: Partial = {}): RecurringInput { + return { + kind: "recurring", + title: "Repo brief", + prompt: "Summarize repo changes.", + context: "fresh", + where: { projectID }, + timezone: "Asia/Shanghai", + rhythm: { kind: "interval", everyMs }, + stop: { kind: "never" }, + ...overrides, + } +} + +function deferred() { + let resolve!: (value: T | PromiseLike) => void + const promise = new Promise((done) => { + resolve = done + }) + return { promise, resolve } +} + +async function waitForRunStates(automationID: string, states: Automation.Run["state"][]) { + const deadline = Date.now() + 1_000 + while (Date.now() < deadline) { + const items = Automation.runs({ automationID }).items + if (items.length >= states.length && states.every((state, index) => items[index]?.state === state)) return items + await Bun.sleep(5) + } + throw new Error(`Timed out waiting for automation run states: ${states.join(", ")}`) +} + +async function waitForRunCount(automationID: string, count: number) { + const deadline = Date.now() + 1_000 + while (Date.now() < deadline) { + const items = Automation.runs({ automationID, limit: 100 }).items + if (items.length >= count) return items + await Bun.sleep(5) + } + throw new Error(`Timed out waiting for automation run count: ${count}`) +} + +describe("automation scheduler", () => { + test("fires a one-shot automation once with unattended execution", async () => { + await withAutomation(async (projectID) => { + const clock = new FakeClock(0) + const attendance: string[] = [] + const scheduler = AutomationScheduler.make({ + clock, + executor: async (input) => { + attendance.push(input.attendance) + return { sessionID: SessionID.descending(), result: "done", cost: 0 } + }, + }) + const definition = Automation.create(oneshotInput(projectID, 1_000), { now: 0 }) + + scheduler.reschedule(definition) + await clock.advance(999) + expect(Automation.runs({ automationID: definition.id }).items).toHaveLength(0) + + await clock.advance(1) + const runs = await waitForRunStates(definition.id, ["succeeded"]) + + expect(runs).toHaveLength(1) + expect(runs[0].triggeredAt).toBe(1_000) + expect(attendance).toEqual(["unattended"]) + await clock.advance(10_000) + expect(Automation.runs({ automationID: definition.id }).items).toHaveLength(1) + scheduler.stop() + }) + }) + + test("schedules existing automations when the scheduler starts", async () => { + await withAutomation(async (projectID) => { + const clock = new FakeClock(0) + const calls: number[] = [] + Automation.create(oneshotInput(projectID, 1_000), { now: 0 }) + + const scheduler = AutomationScheduler.make({ + clock, + executor: async () => { + calls.push(clock.now()) + return { sessionID: SessionID.descending(), result: "done", cost: 0 } + }, + }) + + await clock.advance(1_000) + expect(calls).toEqual([1_000]) + scheduler.stop() + }) + }) + + test("schedules recurring automations created through the automate tool", async () => { + await withAutomation(async (projectID) => { + const clock = new FakeClock(0) + const scheduler = AutomationScheduler.make({ + clock, + executor: async () => ({ sessionID: SessionID.descending(), result: "done", cost: 0 }), + }) + const tool = createAutomateDefinition() + + const result = await Effect.runPromise( + tool.execute( + recurringInput(projectID, 60_000), + { + sessionID: SessionID.descending(), + messageID: MessageID.ascending(), + agent: "build", + abort: new AbortController().signal, + messages: [], + metadata: () => Effect.void, + ask: () => Effect.void, + }, + ), + ) + + const definition = result.metadata.automationDefinition + await clock.advance(60_000) + const runs = await waitForRunStates(definition.id, ["succeeded"]) + + expect(runs).toHaveLength(1) + expect(runs[0].triggeredAt).toBe(60_000) + scheduler.stop() + }) + }) + + test("does not reschedule a one-shot automation after its fire time has run", async () => { + await withAutomation(async (projectID) => { + const clock = new FakeClock(0) + const calls: number[] = [] + const scheduler = AutomationScheduler.make({ + clock, + executor: async () => { + calls.push(clock.now()) + return { sessionID: SessionID.descending(), result: "done", cost: 0 } + }, + }) + const definition = Automation.create(oneshotInput(projectID, 1_000), { now: 0 }) + + scheduler.reschedule(definition) + await clock.advance(1_000) + await waitForRunStates(definition.id, ["succeeded"]) + scheduler.reschedule(definition) + await clock.advance(0) + + expect(calls).toEqual([1_000]) + scheduler.stop() + }) + }) + + test("does not rerun a completed one-shot automation after update or pause resume", async () => { + await withAutomation(async (projectID) => { + const clock = new FakeClock(0) + const calls: number[] = [] + const scheduler = AutomationScheduler.make({ + clock, + executor: async () => { + calls.push(clock.now()) + return { sessionID: SessionID.descending(), result: "done", cost: 0 } + }, + }) + const definition = Automation.create(oneshotInput(projectID, 1_000), { now: 0 }) + + scheduler.reschedule(definition) + await clock.advance(1_000) + await waitForRunStates(definition.id, ["succeeded"]) + + const renamed = Automation.update(definition.id, { title: "Updated one-shot" }, { now: 2_000 }) + scheduler.reschedule(renamed) + const paused = Automation.update(definition.id, { paused: true }, { now: 3_000 }) + scheduler.reschedule(paused) + const resumed = Automation.update(definition.id, { paused: false }, { now: 4_000 }) + scheduler.reschedule(resumed) + await clock.advance(0) + + expect(calls).toEqual([1_000]) + expect(Automation.runs({ automationID: definition.id }).items).toHaveLength(1) + scheduler.stop() + }) + }) + + test("ignores deleted automations when a stale timer fires", async () => { + await withAutomation(async (projectID) => { + const clock = new FakeClock(0) + const calls: number[] = [] + const scheduler = AutomationScheduler.make({ + clock, + executor: async () => { + calls.push(clock.now()) + return { sessionID: SessionID.descending(), result: "done", cost: 0 } + }, + }) + const definition = Automation.create(oneshotInput(projectID, 1_000), { now: 0 }) + + scheduler.reschedule(definition) + Automation.remove(definition.id) + await expect(clock.advance(1_000)).resolves.toBeUndefined() + + expect(calls).toEqual([]) + scheduler.stop() + }) + }) + + test("keeps recurring automation scheduled after an active manual run blocks a fire", async () => { + await withAutomation(async (projectID) => { + const clock = new FakeClock(0) + const releaseManual = deferred<{ sessionID: SessionID; result: string | null; cost?: number | null }>() + const schedulerStarts: number[] = [] + const definition = Automation.create(recurringInput(projectID, 60_000), { now: 0 }) + const scheduler = AutomationScheduler.make({ + clock, + executor: async () => { + schedulerStarts.push(clock.now()) + return { sessionID: SessionID.descending(), result: "scheduled", cost: 0 } + }, + }) + Automation.runNowExecuting(definition.id, { + now: 0, + executor: async () => releaseManual.promise, + }) + await waitForRunStates(definition.id, ["scheduled"]) + + scheduler.reschedule(definition) + await clock.advance(60_000) + await waitForRunStates(definition.id, ["stopped", "scheduled"]) + + releaseManual.resolve({ sessionID: SessionID.descending(), result: "manual", cost: 0 }) + await waitForRunStates(definition.id, ["stopped", "succeeded"]) + await clock.advance(60_000) + await waitForRunStates(definition.id, ["succeeded", "stopped", "succeeded"]) + + expect(schedulerStarts).toEqual([120_000]) + scheduler.stop() + }) + }) + + test("stops an active scheduled run when scheduler stops", async () => { + await withAutomation(async (projectID) => { + const clock = new FakeClock(0) + const sessionID = SessionID.descending() + const started = deferred() + const aborted = deferred() + let sawAbort = false + const scheduler = AutomationScheduler.make({ + clock, + executor: async ({ run, signal }) => { + const running = Automation.markRunStarted(run, sessionID, { now: clock.now() }) + await Automation.publishRunUpdated(running) + signal.addEventListener("abort", () => { + sawAbort = true + aborted.resolve() + }) + started.resolve() + await aborted.promise + return { sessionID, result: "should not complete", cost: 0 } + }, + }) + const definition = Automation.create(oneshotInput(projectID, 1_000), { now: 0 }) + + scheduler.reschedule(definition) + await clock.advance(1_000) + await started.promise + + scheduler.stop() + const runs = await waitForRunStates(definition.id, ["stopped"]) + + expect(sawAbort).toBe(true) + expect(runs[0]).toMatchObject({ + state: "stopped", + sessionID, + stopReason: "cancelled", + }) + }) + }) + + test("stops an active scheduled run when the instance is disposed", async () => { + await withAutomation(async (projectID) => { + const clock = new FakeClock(0) + const releaseRun = deferred<{ sessionID: SessionID; result: string | null; cost?: number | null }>() + let runSignal: AbortSignal | undefined + const scheduler = AutomationScheduler.make({ + clock, + executor: async (input) => { + runSignal = input.signal + return releaseRun.promise + }, + }) + AutomationScheduler.install(scheduler) + const definition = Automation.create(recurringInput(projectID, 60_000), { now: 0 }) + + scheduler.reschedule(definition) + await clock.advance(60_000) + expect(runSignal?.aborted).toBe(false) + + await Instance.dispose({ mode: "force" }) + + expect(runSignal?.aborted).toBe(true) + releaseRun.resolve({ sessionID: SessionID.descending(), result: "done", cost: 0 }) + }) + }) + + test("stops an active scheduled run before maintenance dispose waits for other active runs", async () => { + await withAutomation(async (projectID) => { + const clock = new FakeClock(0) + const releaseRun = deferred<{ sessionID: SessionID; result: string | null; cost?: number | null }>() + let runSignal: AbortSignal | undefined + const scheduler = AutomationScheduler.make({ + clock, + executor: async (input) => { + runSignal = input.signal + return releaseRun.promise + }, + }) + AutomationScheduler.install(scheduler) + const definition = Automation.create(recurringInput(projectID, 60_000), { now: 0 }) + const unrelatedActiveRun = trackActiveRun(Instance.directory) + const releaseUnrelatedRun = await unrelatedActiveRun.promise + + scheduler.reschedule(definition) + await clock.advance(60_000) + expect(runSignal?.aborted).toBe(false) + + await Instance.dispose() + + expect(runSignal?.aborted).toBe(true) + releaseRun.resolve({ sessionID: SessionID.descending(), result: "done", cost: 0 }) + + const nextDefinition = Automation.create(recurringInput(projectID, 60_000, { title: "Next recurring" }), { + now: 60_000, + }) + scheduler.reschedule(nextDefinition) + await clock.advance(60_000) + await waitForRunStates(nextDefinition.id, ["succeeded"]) + + releaseUnrelatedRun() + await Bun.sleep(0) + }) + }) + + test("continues maintenance dispose if scheduler pre-stop fails", async () => { + await withAutomation(async () => { + AutomationScheduler.install({ + stop: () => undefined, + stopOwnedRuns: () => { + throw new Error("pre-stop failed") + }, + reschedule: () => undefined, + cancel: () => undefined, + computeNextFireAt: () => null, + }) + + await expect(Instance.dispose()).resolves.toBeUndefined() + }) + }) + + test("anchors recurring schedule after a long manual run completes", async () => { + await withAutomation(async (projectID) => { + const clock = new FakeClock(0) + const releaseManual = deferred<{ sessionID: SessionID; result: string | null; cost?: number | null }>() + const schedulerStarts: number[] = [] + const definition = Automation.create(recurringInput(projectID, 60_000), { now: 0 }) + const scheduler = AutomationScheduler.make({ + clock, + executor: async () => { + schedulerStarts.push(clock.now()) + return { sessionID: SessionID.descending(), result: "scheduled", cost: 0 } + }, + }) + Automation.runNowExecuting(definition.id, { + now: 0, + executor: async () => releaseManual.promise, + }) + await waitForRunStates(definition.id, ["scheduled"]) + + scheduler.reschedule(definition) + await clock.advance(60_000) + await waitForRunStates(definition.id, ["stopped", "scheduled"]) + + await clock.advance(59_000) + releaseManual.resolve({ sessionID: SessionID.descending(), result: "manual", cost: 0 }) + await waitForRunStates(definition.id, ["stopped", "succeeded"]) + + await clock.advance(999) + expect(schedulerStarts).toEqual([]) + + await clock.advance(59_000) + expect(schedulerStarts).toEqual([]) + + await clock.advance(1) + await waitForRunStates(definition.id, ["succeeded", "stopped", "succeeded"]) + expect(schedulerStarts).toEqual([179_000]) + scheduler.stop() + }) + }) + + test("anchors interval automation after completion and never overlaps", async () => { + await withAutomation(async (projectID) => { + const clock = new FakeClock(0) + const releaseFirst = deferred() + const starts: number[] = [] + const scheduler = AutomationScheduler.make({ + clock, + executor: async () => { + starts.push(clock.now()) + if (starts.length === 1) await releaseFirst.promise + return { sessionID: SessionID.descending(), result: "done", cost: 0 } + }, + }) + const definition = Automation.create(recurringInput(projectID, 60_000), { now: 0 }) + + scheduler.reschedule(definition) + await clock.advance(60_000) + expect(starts).toEqual([60_000]) + + await clock.advance(60_000) + expect(starts).toEqual([60_000]) + + releaseFirst.resolve() + await waitForRunStates(definition.id, ["succeeded"]) + await clock.advance(59_999) + expect(starts).toEqual([60_000]) + + await clock.advance(1) + await waitForRunStates(definition.id, ["succeeded", "succeeded"]) + expect(starts).toEqual([60_000, 180_000]) + scheduler.stop() + }) + }) + + test("keeps the next recurring fire when non-schedule fields change", async () => { + await withAutomation(async (projectID) => { + const clock = new FakeClock(0) + const starts: number[] = [] + const scheduler = AutomationScheduler.make({ + clock, + executor: async () => { + starts.push(clock.now()) + return { sessionID: SessionID.descending(), result: "done", cost: 0 } + }, + }) + const definition = Automation.create(recurringInput(projectID, 60_000), { now: 0 }) + + scheduler.reschedule(definition) + await clock.advance(60_000) + await waitForRunStates(definition.id, ["succeeded"]) + + await clock.advance(59_999) + const updated = Automation.update(definition.id, { title: "Updated title" }, { now: 119_999 }) + await Automation.publishDefinitionUpdated(updated) + await clock.advance(1) + await waitForRunStates(definition.id, ["succeeded", "succeeded"]) + + expect(starts).toEqual([60_000, 120_000]) + scheduler.stop() + }) + }) + + test("does not schedule a recurring timer while its scheduled run is active", async () => { + await withAutomation(async (projectID) => { + const clock = new FakeClock(0) + const releaseFirst = deferred() + const starts: number[] = [] + const scheduler = AutomationScheduler.make({ + clock, + executor: async () => { + starts.push(clock.now()) + if (starts.length === 1) await releaseFirst.promise + return { sessionID: SessionID.descending(), result: "done", cost: 0 } + }, + }) + const definition = Automation.create(recurringInput(projectID, 30_000), { now: 0 }) + + scheduler.reschedule(definition) + await clock.advance(30_000) + expect(starts).toEqual([30_000]) + + await clock.advance(10_000) + const updated = Automation.update(definition.id, { title: "Updated title" }, { now: 40_000 }) + await Automation.publishDefinitionUpdated(updated) + + await clock.advance(30_000) + expect(Automation.runs({ automationID: definition.id }).items.map((run) => run.state)).toEqual(["scheduled"]) + + await clock.advance(20_000) + releaseFirst.resolve() + await waitForRunStates(definition.id, ["succeeded"]) + + await clock.advance(29_999) + expect(starts).toEqual([30_000]) + + await clock.advance(1) + await waitForRunStates(definition.id, ["succeeded", "succeeded"]) + expect(starts).toEqual([30_000, 120_000]) + scheduler.stop() + }) + }) + + test("stops scheduling recurring automation after count limit", async () => { + await withAutomation(async (projectID) => { + const clock = new FakeClock(0) + const starts: number[] = [] + const scheduler = AutomationScheduler.make({ + clock, + executor: async () => { + starts.push(clock.now()) + return { sessionID: SessionID.descending(), result: "done", cost: 0 } + }, + }) + const definition = Automation.create( + recurringInput(projectID, 60_000, { stop: { kind: "count", count: 3 } }), + { now: 0 }, + ) + + scheduler.reschedule(definition) + await clock.advance(60_000) + await waitForRunStates(definition.id, ["succeeded"]) + await clock.advance(60_000) + await waitForRunStates(definition.id, ["succeeded", "succeeded"]) + await clock.advance(60_000) + await waitForRunStates(definition.id, ["succeeded", "succeeded", "succeeded"]) + await clock.advance(60_000) + + expect(starts).toEqual([60_000, 120_000, 180_000]) + expect(Automation.runs({ automationID: definition.id }).items).toHaveLength(3) + scheduler.stop() + }) + }) + + test("cancels a pending recurring timer when manual completion reaches count limit", async () => { + await withAutomation(async (projectID) => { + const clock = new FakeClock(0) + const scheduledStarts: number[] = [] + const scheduler = AutomationScheduler.make({ + clock, + executor: async () => { + scheduledStarts.push(clock.now()) + return { sessionID: SessionID.descending(), result: "scheduled", cost: 0 } + }, + }) + const definition = Automation.create( + recurringInput(projectID, 60_000, { stop: { kind: "count", count: 1 } }), + { now: 0 }, + ) + + scheduler.reschedule(definition) + Automation.runNowExecuting(definition.id, { + now: 30_000, + executor: async () => ({ sessionID: SessionID.descending(), result: "manual", cost: 0 }), + }) + await waitForRunStates(definition.id, ["succeeded"]) + + await clock.advance(60_000) + + expect(scheduledStarts).toEqual([]) + expect(Automation.runs({ automationID: definition.id }).items).toHaveLength(1) + scheduler.stop() + }) + }) + + test("stops scheduling recurring automation after count limit above default page size", async () => { + await withAutomation(async (projectID) => { + const clock = new FakeClock(0) + const starts: number[] = [] + const scheduler = AutomationScheduler.make({ + clock, + executor: async () => { + starts.push(clock.now()) + return { sessionID: SessionID.descending(), result: "done", cost: 0 } + }, + }) + const definition = Automation.create( + recurringInput(projectID, 60_000, { stop: { kind: "count", count: 51 } }), + { now: 0 }, + ) + + scheduler.reschedule(definition) + for (let runCount = 1; runCount <= 51; runCount++) { + await clock.advance(60_000) + await waitForRunCount(definition.id, runCount) + } + await clock.advance(60_000) + + expect(starts).toHaveLength(51) + expect(Automation.runs({ automationID: definition.id, limit: 100 }).items).toHaveLength(51) + scheduler.stop() + }) + }) + + test("does not schedule recurring condition stops without an evaluator", async () => { + await withAutomation(async (projectID) => { + const clock = new FakeClock(0) + const starts: number[] = [] + const scheduler = AutomationScheduler.make({ + clock, + executor: async () => { + starts.push(clock.now()) + return { sessionID: SessionID.descending(), result: "done", cost: 0 } + }, + }) + const definition = Automation.create( + recurringInput(projectID, 60_000, { stop: { kind: "condition", condition: "repo is ready" } }), + { now: 0 }, + ) + + scheduler.reschedule(definition) + await clock.advance(60_000) + + expect(starts).toEqual([]) + expect(Automation.runs({ automationID: definition.id }).items).toHaveLength(0) + scheduler.stop() + }) + }) + + test("keeps recurring automation scheduled after another automation holds the project writer", async () => { + await withAutomation(async (projectID) => { + const clock = new FakeClock(0) + const releaseWriter = deferred<{ sessionID: SessionID; result: string | null; cost?: number | null }>() + const starts: number[] = [] + const blocker = Automation.create(oneshotInput(projectID, 10_000_000), { now: 0 }) + Automation.runNowExecuting(blocker.id, { + now: 0, + executor: async () => releaseWriter.promise, + }) + await waitForRunStates(blocker.id, ["scheduled"]) + const scheduler = AutomationScheduler.make({ + clock, + executor: async () => { + starts.push(clock.now()) + return { sessionID: SessionID.descending(), result: "scheduled", cost: 0 } + }, + }) + const definition = Automation.create(recurringInput(projectID, 60_000), { now: 0 }) + + scheduler.reschedule(definition) + await clock.advance(60_000) + await waitForRunStates(definition.id, ["stopped"]) + + releaseWriter.resolve({ sessionID: SessionID.descending(), result: "manual", cost: 0 }) + await waitForRunStates(blocker.id, ["succeeded"]) + await clock.advance(60_000) + await waitForRunStates(definition.id, ["succeeded", "stopped"]) + + expect(starts).toEqual([120_000]) + scheduler.stop() + }) + }) + + test("does not re-anchor recurring schedule after a manual writer conflict stop", async () => { + await withAutomation(async (projectID) => { + const clock = new FakeClock(0) + const releaseBlocker = deferred<{ sessionID: SessionID; result: string | null; cost?: number | null }>() + const scheduler = AutomationScheduler.make({ + clock, + executor: async () => ({ sessionID: SessionID.descending(), result: "scheduled", cost: 0 }), + }) + const blocker = Automation.create(oneshotInput(projectID, 10_000_000), { now: 0 }) + const definition = Automation.create(recurringInput(projectID, 60_000), { now: 0 }) + + Automation.runNowExecuting(blocker.id, { + now: 0, + executor: async () => releaseBlocker.promise, + }) + await waitForRunStates(blocker.id, ["scheduled"]) + scheduler.reschedule(definition) + + await clock.advance(30_000) + Automation.runNowExecuting(definition.id, { + now: 30_000, + executor: async () => ({ sessionID: SessionID.descending(), result: "manual", cost: 0 }), + }) + await waitForRunStates(definition.id, ["stopped"]) + + await clock.advance(30_000) + + const runs = await waitForRunStates(definition.id, ["stopped", "stopped"]) + expect(runs[0].triggeredAt).toBe(60_000) + releaseBlocker.resolve({ sessionID: SessionID.descending(), result: "blocker done", cost: 0 }) + scheduler.stop() + }) + }) + + test("records a stopped run instead of overlapping an active automation", async () => { + await withAutomation(async (projectID) => { + const clock = new FakeClock(0) + const calls: number[] = [] + const scheduler = AutomationScheduler.make({ + clock, + executor: async () => { + calls.push(clock.now()) + return { sessionID: SessionID.descending(), result: "done", cost: 0 } + }, + }) + const definition = Automation.create(oneshotInput(projectID, 1_000), { now: 0 }) + const active = Automation.runNow(definition.id, { now: 0 }) + Automation.markRunStarted(active, SessionID.descending(), { now: 0 }) + + scheduler.reschedule(definition) + await clock.advance(1_000) + const runs = await waitForRunStates(definition.id, ["stopped", "running"]) + + expect(calls).toEqual([]) + expect(runs[0]).toMatchObject({ + state: "stopped", + stopReason: "previous_run_awaiting_input", + triggeredAt: 1_000, + completedAt: 1_000, + }) + scheduler.stop() + }) + }) + + test("does not fire long one-shot timers before the target time", async () => { + await withAutomation(async (projectID) => { + const clock = new FakeClock(0) + const calls: number[] = [] + const scheduler = AutomationScheduler.make({ + clock, + executor: async () => { + calls.push(clock.now()) + return { sessionID: SessionID.descending(), result: "done", cost: 0 } + }, + }) + const fireAt = 3_000_000_000 + const definition = Automation.create(oneshotInput(projectID, fireAt), { now: 0 }) + + scheduler.reschedule(definition) + await clock.advance(2_147_483_647) + expect(calls).toEqual([]) + + await clock.advance(fireAt - 2_147_483_647) + await waitForRunStates(definition.id, ["succeeded"]) + + expect(calls).toEqual([fireAt]) + scheduler.stop() + }) + }) + + test("ignores an aborted stale recurring task after reschedule", async () => { + await withAutomation(async (projectID) => { + const clock = new FakeClock(0) + const runtime = new ManualRuntime() + const calls: number[] = [] + const scheduler = AutomationScheduler.make({ + clock, + runtime, + executor: async () => { + calls.push(clock.now()) + return { sessionID: SessionID.descending(), result: "done", cost: 0 } + }, + }) + const definition = Automation.create(recurringInput(projectID, 30_000), { now: 0 }) + + scheduler.reschedule(definition) + await clock.advance(30_000) + const updated = Automation.update(definition.id, { rhythm: { kind: "interval", everyMs: 40_000 } }, { now: 30_000 }) + scheduler.reschedule(updated) + + runtime.start(0) + await clock.flush() + expect(calls).toEqual([]) + + runtime.start(1) + await clock.advance(40_000) + await waitForRunStates(definition.id, ["succeeded"]) + expect(calls).toEqual([70_000]) + scheduler.stop() + }) + }) + + test("executes schedules after a short overslept timer", async () => { + await withAutomation(async (projectID) => { + const clock = new OversleepClock(0, 65_000) + const calls: number[] = [] + const scheduler = AutomationScheduler.make({ + clock, + executor: async () => { + calls.push(clock.now()) + return { sessionID: SessionID.descending(), result: "done", cost: 0 } + }, + }) + const definition = Automation.create(oneshotInput(projectID, 60_000), { now: 0 }) + + scheduler.reschedule(definition) + await waitForRunStates(definition.id, ["succeeded"]) + + expect(calls).toEqual([65_000]) + scheduler.stop() + }) + }) + + test("records missed schedules instead of catching up after an overslept timer", async () => { + await withAutomation(async (projectID) => { + const clock = new OversleepClock(0, 180_001) + const calls: number[] = [] + const scheduler = AutomationScheduler.make({ + clock, + executor: async () => { + calls.push(clock.now()) + return { sessionID: SessionID.descending(), result: "done", cost: 0 } + }, + }) + const definition = Automation.create(oneshotInput(projectID, 60_000), { now: 0 }) + + scheduler.reschedule(definition) + const runs = await waitForRunCount(definition.id, 1) + + expect(calls).toEqual([]) + expect(runs[0]).toMatchObject({ + state: "stopped", + stopReason: "missed_schedule", + triggeredAt: 60_000, + completedAt: 180_001, + }) + scheduler.stop() + }) + }) +})