From 705c3ab746d6c113e9c1a3d50718effc1386b60d Mon Sep 17 00:00:00 2001 From: Ryan Vogel <89211796+R44VC0RP@users.noreply.github.com> Date: Tue, 29 Sep 2026 11:57:09 -0400 Subject: [PATCH] feat(core): let agents create recurring and one-shot schedules Adds schedule_create, schedule_list and schedule_cancel tools in the opencode namespace so the agent can schedule prompts back into its own session. schedule_create accepts `every` (30s/15m/2h/1d), `at` (ISO 8601 with offset) for a single run, or both (`at` sets the first run). SessionSchedule now stores multiple schedules per session, keyed by a new `sch_` ID, and persists the absolute next fire time instead of only the interval. This replaces the one-schedule-per-session model and fixes restarts resetting the countdown: one-shots fire at their original time after a restart, and overdue recurring schedules fire once then resume their cadence rather than bursting. `/schedule off` now cancels every schedule in the session. Past `at` times, missing timing and invalid intervals return tool errors. Verified with an isolated managed-mode server and a scripted OpenAI-compatible mock model driving `execute`: create every 5s fired every 5s; `at` fired at 15:55:42.013 for a 15:55:42 target; a one-shot created before a restart fired on time after it (15:56:23.013); list returned both schedules with server time; cancel stopped firing; invalid inputs returned the three expected errors. core and server typecheck. --- packages/core/src/id/id.ts | 1 + packages/core/src/plugin/schedule.ts | 123 ++++++++++++++++++++++-- packages/core/src/session/schedule.ts | 131 +++++++++++++++++--------- 3 files changed, 202 insertions(+), 53 deletions(-) diff --git a/packages/core/src/id/id.ts b/packages/core/src/id/id.ts index f849c0558b0..74eca0c8a5e 100644 --- a/packages/core/src/id/id.ts +++ b/packages/core/src/id/id.ts @@ -9,6 +9,7 @@ const prefixes = { question: "que", part: "prt", pty: "pty", + schedule: "sch", tool: "tool", workspace: "wrk", } as const diff --git a/packages/core/src/plugin/schedule.ts b/packages/core/src/plugin/schedule.ts index 5399f12b2b9..cc8b19d81ef 100644 --- a/packages/core/src/plugin/schedule.ts +++ b/packages/core/src/plugin/schedule.ts @@ -1,15 +1,52 @@ export * as SchedulePlugin from "./schedule.js" +import { ToolFailure } from "@opencode/ai" import { define } from "@opencode/plugin/effect/plugin" -import { Effect } from "effect" +import { Clock, Effect, Schema } from "effect" import { SessionSchedule } from "../session/schedule.js" -const units = { s: 1_000, m: 60_000, h: 3_600_000 } +const CreateInput = Schema.Struct({ + prompt: Schema.String.check(Schema.isMinLength(1)).annotate({ + description: + "Message delivered to this session as a new user message when the schedule fires. Write a complete, self-contained instruction.", + }), + every: Schema.optionalKey(Schema.String).annotate({ + description: "Repeat interval such as 30s, 15m, 2h or 1d. Without `at`, the first run is one interval from now.", + }), + at: Schema.optionalKey(Schema.String).annotate({ + description: + "ISO 8601 timestamp with a timezone offset, e.g. 2026-09-29T15:00:00-04:00. Alone it schedules a single run; with `every` it sets the first run.", + }), +}) + +const CancelInput = Schema.Struct({ + id: Schema.String.annotate({ description: "Schedule ID returned by schedule_create or schedule_list." }), +}) + +const View = Schema.Struct({ + id: Schema.String, + prompt: Schema.String, + next: Schema.String.annotate({ description: "Next run as an ISO 8601 timestamp." }), + every: Schema.optionalKey(Schema.String).annotate({ description: "Repeat interval; absent for a single run." }), +}) + +const ListOutput = Schema.Struct({ + now: Schema.String.annotate({ description: "Current server time as an ISO 8601 timestamp." }), + schedules: Schema.Array(View), +}) + +const view = (info: SessionSchedule.Info) => ({ + id: info.id, + prompt: info.text, + next: new Date(info.next).toISOString(), + ...(info.every === undefined ? {} : { every: `${info.every / 1000}s` }), +}) export const Plugin = define({ id: "opencode.schedule", effect: Effect.fn(function* (ctx) { const schedule = yield* SessionSchedule.Service + yield* ctx.command.transform((editor) => { editor.add({ name: "schedule", @@ -17,15 +54,83 @@ export const Plugin = define({ execute: (input) => Effect.gen(function* () { const text = input.prompt.text.trim() - if (text === "off") return yield* schedule.remove(input.sessionID) - const match = /^(\d+)([smh])\s+([\s\S]+)$/.exec(text) - if (!match) return yield* Effect.fail(new Error("Usage: /schedule <30s|15m|2h> , or /schedule off")) - yield* schedule.set(input.sessionID, { - interval: Number(match[1]) * units[match[2] as keyof typeof units], - text: match[3], - }) + if (text === "off") { + const existing = yield* schedule.list(input.sessionID) + return yield* Effect.forEach(existing, (info) => schedule.cancel(info.id), { discard: true }) + } + const [interval = "", ...rest] = text.split(/\s+/) + const every = SessionSchedule.parseInterval(interval) + const prompt = rest.join(" ") + if (every === undefined || !prompt) + return yield* Effect.fail(new Error("Usage: /schedule <30s|15m|2h> , or /schedule off")) + yield* schedule.create({ sessionID: input.sessionID, text: prompt, every }) }), }) }) + + yield* ctx.tool + .transform((editor) => { + editor.add({ + name: "schedule_create", + description: + "Schedule a prompt to be sent back to the current session later, even if no client has it open. Use `every` for recurring checks (e.g. every 15m), `at` for a single run at an absolute time, or both. Returns the schedule ID.", + input: CreateInput, + output: View, + options: { namespace: "opencode", codemode: true }, + execute: (input, context) => + Effect.gen(function* () { + if (input.every === undefined && input.at === undefined) + return yield* new ToolFailure({ message: "Provide `every`, `at`, or both" }) + const every = input.every === undefined ? undefined : SessionSchedule.parseInterval(input.every) + if (input.every !== undefined && every === undefined) + return yield* new ToolFailure({ message: `Invalid interval ${input.every}; use e.g. 30s, 15m, 2h, 1d` }) + const at = input.at === undefined ? undefined : Date.parse(input.at) + const now = yield* Clock.currentTimeMillis + if (at !== undefined && Number.isNaN(at)) + return yield* new ToolFailure({ message: `Invalid timestamp ${input.at}` }) + if (at !== undefined && at <= now) + return yield* new ToolFailure({ + message: `${input.at} is in the past; the current time is ${new Date(now).toISOString()}`, + }) + const info = yield* schedule.create({ sessionID: context.sessionID, text: input.prompt, at, every }) + return { + output: view(info), + content: `Scheduled ${info.id}: next run ${view(info).next}${every ? `, then every ${input.every}` : ""}.`, + } + }), + }) + editor.add({ + name: "schedule_list", + description: "List the prompts scheduled for the current session, with the current server time.", + input: Schema.Struct({}), + output: ListOutput, + options: { namespace: "opencode", codemode: true }, + execute: (_input, context) => + Effect.gen(function* () { + const infos = yield* schedule.list(context.sessionID) + return { + output: { + now: new Date(yield* Clock.currentTimeMillis).toISOString(), + schedules: infos.map(view), + }, + } + }), + }) + editor.add({ + name: "schedule_cancel", + description: "Cancel a scheduled prompt in the current session.", + input: CancelInput, + output: Schema.Struct({ id: Schema.String }), + options: { namespace: "opencode", codemode: true }, + execute: (input, context) => + Effect.gen(function* () { + const owned = (yield* schedule.list(context.sessionID)).some((info) => info.id === input.id) + if (!owned) return yield* new ToolFailure({ message: `No schedule ${input.id} in this session` }) + yield* schedule.cancel(input.id) + return { output: { id: input.id }, content: `Cancelled ${input.id}.` } + }), + }) + }) + .pipe(Effect.orDie) }), }) diff --git a/packages/core/src/session/schedule.ts b/packages/core/src/session/schedule.ts index 2aec9d6eabb..60b56f996ac 100644 --- a/packages/core/src/session/schedule.ts +++ b/packages/core/src/session/schedule.ts @@ -1,24 +1,50 @@ export * as SessionSchedule from "./schedule.js" -import { Context, Duration, Effect, Fiber, Layer, Option, Schema } from "effect" +import { Clock, Context, Duration, Effect, Fiber, Layer, Option, Schema } from "effect" import { makeGlobalNode } from "@opencode/util/effect/app-node" +import { Identifier } from "../id/id.js" import { KV } from "../kv.js" import { Session } from "../session.js" import { SessionSchema } from "./schema.js" -const Entry = Schema.Struct({ interval: Schema.Number, text: Schema.String }) -export type Entry = typeof Entry.Type -const decodeEntry = Schema.decodeUnknownOption(Entry) +export const Info = Schema.Struct({ + id: Schema.String, + sessionID: SessionSchema.ID, + text: Schema.String, + /** Next fire time in epoch milliseconds. Persisted so restarts keep the cadence. */ + next: Schema.Number, + /** Repeat interval in milliseconds; absent for one-shot schedules. */ + every: Schema.optionalKey(Schema.Number), +}) +export type Info = typeof Info.Type +const decodeInfo = Schema.decodeUnknownOption(Info) const prefix = "session.schedule/" +const units = { s: 1_000, m: 60_000, h: 3_600_000, d: 86_400_000 } + +/** Parses an interval such as `30s`, `15m`, `2h` or `1d` into milliseconds. */ +export function parseInterval(input: string) { + const match = /^(\d+)([smhd])$/.exec(input.trim()) + if (!match) return + const value = Number(match[1]) * units[match[2] as keyof typeof units] + return value > 0 ? value : undefined +} + export interface Interface { - /** Saves a recurring prompt for a Session and (re)starts its timer. */ - readonly set: (sessionID: SessionSchema.ID, entry: Entry) => Effect.Effect - readonly remove: (sessionID: SessionSchema.ID) => Effect.Effect + /** Saves a prompt that fires once at `at`, or every `every` ms starting one interval from now. */ + readonly create: (input: { + sessionID: SessionSchema.ID + text: string + at?: number + every?: number + }) => Effect.Effect + readonly cancel: (id: string) => Effect.Effect + readonly list: (sessionID: SessionSchema.ID) => Effect.Effect /** * Starts timers for every saved schedule. Timers live in this global service rather than in a * Location, so they keep firing while the Session's Location is unloaded; each prompt loads it on - * demand. Inert until called: the managed server calls it once at boot. + * demand. Overdue schedules fire once immediately. Inert until called: the managed server calls it + * once at boot. */ readonly resume: Effect.Effect } @@ -31,52 +57,69 @@ const layer = Layer.effect( const kv = yield* KV.Service const sessions = yield* Session.Service const scope = yield* Effect.scope - const running = new Map>() + const running = new Map>() - const stop = Effect.fnUntraced(function* (sessionID: SessionSchema.ID) { - const fiber = running.get(sessionID) - running.delete(sessionID) + const stop = Effect.fnUntraced(function* (id: string) { + const fiber = running.get(id) + running.delete(id) if (fiber) yield* Fiber.interrupt(fiber) }) - const start = Effect.fnUntraced(function* (sessionID: SessionSchema.ID, entry: Entry) { - yield* stop(sessionID) - const loop: Effect.Effect = Effect.gen(function* () { - yield* Effect.sleep(Duration.millis(entry.interval)) - const exists = yield* sessions.prompt({ sessionID, text: entry.text, delivery: "queue" }).pipe( - Effect.as(true), - Effect.catchTag("Session.NotFoundError", () => Effect.succeed(false)), - Effect.catchCause((cause) => - Effect.logWarning("scheduled prompt failed", { sessionID, cause }).pipe(Effect.as(true)), - ), - ) - // A deleted Session ends its schedule instead of retrying forever. - if (!exists) return yield* kv.remove(prefix + sessionID) - return yield* loop - }) - running.set(sessionID, yield* Effect.forkIn(loop, scope)) + const start = Effect.fnUntraced(function* (info: Info) { + yield* stop(info.id) + const loop = (current: Info): Effect.Effect => + Effect.gen(function* () { + const wait = current.next - (yield* Clock.currentTimeMillis) + if (wait > 0) yield* Effect.sleep(Duration.millis(wait)) + const exists = yield* sessions + .prompt({ sessionID: current.sessionID, text: current.text, delivery: "queue" }) + .pipe( + Effect.as(true), + Effect.catchTag("Session.NotFoundError", () => Effect.succeed(false)), + Effect.catchCause((cause) => + Effect.logWarning("scheduled prompt failed", { id: current.id, cause }).pipe(Effect.as(true)), + ), + ) + // One-shot schedules and schedules of deleted Sessions end here. + if (!exists || current.every === undefined) { + running.delete(current.id) + return yield* kv.remove(prefix + current.id) + } + // Skip occurrences missed while the server was down instead of firing them in a burst. + const now = yield* Clock.currentTimeMillis + const next = { ...current, next: Math.max(current.next + current.every, now + current.every) } + yield* kv.set(prefix + current.id, next) + return yield* loop(next) + }) + running.set(info.id, yield* Effect.forkIn(loop(info), scope)) + }) + + const all = Effect.gen(function* () { + const saved = yield* kv.scan({ prefix, limit: 1000 }) + return saved.entries.flatMap((item) => Option.toArray(decodeInfo(item.value))) }) return Service.of({ - set: Effect.fnUntraced(function* (sessionID, entry) { - yield* kv.set(prefix + sessionID, entry) - yield* start(sessionID, entry) + create: Effect.fnUntraced(function* (input) { + const now = yield* Clock.currentTimeMillis + const info: Info = { + id: Identifier.ascending("schedule"), + sessionID: input.sessionID, + text: input.text, + next: input.at ?? now + (input.every ?? 0), + ...(input.every === undefined ? {} : { every: input.every }), + } + yield* kv.set(prefix + info.id, info) + yield* start(info) + return info }), - remove: Effect.fnUntraced(function* (sessionID) { - yield* stop(sessionID) - yield* kv.remove(prefix + sessionID) + cancel: Effect.fnUntraced(function* (id) { + yield* stop(id) + yield* kv.remove(prefix + id) }), + list: (sessionID) => all.pipe(Effect.map((infos) => infos.filter((info) => info.sessionID === sessionID))), resume: Effect.gen(function* () { - const saved = yield* kv.scan({ prefix }) - yield* Effect.forEach( - saved.entries, - (item) => { - const entry = decodeEntry(item.value) - if (Option.isNone(entry)) return Effect.void - return start(SessionSchema.ID.make(item.key.slice(prefix.length)), entry.value) - }, - { discard: true }, - ) + yield* Effect.forEach(yield* all, start, { discard: true }) }), }) }),