feat(core): run session schedules as global background jobs

Moves schedule timers out of the per-Location plugin into a new global
SessionSchedule service. Previously timers lived in the Location's plugin
scope, so they stopped on server restart until something reloaded the
Location, and stopped whenever the Location was evicted after 60 minutes
of inactivity. Timers now live in the global service and call
Session.prompt, which loads the Location on demand, so schedules fire
even when the session is not open.

- SessionSchedule persists entries in KV (session.schedule/<id>) and
  exposes set/remove/resume. A deleted session ends its schedule.
- The managed server calls resume at boot next to restart continuity;
  unmanaged servers sharing the database do not restore schedules.
- The opencode.schedule plugin now only registers /schedule and
  delegates to the service; the per-Location restore logic is removed.

Verified: core and server typecheck; against an isolated managed-mode
server, a 5s schedule fired every 5s, resumed after a restart with no
client touching the session, and `/schedule off` stopped the
boot-restored timer. Server tests: 3 OpenAI OAuth port tests fail
identically without this change.
This commit is contained in:
Ryan Vogel
2026-09-29 11:43:20 -04:00
parent e0709f8de0
commit a5ceab882e
5 changed files with 99 additions and 50 deletions
+3
View File
@@ -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,
+8 -50
View File
@@ -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<Session.ID, Fiber.Fiber<void>>()
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> <prompt>, 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],
})
}),
})
})
+83
View File
@@ -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<void>
readonly remove: (sessionID: SessionSchema.ID) => Effect.Effect<void>
/**
* 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<void>
}
export class Service extends Context.Service<Service, Interface>()("@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<SessionSchema.ID, Fiber.Fiber<void>>()
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<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))
})
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] })
+3
View File
@@ -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* <E, R>(
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
+2
View File
@@ -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)