diff --git a/packages/core/src/plugin/internal.ts b/packages/core/src/plugin/internal.ts index 2080450cdd6..67290e71892 100644 --- a/packages/core/src/plugin/internal.ts +++ b/packages/core/src/plugin/internal.ts @@ -61,6 +61,7 @@ import { Ripgrep } from "../ripgrep.js" import { Session } from "../session.js" import { SessionCompaction } from "../session/compaction.js" import { SessionInstructions } from "../session/instructions.js" +import { SessionSchedule } from "../session/schedule.js" import { Shell } from "../shell.js" import { ShellSelect } from "../shell/select.js" import { Snapshot } from "../snapshot.js" @@ -143,6 +144,7 @@ const services = [ Session.Service, SessionCompaction.Service, SessionInstructions.Service, + SessionSchedule.Service, Shell.Service, ShellSelect.Service, Snapshot.Service, @@ -196,6 +198,7 @@ export const requirements = LayerNode.group([ Session.node, SessionCompaction.node, SessionInstructions.node, + SessionSchedule.node, Shell.node, ShellSelect.node, Snapshot.node, diff --git a/packages/core/src/plugin/schedule.ts b/packages/core/src/plugin/schedule.ts index ce4e1b7055c..5399f12b2b9 100644 --- a/packages/core/src/plugin/schedule.ts +++ b/packages/core/src/plugin/schedule.ts @@ -1,54 +1,15 @@ export * as SchedulePlugin from "./schedule.js" import { define } from "@opencode/plugin/effect/plugin" -import { Session } from "@opencode/schema/session" -import { Duration, Effect, Fiber, Option, Schema, Scope } from "effect" +import { Effect } from "effect" +import { SessionSchedule } from "../session/schedule.js" -const Entry = Schema.Struct({ interval: Schema.Number, text: Schema.String }) -const decodeEntry = Schema.decodeUnknownOption(Entry) const units = { s: 1_000, m: 60_000, h: 3_600_000 } export const Plugin = define({ id: "opencode.schedule", effect: Effect.fn(function* (ctx) { - const scope = yield* Scope.Scope - const running = new Map>() - - const stop = Effect.fn("SchedulePlugin.stop")(function* (sessionID: Session.ID) { - const fiber = running.get(sessionID) - running.delete(sessionID) - if (fiber) yield* Fiber.interrupt(fiber) - }) - - const start = Effect.fn("SchedulePlugin.start")(function* (sessionID: Session.ID, entry: typeof Entry.Type) { - yield* stop(sessionID) - const fiber = yield* Effect.sleep(Duration.millis(entry.interval)).pipe( - Effect.andThen(ctx.session.prompt({ sessionID, text: entry.text, delivery: "queue" })), - Effect.catchCause((cause) => Effect.logWarning("scheduled prompt failed", { sessionID, cause })), - Effect.forever, - Effect.forkIn(scope), - ) - running.set(sessionID, fiber) - }) - - // Schedules persist in global plugin storage; each location restarts only its own sessions' schedules. - const saved = yield* ctx.storage.scan({ prefix: "" }) - yield* Effect.forEach( - saved.entries, - (item) => - Effect.gen(function* () { - const entry = decodeEntry(item.value) - if (Option.isNone(entry)) return - const sessionID = Session.ID.make(item.key) - const session = yield* ctx.session.get({ sessionID }).pipe(Effect.option) - if (Option.isNone(session)) return - if (session.value.location.directory !== ctx.location.directory) return - if (session.value.location.workspaceID !== ctx.location.workspaceID) return - yield* start(sessionID, entry.value) - }), - { discard: true }, - ) - + const schedule = yield* SessionSchedule.Service yield* ctx.command.transform((editor) => { editor.add({ name: "schedule", @@ -56,16 +17,13 @@ export const Plugin = define({ execute: (input) => Effect.gen(function* () { const text = input.prompt.text.trim() - if (text === "off") { - yield* stop(input.sessionID) - yield* ctx.storage.remove(input.sessionID) - return - } + 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")) - const entry = { interval: Number(match[1]) * units[match[2] as keyof typeof units], text: match[3] } - yield* ctx.storage.set(input.sessionID, entry) - yield* start(input.sessionID, entry) + yield* schedule.set(input.sessionID, { + interval: Number(match[1]) * units[match[2] as keyof typeof units], + text: match[3], + }) }), }) }) diff --git a/packages/core/src/session/schedule.ts b/packages/core/src/session/schedule.ts new file mode 100644 index 00000000000..d6ec9b6bbe1 --- /dev/null +++ b/packages/core/src/session/schedule.ts @@ -0,0 +1,83 @@ +export * as SessionSchedule from "./schedule.js" + +import { Context, Duration, Effect, Fiber, Layer, Option, Schema } from "effect" +import { makeGlobalNode } from "@opencode/util/effect/app-node" +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) +const prefix = "session.schedule/" + +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 + /** + * 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. + */ + readonly resume: Effect.Effect +} + +export class Service extends Context.Service()("@opencode/SessionSchedule") {} + +const layer = Layer.effect( + Service, + Effect.gen(function* () { + const kv = yield* KV.Service + const sessions = yield* Session.Service + const scope = yield* Effect.scope + const running = new Map>() + + const stop = Effect.fnUntraced(function* (sessionID: SessionSchema.ID) { + const fiber = running.get(sessionID) + running.delete(sessionID) + 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)) + }) + + return Service.of({ + set: Effect.fnUntraced(function* (sessionID, entry) { + yield* kv.set(prefix + sessionID, entry) + yield* start(sessionID, entry) + }), + remove: Effect.fnUntraced(function* (sessionID) { + yield* stop(sessionID) + yield* kv.remove(prefix + 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 }, + ) + }), + }) + }), +) + +export const node = makeGlobalNode({ service: Service, layer, deps: [KV.node, Session.node] }) diff --git a/packages/server/src/process.ts b/packages/server/src/process.ts index 1c11586d630..11a0661c818 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 { SessionSchedule } from "@opencode/core/session/schedule" import { InstallationEvent } from "@opencode/schema/installation-event" import { hasPtyConnectTicketURL } from "@opencode/protocol/groups/pty" import { hasPersistentPtyConnectTicketURL } from "@opencode/protocol/groups/persistent-pty" @@ -105,6 +106,8 @@ export const start = Effect.fn("ServerProcess.start")(function* ( yield* installRestartContinuity(Context.get(context, SessionRestart.Service)).pipe( Effect.provideService(Scope.Scope, applicationScope), ) + // Only the managed server restores saved schedules, so extra servers sharing the database never double-fire. + yield* Context.get(context, SessionSchedule.Service).resume } 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 c4f746f1d82..babbfe5297a 100644 --- a/packages/server/src/routes.ts +++ b/packages/server/src/routes.ts @@ -27,6 +27,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 { SessionSchedule } from "@opencode/core/session/schedule" import { PluginUpdate } from "@opencode/core/plugin/update" import { SdkPlugins } from "@opencode/core/plugin/sdk" import { WellKnown } from "@opencode/core/wellknown" @@ -73,6 +74,7 @@ const applicationServiceNodes = [ LocationServiceMap.node, LocationActivity.node, SessionRestart.node, + SessionSchedule.node, Workspace.node, ] as const const applicationServices = LayerNode.group(applicationServiceNodes)