diff --git a/packages/client/src/promise/generated/types.ts b/packages/client/src/promise/generated/types.ts index 9850eb46c8c9..09514cf4ee17 100644 --- a/packages/client/src/promise/generated/types.ts +++ b/packages/client/src/promise/generated/types.ts @@ -2186,6 +2186,7 @@ export type ConfigEntry = share?: "manual" | "auto" | "disabled" enterprise?: { url?: string } username?: string + timezone?: string permissions?: PermissionRuleset policies?: Array<{ action: "provider.use" | "tool.use" | "integration.use" @@ -2345,6 +2346,7 @@ export type ConfigEntry = todo?: boolean background_compaction?: boolean media_blob?: boolean + schedules?: boolean subagent_depth?: number location_idle_timeout?: string | false policies?: Array<{ diff --git a/packages/core/schema.json b/packages/core/schema.json index 711154c879ad..cb027faaa3ea 100644 --- a/packages/core/schema.json +++ b/packages/core/schema.json @@ -1,8 +1,8 @@ { "version": "7", "dialect": "sqlite", - "id": "44bd8ca5-689a-410e-a46a-958eb5accd97", - "prevIds": ["f1f14616-b6a1-483d-87b0-d5b7a9132580"], + "id": "23940db2-8b4e-4a5e-a404-70d785bd788c", + "prevIds": ["44bd8ca5-689a-410e-a46a-958eb5accd97"], "ddl": [ { "name": "account_state", @@ -44,6 +44,10 @@ "name": "project", "entityType": "tables" }, + { + "name": "schedule", + "entityType": "tables" + }, { "name": "blob_ref", "entityType": "tables" @@ -744,6 +748,176 @@ "entityType": "columns", "table": "project" }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "id", + "entityType": "columns", + "table": "schedule" + }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "session_id", + "entityType": "columns", + "table": "schedule" + }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "name", + "entityType": "columns", + "table": "schedule" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "action", + "entityType": "columns", + "table": "schedule" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "when", + "entityType": "columns", + "table": "schedule" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "tz", + "entityType": "columns", + "table": "schedule" + }, + { + "type": "text", + "notNull": true, + "autoincrement": false, + "default": "'active'", + "generated": null, + "name": "status", + "entityType": "columns", + "table": "schedule" + }, + { + "type": "integer", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "next_due_at", + "entityType": "columns", + "table": "schedule" + }, + { + "type": "integer", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "ends_at", + "entityType": "columns", + "table": "schedule" + }, + { + "type": "integer", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "remaining_runs", + "entityType": "columns", + "table": "schedule" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": "0", + "generated": null, + "name": "run_count", + "entityType": "columns", + "table": "schedule" + }, + { + "type": "integer", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "last_fired_at", + "entityType": "columns", + "table": "schedule" + }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "last_message_id", + "entityType": "columns", + "table": "schedule" + }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "last_outcome", + "entityType": "columns", + "table": "schedule" + }, + { + "type": "integer", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "retry_at", + "entityType": "columns", + "table": "schedule" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "time_created", + "entityType": "columns", + "table": "schedule" + }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "time_updated", + "entityType": "columns", + "table": "schedule" + }, { "type": "text", "notNull": true, @@ -1618,6 +1792,17 @@ "entityType": "fks", "table": "project_directory" }, + { + "columns": ["session_id"], + "tableTo": "session_v2", + "columnsTo": ["id"], + "onUpdate": "NO ACTION", + "onDelete": "CASCADE", + "nameExplicit": false, + "name": "fk_schedule_session_id_session_v2_id_fk", + "entityType": "fks", + "table": "schedule" + }, { "columns": ["session_id"], "tableTo": "session_v2", @@ -1797,6 +1982,13 @@ "table": "project", "entityType": "pks" }, + { + "columns": ["id"], + "nameExplicit": false, + "name": "schedule_pk", + "table": "schedule", + "entityType": "pks" + }, { "columns": ["hash"], "nameExplicit": false, @@ -1908,6 +2100,34 @@ "entityType": "indexes", "table": "permission" }, + { + "columns": [ + { + "value": "session_id", + "isExpression": false + } + ], + "isUnique": false, + "where": null, + "origin": "manual", + "name": "schedule_session_idx", + "entityType": "indexes", + "table": "schedule" + }, + { + "columns": [ + { + "value": "next_due_at", + "isExpression": false + } + ], + "isUnique": false, + "where": "\"schedule\".\"status\" = 'active' and \"schedule\".\"next_due_at\" is not null", + "origin": "manual", + "name": "schedule_due_idx", + "entityType": "indexes", + "table": "schedule" + }, { "columns": [ { diff --git a/packages/core/src/config/normalize.ts b/packages/core/src/config/normalize.ts index 9d693086e12d..6e0e1dd588bb 100644 --- a/packages/core/src/config/normalize.ts +++ b/packages/core/src/config/normalize.ts @@ -210,6 +210,7 @@ export function normalize(input: unknown): Result { share: Info.fields.share, enterprise: Info.fields.enterprise, username: Info.fields.username, + timezone: Info.fields.timezone, snapshots: Info.fields.snapshots, media: Info.fields.media, tool_output: Info.fields.tool_output, @@ -494,6 +495,15 @@ function normalizeExperimental( ) if (value !== undefined) result.media_blob = value } + if (own(experimental, "schedules")) { + const value = decodeEncoded( + ConfigExperimental.Info.fields.schedules, + experimental.schedules, + ["experimental", "schedules"], + diagnostics, + ) + if (value !== undefined) result.schedules = value + } if (own(experimental, "subagent_depth")) { const value = decodeEncoded( ConfigExperimental.Info.fields.subagent_depth, diff --git a/packages/core/src/database/migration.gen.ts b/packages/core/src/database/migration.gen.ts index 712c88bad4a1..575a1bd64a58 100644 --- a/packages/core/src/database/migration.gen.ts +++ b/packages/core/src/database/migration.gen.ts @@ -53,6 +53,7 @@ import m50 from "./migration/20261024002554_session_allowance.js" import m51 from "./migration/20261024175301_drop_session_allowance.js" import m52 from "./migration/20261025123711_blob.js" import m53 from "./migration/20261025154125_drop_session_pending.js" +import m54 from "./migration/20261026223005_schedule.js" export const migrations = [ m00, @@ -109,4 +110,5 @@ export const migrations = [ m51, m52, m53, + m54, ] satisfies DatabaseMigration.Migration[] diff --git a/packages/core/src/database/migration/20261026223005_schedule.ts b/packages/core/src/database/migration/20261026223005_schedule.ts new file mode 100644 index 000000000000..18069a50754d --- /dev/null +++ b/packages/core/src/database/migration/20261026223005_schedule.ts @@ -0,0 +1,38 @@ +import { Effect } from "effect" +import type { DatabaseMigration } from "../migration.js" + +const migration: DatabaseMigration.Migration = { + id: "20261026223005_schedule", + up(tx) { + return Effect.gen(function* () { + yield* tx.run(` + CREATE TABLE \`schedule\` ( + \`id\` text PRIMARY KEY, + \`session_id\` text, + \`name\` text, + \`action\` text NOT NULL, + \`when\` text NOT NULL, + \`tz\` text NOT NULL, + \`status\` text DEFAULT 'active' NOT NULL, + \`next_due_at\` integer, + \`ends_at\` integer, + \`remaining_runs\` integer, + \`run_count\` integer DEFAULT 0 NOT NULL, + \`last_fired_at\` integer, + \`last_message_id\` text, + \`last_outcome\` text, + \`retry_at\` integer, + \`time_created\` integer NOT NULL, + \`time_updated\` integer NOT NULL, + CONSTRAINT \`fk_schedule_session_id_session_v2_id_fk\` FOREIGN KEY (\`session_id\`) REFERENCES \`session_v2\`(\`id\`) ON DELETE CASCADE + ); + `) + yield* tx.run(`CREATE INDEX \`schedule_session_idx\` ON \`schedule\` (\`session_id\`);`) + yield* tx.run( + `CREATE INDEX \`schedule_due_idx\` ON \`schedule\` (\`next_due_at\`) WHERE "schedule"."status" = 'active' and "schedule"."next_due_at" is not null;`, + ) + }) + }, +} + +export default migration diff --git a/packages/core/src/database/schema.gen.ts b/packages/core/src/database/schema.gen.ts index 9de9d86990c3..0e09e8306641 100644 --- a/packages/core/src/database/schema.gen.ts +++ b/packages/core/src/database/schema.gen.ts @@ -116,6 +116,28 @@ const schema: Omit = { \`commands\` text ); `) + yield* tx.run(` + CREATE TABLE \`schedule\` ( + \`id\` text PRIMARY KEY, + \`session_id\` text, + \`name\` text, + \`action\` text NOT NULL, + \`when\` text NOT NULL, + \`tz\` text NOT NULL, + \`status\` text DEFAULT 'active' NOT NULL, + \`next_due_at\` integer, + \`ends_at\` integer, + \`remaining_runs\` integer, + \`run_count\` integer DEFAULT 0 NOT NULL, + \`last_fired_at\` integer, + \`last_message_id\` text, + \`last_outcome\` text, + \`retry_at\` integer, + \`time_created\` integer NOT NULL, + \`time_updated\` integer NOT NULL, + CONSTRAINT \`fk_schedule_session_id_session_v2_id_fk\` FOREIGN KEY (\`session_id\`) REFERENCES \`session_v2\`(\`id\`) ON DELETE CASCADE + ); + `) yield* tx.run(` CREATE TABLE \`blob_ref\` ( \`session_id\` text NOT NULL, @@ -255,6 +277,10 @@ const schema: Omit = { yield* tx.run( `CREATE UNIQUE INDEX \`permission_project_action_resource_idx\` ON \`permission\` (\`project_id\`,\`action\`,\`resource\`);`, ) + yield* tx.run(`CREATE INDEX \`schedule_session_idx\` ON \`schedule\` (\`session_id\`);`) + yield* tx.run( + `CREATE INDEX \`schedule_due_idx\` ON \`schedule\` (\`next_due_at\`) WHERE "schedule"."status" = 'active' and "schedule"."next_due_at" is not null;`, + ) yield* tx.run(`CREATE INDEX \`blob_ref_hash_idx\` ON \`blob_ref\` (\`hash\`);`) yield* tx.run(`CREATE INDEX \`blob_ref_session_seq_idx\` ON \`blob_ref\` (\`session_id\`,\`seq\`);`) yield* tx.run( diff --git a/packages/core/src/schedule.ts b/packages/core/src/schedule.ts new file mode 100644 index 000000000000..1c5a8aae48f2 --- /dev/null +++ b/packages/core/src/schedule.ts @@ -0,0 +1,470 @@ +export * as Schedule from "./schedule.js" + +import { and, asc, eq, gt, isNull, lte, min, or } from "drizzle-orm" +import { Cause, Clock, Context, Cron, DateTime, Effect, Latch, Layer, Option, Result, Schema } from "effect" +import { makeGlobalNode } from "@opencode/util/effect/app-node" +import { Schedule as ScheduleSchema } from "@opencode/schema/schedule" +import { Config } from "./config.js" +import { Database } from "./database/database.js" +import { Instance } from "./instance/service.js" +import { Session } from "./session.js" +import { SessionExecution } from "./session/execution.js" +import { SessionMessage } from "./session/message.js" +import { SessionSchema } from "./session/schema.js" +import { SessionInboxTable } from "./session/sql.js" +import { ScheduleTable } from "./schedule/sql.js" + +export const ID = ScheduleSchema.ID +export type ID = ScheduleSchema.ID +export const Info = ScheduleSchema.Info +export type Info = ScheduleSchema.Info +export const CreateInput = ScheduleSchema.CreateInput +export type CreateInput = ScheduleSchema.CreateInput + +export class NotFoundError extends Schema.TaggedError()("Schedule.NotFoundError", { + scheduleID: ID, +}) {} + +export class InvalidError extends Schema.TaggedError()("Schedule.InvalidError", { + reason: Schema.String, +}) { + override get message() { + return this.reason + } +} + +export class DisabledError extends Schema.TaggedError()("Schedule.DisabledError", {}) { + override get message() { + return "Scheduled tasks are disabled; set experimental.schedules in config" + } +} + +export interface Interface { + readonly create: (input: CreateInput) => Effect.Effect + readonly list: (input?: { readonly sessionID?: SessionSchema.ID }) => Effect.Effect + readonly get: (id: ID) => Effect.Effect + readonly pause: (id: ID) => Effect.Effect + readonly resume: (id: ID) => Effect.Effect + readonly remove: (id: ID) => Effect.Effect + /** Fires every schedule that is due now, at most once each. */ + readonly tick: Effect.Effect + /** + * Runs `tick` whenever a schedule falls due, until interrupted. All state lives in the table, so exactly one + * process may run this: the managed server. + */ + readonly run: Effect.Effect +} + +export class Service extends Context.Service()("@opencode/Schedule") {} + +type Row = typeof ScheduleTable.$inferSelect + +const MINUTE = 60_000 +const HOUR = 60 * MINUTE +const MIN_INTERVAL = MINUTE +/** A firing this far past its due time is marked late. */ +const LATE_AFTER = 60_000 +/** The loop re-reads the table at least this often, so clock jumps and missed doorbells are caught. */ +const MAX_SLEEP = 30_000 +/** How long a due schedule waits before its location's flag is read again, while schedules are off there. */ +const DISABLED_RETRY = HOUR +/** How long a schedule whose prompt failed waits before trying again. */ +const FAILED_RETRY = 5 * MINUTE + +const duration = (value: string) => + Option.match(ScheduleSchema.durationMillis(value), { + onNone: () => Effect.fail(new InvalidError({ reason: `Invalid duration: ${value}` })), + onSome: (millis) => + millis > 0 + ? Effect.succeed(millis) + : Effect.fail(new InvalidError({ reason: `Duration must be positive: ${value}` })), + }) + +/** Stored durations were validated on create. */ +const millis = (value: string) => + Option.getOrThrowWith(ScheduleSchema.durationMillis(value), () => new Error(`Invalid duration: ${value}`)) + +const OFFSET = /\d(?:z|[+-]\d{2}:?\d{2})$/i + +/** Parses an instant with an offset, or a wall-clock time in `zone`. */ +const parseAt = (input: string, zone: DateTime.TimeZone) => + Option.map( + OFFSET.test(input) ? DateTime.make(input) : DateTime.makeZoned(input, { timeZone: zone, adjustForTimeZone: true }), + DateTime.toEpochMillis, + ) + +const cronNext = Option.liftThrowable(Cron.next) + +/** + * The first slot strictly after `now`, or undefined if the expression never matches a date. When clocks fall back, + * Cron.next maps a repeated wall-clock time to its first occurrence, which is already past during the repeated hour, + * so no slot fires twice: step past each such slot until one lands after `now`. + */ +const nextCron = (cron: Cron.Cron, now: number, from = now): number | undefined => { + const next = cronNext(cron, new Date(from)) + if (Option.isNone(next)) return undefined + const at = next.value.getTime() + return at > now ? at : nextCron(cron, now, at + HOUR) +} + +/** The first slot strictly after `now`, or undefined for one-shots. */ +const following = (row: Pick, now: number) => { + switch (row.when.type) { + case "in": + case "at": + return undefined + case "every": { + // Keep the phase: skip every missed slot rather than firing them in a burst. + const every = millis(row.when.every) + const due = row.next_due_at ?? now + return due + (Math.floor((now - due) / every) + 1) * every + } + case "cron": + return nextCron(Cron.parseUnsafe(row.when.cron, row.tz), now) + } +} + +const finished = (row: Pick, next: number | undefined, remaining: number | null) => + next === undefined || (remaining !== null && remaining <= 0) || (row.ends_at !== null && next > row.ends_at) + +const optionalTime = (value: number | null) => (value === null ? undefined : DateTime.makeUnsafe(value)) + +const toInfo = (row: Row): Info => ({ + id: row.id, + sessionID: row.session_id ?? undefined, + name: row.name ?? undefined, + when: row.when, + tz: row.tz, + action: row.action, + status: row.status, + nextDueAt: optionalTime(row.next_due_at), + endsAt: optionalTime(row.ends_at), + remainingRuns: row.remaining_runs ?? undefined, + runCount: row.run_count, + lastFiredAt: optionalTime(row.last_fired_at), + lastOutcome: row.last_outcome ?? undefined, + time: { created: DateTime.makeUnsafe(row.time_created), updated: DateTime.makeUnsafe(row.time_updated) }, +}) + +const layer = Layer.effect( + Service, + Effect.gen(function* () { + const { db } = yield* Database.Service + const sessions = yield* Session.Service + const instances = yield* Instance.Service + const execution = yield* SessionExecution.Service + const doorbell = yield* Latch.make() + + // Config is per location, so a schedule follows the config where its Session lives. + const settings = (session: SessionSchema.Info) => + Effect.gen(function* () { + const config = yield* Effect.serviceOption(Config.Service) + if (Option.isNone(config)) return { enabled: false, timezone: undefined } + const entries = yield* config.value.entries() + return { + enabled: Config.latest(entries, "experimental")?.schedules === true, + timezone: Config.latest(entries, "timezone"), + } + }).pipe(instances.provide(session)) + + const find = (id: ID) => + db + .select() + .from(ScheduleTable) + .where(eq(ScheduleTable.id, id)) + .get() + .pipe( + Effect.orDie, + Effect.flatMap((row) => (row ? Effect.succeed(row) : Effect.fail(new NotFoundError({ scheduleID: id })))), + ) + + const update = (row: Row, values: Partial) => + db + .update(ScheduleTable) + .set(values) + .where(and(eq(ScheduleTable.id, row.id), eq(ScheduleTable.status, "active"))) + .run() + .pipe(Effect.orDie) + + const pending = (id: SessionMessage.ID) => + db + .select({ id: SessionInboxTable.id }) + .from(SessionInboxTable) + .where(eq(SessionInboxTable.id, id)) + .get() + .pipe( + Effect.orDie, + Effect.map((row) => row !== undefined), + ) + + const fire = Effect.fnUntraced(function* (row: Row, sessionID: SessionSchema.ID, due: number, now: number) { + const run = row.run_count + 1 + // The id is derived from the run, so retrying after a crash reconciles with the earlier admission. + const id = SessionMessage.ID.make(`msg_${row.id}_${run}`) + const late = now - due >= LATE_AFTER + yield* sessions.prompt({ + sessionID, + id, + text: row.action.prompt, + delivery: "steer", + metadata: { schedule: { id: row.id, name: row.name ?? undefined, run, dueAt: due, late } }, + }) + const next = following(row, now) + const remaining = row.remaining_runs === null ? null : row.remaining_runs - 1 + const done = finished(row, next, remaining) + // Record the run even if a pause landed during the prompt, or the next firing would reuse this message id. + yield* db + .update(ScheduleTable) + .set({ + run_count: run, + last_message_id: id, + last_fired_at: now, + last_outcome: late ? "late" : "fired", + retry_at: null, + remaining_runs: remaining, + ...(done ? { status: "done" as const } : {}), + next_due_at: done ? null : next, + time_updated: now, + }) + .where(eq(ScheduleTable.id, row.id)) + .run() + .pipe(Effect.orDie) + }) + + const step = Effect.fnUntraced(function* (row: Row, now: number) { + const due = row.next_due_at + if (due === null || row.session_id === null) return + const session = yield* sessions.get(row.session_id) + if (!(yield* settings(session)).enabled) { + // Reading the flag boots the Session's location, so check again in an hour rather than every tick. The row + // keeps its due time, so once the flag is back on it catches up like any missed slot. + yield* update(row, { last_outcome: "disabled", retry_at: now + DISABLED_RETRY, time_updated: now }) + return + } + if (row.last_message_id !== null && (yield* pending(row.last_message_id))) { + // The wake that followed admission may have been lost; waking again is harmless. + yield* execution.wake(session.id) + const next = following(row, now) + const done = finished(row, next, row.remaining_runs) + yield* update(row, { + last_outcome: "skipped", + retry_at: null, + status: done ? "done" : "active", + next_due_at: done ? null : next, + time_updated: now, + }) + return + } + yield* fire(row, session.id, due, now) + }) + + const tickAt = Effect.fnUntraced(function* (now: number) { + const due = yield* db + .select() + .from(ScheduleTable) + .where( + and( + eq(ScheduleTable.status, "active"), + lte(ScheduleTable.next_due_at, now), + or(isNull(ScheduleTable.retry_at), lte(ScheduleTable.retry_at, now)), + ), + ) + .orderBy(asc(ScheduleTable.next_due_at)) + .all() + .pipe(Effect.orDie) + yield* Effect.forEach( + due, + (row) => + step(row, now).pipe( + Effect.catchCause((cause) => + Cause.hasInterruptsOnly(cause) + ? Effect.interrupt + : // Back off rather than retrying a failing prompt on every tick. + Effect.logWarning("schedule failed to fire", { scheduleID: row.id, cause }).pipe( + Effect.andThen( + update(row, { last_outcome: "failed", retry_at: now + FAILED_RETRY, time_updated: now }), + ), + ), + ), + ), + { discard: true }, + ) + }, Effect.withSpan("Schedule.tick")) + + const tick = Effect.flatMap(Clock.currentTimeMillis, tickAt) + + // Rows due by `from` that could not fire (a failed prompt, a retry_at not yet reached) are rechecked after at most + // MAX_SLEEP, not in a spin. + const nextWake = Effect.fnUntraced(function* (from: number) { + const row = yield* db + .select({ at: min(ScheduleTable.next_due_at) }) + .from(ScheduleTable) + .where(and(eq(ScheduleTable.status, "active"), gt(ScheduleTable.next_due_at, from))) + .get() + .pipe(Effect.orDie) + if (row?.at == null) return MAX_SLEEP + return Math.max(0, Math.min(MAX_SLEEP, row.at - (yield* Clock.currentTimeMillis))) + }) + + // A crash between admitting a prompt and its Session claiming the work leaves the prompt unclaimed and the + // Session asleep, so wake every Session whose last scheduled prompt is still pending. + const rewake = db + .selectDistinct({ sessionID: ScheduleTable.session_id }) + .from(ScheduleTable) + .innerJoin(SessionInboxTable, eq(SessionInboxTable.id, ScheduleTable.last_message_id)) + .all() + .pipe( + Effect.orDie, + Effect.flatMap((rows) => + Effect.forEach(rows, (row) => (row.sessionID === null ? Effect.void : execution.wake(row.sessionID)), { + discard: true, + }), + ), + ) + + const loop = Effect.gen(function* () { + yield* doorbell.close + const now = yield* Clock.currentTimeMillis + yield* tickAt(now).pipe(Effect.catchCause((cause) => Effect.logError("schedule tick failed", { cause }))) + const wait = yield* nextWake(now) + if (wait > 0) yield* doorbell.await.pipe(Effect.timeoutOption(wait)) + }) + + const run = rewake.pipe( + Effect.catchCause((cause) => Effect.logError("schedule rewake failed", { cause })), + Effect.andThen(Effect.forever(loop)), + ) + + const resolveZone = (tz: string | undefined) => { + if (tz === undefined) return Effect.succeed(DateTime.zoneMakeLocal()) + return Option.match(DateTime.zoneMakeNamed(tz), { + onNone: () => Effect.fail(new InvalidError({ reason: `Unknown time zone: ${tz}` })), + onSome: Effect.succeed, + }) + } + + const firstDue = Effect.fnUntraced(function* (when: ScheduleSchema.When, zone: DateTime.TimeZone, now: number) { + switch (when.type) { + case "in": + return now + (yield* duration(when.in)) + case "at": { + const at = parseAt(when.at, zone) + if (Option.isNone(at)) return yield* new InvalidError({ reason: `Invalid time: ${when.at}` }) + if (at.value <= now) return yield* new InvalidError({ reason: `Time is in the past: ${when.at}` }) + return at.value + } + case "every": { + const every = yield* duration(when.every) + if (every < MIN_INTERVAL) return yield* new InvalidError({ reason: "every must be at least 1 minute" }) + return now + every + } + case "cron": { + const cron = Cron.parse(when.cron, zone) + if (Result.isFailure(cron)) return yield* new InvalidError({ reason: cron.failure.message }) + if (cron.success.seconds.size !== 1) + return yield* new InvalidError({ reason: "cron must fire at most once a minute" }) + const next = nextCron(cron.success, now) + if (next === undefined) return yield* new InvalidError({ reason: `cron never fires: ${when.cron}` }) + return next + } + } + }) + + const endsAt = Effect.fnUntraced(function* ( + ends: ScheduleSchema.Ends | undefined, + zone: DateTime.TimeZone, + now: number, + ) { + const after = ends?.after === undefined ? undefined : now + (yield* duration(ends.after)) + if (ends?.at === undefined) return after ?? null + const at = parseAt(ends.at, zone) + if (Option.isNone(at)) return yield* new InvalidError({ reason: `Invalid time: ${ends.at}` }) + return Math.min(at.value, after ?? at.value) + }) + + return Service.of({ + create: Effect.fn("Schedule.create")(function* (input) { + const session = yield* sessions.get(input.sessionID) + const config = yield* settings(session).pipe(Effect.orDie) + if (!config.enabled) return yield* new DisabledError() + const zone = yield* resolveZone(input.when.tz ?? config.timezone) + const now = yield* Clock.currentTimeMillis + const due = yield* firstDue(input.when, zone, now) + if (input.ends?.times !== undefined && input.ends.times < 1) + return yield* new InvalidError({ reason: "ends.times must be at least 1" }) + const ends = yield* endsAt(input.ends, zone, now) + if (ends !== null && due > ends) + return yield* new InvalidError({ reason: "The schedule ends before it first fires" }) + const row = yield* db + .insert(ScheduleTable) + .values({ + id: ID.create(), + session_id: session.id, + name: input.name ?? null, + action: input.action, + when: input.when, + tz: DateTime.zoneToString(zone), + next_due_at: due, + ends_at: ends, + remaining_runs: input.ends?.times ?? null, + time_created: now, + time_updated: now, + }) + .returning() + .get() + .pipe(Effect.orDie) + yield* doorbell.open + return toInfo(row) + }), + list: Effect.fn("Schedule.list")(function* (input) { + const rows = yield* db + .select() + .from(ScheduleTable) + .where(input?.sessionID === undefined ? undefined : eq(ScheduleTable.session_id, input.sessionID)) + .orderBy(asc(ScheduleTable.time_created), asc(ScheduleTable.id)) + .all() + .pipe(Effect.orDie) + return rows.map(toInfo) + }), + get: Effect.fn("Schedule.get")(function* (id) { + return toInfo(yield* find(id)) + }), + pause: Effect.fn("Schedule.pause")(function* (id) { + const row = yield* find(id) + if (row.status !== "active") return toInfo(row) + const now = yield* Clock.currentTimeMillis + yield* update(row, { status: "paused", time_updated: now }) + return toInfo(yield* find(id)) + }), + resume: Effect.fn("Schedule.resume")(function* (id) { + const row = yield* find(id) + if (row.status !== "paused") return toInfo(row) + const now = yield* Clock.currentTimeMillis + // A one-shot that fell due while paused fires once, late; a repeating schedule resumes at its next slot. + const next = + row.next_due_at === null || row.next_due_at > now ? row.next_due_at : (following(row, now) ?? row.next_due_at) + const done = next === null || finished(row, next, row.remaining_runs) + yield* db + .update(ScheduleTable) + .set({ status: done ? "done" : "active", next_due_at: done ? null : next, retry_at: null, time_updated: now }) + .where(and(eq(ScheduleTable.id, id), eq(ScheduleTable.status, "paused"))) + .run() + .pipe(Effect.orDie) + yield* doorbell.open + return toInfo(yield* find(id)) + }), + remove: Effect.fn("Schedule.remove")(function* (id) { + yield* find(id) + yield* db.delete(ScheduleTable).where(eq(ScheduleTable.id, id)).run().pipe(Effect.orDie) + }), + tick, + run, + }) + }), +) + +export const node = makeGlobalNode({ + service: Service, + layer, + deps: [Database.node, Session.node, SessionExecution.node, Instance.node], +}) diff --git a/packages/core/src/schedule/sql.ts b/packages/core/src/schedule/sql.ts new file mode 100644 index 000000000000..bd24fbb968fd --- /dev/null +++ b/packages/core/src/schedule/sql.ts @@ -0,0 +1,39 @@ +import { sql } from "drizzle-orm" +import { index, integer, sqliteTable, text } from "drizzle-orm/sqlite-core" +import type { Schedule } from "@opencode/schema/schedule" +import { Timestamps } from "../database/schema.sql.js" +import type { SessionMessage } from "../session/message.js" +import type { SessionSchema } from "../session/schema.js" +import { SessionTable } from "../session/sql.js" + +export const ScheduleTable = sqliteTable( + "schedule", + { + id: text().$type().primaryKey(), + /** The Session a prompt action wakes. */ + session_id: text() + .$type() + .references(() => SessionTable.id, { onDelete: "cascade" }), + name: text(), + action: text({ mode: "json" }).$type().notNull(), + when: text({ mode: "json" }).$type().notNull(), + tz: text().notNull(), + status: text().$type().notNull().default("active"), + next_due_at: integer(), + ends_at: integer(), + remaining_runs: integer(), + run_count: integer().notNull().default(0), + last_fired_at: integer(), + last_message_id: text().$type(), + last_outcome: text().$type(), + /** The loop leaves a due row alone until then, e.g. while its location has schedules turned off. */ + retry_at: integer(), + ...Timestamps, + }, + (table) => [ + index("schedule_session_idx").on(table.session_id), + index("schedule_due_idx") + .on(table.next_due_at) + .where(sql`${table.status} = 'active' and ${table.next_due_at} is not null`), + ], +) diff --git a/packages/core/test/schedule.test.ts b/packages/core/test/schedule.test.ts new file mode 100644 index 000000000000..09515e859b94 --- /dev/null +++ b/packages/core/test/schedule.test.ts @@ -0,0 +1,693 @@ +import { describe, expect } from "bun:test" +import type { FileSystem } from "@opencode/core/filesystem" +import { DateTime, Effect, Fiber, Layer, LayerMap, Stream } from "effect" +import { TestClock } from "effect/testing" +import { eq } from "drizzle-orm" +import { Database } from "@opencode/core/database/database" +import { AppNodeBuilder } from "@opencode/core/effect/app-node-builder" +import { LayerNode } from "@opencode/util/effect/layer-node" +import { makeGlobalNode } from "@opencode/util/effect/app-node" +import { Bus } from "@opencode/core/bus" +import { Location } from "@opencode/schema/location" +import { Project } from "@opencode/core/project" +import { ProjectTable } from "@opencode/core/project/sql" +import { AbsolutePath } from "@opencode/core/schema" +import { Schedule } from "@opencode/core/schedule" +import { ScheduleTable } from "@opencode/core/schedule/sql" +import { Session } from "@opencode/core/session" +import { SessionMessage } from "@opencode/core/session/message" +import { SessionProjector } from "@opencode/core/session/projector" +import { SessionExecution } from "@opencode/core/session/execution" +import { SessionInbox } from "@opencode/core/session/inbox" +import { SessionInboxTable, SessionTable } from "@opencode/core/session/sql" +import { SessionStore } from "@opencode/core/session/store" +import { LocationServiceMap } from "@opencode/core/location-service-map" +import type { LocationServices } from "@opencode/core/location-services" +import { Plugin } from "@opencode/core/plugin" +import { PluginHooks } from "@opencode/core/plugin/hooks" +import { Snapshot } from "@opencode/core/snapshot" +import { Skill } from "@opencode/core/skill" +import { Config } from "@opencode/core/config" +import { Document, Info } from "@opencode/schema/config" +import { ConfigExperimental } from "@opencode/schema/config/experimental" +import { testEffect } from "./lib/effect" + +const enabledDirectory = "/enabled" +const disabledDirectory = "/disabled" +const toggleDirectory = "/toggle" + +const wakes: Session.ID[] = [] +/** Runs inside the next wake, which happens while a schedule is firing. */ +let duringWake: Effect.Effect = Effect.void +const execution = Layer.succeed( + SessionExecution.Service, + SessionExecution.Service.of({ + active: Effect.sync(() => new Set()), + isActive: () => Effect.succeed(false), + resume: () => Effect.void, + interrupt: () => Effect.succeed(false), + wake: (sessionID) => + Effect.suspend(() => { + wakes.push(sessionID) + const effect = duringWake + duringWake = Effect.void + return effect + }), + awaitIdle: () => Effect.void, + }), +) +const enabledEntries = [ + new Document({ + type: "document", + info: new Info({ + timezone: "America/New_York", + experimental: new ConfigExperimental.Info({ schedules: true }), + }), + }), +] +const enabled = Config.testLayer(enabledEntries) +/** A location whose flag the test turns on, counting every config read. */ +const toggle = { on: false, reads: 0 } +const toggled = Layer.succeed( + Config.Service, + Config.Service.of({ + entries: () => + Effect.sync(() => { + toggle.reads++ + return toggle.on ? enabledEntries : [] + }), + changes: () => Stream.empty, + }), +) +const locations = makeGlobalNode({ + service: LocationServiceMap.Service, + layer: Layer.effect( + LocationServiceMap.Service, + Effect.gen(function* () { + const bus = yield* Bus.Service + return yield* LayerMap.make( + (ref: Location.Ref) => + // oxlint-disable-next-line typescript-eslint/no-unsafe-type-assertion + Layer.mergeAll( + ref.directory === enabledDirectory ? enabled : ref.directory === toggleDirectory ? toggled : Layer.empty, + LayerNode.compile(LayerNode.group([PluginHooks.node, Skill.node]), { + replacements: [Bus.node.replace(Layer.succeed(Bus.Service, bus))], + }), + Layer.mock(Snapshot.Service, { capture: () => Effect.undefined, restore: () => Effect.void }), + Layer.mock(Plugin.Service, { awaitActivation: Effect.void }), + ).pipe(Layer.fresh) as unknown as Layer.Layer, + ) + }), + ), + deps: [Bus.node], +}) +const it = testEffect( + AppNodeBuilder.build( + LayerNode.group([Database.node, Bus.node, SessionProjector.node, SessionStore.node, Session.node, Schedule.node]), + [ + Bus.node.replace(Bus.configured({ persist: true })), + SessionExecution.node.replace(execution), + LocationServiceMap.node.replace(locations), + ], + ), +) + +const sessionID = Session.ID.make("ses_schedule_test") + +const setup = (directory = enabledDirectory) => + Effect.gen(function* () { + const { db } = yield* Database.Service + yield* db + .insert(ProjectTable) + .values({ id: Project.ID.global, worktree: AbsolutePath.make("/"), sandboxes: [] }) + .onConflictDoNothing() + .run() + .pipe(Effect.orDie) + yield* db + .insert(SessionTable) + .values({ id: sessionID, project_id: Project.ID.global, slug: "test", directory, title: "test", version: "test" }) + .run() + .pipe(Effect.orDie) + wakes.length = 0 + duringWake = Effect.void + toggle.on = false + toggle.reads = 0 + return yield* Schedule.Service + }) + +const setTime = (iso: string) => TestClock.setTime(Date.parse(iso)) +const pending = Database.Service.use(({ db }) => + db + .select({ id: SessionInboxTable.id }) + .from(SessionInboxTable) + .all() + .pipe( + Effect.orDie, + Effect.map((rows) => rows.map((row) => row.id)), + ), +) +/** Moves pending prompts into the transcript, as a running Session would. */ +const deliver = Effect.gen(function* () { + const { db } = yield* Database.Service + yield* SessionInbox.promote(db, yield* Bus.Service, sessionID, "steer") +}) +const messageID = (schedule: Schedule.Info, run: number) => SessionMessage.ID.make(`msg_${schedule.id}_${run}`) +const iso = (time: DateTime.Utc | undefined) => time && DateTime.formatIso(time) + +describe("Schedule", () => { + it.effect("fires an in schedule once", () => + Effect.gen(function* () { + const schedules = yield* setup() + const schedule = yield* schedules.create({ + sessionID, + when: { type: "in", in: "5 minutes" }, + action: { type: "prompt", prompt: "check the build" }, + }) + expect(schedule.tz).toBe("America/New_York") + + yield* TestClock.adjust("4 minutes") + yield* schedules.tick + expect(yield* pending).toEqual([]) + + yield* TestClock.adjust("1 minute") + yield* schedules.tick + expect(yield* pending).toEqual([messageID(schedule, 1)]) + expect(wakes).toEqual([sessionID]) + const { db } = yield* Database.Service + expect(yield* SessionInbox.find(db, messageID(schedule, 1))).toMatchObject({ + type: "user", + delivery: "steer", + payload: { text: "check the build" }, + }) + expect(yield* schedules.get(schedule.id)).toMatchObject({ + status: "done", + runCount: 1, + lastOutcome: "fired", + nextDueAt: undefined, + }) + + yield* TestClock.adjust("10 minutes") + yield* schedules.tick + expect(yield* pending).toEqual([messageID(schedule, 1)]) + }), + ) + + it.effect("fires an at schedule at the wall-clock time in its zone", () => + Effect.gen(function* () { + yield* setTime("2026-11-01T12:00:00Z") + const schedules = yield* setup() + const schedule = yield* schedules.create({ + sessionID, + when: { type: "at", at: "2026-11-02T09:30", tz: "Europe/Paris" }, + action: { type: "prompt", prompt: "standup" }, + }) + expect(iso(schedule.nextDueAt)).toBe("2026-11-02T08:30:00.000Z") + expect( + yield* Effect.flip( + schedules.create({ + sessionID, + when: { type: "at", at: "2026-11-01T11:00:00Z" }, + action: { type: "prompt", prompt: "too late" }, + }), + ), + ).toMatchObject({ _tag: "Schedule.InvalidError" }) + + yield* setTime("2026-11-02T08:30:00Z") + yield* schedules.tick + expect(yield* pending).toEqual([messageID(schedule, 1)]) + expect((yield* schedules.get(schedule.id)).status).toBe("done") + }), + ) + + it.effect("stops an every schedule after ends.times", () => + Effect.gen(function* () { + const schedules = yield* setup() + const schedule = yield* schedules.create({ + sessionID, + when: { type: "every", every: "1 minute" }, + ends: { times: 2 }, + action: { type: "prompt", prompt: "poll" }, + }) + for (const _ of [1, 2, 3]) { + yield* TestClock.adjust("1 minute") + yield* schedules.tick + yield* deliver + } + expect(yield* schedules.get(schedule.id)).toMatchObject({ status: "done", runCount: 2, remainingRuns: 0 }) + expect(wakes).toEqual([sessionID, sessionID]) + }), + ) + + it.effect("stops an every schedule at ends.at", () => + Effect.gen(function* () { + yield* setTime("2026-10-26T12:00:00Z") + const schedules = yield* setup() + const schedule = yield* schedules.create({ + sessionID, + when: { type: "every", every: "10 minutes" }, + ends: { at: "2026-10-26T12:25:00Z" }, + action: { type: "prompt", prompt: "poll" }, + }) + for (const _ of [1, 2, 3]) { + yield* TestClock.adjust("10 minutes") + yield* schedules.tick + yield* deliver + } + expect(yield* schedules.get(schedule.id)).toMatchObject({ status: "done", runCount: 2 }) + }), + ) + + it.effect("keeps cron at the same wall-clock time across a DST change", () => + Effect.gen(function* () { + yield* setTime("2026-10-31T12:00:00Z") + const schedules = yield* setup() + const schedule = yield* schedules.create({ + sessionID, + when: { type: "cron", cron: "30 9 * * *", tz: "America/New_York" }, + action: { type: "prompt", prompt: "morning" }, + }) + expect(iso(schedule.nextDueAt)).toBe("2026-10-31T13:30:00.000Z") + + yield* setTime("2026-10-31T13:30:00Z") + yield* schedules.tick + expect(iso((yield* schedules.get(schedule.id)).nextDueAt)).toBe("2026-11-01T14:30:00.000Z") + }), + ) + + it.effect("catches up once after downtime, marked late, and keeps the phase", () => + Effect.gen(function* () { + yield* setTime("2026-10-26T12:00:00Z") + const schedules = yield* setup() + const schedule = yield* schedules.create({ + sessionID, + when: { type: "every", every: "10 minutes" }, + action: { type: "prompt", prompt: "poll" }, + }) + + yield* setTime("2026-10-26T13:05:00Z") + yield* schedules.tick + yield* schedules.tick + expect(yield* pending).toEqual([messageID(schedule, 1)]) + const { db } = yield* Database.Service + expect(yield* SessionInbox.find(db, messageID(schedule, 1))).toMatchObject({ + payload: { metadata: { schedule: { id: schedule.id, run: 1, late: true } } }, + }) + expect(yield* schedules.get(schedule.id)).toMatchObject({ runCount: 1, lastOutcome: "late" }) + expect(iso((yield* schedules.get(schedule.id)).nextDueAt)).toBe("2026-10-26T13:10:00.000Z") + }), + ) + + it.effect("skips a firing while the previous prompt is unconsumed", () => + Effect.gen(function* () { + const schedules = yield* setup() + const schedule = yield* schedules.create({ + sessionID, + when: { type: "every", every: "1 minute" }, + action: { type: "prompt", prompt: "poll" }, + }) + yield* TestClock.adjust("1 minute") + yield* schedules.tick + yield* TestClock.adjust("1 minute") + yield* schedules.tick + expect(yield* pending).toEqual([messageID(schedule, 1)]) + expect(yield* schedules.get(schedule.id)).toMatchObject({ runCount: 1, lastOutcome: "skipped" }) + // The skip wakes the Session again in case the wake after admission was lost. + expect(wakes).toEqual([sessionID, sessionID]) + + yield* deliver + yield* TestClock.adjust("1 minute") + yield* schedules.tick + expect(yield* pending).toEqual([messageID(schedule, 2)]) + }), + ) + + it.effect("does not duplicate a prompt admitted before a crash", () => + Effect.gen(function* () { + const schedules = yield* setup() + const sessions = yield* Session.Service + const schedule = yield* schedules.create({ + sessionID, + when: { type: "in", in: "1 minute" }, + action: { type: "prompt", prompt: "once" }, + }) + yield* TestClock.adjust("1 minute") + // The previous process admitted run 1 and died before advancing the row. + yield* sessions.prompt({ sessionID, id: messageID(schedule, 1), text: "once", resume: false }) + + yield* schedules.tick + expect(yield* pending).toEqual([messageID(schedule, 1)]) + expect(wakes).toEqual([sessionID]) + expect(yield* schedules.get(schedule.id)).toMatchObject({ status: "done", runCount: 1 }) + }), + ) + + it.effect("removes a Session's schedules with the Session", () => + Effect.gen(function* () { + const schedules = yield* setup() + yield* schedules.create({ + sessionID, + when: { type: "every", every: "1 hour" }, + action: { type: "prompt", prompt: "poll" }, + }) + const { db } = yield* Database.Service + yield* db.delete(SessionTable).where(eq(SessionTable.id, sessionID)).run().pipe(Effect.orDie) + expect(yield* schedules.list()).toEqual([]) + }), + ) + + it.effect("does nothing when experimental.schedules is off", () => + Effect.gen(function* () { + const schedules = yield* setup(disabledDirectory) + expect( + yield* Effect.flip( + schedules.create({ + sessionID, + when: { type: "in", in: "1 minute" }, + action: { type: "prompt", prompt: "poll" }, + }), + ), + ).toMatchObject({ _tag: "Schedule.DisabledError" }) + + const { db } = yield* Database.Service + yield* db + .insert(ScheduleTable) + .values({ + id: Schedule.ID.create(), + session_id: sessionID, + action: { type: "prompt", prompt: "poll" }, + when: { type: "in", in: "1 minute" }, + tz: "UTC", + next_due_at: 0, + time_created: 0, + time_updated: 0, + }) + .run() + .pipe(Effect.orDie) + yield* TestClock.adjust("1 minute") + yield* schedules.tick + expect(yield* pending).toEqual([]) + expect((yield* schedules.list())[0]).toMatchObject({ runCount: 0 }) + }), + ) + + it.effect("rechecks a schedule hourly while its location has the flag off, then catches up once it is on", () => + Effect.gen(function* () { + const schedules = yield* setup(toggleDirectory) + const { db } = yield* Database.Service + const id = Schedule.ID.create() + yield* db + .insert(ScheduleTable) + .values({ + id, + session_id: sessionID, + action: { type: "prompt", prompt: "poll" }, + when: { type: "every", every: "1 minute" }, + tz: "UTC", + next_due_at: 60_000, + time_created: 0, + time_updated: 0, + }) + .run() + .pipe(Effect.orDie) + yield* TestClock.adjust("1 minute") + yield* schedules.tick + expect(yield* schedules.get(id)).toMatchObject({ + status: "active", + runCount: 0, + lastOutcome: "disabled", + nextDueAt: DateTime.makeUnsafe(60_000), + }) + const reads = toggle.reads + expect(reads).toBeGreaterThan(0) + + // Within the hour the location is not booted to read the flag again. + yield* TestClock.adjust("59 minutes") + yield* schedules.tick + expect(toggle.reads).toBe(reads) + expect(yield* pending).toEqual([]) + + // Once the flag is on and the hour has passed, the missed slot fires once, late, and keeps the phase. + toggle.on = true + yield* TestClock.adjust("2 minutes") + yield* schedules.tick + expect(yield* pending).toEqual([SessionMessage.ID.make(`msg_${id}_1`)]) + expect(yield* schedules.get(id)).toMatchObject({ + status: "active", + runCount: 1, + lastOutcome: "late", + nextDueAt: DateTime.makeUnsafe(63 * 60_000), + }) + }), + ) + + it.effect("pauses and resumes at the next slot", () => + Effect.gen(function* () { + const schedules = yield* setup() + const schedule = yield* schedules.create({ + sessionID, + when: { type: "every", every: "1 minute" }, + action: { type: "prompt", prompt: "poll" }, + }) + expect((yield* schedules.pause(schedule.id)).status).toBe("paused") + yield* TestClock.adjust("150 seconds") + yield* schedules.tick + expect(yield* pending).toEqual([]) + + const resumed = yield* schedules.resume(schedule.id) + expect(resumed.status).toBe("active") + expect(iso(resumed.nextDueAt)).toBe(new Date(180_000).toISOString()) + yield* schedules.remove(schedule.id) + expect(yield* Effect.flip(schedules.get(schedule.id))).toMatchObject({ _tag: "Schedule.NotFoundError" }) + }), + ) + + it.effect("keeps the run count when paused while a prompt is being admitted", () => + Effect.gen(function* () { + const schedules = yield* setup() + const schedule = yield* schedules.create({ + sessionID, + when: { type: "every", every: "1 minute" }, + action: { type: "prompt", prompt: "poll" }, + }) + yield* TestClock.adjust("1 minute") + duringWake = schedules.pause(schedule.id).pipe(Effect.orDie, Effect.asVoid) + yield* schedules.tick + expect(yield* schedules.get(schedule.id)).toMatchObject({ status: "paused", runCount: 1 }) + yield* deliver + + yield* schedules.resume(schedule.id) + yield* TestClock.adjust("1 minute") + yield* schedules.tick + expect(yield* pending).toEqual([messageID(schedule, 2)]) + }), + ) + + it.effect("run sleeps until the next schedule is due", () => + Effect.gen(function* () { + const schedules = yield* setup() + const fiber = yield* Effect.forkChild(schedules.run) + const schedule = yield* schedules.create({ + sessionID, + when: { type: "in", in: "10 seconds" }, + action: { type: "prompt", prompt: "soon" }, + }) + yield* TestClock.adjust("9 seconds") + expect(yield* pending).toEqual([]) + yield* TestClock.adjust("1 second") + yield* Effect.yieldNow + expect(yield* pending).toEqual([messageID(schedule, 1)]) + yield* Fiber.interrupt(fiber) + }), + ) + + it.effect("fires the last slot when the tick lands just after ends_at", () => + Effect.gen(function* () { + const schedules = yield* setup() + const schedule = yield* schedules.create({ + sessionID, + when: { type: "every", every: "10 minutes" }, + ends: { after: "20 minutes" }, + action: { type: "prompt", prompt: "poll" }, + }) + yield* TestClock.adjust("10 minutes") + yield* schedules.tick + yield* deliver + yield* TestClock.adjust(600_001) + yield* schedules.tick + expect(yield* pending).toEqual([messageID(schedule, 2)]) + expect(yield* schedules.get(schedule.id)).toMatchObject({ status: "done", runCount: 2, lastOutcome: "fired" }) + }), + ) + + // On 2026-11-01 New York falls back at 06:00Z, so 1:00-1:59 runs twice: 05:00Z EDT, then 06:00Z EST. + const cron = (schedules: Schedule.Interface, expression: string) => + schedules.create({ + sessionID, + when: { type: "cron", cron: expression, tz: "America/New_York" }, + action: { type: "prompt", prompt: expression }, + }) + + it.effect("does not fire a step cron again in the repeated hour", () => + Effect.gen(function* () { + yield* setTime("2026-11-01T05:50:00Z") + const schedules = yield* setup() + const schedule = yield* cron(schedules, "*/10 * * * *") + // 1:50 EDT fired the last 1:xx slot; the next is 2:00 EST, an hour and ten minutes later. + expect(iso(schedule.nextDueAt)).toBe("2026-11-01T07:00:00.000Z") + + // Created inside the repeated hour, the next slot is the first one after it, not three hours on. + yield* setTime("2026-11-01T06:35:00Z") + expect(iso((yield* cron(schedules, "*/10 * * * *")).nextDueAt)).toBe("2026-11-01T07:00:00.000Z") + expect(iso((yield* cron(schedules, "0,30 * * * *")).nextDueAt)).toBe("2026-11-01T07:00:00.000Z") + expect(iso((yield* cron(schedules, "0 * * * *")).nextDueAt)).toBe("2026-11-01T07:00:00.000Z") + }), + ) + + it.effect("fires a fixed-time cron once in the repeated hour", () => + Effect.gen(function* () { + yield* setTime("2026-11-01T05:00:00Z") + const schedules = yield* setup() + const schedule = yield* cron(schedules, "30 1 * * *") + expect(iso(schedule.nextDueAt)).toBe("2026-11-01T05:30:00.000Z") + + yield* setTime("2026-11-01T05:30:00Z") + yield* schedules.tick + expect(iso((yield* schedules.get(schedule.id)).nextDueAt)).toBe("2026-11-02T06:30:00.000Z") + + yield* setTime("2026-11-01T06:05:00Z") + expect(iso((yield* cron(schedules, "30 1 * * *")).nextDueAt)).toBe("2026-11-02T06:30:00.000Z") + }), + ) + + it.effect("moves cron past the skipped hour when clocks spring forward", () => + Effect.gen(function* () { + // On 2026-03-08 New York jumps from 2:00 EST to 3:00 EDT at 07:00Z. + yield* setTime("2026-03-08T06:55:00Z") + const schedules = yield* setup() + expect(iso((yield* cron(schedules, "*/10 * * * *")).nextDueAt)).toBe("2026-03-08T07:00:00.000Z") + expect(iso((yield* cron(schedules, "30 2 * * *")).nextDueAt)).toBe("2026-03-08T07:30:00.000Z") + }), + ) + + it.effect("rejects a cron that never fires", () => + Effect.gen(function* () { + const schedules = yield* setup() + const error = yield* Effect.flip( + schedules.create({ + sessionID, + when: { type: "cron", cron: "0 0 31 2 *" }, + action: { type: "prompt", prompt: "never" }, + }), + ) + expect(error).toMatchObject({ _tag: "Schedule.InvalidError" }) + }), + ) + + it.effect("wakes a Session on start when its scheduled prompt was never claimed", () => + Effect.gen(function* () { + const schedules = yield* setup() + yield* schedules.create({ + sessionID, + when: { type: "in", in: "1 minute" }, + action: { type: "prompt", prompt: "once" }, + }) + yield* TestClock.adjust("1 minute") + yield* schedules.tick + // The process died before the Session claimed the prompt, so that wake was lost. + wakes.length = 0 + const fiber = yield* Effect.forkChild(schedules.run) + yield* Effect.yieldNow + expect(wakes).toEqual([sessionID]) + yield* Fiber.interrupt(fiber) + }), + ) + + it.effect("parses at times with offsets or in the schedule's zone", () => + Effect.gen(function* () { + yield* setTime("2026-11-01T12:00:00Z") + const schedules = yield* setup() + const at = (value: string) => + schedules + .create({ sessionID, when: { type: "at", at: value }, action: { type: "prompt", prompt: "at" } }) + .pipe(Effect.map((schedule) => iso(schedule.nextDueAt))) + expect(yield* at("2026-11-02T09:30:00-0500")).toBe("2026-11-02T14:30:00.000Z") + expect(yield* at("2026-11-02T09:30:00-05:00")).toBe("2026-11-02T14:30:00.000Z") + expect(yield* at("2026-11-02T09:30")).toBe("2026-11-02T14:30:00.000Z") + expect(yield* at("2026-11-02 09:30:00-05:00")).toBe("2026-11-02T14:30:00.000Z") + expect(yield* at("2026-11-02 14:30:00Z")).toBe("2026-11-02T14:30:00.000Z") + expect(yield* at("2026-11-02 09:30")).toBe("2026-11-02T14:30:00.000Z") + }), + ) + + it.effect("run ticks again at once when a schedule fell due while it was firing", () => + Effect.gen(function* () { + const schedules = yield* setup() + const first = yield* schedules.create({ + sessionID, + when: { type: "in", in: "10 seconds" }, + action: { type: "prompt", prompt: "first" }, + }) + const second = yield* schedules.create({ + sessionID, + when: { type: "in", in: "10500 millis" }, + action: { type: "prompt", prompt: "second" }, + }) + // Firing the first schedule takes a second. + duringWake = TestClock.adjust("1 second") + const fiber = yield* Effect.forkChild(schedules.run) + yield* Effect.yieldNow + yield* TestClock.adjust("10 seconds") + // Without moving the clock again, the loop must tick a second time on its own. + const ids = yield* Effect.yieldNow.pipe( + Effect.andThen(pending), + Effect.repeat({ until: (ids) => ids.length >= 2, times: 1000 }), + ) + expect(ids).toEqual([messageID(first, 1), messageID(second, 1)]) + yield* Fiber.interrupt(fiber) + }), + ) + + it.effect("backs off for five minutes when a prompt fails, then fires once", () => + Effect.gen(function* () { + const schedules = yield* setup() + const schedule = yield* schedules.create({ + sessionID, + when: { type: "in", in: "1 minute" }, + action: { type: "prompt", prompt: "flaky" }, + }) + yield* TestClock.adjust("1 minute") + duringWake = Effect.die("wake failed") + yield* schedules.tick + expect(yield* schedules.get(schedule.id)).toMatchObject({ status: "active", runCount: 0, lastOutcome: "failed" }) + + // Not retried on every tick. + wakes.length = 0 + yield* TestClock.adjust("4 minutes") + yield* schedules.tick + expect(wakes).toEqual([]) + + yield* TestClock.adjust("1 minute") + yield* schedules.tick + expect(wakes).toEqual([sessionID]) + expect(yield* pending).toEqual([messageID(schedule, 1)]) + expect(yield* schedules.get(schedule.id)).toMatchObject({ status: "done", runCount: 1, lastOutcome: "late" }) + }), + ) + + it.effect("rejects a zero duration or ends.times below 1", () => + Effect.gen(function* () { + const schedules = yield* setup() + const create = (input: Omit) => + Effect.flip(schedules.create({ sessionID, action: { type: "prompt", prompt: "x" }, ...input })) + expect(yield* create({ when: { type: "in", in: "0 seconds" } })).toMatchObject({ + _tag: "Schedule.InvalidError", + reason: "Duration must be positive: 0 seconds", + }) + expect(yield* create({ when: { type: "every", every: "1 hour" }, ends: { times: 0 } })).toMatchObject({ + _tag: "Schedule.InvalidError", + reason: "ends.times must be at least 1", + }) + expect(yield* schedules.list({ sessionID })).toEqual([]) + }), + ) +}) diff --git a/packages/protocol/openapi.json b/packages/protocol/openapi.json index b4cd7f1b3a85..1a190b249e60 100644 --- a/packages/protocol/openapi.json +++ b/packages/protocol/openapi.json @@ -15036,6 +15036,9 @@ "username": { "type": "string" }, + "timezone": { + "type": "string" + }, "permissions": { "$ref": "#/components/schemas/Permission.Ruleset" }, @@ -15315,6 +15318,9 @@ "media_blob": { "type": "boolean" }, + "schedules": { + "type": "boolean" + }, "subagent_depth": { "type": "integer", "minimum": 0 diff --git a/packages/schema/src/config.ts b/packages/schema/src/config.ts index 732947eb3fe9..e78cbae0f8e3 100644 --- a/packages/schema/src/config.ts +++ b/packages/schema/src/config.ts @@ -52,6 +52,9 @@ export class Info extends Schema.Class("Config.Info")({ username: Schema.String.pipe(optional).annotate({ description: "Username displayed in conversations and used for telemetry identity", }), + timezone: Schema.String.pipe(optional).annotate({ + description: "IANA time zone for schedules, e.g. America/Los_Angeles. Defaults to the server's local zone", + }), permissions: Permission.Ruleset.pipe(optional).annotate({ description: "Ordered tool permission rules applied to agent tool use", }), diff --git a/packages/schema/src/config/experimental.ts b/packages/schema/src/config/experimental.ts index 633a03e58d46..2541219fa5e9 100644 --- a/packages/schema/src/config/experimental.ts +++ b/packages/schema/src/config/experimental.ts @@ -19,6 +19,9 @@ export class Info extends Schema.Class("ConfigExperimental.Info")({ description: "Store new session media once in a deduplicated blob table instead of inline in every message row. Defaults to false.", }), + schedules: Schema.Boolean.pipe(optional).annotate({ + description: "Enable experimental scheduled tasks that wake a session with a prompt. Defaults to false.", + }), subagent_depth: NonNegativeInt.pipe(optional).annotate({ description: "Maximum subagent nesting depth. Defaults to 1.", }), diff --git a/packages/schema/src/schedule.ts b/packages/schema/src/schedule.ts new file mode 100644 index 000000000000..b962b85a095e --- /dev/null +++ b/packages/schema/src/schedule.ts @@ -0,0 +1,109 @@ +export * as Schedule from "./schedule.js" + +import { Duration, Option, Schema } from "effect" +import { ascending } from "./identifier.js" +import { DateTimeUtcFromMillis, NonNegativeInt, PositiveInt, optional, statics } from "./schema.js" +import { SessionID } from "./session-id.js" + +export const ID = Schema.String.check(Schema.isStartingWith("sch_")).pipe( + Schema.brand("Schedule.ID"), + Schema.annotate({ identifier: "Schedule.ID" }), + statics((schema) => ({ + create: () => schema.make("sch_" + ascending()), + })), +) +export type ID = typeof ID.Type + +const decodeDuration = Schema.decodeUnknownOption(Schema.DurationFromString) + +/** Milliseconds in a duration string, or `None` when it is not one. */ +export const durationMillis = (value: string) => decodeDuration(value).pipe(Option.map(Duration.toMillis)) + +/** A duration such as "90 seconds", "5 minutes" or "2 hours". */ +export const DurationString = Schema.String.check( + Schema.makeFilter((value: string) => + Option.match(durationMillis(value), { + onNone: () => `expected a duration such as "5 minutes", got ${JSON.stringify(value)}`, + onSome: (millis) => millis > 0 || "expected a positive duration", + }), + ), +).annotate({ identifier: "Schedule.Duration" }) + +const zone = { + tz: Schema.String.pipe(optional).annotate({ + description: "IANA time zone. Defaults to the configured timezone, then the server's local zone.", + }), +} + +export const In = Schema.Struct({ type: Schema.tag("in"), in: DurationString, ...zone }).annotate({ + identifier: "Schedule.When.In", +}) +/** An instant with an offset, or a wall-clock time in `tz`, e.g. "2026-10-27T09:00". */ +export const At = Schema.Struct({ type: Schema.tag("at"), at: Schema.String, ...zone }).annotate({ + identifier: "Schedule.When.At", +}) +export const Every = Schema.Struct({ type: Schema.tag("every"), every: DurationString, ...zone }).annotate({ + identifier: "Schedule.When.Every", +}) +export const Cron = Schema.Struct({ type: Schema.tag("cron"), cron: Schema.String, ...zone }).annotate({ + identifier: "Schedule.When.Cron", +}) + +export const When = Schema.Union([In, At, Every, Cron]).pipe( + Schema.toTaggedUnion("type"), + Schema.annotate({ identifier: "Schedule.When" }), +) +export type When = typeof When.Type + +/** Stops a repeating schedule at whichever limit comes first. */ +export const Ends = Schema.Struct({ + after: DurationString.pipe(optional), + at: Schema.String.pipe(optional), + times: PositiveInt.pipe(optional), +}).annotate({ identifier: "Schedule.Ends" }) +export type Ends = typeof Ends.Type + +export const PromptAction = Schema.Struct({ type: Schema.tag("prompt"), prompt: Schema.String }).annotate({ + identifier: "Schedule.Action.Prompt", +}) + +export const Action = Schema.Union([PromptAction]).pipe( + Schema.toTaggedUnion("type"), + Schema.annotate({ identifier: "Schedule.Action" }), +) +export type Action = typeof Action.Type + +export const Status = Schema.Literals(["active", "paused", "done"]).annotate({ identifier: "Schedule.Status" }) +export type Status = typeof Status.Type + +export const Outcome = Schema.Literals(["fired", "late", "skipped", "disabled", "failed"]).annotate({ + identifier: "Schedule.Outcome", +}) +export type Outcome = typeof Outcome.Type + +export const Info = Schema.Struct({ + id: ID, + sessionID: SessionID.pipe(optional), + name: Schema.String.pipe(optional), + when: When, + tz: Schema.String, + action: Action, + status: Status, + nextDueAt: DateTimeUtcFromMillis.pipe(optional), + endsAt: DateTimeUtcFromMillis.pipe(optional), + remainingRuns: NonNegativeInt.pipe(optional), + runCount: NonNegativeInt, + lastFiredAt: DateTimeUtcFromMillis.pipe(optional), + lastOutcome: Outcome.pipe(optional), + time: Schema.Struct({ created: DateTimeUtcFromMillis, updated: DateTimeUtcFromMillis }), +}).annotate({ identifier: "Schedule.Info" }) +export type Info = typeof Info.Type + +export const CreateInput = Schema.Struct({ + sessionID: SessionID, + name: Schema.String.pipe(optional), + when: When, + ends: Ends.pipe(optional), + action: Action, +}).annotate({ identifier: "Schedule.CreateInput" }) +export type CreateInput = typeof CreateInput.Type diff --git a/packages/server/src/process.ts b/packages/server/src/process.ts index 3abd8b815246..5ae353d3dff5 100644 --- a/packages/server/src/process.ts +++ b/packages/server/src/process.ts @@ -3,6 +3,7 @@ export * as ServerProcess from "./process" import { NodeHttpServer } from "@effect/platform-node" import { Bus } from "@opencode/core/bus" import { SessionRestart } from "@opencode/core/session/execution/restart" +import { Schedule } from "@opencode/core/schedule" import { InstallationEvent } from "@opencode/schema/installation-event" import { hasPtyConnectTicketURL } from "@opencode/protocol/groups/pty" import { hasPersistentPtyConnectTicketURL } from "@opencode/protocol/groups/persistent-pty" @@ -109,6 +110,8 @@ export const start = Effect.fn("ServerProcess.start")(function* ( yield* installRestartContinuity(Context.get(context, SessionRestart.Service)).pipe( Effect.provideService(Scope.Scope, applicationScope), ) + // Schedules fire from one process only; the table holds all their state. + yield* Effect.forkIn(Context.get(context, Schedule.Service).run, applicationScope) } yield* Ref.set(application, Option.some(Context.get(context, HttpRouter.HttpRouter).asHttpEffect())) yield* status.ready diff --git a/packages/server/src/routes.ts b/packages/server/src/routes.ts index 19122cfe789f..e3ae6830b196 100644 --- a/packages/server/src/routes.ts +++ b/packages/server/src/routes.ts @@ -29,6 +29,7 @@ import { LocationServiceMap } from "@opencode/core/location-service-map" import { LocationActivity } from "@opencode/core/location-activity" import { ModelsDev } from "@opencode/core/models-dev" import { SessionRestart } from "@opencode/core/session/execution/restart" +import { Schedule } from "@opencode/core/schedule" import { PluginUpdate } from "@opencode/core/plugin/update" import { SdkPlugins } from "@opencode/core/plugin/sdk" import { WellKnown } from "@opencode/core/wellknown" @@ -77,6 +78,7 @@ const applicationServiceNodes = [ LocationServiceMap.node, LocationActivity.node, SessionRestart.node, + Schedule.node, Workspace.node, FileWatch.node, ] as const