mirror of
https://gitee.com/zhangsken/opencode.git
synced 2026-10-16 10:30:01 +08:00
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.
This commit is contained in:
@@ -9,6 +9,7 @@ const prefixes = {
|
||||
question: "que",
|
||||
part: "prt",
|
||||
pty: "pty",
|
||||
schedule: "sch",
|
||||
tool: "tool",
|
||||
workspace: "wrk",
|
||||
} as const
|
||||
|
||||
@@ -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> <prompt>, 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> <prompt>, 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)
|
||||
}),
|
||||
})
|
||||
|
||||
@@ -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<void>
|
||||
readonly remove: (sessionID: SessionSchema.ID) => Effect.Effect<void>
|
||||
/** 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<Info>
|
||||
readonly cancel: (id: string) => Effect.Effect<void>
|
||||
readonly list: (sessionID: SessionSchema.ID) => Effect.Effect<readonly Info[]>
|
||||
/**
|
||||
* 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<void>
|
||||
}
|
||||
@@ -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<SessionSchema.ID, Fiber.Fiber<void>>()
|
||||
const running = new Map<string, Fiber.Fiber<void>>()
|
||||
|
||||
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<void> = 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<void> =>
|
||||
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 })
|
||||
}),
|
||||
})
|
||||
}),
|
||||
|
||||
Reference in New Issue
Block a user