diff --git a/.changeset/scheduled-wakeup.md b/.changeset/scheduled-wakeup.md new file mode 100644 index 00000000000..fa4e6f18e09 --- /dev/null +++ b/.changeset/scheduled-wakeup.md @@ -0,0 +1,5 @@ +--- +"@kilocode/cli": minor +--- + +Support scheduling and cancelling future wakeups: the agent can ask to resume a session at a later time, see what it scheduled, and cancel a pending wakeup before it fires. diff --git a/packages/opencode/src/effect/app-runtime.ts b/packages/opencode/src/effect/app-runtime.ts index 0a2acfcaf33..62f8b7913cf 100644 --- a/packages/opencode/src/effect/app-runtime.ts +++ b/packages/opencode/src/effect/app-runtime.ts @@ -57,6 +57,7 @@ import { RuntimeFlags } from "@/effect/runtime-flags" import { Notebook } from "@/kilocode/notebook/service" import { SessionDrain } from "@/kilocode/session/drain" import { AgentManager } from "@/kilocode/agent-manager/service" +import { Wakeup } from "@/kilocode/wakeup" // kilocode_change end import { EventV2Bridge } from "@/event-v2-bridge" import { LayerNode } from "@opencode-ai/core/effect/layer-node" @@ -77,6 +78,7 @@ const kilo = LayerNode.group([ AgentManager.node, Notebook.node, SessionDrain.node, + Wakeup.node, memory, ]) // kilocode_change end diff --git a/packages/opencode/src/kilocode/bootstrap.ts b/packages/opencode/src/kilocode/bootstrap.ts index c7d23221c8c..0d82f2c120b 100644 --- a/packages/opencode/src/kilocode/bootstrap.ts +++ b/packages/opencode/src/kilocode/bootstrap.ts @@ -12,12 +12,14 @@ import { SessionSummary } from "@/session/summary" import { SessionExport } from "@/kilocode/session-export" import { createWorkspaceProvider } from "@/kilocode/session-export/workspace-provider" import { Instance } from "@/kilocode/instance" +import { InstanceRef } from "@/effect/instance-ref" import { Identity } from "@kilocode/kilo-telemetry" import { MemoryLifecycle } from "@/kilocode/memory/turn" import { MemoryService } from "@kilocode/kilo-memory/effect/service" import { MemoryEvents } from "@/kilocode/memory/events" import { installMemoryRuntime } from "@/kilocode/memory/runtime" import { KiloToolRegistry } from "@/kilocode/tool/registry" +import { Wakeup } from "@/kilocode/wakeup" import { LayerNode } from "@opencode-ai/core/effect/layer-node" import { KilocodeWatcher } from "@/kilocode/watcher" import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder" // kilocode_change @@ -43,6 +45,7 @@ export namespace KilocodeBootstrap { const provider = yield* Provider.Service const memory = yield* MemoryService.Service const watcher = yield* KilocodeWatcher.Service + const wake = yield* Wakeup.Service const init = Effect.fn("KilocodeBootstrap.init")(function* () { yield* watcher.init() @@ -55,6 +58,16 @@ export namespace KilocodeBootstrap { yield* bus.subscribeCallback(MemoryEvents.Updated, (evt) => KiloToolRegistry.invalidateMemoryEnabled(evt.properties.directory), ) + // Re-arm this directory's persisted wakeups on every instance start: overdue ones + // fire immediately, the rest get their timers. A failure must not block bootstrap. + const inst = yield* InstanceRef + if (inst) { + yield* wake.adopt(inst.directory).pipe( + Effect.catchCause((cause) => + Effect.sync(() => log.warn("wakeup adopt failed", { err: Cause.squash(cause) })), + ), + ) + } // Session export bootstrap. yield* Effect.gen(function* () { if (!SessionExport.enabled) return @@ -105,6 +118,7 @@ export namespace KilocodeBootstrap { MemoryService.layer, Bus.defaultLayer, KilocodeWatcher.defaultLayer, + AppNodeBuilder.build(Wakeup.node), ]), ) @@ -114,7 +128,16 @@ export namespace KilocodeBootstrap { LayerNode.make({ service: Service, layer, - deps: [KiloSessions.node, Session.node, SessionSummary.node, Provider.node, memory, Bus.node, watcher], + deps: [ + KiloSessions.node, + Session.node, + SessionSummary.node, + Provider.node, + memory, + Bus.node, + watcher, + Wakeup.node, + ], }), ) } diff --git a/packages/opencode/src/kilocode/session/control.ts b/packages/opencode/src/kilocode/session/control.ts index 951b641b122..1f3024189cc 100644 --- a/packages/opencode/src/kilocode/session/control.ts +++ b/packages/opencode/src/kilocode/session/control.ts @@ -63,6 +63,13 @@ export namespace KiloSessionControl { }), ) - return { begin, stop } + // Whether the session is between a stop and the next resuming prompt. A + // synthetic background prompt does not clear it, so a caller that cannot + // accept a dropped turn (a scheduled wakeup) reads this first. + const paused = Effect.fn("KiloSessionControl.paused")(function* (id: SessionID) { + return (yield* get(id)).paused + }) + + return { begin, stop, paused } }) } diff --git a/packages/opencode/src/kilocode/tool/cancel-wakeup.ts b/packages/opencode/src/kilocode/tool/cancel-wakeup.ts new file mode 100644 index 00000000000..d0742947925 --- /dev/null +++ b/packages/opencode/src/kilocode/tool/cancel-wakeup.ts @@ -0,0 +1,95 @@ +import { Wakeup } from "@/kilocode/wakeup" +import { Tool } from "@/tool/tool" +import { Effect, Schema } from "effect" +import DESCRIPTION from "./cancel-wakeup.txt" + +const Action = Schema.Literals(["list", "cancel"]) +type Action = Schema.Schema.Type + +export const Params = Schema.Struct({ + action: Action.annotate({ description: "Operation to perform" }), + id: Schema.optional(Schema.String).annotate({ + description: "Required for cancel. Id of the pending wakeup to cancel.", + }), +}).check( + Schema.makeFilter((params: { action: Action; id?: string }) => { + if (params.action !== "cancel") return undefined + if (params.id?.trim()) return undefined + return "id is required when action is cancel" + }), +) +export type Params = Schema.Schema.Type + +export type Meta = { + id?: Wakeup.ID + count?: number + cancelled?: boolean +} + +/** Whole-unit countdown to the due time, e.g. `in 5m`. */ +function relative(dueAt: number, now: number) { + const delta = Math.max(0, dueAt - now) + if (delta < 60_000) return `in ${Math.max(1, Math.round(delta / 1_000))}s` + if (delta < 3_600_000) return `in ${Math.round(delta / 60_000)}m` + if (delta < 86_400_000) return `in ${Math.round(delta / 3_600_000)}h` + return `in ${Math.round(delta / 86_400_000)}d` +} + +function excerpt(text: string, max = 80) { + const flat = text.replace(/\s+/g, " ").trim() + return flat.length > max ? `${flat.slice(0, max - 1)}…` : flat +} + +function line(info: Wakeup.Info, now: number) { + const due = new Date(info.dueAt).toISOString() + return `${info.id} due ${due} (${relative(info.dueAt, now)}) ${excerpt(info.reason ?? info.prompt)}` +} + +export const CancelWakeupTool = Tool.define( + "cancel_wakeup", + Effect.gen(function* () { + const wake = yield* Wakeup.Service + return { + description: DESCRIPTION, + parameters: Params, + execute: (params, ctx) => + Effect.gen(function* () { + if (params.action === "list") { + const list = yield* wake.list({ sessionID: ctx.sessionID }) + return { + title: "Scheduled wakeups", + output: list.length + ? list.map((info) => line(info, Date.now())).join("\n") + : "No pending wakeups for this session.", + metadata: { count: list.length }, + } + } + + const id = params.id?.trim() + if (!id) { + return { + title: "Invalid wakeup input", + output: "id is required when action is cancel", + metadata: {}, + } + } + + // Cancel is idempotent: an already-fired, already-cancelled, or + // unknown id is reported, never thrown. + const removed = yield* wake.cancel(id as Wakeup.ID, ctx.sessionID) + if (!removed) { + return { + title: "No pending wakeup", + output: `No pending wakeup with id ${id}.`, + metadata: { id: id as Wakeup.ID }, + } + } + return { + title: "Cancelled wakeup", + output: `Cancelled wakeup ${removed.id} (${new Date(removed.dueAt).toISOString()}).`, + metadata: { id: removed.id, cancelled: true }, + } + }), + } + }), +) diff --git a/packages/opencode/src/kilocode/tool/cancel-wakeup.txt b/packages/opencode/src/kilocode/tool/cancel-wakeup.txt new file mode 100644 index 00000000000..18911d205f7 --- /dev/null +++ b/packages/opencode/src/kilocode/tool/cancel-wakeup.txt @@ -0,0 +1,12 @@ +Review and cancel the wakeups scheduled for this session. + +Use this tool to: +- List the wakeups you scheduled, with their id, due time, and reason (or the scheduled prompt when you gave no reason) +- Cancel a wakeup you no longer need, by its id + +Call it with action "list" first to find the id, then action "cancel" with that id. +Cancelling an id that is already gone is safe: the tool reports it and does not fail. + +Do NOT use this tool: +- As a poll loop while waiting — the harness wakes you when the wakeup fires +- To cancel a wakeup that has already fired — it is gone, and the cancel is a no-op diff --git a/packages/opencode/src/kilocode/tool/registry.ts b/packages/opencode/src/kilocode/tool/registry.ts index 37ff5b992ff..0ee2892c7b5 100644 --- a/packages/opencode/src/kilocode/tool/registry.ts +++ b/packages/opencode/src/kilocode/tool/registry.ts @@ -5,6 +5,7 @@ import { AgentManagerTool } from "./agent-manager" import { BackgroundProcessTool } from "./background-process" import { BoardReadTool, BoardPostTool } from "./board" import { BrowserOpenTool } from "./browser-open" +import { CancelWakeupTool } from "./cancel-wakeup" import { ChartTool } from "./chart" import { GenerateImageTool } from "./generate-image" import { NotebookEditTool, NotebookExecuteTool, NotebookReadTool } from "./notebook-host" @@ -12,6 +13,7 @@ import { MemoryRecallTool } from "./memory-recall" import { MemorySaveTool } from "./memory-save" import { NotifyUserTool } from "./notify-user" import { OpenPlanTool } from "./open-plan" +import { ScheduleWakeupTool } from "./schedule-wakeup" import { SendFileTool } from "./send-file" import * as Tool from "../../tool/tool" import { Flag } from "@opencode-ai/core/flag/flag" @@ -89,6 +91,9 @@ export namespace KiloToolRegistry { const notify = yield* NotifyUserTool.pipe(Effect.provideService(KiloSessions.Service, sessions)) const openPlan = yield* OpenPlanTool const send = yield* SendFileTool + // Wakeup.Service is provided by Wakeup.node in the tool-registry node graph. + const schedule = yield* ScheduleWakeupTool + const cancel = yield* CancelWakeupTool const board = yield* Effect.all({ boardRead: BoardReadTool, boardPost: BoardPostTool, @@ -108,6 +113,8 @@ export namespace KiloToolRegistry { notify, openPlan, send, + schedule, + cancel, ...board, } const tools = yield* Effect.all({ @@ -128,6 +135,8 @@ export namespace KiloToolRegistry { notify, openPlan, send, + schedule, + cancel, ...board, ...tools, } @@ -150,6 +159,8 @@ export namespace KiloToolRegistry { notify: Tool.Info openPlan?: Tool.Info send: Tool.Info + schedule?: Tool.Info + cancel?: Tool.Info boardRead?: Tool.Info goalReport?: Tool.Info boardPost?: Tool.Info @@ -174,6 +185,8 @@ export namespace KiloToolRegistry { send: Tool.init(tools.send), }) const openPlan = tools.openPlan ? yield* Tool.init(tools.openPlan) : undefined + const schedule = tools.schedule ? yield* Tool.init(tools.schedule) : undefined + const cancel = tools.cancel ? yield* Tool.init(tools.cancel) : undefined const report = tools.goalReport ? { goalReport: yield* Tool.init(tools.goalReport) } : {} const board = tools.boardRead && tools.boardPost @@ -197,6 +210,8 @@ export namespace KiloToolRegistry { ...notebooks, semantic, openPlan, + schedule, + cancel, notify: base.notify, send: base.send, } @@ -262,6 +277,8 @@ export namespace KiloToolRegistry { notify: Tool.Def openPlan?: Tool.Def send: Tool.Def + schedule?: Tool.Def + cancel?: Tool.Def boardRead?: Tool.Def goalReport?: Tool.Def boardPost?: Tool.Def @@ -293,6 +310,8 @@ export namespace KiloToolRegistry { tools.recall, ...(Flag.KILO_CLIENT === "vscode" ? [tools.chart] : []), ...(Flag.KILO_CLIENT === "cli" || Flag.KILO_CLIENT === "vscode" ? [tools.process] : []), + ...((Flag.KILO_CLIENT === "cli" || Flag.KILO_CLIENT === "vscode") && tools.schedule ? [tools.schedule] : []), + ...((Flag.KILO_CLIENT === "cli" || Flag.KILO_CLIENT === "vscode") && tools.cancel ? [tools.cancel] : []), ...(Flag.KILO_CLIENT === "vscode" || cfg.experimental?.task_model_selection === true ? [tools.managerModels] : []), diff --git a/packages/opencode/src/kilocode/tool/schedule-wakeup.ts b/packages/opencode/src/kilocode/tool/schedule-wakeup.ts new file mode 100644 index 00000000000..9b94d3dc45c --- /dev/null +++ b/packages/opencode/src/kilocode/tool/schedule-wakeup.ts @@ -0,0 +1,99 @@ +import { Wakeup } from "@/kilocode/wakeup" +import { InstanceState } from "@/effect/instance-state" +import { Tool } from "@/tool/tool" +import { Effect, Schema } from "effect" +import DESCRIPTION from "./schedule-wakeup.txt" + +export const Params = Schema.Struct({ + prompt: Schema.String.annotate({ + description: "Text to resume this session with when the wakeup fires.", + }), + delay: Schema.optional(Schema.String).annotate({ + description: "Relative span from now, e.g. 30s, 5m, 2h, 1d. A bare number is seconds.", + }), + when: Schema.optional(Schema.String).annotate({ + description: "Absolute ISO-8601 date-time, e.g. 2026-09-13T14:30:00Z. An offset makes it absolute; without one the host timezone applies.", + }), + reason: Schema.optional(Schema.String).annotate({ + description: "Short label shown in the sidebar and in cancel_wakeup's list.", + }), +}) +export type Params = Schema.Schema.Type + +export type Meta = { + id?: Wakeup.ID + dueAt?: number + prompt?: string +} + +/** Whole-unit countdown to the due time, e.g. `in 5m`. */ +function relative(dueAt: number, now: number) { + const delta = Math.max(0, dueAt - now) + if (delta < 60_000) return `in ${Math.max(1, Math.round(delta / 1_000))}s` + if (delta < 3_600_000) return `in ${Math.round(delta / 60_000)}m` + if (delta < 86_400_000) return `in ${Math.round(delta / 3_600_000)}h` + return `in ${Math.round(delta / 86_400_000)}d` +} + +function invalid(message: string) { + return { + title: "Invalid wakeup input", + output: message, + metadata: {}, + } +} + +function tooMany() { + return { + title: "Too many scheduled wakeups", + output: `Too many scheduled wakeups: this session already holds the maximum of ${Wakeup.MAX_PER_SESSION} pending wakeups. Cancel one with cancel_wakeup before scheduling another.`, + metadata: {}, + } +} + +function created(info: Wakeup.Info, now: number, input: Params) { + const due = new Date(info.dueAt).toISOString() + // The schedule's own clock is the reference for the clamp check: a fresh + // Date.now() drifts a few milliseconds off resolve()'s base. + const clamped = Wakeup.clampNotice(input, info.dueAt, info.created) + return { + title: `Scheduled wakeup ${info.id}`, + output: [ + `Scheduled wakeup ${info.id}, due ${due} (${relative(info.dueAt, now)}).${clamped ? ` ${clamped}` : ""}`, + `When it fires this session resumes with: ${info.prompt}`, + ].join("\n"), + metadata: { id: info.id, dueAt: info.dueAt, prompt: info.prompt }, + } +} + +export const ScheduleWakeupTool = Tool.define( + "schedule_wakeup", + Effect.gen(function* () { + const wake = yield* Wakeup.Service + return { + description: DESCRIPTION, + parameters: Params, + execute: (params, ctx) => + Effect.gen(function* () { + const inst = yield* InstanceState.context + return yield* wake + .schedule({ + sessionID: ctx.sessionID, + directory: inst.directory, + prompt: params.prompt, + delay: params.delay, + when: params.when, + reason: params.reason, + }) + .pipe( + Effect.map((info) => created(info, Date.now(), params)), + Effect.catchTags({ + "Wakeup.InvalidTime": (err) => Effect.succeed(invalid(err.message)), + "Wakeup.PastTime": (err) => Effect.succeed(invalid(err.message)), + "Wakeup.TooMany": () => Effect.succeed(tooMany()), + }), + ) + }), + } + }), +) diff --git a/packages/opencode/src/kilocode/tool/schedule-wakeup.txt b/packages/opencode/src/kilocode/tool/schedule-wakeup.txt new file mode 100644 index 00000000000..73787af8b2b --- /dev/null +++ b/packages/opencode/src/kilocode/tool/schedule-wakeup.txt @@ -0,0 +1,16 @@ +Schedule a wakeup for yourself: a point in the future when the harness resumes this session with the prompt you give it. + +Use this tool to defer your own continuation to a future time, when you cannot keep the turn open: +- Waiting on a build, deploy, or CI window that finishes outside a blocking command. +- A later check-in on something that changes slowly. +- A reminder to poll a long-running task after a sensible interval. + +Do NOT use this tool for short waits that a blocking shell command covers, for anything inside the next few seconds, for periodic or repeating schedules, or for more than the per-session cap. For a short wait, run the wait as a normal blocking shell command with its `timeout` raised instead. + +Time: +- Give exactly one of `when` or `delay`. +- `when` is an absolute ISO-8601 date-time, e.g. `2026-09-13T14:30:00Z`. With an explicit offset (`Z` or `+02:00`) it is absolute; without one, the host's local timezone applies. +- `delay` is a relative span, e.g. `30s`, `5m`, `2h`, `1d`; a bare number is seconds. +- Clamps: a delay under 10 seconds is raised to the 10-second minimum, a time more than 7 days out is pulled back to the 7-day maximum, and a time at or before now is rejected. A session may hold at most 10 pending wakeups; cancel one with `cancel_wakeup` first when it is full. + +After a successful call, the tool reports the wakeup `id`, its absolute due time, and the relative delay. Keep that id to list or cancel the wakeup later. diff --git a/packages/opencode/src/kilocode/wakeup/index.ts b/packages/opencode/src/kilocode/wakeup/index.ts new file mode 100644 index 00000000000..36251fe260a --- /dev/null +++ b/packages/opencode/src/kilocode/wakeup/index.ts @@ -0,0 +1,191 @@ +import { KiloShutdown } from "@/kilocode/cli/shutdown" +import { SessionID } from "@/session/schema" +import { Storage } from "@/storage/storage" +import { LayerNode } from "@opencode-ai/core/effect/layer-node" +import { Context, Effect, Fiber, Layer, Semaphore } from "effect" +import { fireLayer, text as wakeupText } from "./resume" +import * as schema from "./schema" + +export namespace Wakeup { + export const MIN_DELAY_MS = schema.MIN_DELAY_MS + export const MAX_HORIZON_MS = schema.MAX_HORIZON_MS + export const MAX_PER_SESSION = schema.MAX_PER_SESSION + export const ID = schema.ID + export type ID = schema.ID + export const Info = schema.Info + export type Info = schema.Info + export const Input = schema.Input + export type Input = schema.Input + export const InvalidTime = schema.InvalidTime + export type InvalidTime = schema.InvalidTime + export const PastTime = schema.PastTime + export type PastTime = schema.PastTime + export const TooMany = schema.TooMany + export type TooMany = schema.TooMany + export const Fire = schema.Fire + export type Fire = schema.Fire + export const resolve = schema.resolve + export const clampNotice = schema.clampNotice + export const text = wakeupText + + export interface Interface { + readonly schedule: (input: Input) => Effect.Effect + readonly list: (input?: { sessionID?: SessionID }) => Effect.Effect + readonly cancel: (id: ID, sessionID?: SessionID) => Effect.Effect + readonly adopt: (directory: string) => Effect.Effect + } + + export class Service extends Context.Service()("@kilocode/Wakeup") {} + + const key = (info: { sessionID: SessionID; id: ID }) => ["wakeup", String(info.sessionID), String(info.id)] + + export const layer = Layer.effect( + Service, + Effect.gen(function* () { + const storage = yield* Storage.Service + const fire = yield* Fire + // Timers live in the service scope, so tearing the layer down stops them. + const scope = yield* Effect.scope + const timers = new Map>() + const entries = new Map() + // Ids whose persistence was already dropped and whose resume is in flight. + // `adopt` must not re-fire one of these while the slow turn runs. + const firing = new Set() + // Serializes the count-and-write in `schedule` so two concurrent schedulers + // cannot both pass the cap. + const gate = Semaphore.makeUnsafe(1) + + const stop = () => { + for (const fiber of timers.values()) fiber.interruptUnsafe() + timers.clear() + } + const unregister = KiloShutdown.register(stop) + yield* Effect.addFinalizer(() => + Effect.sync(() => { + unregister() + stop() + }), + ) + + const read = (target: string[]) => + storage.read(target).pipe(Effect.catch(() => Effect.succeed(undefined))) + + const lookup = Effect.fnUntraced(function* (id: ID) { + const known = entries.get(id) + if (known) return known + const keys = yield* storage.list(["wakeup"]).pipe(Effect.catch(() => Effect.succeed([] as string[][]))) + for (const target of keys) { + if (target.at(-1) !== id) continue + const info = yield* read(target) + if (info) return info + } + return undefined + }) + + const fireNow = (info: Info, inPlace = false) => + Effect.gen(function* () { + if (firing.has(info.id)) return + firing.add(info.id) + // The guard release belongs only to the branch that acquired it: an + // early return above must not clear an in-flight fire's guard. + yield* Effect.gen(function* () { + entries.delete(info.id) + timers.delete(info.id) + // Drop the persistence before the resume: the model turn can be slow, + // and a concurrent `adopt` that still sees the file would fire twice. + yield* storage.remove(key(info)).pipe(Effect.ignore) + yield* fire + .run(info, { inPlace }) + .pipe(Effect.catchCause((cause) => Effect.logError("wakeup fire failed", { id: info.id, cause }))) + }).pipe(Effect.ensuring(Effect.sync(() => firing.delete(info.id)))) + }) + + const arm = (info: Info) => + Effect.gen(function* () { + const delay = Math.max(0, info.dueAt - Date.now()) + const fiber = yield* Effect.forkIn( + Effect.sleep(`${delay} millis`).pipe(Effect.andThen(fireNow(info))), + scope, + ) + timers.set(info.id, fiber) + }) + + const list = Effect.fn("Wakeup.list")(function* (input?: { sessionID?: SessionID }) { + const found = new Map(entries) + const prefix = input?.sessionID ? ["wakeup", String(input.sessionID)] : ["wakeup"] + const keys = yield* storage.list(prefix).pipe(Effect.catch(() => Effect.succeed([] as string[][]))) + for (const target of keys) { + const info = yield* read(target) + if (info && !found.has(info.id)) found.set(info.id, info) + } + return Array.from(found.values()) + .filter((info) => !input?.sessionID || info.sessionID === input.sessionID) + .toSorted((a, b) => a.dueAt - b.dueAt || a.id.localeCompare(b.id)) + }) + + const schedule = Effect.fn("Wakeup.schedule")(function* (input: Input) { + return yield* gate.withPermits(1)( + Effect.gen(function* () { + const now = Date.now() + const dueAt = yield* schema.resolve(input, now) + // Count only wakeups that still parse: an unreadable file must not + // hold a slot, and the count and the write must be one critical section. + const pending = yield* list({ sessionID: input.sessionID }) + if (pending.length >= MAX_PER_SESSION) { + return yield* new TooMany({ message: `A session can hold at most ${MAX_PER_SESSION} pending wakeups` }) + } + const info: Info = { + id: ID.ascending(), + sessionID: input.sessionID, + directory: input.directory, + prompt: input.prompt, + reason: input.reason, + agent: input.agent, + dueAt, + created: now, + } + yield* storage.write(key(info), info).pipe(Effect.orDie) + entries.set(info.id, info) + yield* arm(info) + return info + }), + ) + }) + + const cancel = Effect.fn("Wakeup.cancel")(function* (id: ID, sessionID?: SessionID) { + const info = yield* lookup(id) + if (!info || (sessionID && info.sessionID !== sessionID)) return undefined + const fiber = timers.get(id) + if (fiber) { + timers.delete(id) + yield* Fiber.interrupt(fiber) + } + entries.delete(id) + yield* storage.remove(key(info)).pipe(Effect.ignore) + return info + }) + + const adopt = Effect.fn("Wakeup.adopt")(function* (directory: string) { + const keys = yield* storage.list(["wakeup"]).pipe(Effect.catch(() => Effect.succeed([] as string[][]))) + for (const target of keys) { + const info = yield* read(target) + if (!info || info.directory !== directory) continue + if (entries.has(info.id) || timers.has(info.id) || firing.has(info.id)) continue + entries.set(info.id, info) + // Adopt runs inside the directory's bootstrap, so it must resume in + // place; `provide` would await the in-flight load and deadlock. + if (info.dueAt <= Date.now()) yield* fireNow(info, true) + else yield* arm(info) + } + }) + + return Service.of({ schedule, list, cancel, adopt }) + }), + ) + + export const defaultLayer = layer.pipe(Layer.provide(fireLayer)) + + export const node = LayerNode.make({ service: Service, layer: defaultLayer, deps: [Storage.node] }) +} + +export * from "./schema" diff --git a/packages/opencode/src/kilocode/wakeup/resume.ts b/packages/opencode/src/kilocode/wakeup/resume.ts new file mode 100644 index 00000000000..ce021a3d39c --- /dev/null +++ b/packages/opencode/src/kilocode/wakeup/resume.ts @@ -0,0 +1,86 @@ +import { Instance, provide } from "@/kilocode/instance" +import { InstanceRef } from "@/effect/instance-ref" +import * as Log from "@opencode-ai/core/util/log" +import type { InstanceContext } from "@/project/instance-context" +import { Effect, Layer } from "effect" +import { Fire, type Info } from "./schema" + +const log = Log.create({ service: "wakeup" }) + +/** The prompt the model sees when a wakeup fires: the scheduled text plus wakeup context. */ +export function text(info: Info): string { + return `[scheduled wakeup] ${info.prompt}\n\n(No user is present. You scheduled this wakeup yourself as ${info.id}, due ${new Date(info.dueAt).toISOString()}.)` +} + +async function resume(info: Info, inst?: InstanceContext, inPlace = false) { + try { + const [{ AppRuntime }, { Session }, { SessionPrompt }] = await Promise.all([ + import("@/effect/app-runtime"), + import("@/session/session"), + import("@/session/prompt"), + ]) + const fn = async () => { + await AppRuntime.runPromise(Session.Service.use((svc) => svc.get(info.sessionID))) + // The prompt path drops a synthetic turn while the session is paused, so + // the wake would vanish without a trace. Refuse it here and log instead. + const paused = await AppRuntime.runPromise(SessionPrompt.Service.use((svc) => svc.paused(info.sessionID))) + if (paused) { + log.error("wakeup could not resume session", { + id: info.id, + sessionID: info.sessionID, + directory: info.directory, + reason: "session is paused", + }) + return + } + // Fork the turn so the firing timer never blocks on the model running. + await AppRuntime.runPromise( + SessionPrompt.Service.use((svc) => + Effect.forkDetach( + svc + .prompt({ + sessionID: info.sessionID, + agent: info.agent, + parts: [ + { + type: "text", + text: text(info), + synthetic: true, + metadata: { background: true, wakeup: true, wakeupID: info.id }, + }, + ], + }) + .pipe(Effect.catchCause((cause) => Effect.logError("wakeup prompt failed", { id: info.id, cause }))), + ), + ), + ) + } + // An overdue wake fires from `adopt` while its directory's instance is still + // bootstrapping. Re-entering `provide` would await that very load and + // deadlock, so that path resumes in place. A timer fire happens after + // bootstrap, so it re-resolves the instance and picks up a reload. + if (inPlace && inst && inst.directory === info.directory) { + await Instance.restore(inst, fn) + return + } + await provide({ directory: info.directory, fn }) + } catch (err) { + log.error("wakeup could not resume session", { + id: info.id, + sessionID: info.sessionID, + directory: info.directory, + err, + }) + } +} + +export const fireLayer = Layer.succeed( + Fire, + Fire.of({ + run: (info, options) => + Effect.gen(function* () { + const inst = yield* InstanceRef + yield* Effect.promise(() => resume(info, inst, options?.inPlace === true)) + }), + }), +) diff --git a/packages/opencode/src/kilocode/wakeup/schema.ts b/packages/opencode/src/kilocode/wakeup/schema.ts new file mode 100644 index 00000000000..99058a91ab1 --- /dev/null +++ b/packages/opencode/src/kilocode/wakeup/schema.ts @@ -0,0 +1,150 @@ +import { Identifier } from "@/id/id" +import { SessionID } from "@/session/schema" +import { NonNegativeInt, optionalOmitUndefined, withStatics } from "@opencode-ai/core/schema" +import { zod, ZodOverride } from "@opencode-ai/core/effect-zod" +import { Context, Effect, Schema, Types } from "effect" +import z from "zod" + +/** A `delay` under this is raised to it; an absolute `when` is honored as given. */ +export const MIN_DELAY_MS = 10_000 +/** A scheduled wakeup never fires further out than seven days. */ +export const MAX_HORIZON_MS = 7 * 24 * 60 * 60 * 1000 +/** One session may hold at most this many pending wakeups. */ +export const MAX_PER_SESSION = 10 + +const idSchema = Schema.String.annotate({ [ZodOverride]: z.string().startsWith("wku") }).pipe( + Schema.brand("WakeupID"), +) +export type ID = typeof idSchema.Type +export const ID = idSchema.pipe( + withStatics((schema: typeof idSchema) => ({ + ascending: (id?: string) => { + if (id && !id.startsWith("wku")) throw new Error(`Wakeup ID must start with wku: ${id}`) + return schema.make(id ?? Identifier.create("wku", "ascending")) + }, + zod: zod(schema), + })), +) + +export const Info = Schema.Struct({ + id: ID, + sessionID: SessionID, + directory: Schema.String, + prompt: Schema.String, + reason: optionalOmitUndefined(Schema.String), + agent: optionalOmitUndefined(Schema.String), + dueAt: NonNegativeInt, + created: NonNegativeInt, +}) + .annotate({ identifier: "WakeupInfo" }) + .pipe(withStatics((s) => ({ zod: zod(s) }))) +export type Info = Types.DeepMutable> + +export const Input = Schema.Struct({ + sessionID: SessionID, + directory: Schema.String, + prompt: Schema.String, + when: optionalOmitUndefined(Schema.String), + delay: optionalOmitUndefined(Schema.String), + reason: optionalOmitUndefined(Schema.String), + agent: optionalOmitUndefined(Schema.String), +}) + .annotate({ identifier: "WakeupInput" }) + .pipe(withStatics((s) => ({ zod: zod(s) }))) +export type Input = Types.DeepMutable> + +/** Neither a usable `when` nor a usable `delay` was supplied. */ +export class InvalidTime extends Schema.TaggedErrorClass()("Wakeup.InvalidTime", { + message: Schema.String, +}) {} + +/** The requested time is at or before now. */ +export class PastTime extends Schema.TaggedErrorClass()("Wakeup.PastTime", { + message: Schema.String, +}) {} + +/** The session already holds the maximum number of pending wakeups. */ +export class TooMany extends Schema.TaggedErrorClass()("Wakeup.TooMany", { + message: Schema.String, +}) {} + +/** The resume boundary: the service fires through this so tests can stub it. */ +export class Fire extends Context.Service< + Fire, + { readonly run: (info: Info, options?: { inPlace?: boolean }) => Effect.Effect } +>()("@kilocode/WakeupFire") {} + +// ISO-8601 date-time. The offset is optional; when present it is absolute, and +// when omitted `Date.parse` interprets the wall clock in the host timezone. +const WHEN = /^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}(:\d{2}(\.\d{1,9})?)?(Z|[+-]\d{2}:?\d{2})?$/ +// `30`, `30s`, `5m`, `2h`, `1d` — bare numbers are seconds. +const DELAY = /^\s*(\d+)\s*([smhd])?\s*$/i + +function parseWhen(input: string) { + if (!WHEN.test(input)) return undefined + const value = Date.parse(input) + return Number.isFinite(value) ? value : undefined +} + +function parseDelay(input: string) { + const match = DELAY.exec(input) + if (!match) return undefined + const value = Number(match[1]) + if (!Number.isFinite(value)) return undefined + const unit = (match[2] ?? "s").toLowerCase() + const scale = unit === "s" ? 1_000 : unit === "m" ? 60_000 : unit === "h" ? 3_600_000 : 86_400_000 + return value * scale +} + +/** + * Resolve the requested time to an absolute epoch, applying the clamps: + * exactly one of `when`/`delay`; a past time is rejected; a positive delay + * below the minimum clamps up to it; anything beyond the horizon clamps down. + */ +export function resolve( + input: { when?: string; delay?: string }, + now = Date.now(), +): Effect.Effect { + const when = input.when != null && input.when !== "" ? input.when : undefined + const delay = input.delay != null && input.delay !== "" ? input.delay : undefined + if ((when !== undefined) === (delay !== undefined)) { + return Effect.fail(new InvalidTime({ message: "Provide exactly one of when or delay" })) + } + if (when !== undefined) { + const target = parseWhen(when) + if (target === undefined) return Effect.fail(new InvalidTime({ message: `Invalid time: ${when}` })) + if (target <= now) return Effect.fail(new PastTime({ message: `Wakeup time is not in the future: ${when}` })) + return Effect.succeed(Math.min(target, now + MAX_HORIZON_MS)) + } + const span = parseDelay(delay as string) + if (span === undefined) return Effect.fail(new InvalidTime({ message: `Invalid delay: ${delay}` })) + if (span <= 0) return Effect.fail(new PastTime({ message: `Wakeup delay is not in the future: ${delay}` })) + return Effect.succeed(Math.min(Math.max(now + span, now + MIN_DELAY_MS), now + MAX_HORIZON_MS)) +} + +/** + * The clamp that applied to a resolved schedule, as one model-facing sentence, + * or undefined when the request was honored as given. Tools echo it so the + * model knows its requested time was adjusted. + */ +export function clampNotice( + input: { when?: string; delay?: string }, + dueAt: number, + now = Date.now(), +): string | undefined { + const when = input.when != null && input.when !== "" ? input.when : undefined + const delay = input.delay != null && input.delay !== "" ? input.delay : undefined + if (when !== undefined && dueAt - now === MAX_HORIZON_MS && (parseWhen(when) ?? 0) > now + MAX_HORIZON_MS) { + return `Requested when: "${when}" is beyond the 7-day horizon and was pulled back to it.` + } + if (delay === undefined) return undefined + const span = parseDelay(delay) + if (span === undefined) return undefined + if (span < MIN_DELAY_MS && dueAt - now === MIN_DELAY_MS) { + return `Requested delay: "${delay}" is under the 10-second minimum and was raised to it.` + } + if (span > MAX_HORIZON_MS && dueAt - now === MAX_HORIZON_MS) { + return `Requested delay: "${delay}" is beyond the 7-day horizon and was pulled back to it.` + } + return undefined +} diff --git a/packages/opencode/src/session/prompt.ts b/packages/opencode/src/session/prompt.ts index 11ea4f1c782..5ad91cc9d0a 100644 --- a/packages/opencode/src/session/prompt.ts +++ b/packages/opencode/src/session/prompt.ts @@ -146,6 +146,7 @@ function isOrphanedInterruptedTool(part: SessionV1.ToolPart) { export interface Interface { readonly cancel: (sessionID: SessionID, scope?: KiloSessionControl.AbortScope) => Effect.Effect // kilocode_change + readonly paused: (sessionID: SessionID) => Effect.Effect // kilocode_change - wakeup resume refuses a paused session instead of dropping its turn readonly prompt: (input: PromptInput) => Effect.Effect readonly loop: (input: LoopInput) => Effect.Effect readonly shell: (input: ShellInput) => Effect.Effect @@ -2569,6 +2570,7 @@ export const layer = Layer.effect( return Service.of({ cancel, + paused: (id) => control.paused(id), // kilocode_change - wakeup resume reads it before forking a turn prompt, loop: (input) => loop(input).pipe(Effect.orDie), shell, diff --git a/packages/opencode/src/tool/registry.ts b/packages/opencode/src/tool/registry.ts index bd2c9bf94b1..d6dcdb5b148 100644 --- a/packages/opencode/src/tool/registry.ts +++ b/packages/opencode/src/tool/registry.ts @@ -33,6 +33,7 @@ import { WebSearchTool } from "./websearch" import { KiloToolRegistry } from "../kilocode/tool/registry" // kilocode_change import { Notebook } from "@/kilocode/notebook/service" // kilocode_change import { AgentManager } from "@/kilocode/agent-manager/service" // kilocode_change +import { Wakeup } from "@/kilocode/wakeup" // kilocode_change import { SessionDrain } from "@/kilocode/session/drain" // kilocode_change import { RepoOverviewTool } from "@/kilocode/tool/repo-overview" // kilocode_change import { RepoCloneTool } from "./repo_clone" // kilocode_change @@ -541,6 +542,7 @@ export const node = LayerNode.suspend(() => Notebook.node, RepositoryCache.node, KiloSessions.node, + Wakeup.node, // kilocode_change - provides Wakeup.Service to the schedule_wakeup/cancel_wakeup tools ], }), ) diff --git a/packages/opencode/test/kilocode/tool-registry-indexing.test.ts b/packages/opencode/test/kilocode/tool-registry-indexing.test.ts index afe8a0a8667..604b298378d 100644 --- a/packages/opencode/test/kilocode/tool-registry-indexing.test.ts +++ b/packages/opencode/test/kilocode/tool-registry-indexing.test.ts @@ -6,6 +6,7 @@ import { Agent } from "../../src/agent/agent" import { Bus } from "../../src/bus" import { KiloIndexing } from "../../src/kilocode/indexing" import { KilocodeBootstrap } from "../../src/kilocode/bootstrap" +import { Wakeup } from "../../src/kilocode/wakeup" import { KilocodeWatcher } from "../../src/kilocode/watcher" import { KiloSessions } from "../../src/kilo-sessions/kilo-sessions" import { KiloMemory } from "@kilocode/kilo-memory/effect" @@ -519,6 +520,15 @@ describe("kilocode tool registry indexing", () => { const summary = Layer.succeed(SessionSummary.Service, {} as SessionSummary.Interface) const provider = Layer.succeed(Provider.Service, {} as Provider.Interface) const watcher = Layer.succeed(KilocodeWatcher.Service, KilocodeWatcher.Service.of({ init: () => Effect.void })) + const wakeup = Layer.succeed( + Wakeup.Service, + Wakeup.Service.of({ + schedule: () => Effect.die(new Error("wakeup schedule is not used by this test")), + list: () => Effect.succeed([]), + cancel: () => Effect.succeed(undefined), + adopt: () => Effect.void, + }), + ) const indexing = spyOn(KiloIndexing, "init").mockRejectedValue(err) const warn = spyOn(logger, "warn").mockImplementation(() => {}) @@ -526,7 +536,9 @@ describe("kilocode tool registry indexing", () => { await Effect.runPromise( KilocodeBootstrap.Service.use((svc) => svc.init()).pipe( Effect.provide( - KilocodeBootstrap.layer.pipe(Layer.provide([sessions, bus, memory, session, summary, provider, watcher])), + KilocodeBootstrap.layer.pipe( + Layer.provide([sessions, bus, memory, session, summary, provider, watcher, wakeup]), + ), ), Effect.scoped, ), diff --git a/packages/opencode/test/kilocode/tool/cancel-wakeup.test.ts b/packages/opencode/test/kilocode/tool/cancel-wakeup.test.ts new file mode 100644 index 00000000000..34f5e421ffd --- /dev/null +++ b/packages/opencode/test/kilocode/tool/cancel-wakeup.test.ts @@ -0,0 +1,153 @@ +import { describe, expect, test } from "bun:test" +import fs from "fs" +import { rm } from "fs/promises" +import os from "os" +import path from "path" +import { Effect, Layer } from "effect" +import { LayerNode } from "@opencode-ai/core/effect/layer-node" +import { FSUtil } from "@opencode-ai/core/fs-util" +import { Agent } from "@/agent/agent" +import { Git } from "@/git" +import { Wakeup } from "@/kilocode/wakeup" +import { CancelWakeupTool, type Meta, Params } from "@/kilocode/tool/cancel-wakeup" +import { MessageID, SessionID } from "@/session/schema" +import { Storage } from "@/storage/storage" +import * as Truncate from "@/tool/truncate" +import type { Tool } from "@/tool/tool" + +const agentInfo = { + name: "code", + mode: "primary", + options: {}, + permission: {}, +} as Agent.Info + +const agents = Agent.Service.of({ + get: () => Effect.succeed(agentInfo), + list: () => Effect.succeed([agentInfo]), + defaultInfo: () => Effect.succeed(agentInfo), + defaultAgent: () => Effect.succeed("code"), + generate: () => Effect.succeed({ identifier: "code", whenToUse: "", systemPrompt: "" }), +}) + +const truncate = Truncate.Service.of({ + cleanup: () => Effect.void, + write: () => Effect.succeed(""), + output: (text) => Effect.succeed({ content: text as string, truncated: false }), + limits: () => Effect.succeed({ maxLines: Truncate.MAX_LINES, maxBytes: Truncate.MAX_BYTES }), +}) + +const ctx: Tool.Context = { + sessionID: SessionID.make("ses_test"), + messageID: MessageID.make("msg_test"), + callID: "call_test", + agent: "code", + abort: new AbortController().signal, + messages: [], + metadata: () => Effect.void, + ask: () => Effect.void, +} + +const fire = Layer.succeed(Wakeup.Fire, Wakeup.Fire.of({ run: () => Effect.void })) + +function makeLayer(dir: string) { + const storage = Storage.layerFromDir(path.join(dir, "storage")).pipe( + Layer.provide(LayerNode.compile(LayerNode.group([FSUtil.node, Git.node]))), + ) + return Layer.mergeAll( + Wakeup.layer.pipe(Layer.provide(Layer.merge(storage, fire))), + Layer.succeed(Agent.Service, agents), + Layer.succeed(Truncate.Service, truncate), + ) +} + +type ToolDef = Tool.DefWithoutID + +/** Build the tool against a fresh, isolated Wakeup service and run one body. */ +async function run(fn: (tool: ToolDef, wake: Wakeup.Interface, dir: string) => Effect.Effect): Promise { + const dir = fs.mkdtempSync(path.join(os.tmpdir(), "opencode-cancel-wakeup-")) + try { + return await Effect.runPromise( + Effect.gen(function* () { + const wake = yield* Wakeup.Service + const info = yield* CancelWakeupTool + const tool = yield* info.init() + return yield* fn(tool, wake, dir) + }).pipe(Effect.provide(makeLayer(dir))), + ) + } finally { + await rm(dir, { recursive: true, force: true }) + } +} + +const schedule = (wake: Wakeup.Interface, dir: string, delay = "1m") => + wake.schedule({ sessionID: ctx.sessionID, directory: dir, prompt: "check the build", delay }).pipe(Effect.orDie) + +describe("cancel_wakeup tool", () => { + test("describes the list as showing the reason, not the raw prompt", () => + run((tool) => + Effect.gen(function* () { + expect(tool.description).toContain("reason") + }), + )) + + test("lists no wakeups on an empty store", () => + run((tool) => + Effect.gen(function* () { + const result = yield* tool.execute({ action: "list" }, ctx) + + expect(result.title).toBe("Scheduled wakeups") + expect(result.output).toBe("No pending wakeups for this session.") + expect(result.metadata.count).toBe(0) + }), + )) + + test("lists a scheduled wakeup by id, due time, and prompt", () => + run((tool, wake, dir) => + Effect.gen(function* () { + const info = yield* schedule(wake, dir) + + const result = yield* tool.execute({ action: "list" }, ctx) + + expect(result.output).toContain(info.id) + expect(result.output).toContain(new Date(info.dueAt).toISOString()) + expect(result.output).toContain("check the build") + expect(result.metadata.count).toBe(1) + }), + )) + + test("cancels a pending wakeup and reports its id", () => + run((tool, wake, dir) => + Effect.gen(function* () { + const info = yield* schedule(wake, dir) + + const result = yield* tool.execute({ action: "cancel", id: info.id }, ctx) + + expect(result.output).toBe(`Cancelled wakeup ${info.id} (${new Date(info.dueAt).toISOString()}).`) + expect(yield* wake.list({ sessionID: ctx.sessionID })).toEqual([]) + }), + )) + + test("reports an already-gone wakeup without erroring", () => + run((tool, wake, dir) => + Effect.gen(function* () { + const info = yield* schedule(wake, dir) + yield* tool.execute({ action: "cancel", id: info.id }, ctx) + + const again = yield* tool.execute({ action: "cancel", id: info.id }, ctx) + + expect(again.output).toBe(`No pending wakeup with id ${info.id}.`) + expect(again.metadata.cancelled).toBeUndefined() + }), + )) + + test("rejects cancel without an id", async () => { + await expect( + run((tool) => + Effect.gen(function* () { + return yield* tool.execute({ action: "cancel" }, ctx) + }), + ), + ).rejects.toBeDefined() + }) +}) diff --git a/packages/opencode/test/kilocode/tool/schedule-wakeup.test.ts b/packages/opencode/test/kilocode/tool/schedule-wakeup.test.ts new file mode 100644 index 00000000000..9ccfa633587 --- /dev/null +++ b/packages/opencode/test/kilocode/tool/schedule-wakeup.test.ts @@ -0,0 +1,217 @@ +import { describe, expect, test } from "bun:test" +import fs from "fs" +import { rm } from "fs/promises" +import os from "os" +import path from "path" +import { Effect, Layer } from "effect" +import { LayerNode } from "@opencode-ai/core/effect/layer-node" +import { FSUtil } from "@opencode-ai/core/fs-util" +import { Agent } from "@/agent/agent" +import { InstanceRef } from "@/effect/instance-ref" +import { Git } from "@/git" +import { Wakeup } from "@/kilocode/wakeup" +import { ScheduleWakeupTool, type Meta, Params } from "@/kilocode/tool/schedule-wakeup" +import { MessageID, SessionID } from "@/session/schema" +import { Storage } from "@/storage/storage" +import * as Truncate from "@/tool/truncate" +import type { Tool } from "@/tool/tool" + +const agentInfo = { + name: "code", + mode: "primary", + options: {}, + permission: {}, +} as Agent.Info + +const agents = Agent.Service.of({ + get: () => Effect.succeed(agentInfo), + list: () => Effect.succeed([agentInfo]), + defaultInfo: () => Effect.succeed(agentInfo), + defaultAgent: () => Effect.succeed("code"), + generate: () => Effect.succeed({ identifier: "code", whenToUse: "", systemPrompt: "" }), +}) + +const truncate = Truncate.Service.of({ + cleanup: () => Effect.void, + write: () => Effect.succeed(""), + output: (text) => Effect.succeed({ content: text as string, truncated: false }), + limits: () => Effect.succeed({ maxLines: Truncate.MAX_LINES, maxBytes: Truncate.MAX_BYTES }), +}) + +const ctx: Tool.Context = { + sessionID: SessionID.make("ses_test"), + messageID: MessageID.make("msg_test"), + callID: "call_test", + agent: "code", + abort: new AbortController().signal, + messages: [], + metadata: () => Effect.void, + ask: () => Effect.void, +} + +const fire = Layer.succeed(Wakeup.Fire, Wakeup.Fire.of({ run: () => Effect.void })) + +function makeLayer(dir: string) { + const storage = Storage.layerFromDir(path.join(dir, "storage")).pipe( + Layer.provide(LayerNode.compile(LayerNode.group([FSUtil.node, Git.node]))), + ) + return Layer.mergeAll( + Layer.succeed(InstanceRef, { directory: dir, worktree: dir, project: {} as any }), + Wakeup.layer.pipe(Layer.provide(Layer.merge(storage, fire))), + Layer.succeed(Agent.Service, agents), + Layer.succeed(Truncate.Service, truncate), + ) +} + +type ToolDef = Tool.DefWithoutID + +/** The success shape: a scheduled wakeup carries id, dueAt, and prompt. */ +function scheduled(result: { metadata: Meta; output: string }) { + const { id, dueAt, prompt } = result.metadata + if (id === undefined || dueAt === undefined || prompt === undefined) { + throw new Error(`expected a scheduled wakeup, got: ${result.output}`) + } + return { id, dueAt, prompt } +} + +/** Build the tool against a fresh, isolated Wakeup service and run one body. */ +async function run( + fn: (tool: ToolDef, wake: Wakeup.Interface, dir: string, info: Tool.Info) => Effect.Effect, +): Promise { + const dir = fs.mkdtempSync(path.join(os.tmpdir(), "opencode-schedule-wakeup-")) + try { + return await Effect.runPromise( + Effect.gen(function* () { + const wake = yield* Wakeup.Service + const info = yield* ScheduleWakeupTool + const tool = yield* info.init() + return yield* fn(tool, wake, dir, info) + }).pipe(Effect.provide(makeLayer(dir))), + ) + } finally { + await rm(dir, { recursive: true, force: true }) + } +} + +describe("schedule_wakeup tool", () => { + test("registers as schedule_wakeup with when-not-to-use guidance", () => + run((tool, _wake, _dir, info) => + Effect.gen(function* () { + expect(info.id).toBe("schedule_wakeup") + expect(tool.description).toContain( + "Do NOT use this tool for short waits that a blocking shell command covers", + ) + expect(tool.description).toContain("10 seconds") + }), + )) + + test("schedules a wakeup and reports the id, due time, and prompt", () => + run((tool, wake, _dir) => + Effect.gen(function* () { + const before = Date.now() + + const result = yield* tool.execute({ prompt: "check the deploy", delay: "5m", reason: "deploy" }, ctx) + + const { id, dueAt, prompt } = scheduled(result) + expect(id).toMatch(/^wku/) + expect(result.output).toContain(id) + expect(result.output).toContain(new Date(dueAt).toISOString()) + expect(result.output).toContain("check the deploy") + expect(dueAt).toBeGreaterThanOrEqual(before + 5 * 60_000) + expect(prompt).toBe("check the deploy") + + const list = yield* wake.list({ sessionID: ctx.sessionID }) + expect(list).toHaveLength(1) + expect(list[0]?.prompt).toBe("check the deploy") + expect(result.output).not.toContain("Requested") + }), + )) + + test("accepts an absolute when and resolves it to the due time", () => + run((tool, _wake, _dir) => + Effect.gen(function* () { + const when = new Date(Date.now() + 60 * 60_000) + + const result = yield* tool.execute({ prompt: "later", when: when.toISOString() }, ctx) + + const { dueAt } = scheduled(result) + expect(result.title).toMatch(/^Scheduled wakeup wku/) + expect(dueAt).toBeGreaterThanOrEqual(when.getTime() - 2_000) + expect(result.output).toContain(when.toISOString()) + }), + )) + + test("returns an Invalid wakeup input result for a past when", () => + run((tool, wake, _dir) => + Effect.gen(function* () { + const result = yield* tool.execute( + { prompt: "too late", when: new Date(Date.now() - 60_000).toISOString() }, + ctx, + ) + + expect(result.title).toBe("Invalid wakeup input") + expect(result.metadata.id).toBeUndefined() + expect(yield* wake.list({ sessionID: ctx.sessionID })).toEqual([]) + }), + )) + + test("returns an Invalid wakeup input result when neither when nor delay is given", () => + run((tool, _wake, _dir) => + Effect.gen(function* () { + const result = yield* tool.execute({ prompt: "when?" }, ctx) + + expect(result.title).toBe("Invalid wakeup input") + expect(result.output).toContain("exactly one of when or delay") + }), + )) + + test("clamps a sub-minimum delay up to the minimum", () => + run((tool, _wake, _dir) => + Effect.gen(function* () { + const before = Date.now() + + const result = yield* tool.execute({ prompt: "soon", delay: "1s" }, ctx) + + const { dueAt } = scheduled(result) + const ahead = dueAt - before + expect(ahead).toBeGreaterThanOrEqual(Wakeup.MIN_DELAY_MS) + expect(ahead).toBeLessThan(Wakeup.MIN_DELAY_MS + 5_000) + expect(result.output).toContain(`Requested delay: "1s" is under the 10-second minimum`) + }), + )) + + test("clamps a beyond-horizon when down to the horizon", () => + run((tool, _wake, _dir) => + Effect.gen(function* () { + const before = Date.now() + + const result = yield* tool.execute( + { prompt: "far", when: new Date(Date.now() + Wakeup.MAX_HORIZON_MS * 2).toISOString() }, + ctx, + ) + + const { dueAt } = scheduled(result) + expect(dueAt - before).toBeLessThanOrEqual(Wakeup.MAX_HORIZON_MS + 2_000) + expect(result.output).toContain("Requested when:") + expect(result.output).toContain("7-day horizon") + }), + )) + + test("reports the cap when the session already holds the maximum", () => + run((tool, _wake, _dir) => + Effect.gen(function* () { + for (let i = 0; i < Wakeup.MAX_PER_SESSION; i++) { + const result = yield* tool.execute({ prompt: `wakeup ${i}`, delay: "1m" }, ctx) + expect(result.metadata.id).toBeString() + } + + const result = yield* tool.execute({ prompt: "one more", delay: "1m" }, ctx) + + expect(result.title).toBe("Too many scheduled wakeups") + expect(result.output).toContain("Too many scheduled wakeups") + expect(result.output).toContain(`maximum of ${Wakeup.MAX_PER_SESSION} pending wakeups`) + expect(result.output).toContain("cancel_wakeup") + expect(result.metadata.id).toBeUndefined() + }), + )) +}) diff --git a/packages/opencode/test/kilocode/wakeup/wakeup-resume.test.ts b/packages/opencode/test/kilocode/wakeup/wakeup-resume.test.ts new file mode 100644 index 00000000000..fecc705ae11 --- /dev/null +++ b/packages/opencode/test/kilocode/wakeup/wakeup-resume.test.ts @@ -0,0 +1,223 @@ +import { afterAll, describe, expect, test } from "bun:test" +import fs from "fs" +import { rm } from "fs/promises" +import os from "os" +import path from "path" +import { Effect } from "effect" +import * as Log from "@opencode-ai/core/util/log" +import { AppRuntime } from "@/effect/app-runtime" +import { InstanceRef } from "@/effect/instance-ref" +import { Wakeup } from "@/kilocode/wakeup" +import { InstanceStore } from "@/project/instance-store" +import { Session } from "@/session/session" +import { SessionPrompt } from "@/session/prompt" +import { pollWithTimeout } from "../../lib/effect" + +const model = { + name: "Test Model", + tool_call: true, + attachment: true, + modalities: { input: ["text", "image"], output: ["text"] }, + limit: { context: 100000, output: 10000 }, +} + +// The exact `chat.completion.chunk` frame shape the other session tests use. +function line(input: unknown) { + return `data: ${JSON.stringify(input)}\n\n` +} + +function chunk(input: { delta?: Record; finish?: string }) { + return { + id: "chatcmpl-wakeup-resume-test", + object: "chat.completion.chunk", + choices: [ + { + delta: input.delta ?? {}, + ...(input.finish ? { finish_reason: input.finish } : {}), + }, + ], + } +} + +function reply(text: string) { + const enc = new TextEncoder() + return new ReadableStream({ + start(ctrl) { + ctrl.enqueue(enc.encode(line(chunk({ delta: { role: "assistant" } })))) + ctrl.enqueue(enc.encode(line(chunk({ delta: { content: text } })))) + ctrl.enqueue(enc.encode(line(chunk({ finish: "stop" })))) + ctrl.enqueue(enc.encode("data: [DONE]\n\n")) + ctrl.close() + }, + }) +} + +// The runtime holds the wakeup timer's scope; dispose it once for the file. +afterAll(async () => { + await AppRuntime.dispose() +}) + +function config(baseURL: string) { + return JSON.stringify({ + model: "test/test-model", + small_model: "test/test-model", + enabled_providers: ["test"], + formatter: false, + lsp: false, + provider: { + test: { + name: "Test", + npm: "@ai-sdk/openai-compatible", + options: { apiKey: "test-key", baseURL }, + models: { "test-model": model }, + }, + }, + }) +} + +describe("wakeup resume", () => { + test("an armed wakeup fires, resumes the session with its prompt, and clears the entry", async () => { + const bodies: string[] = [] + const server = Bun.serve({ + port: 0, + async fetch(req) { + const url = new URL(req.url) + if (!url.pathname.endsWith("/chat/completions")) return new Response("not found", { status: 404 }) + bodies.push(await req.text()) + return new Response(reply("woke up"), { + status: 200, + headers: { "Content-Type": "text/event-stream" }, + }) + }, + }) + + const base = fs.realpathSync(os.tmpdir()) + const dir = fs.mkdtempSync(path.join(base, "opencode-wakeup-resume-")) + try { + await Bun.write(path.join(dir, "opencode.json"), config(`${server.url.origin}/v1`)) + + const ctx = await AppRuntime.runPromise(InstanceStore.Service.use((store) => store.load({ directory: dir }))) + const session = await AppRuntime.runPromise( + Session.Service.use((svc) => svc.create({ title: "Wakeup resume" })).pipe( + Effect.provideService(InstanceRef, ctx), + ), + ) + + const info = await AppRuntime.runPromise( + Wakeup.Service.use((wake) => + wake.schedule({ + sessionID: session.id, + directory: dir, + prompt: "poll the deploy", + when: new Date(Date.now() + 1200).toISOString(), + }), + ).pipe(Effect.provideService(InstanceRef, ctx)), + ) + + await Effect.runPromise( + pollWithTimeout( + Effect.sync(() => + bodies.some((body) => body.includes("[scheduled wakeup]") && body.includes("poll the deploy")) + ? true + : undefined, + ), + "the wakeup prompt never reached the model", + "8 seconds", + ), + ) + + const pending = await AppRuntime.runPromise( + Wakeup.Service.use((wake) => wake.list({ sessionID: session.id })).pipe( + Effect.provideService(InstanceRef, ctx), + ), + ) + expect(pending.map((item) => item.id)).not.toContain(info.id) + } finally { + await server.stop(true) + await rm(dir, { recursive: true, force: true }) + } + }, 30_000) + + test("a paused session logs the wakeup as unresumable instead of dropping it silently", async () => { + const bodies: string[] = [] + const server = Bun.serve({ + port: 0, + async fetch(req) { + const url = new URL(req.url) + if (!url.pathname.endsWith("/chat/completions")) return new Response("not found", { status: 404 }) + bodies.push(await req.text()) + return new Response(reply("woke up"), { + status: 200, + headers: { "Content-Type": "text/event-stream" }, + }) + }, + }) + + const base = fs.realpathSync(os.tmpdir()) + const dir = fs.mkdtempSync(path.join(base, "opencode-wakeup-paused-")) + try { + await Bun.write(path.join(dir, "opencode.json"), config(`${server.url.origin}/v1`)) + + const ctx = await AppRuntime.runPromise(InstanceStore.Service.use((store) => store.load({ directory: dir }))) + const session = await AppRuntime.runPromise( + Session.Service.use((svc) => svc.create({ title: "Wakeup paused" })).pipe( + Effect.provideService(InstanceRef, ctx), + ), + ) + + // Abort the idle session through the same pause path the UI uses. + await AppRuntime.runPromise( + SessionPrompt.Service.use((svc) => svc.cancel(session.id)).pipe(Effect.provideService(InstanceRef, ctx)), + ) + const paused = await AppRuntime.runPromise( + SessionPrompt.Service.use((svc) => svc.paused(session.id)).pipe(Effect.provideService(InstanceRef, ctx)), + ) + expect(paused).toBe(true) + + // The wakeup logger is a cached `Log.create` object, so patch the same + // instance resume.ts holds; stderr is not reliable once another test + // redirects the log stream to a file. + const wakeLog = Log.create({ service: "wakeup" }) + const errors: Array<{ message?: unknown; extra?: Record }> = [] + const originalLog = wakeLog.error.bind(wakeLog) + wakeLog.error = ((message?: unknown, extra?: Record) => { + errors.push({ message, extra }) + }) as typeof wakeLog.error + try { + await AppRuntime.runPromise( + Wakeup.Service.use((wake) => + wake.schedule({ + sessionID: session.id, + directory: dir, + prompt: "should be refused", + when: new Date(Date.now() + 1200).toISOString(), + }), + ).pipe(Effect.provideService(InstanceRef, ctx)), + ) + + await Effect.runPromise( + pollWithTimeout( + Effect.sync(() => + errors.some( + (entry) => + entry.message === "wakeup could not resume session" && entry.extra?.reason === "session is paused", + ) + ? true + : undefined, + ), + "the paused wakeup was dropped without an error log", + "8 seconds", + ), + ) + } finally { + wakeLog.error = originalLog + } + + // The wake never reached the model. + expect(bodies.some((body) => body.includes("[scheduled wakeup]"))).toBe(false) + } finally { + await server.stop(true) + await rm(dir, { recursive: true, force: true }) + } + }, 30_000) +}) diff --git a/packages/opencode/test/kilocode/wakeup/wakeup.test.ts b/packages/opencode/test/kilocode/wakeup/wakeup.test.ts new file mode 100644 index 00000000000..51dd4e7d146 --- /dev/null +++ b/packages/opencode/test/kilocode/wakeup/wakeup.test.ts @@ -0,0 +1,505 @@ +import { describe, expect } from "bun:test" +import fs from "fs" +import { rm } from "fs/promises" +import os from "os" +import path from "path" +import { Context, Effect, Exit, Layer } from "effect" +import { LayerNode } from "@opencode-ai/core/effect/layer-node" +import { FSUtil } from "@opencode-ai/core/fs-util" +import { Git } from "@/git" +import { Wakeup } from "@/kilocode/wakeup" +import { SessionID } from "@/session/schema" +import { Storage } from "@/storage/storage" +import { pollWithTimeout, testEffect } from "../../lib/effect" + +type FireMode = { inPlace?: boolean } | undefined + +const Recorder = Context.Service<{ + calls: Wakeup.Info[] + modes: FireMode[] + reenter: Effect.Effect +}>("@test/WakeupRecorder") +const TestDir = Context.Service<{ dir: string }>("@test/WakeupDir") + +const storageLayer = (dir: string) => + Storage.layerFromDir(path.join(dir, "storage")).pipe( + Layer.provide(LayerNode.compile(LayerNode.group([FSUtil.node, Git.node]))), + ) + +const fireLayer = (calls: Wakeup.Info[]) => + Layer.succeed( + Wakeup.Fire, + Wakeup.Fire.of({ + run: (info) => + Effect.sync(() => { + calls.push(info) + }), + }), + ) + +const recorderFire = Layer.effect( + Wakeup.Fire, + Effect.gen(function* () { + const recorder = yield* Recorder + return Wakeup.Fire.of({ + run: (info, options) => + Effect.gen(function* () { + // Lets a test re-enter the service while a fire is in flight. + yield* recorder.reenter + recorder.calls.push(info) + recorder.modes.push(options) + }), + }) + }), +) + +// Layer.fresh: without it Effect's in-test layer cache hands nested builds the +// outer test's storage and Fire, so a "restart" would share the first process. +const serviceLayer = (dir: string, fire: Layer.Layer) => + Layer.fresh(Wakeup.layer.pipe(Layer.provide(Layer.merge(storageLayer(dir), fire)))) + +const dirLayer = Layer.effect( + TestDir, + Effect.acquireRelease( + Effect.sync(() => ({ dir: fs.mkdtempSync(path.join(os.tmpdir(), "opencode-wakeup-")) })), + ({ dir }) => + Effect.promise(() => + rm(dir, { recursive: true, force: true }).catch(() => { + // best effort cleanup of a temp directory + }), + ), + ), +) + +const wakeupLayer = Layer.unwrap( + Effect.gen(function* () { + const { dir } = yield* TestDir + const recorder = Layer.effect( + Recorder, + Effect.sync(() => ({ calls: [] as Wakeup.Info[], modes: [] as FireMode[], reenter: Effect.void })), + ) + return Layer.provideMerge(serviceLayer(dir, recorderFire), recorder) + }), +) + +const it = testEffect(Layer.provideMerge(wakeupLayer, dirLayer)) + +const session = () => SessionID.descending() + +function info(over: Partial = {}): Wakeup.Info { + const now = Date.now() + return { + id: Wakeup.ID.ascending(), + sessionID: session(), + directory: "/tmp/example", + prompt: "persisted", + dueAt: now + 60_000, + created: now, + ...over, + } +} + +/** Write a wakeup straight to the file-backed store, bypassing schedule(). */ +function persist(dir: string, value: Wakeup.Info) { + const file = path.join(dir, "storage", "wakeup", String(value.sessionID), `${value.id}.json`) + fs.mkdirSync(path.dirname(file), { recursive: true }) + fs.writeFileSync(file, JSON.stringify(value)) +} + +/** A file that `read` cannot parse; it must not hold a wakeup slot. */ +function corrupt(dir: string, sessionID: SessionID) { + const file = path.join(dir, "storage", "wakeup", String(sessionID), "wku_corrupt.json") + fs.mkdirSync(path.dirname(file), { recursive: true }) + fs.writeFileSync(file, "{ not json") +} + +describe("Wakeup", () => { + it.effect("schedules and lists a wakeup", () => + Effect.gen(function* () { + const wake = yield* Wakeup.Service + const dir = (yield* TestDir).dir + const sessionID = session() + + const info = yield* wake.schedule({ sessionID, directory: dir, prompt: "check the build", delay: "1m" }) + const list = yield* wake.list({ sessionID }) + + expect(list.map((item) => item.id)).toEqual([info.id]) + expect(list[0]?.prompt).toBe("check the build") + expect(list[0]?.dueAt).toBeGreaterThan(info.created) + }), + ) + + it.effect("cancels a pending wakeup and is idempotent", () => + Effect.gen(function* () { + const wake = yield* Wakeup.Service + const dir = (yield* TestDir).dir + const sessionID = session() + + const info = yield* wake.schedule({ sessionID, directory: dir, prompt: "later", delay: "1m" }) + const removed = yield* wake.cancel(info.id) + + expect(removed?.id).toBe(info.id) + expect(yield* wake.list({ sessionID })).toEqual([]) + expect(yield* wake.cancel(info.id)).toBeUndefined() + }), + ) + + it.effect("rejects a wakeup in the past", () => + Effect.gen(function* () { + const wake = yield* Wakeup.Service + const dir = (yield* TestDir).dir + + const err = yield* Effect.flip( + wake.schedule({ + sessionID: session(), + directory: dir, + prompt: "nope", + when: new Date(Date.now() - 1_000).toISOString(), + }), + ) + + expect(err).toBeInstanceOf(Wakeup.PastTime) + }), + ) + + it.effect("requires exactly one of when or delay", () => + Effect.gen(function* () { + const wake = yield* Wakeup.Service + const dir = (yield* TestDir).dir + + const missing = yield* Effect.flip(wake.schedule({ sessionID: session(), directory: dir, prompt: "nope" })) + expect(missing).toBeInstanceOf(Wakeup.InvalidTime) + + const both = yield* Effect.flip( + wake.schedule({ + sessionID: session(), + directory: dir, + prompt: "nope", + when: new Date(Date.now() + 60_000).toISOString(), + delay: "1m", + }), + ) + expect(both).toBeInstanceOf(Wakeup.InvalidTime) + + const malformed = yield* Effect.flip( + wake.schedule({ sessionID: session(), directory: dir, prompt: "nope", delay: "soon" }), + ) + expect(malformed).toBeInstanceOf(Wakeup.InvalidTime) + }), + ) + + it.effect("clamps a sub-minimum delay up to the minimum", () => + Effect.gen(function* () { + const wake = yield* Wakeup.Service + const dir = (yield* TestDir).dir + + const info = yield* wake.schedule({ sessionID: session(), directory: dir, prompt: "soon", delay: "1s" }) + expect(info.dueAt - info.created).toBe(Wakeup.MIN_DELAY_MS) + }), + ) + + it.effect("clamps a wakeup beyond the horizon down to the horizon", () => + Effect.gen(function* () { + const wake = yield* Wakeup.Service + const dir = (yield* TestDir).dir + + const info = yield* wake.schedule({ + sessionID: session(), + directory: dir, + prompt: "far", + when: new Date(Date.now() + Wakeup.MAX_HORIZON_MS * 2).toISOString(), + }) + expect(info.dueAt - info.created).toBe(Wakeup.MAX_HORIZON_MS) + }), + ) + + it.effect("accepts ten pending wakeups and rejects the eleventh", () => + Effect.gen(function* () { + const wake = yield* Wakeup.Service + const dir = (yield* TestDir).dir + const sessionID = session() + + for (let index = 0; index < Wakeup.MAX_PER_SESSION; index++) { + yield* wake.schedule({ sessionID, directory: dir, prompt: `wake ${index}`, delay: "1m" }) + } + expect(yield* wake.list({ sessionID })).toHaveLength(Wakeup.MAX_PER_SESSION) + + const err = yield* Effect.flip(wake.schedule({ sessionID, directory: dir, prompt: "overflow", delay: "1m" })) + expect(err).toBeInstanceOf(Wakeup.TooMany) + }), + ) + + it.effect("lists a persisted wakeup that was never scheduled in this process", () => + Effect.gen(function* () { + const wake = yield* Wakeup.Service + const dir = (yield* TestDir).dir + const persisted = info({ directory: dir }) + persist(dir, persisted) + + const list = yield* wake.list({ sessionID: persisted.sessionID }) + + expect(list.map((item) => item.id)).toEqual([persisted.id]) + }), + ) + + it.effect("scopes cancel to the caller's session", () => + Effect.gen(function* () { + const wake = yield* Wakeup.Service + const dir = (yield* TestDir).dir + const owner = session() + const other = session() + + const created = yield* wake.schedule({ sessionID: owner, directory: dir, prompt: "mine", delay: "1m" }) + + expect(yield* wake.cancel(created.id, other)).toBeUndefined() + expect((yield* wake.list({ sessionID: owner })).map((item) => item.id)).toEqual([created.id]) + expect((yield* wake.cancel(created.id, owner))?.id).toBe(created.id) + }), + ) + + it.effect("does not let a corrupt persisted file consume a slot", () => + Effect.gen(function* () { + const wake = yield* Wakeup.Service + const dir = (yield* TestDir).dir + const sessionID = session() + corrupt(dir, sessionID) + + for (let index = 0; index < Wakeup.MAX_PER_SESSION; index++) { + yield* wake.schedule({ sessionID, directory: dir, prompt: `wake ${index}`, delay: "1m" }) + } + + expect(yield* wake.list({ sessionID })).toHaveLength(Wakeup.MAX_PER_SESSION) + const err = yield* Effect.flip(wake.schedule({ sessionID, directory: dir, prompt: "overflow", delay: "1m" })) + expect(err).toBeInstanceOf(Wakeup.TooMany) + }), + ) + + it.effect("enforces the cap under concurrent scheduling", () => + Effect.gen(function* () { + const wake = yield* Wakeup.Service + const dir = (yield* TestDir).dir + const sessionID = session() + + const results = yield* Effect.forEach( + Array.from({ length: Wakeup.MAX_PER_SESSION + 5 }), + (_, index) => Effect.exit(wake.schedule({ sessionID, directory: dir, prompt: `wake ${index}`, delay: "1m" })), + { concurrency: "unbounded" }, + ) + + expect(results.filter((exit) => Exit.isSuccess(exit))).toHaveLength(Wakeup.MAX_PER_SESSION) + expect(yield* wake.list({ sessionID })).toHaveLength(Wakeup.MAX_PER_SESSION) + }), + ) + + it.effect("does not fire a wakeup twice when adopt runs during its fire", () => + Effect.gen(function* () { + const wake = yield* Wakeup.Service + const recorder = yield* Recorder + const dir = (yield* TestDir).dir + const persisted = info({ directory: dir, dueAt: Date.now() - 1_000, created: Date.now() - 2_000 }) + + recorder.reenter = Effect.suspend(() => wake.adopt(dir)) + persist(dir, persisted) + + yield* wake.adopt(dir) + + expect(recorder.calls.map((item) => item.id)).toEqual([persisted.id]) + expect(recorder.modes).toEqual([{ inPlace: true }]) + }), + ) + + it.effect("keeps the in-flight guard while the persisted wakeup is visible again", () => + Effect.gen(function* () { + const wake = yield* Wakeup.Service + const recorder = yield* Recorder + const dir = (yield* TestDir).dir + const persisted = info({ directory: dir, dueAt: Date.now() - 1_000, created: Date.now() - 2_000 }) + + // Re-create the record while the first fire is resuming, then re-enter + // adopt: the in-flight guard, not the removed file, must stop a re-fire. + recorder.reenter = Effect.suspend(() => { + persist(dir, persisted) + return wake.adopt(dir) + }) + persist(dir, persisted) + + yield* wake.adopt(dir) + + expect(recorder.calls.map((item) => item.id)).toEqual([persisted.id]) + expect(recorder.modes).toEqual([{ inPlace: true }]) + }), + ) + + it.effect("releases the guard once a fire completes", () => + Effect.gen(function* () { + const wake = yield* Wakeup.Service + const recorder = yield* Recorder + const dir = (yield* TestDir).dir + const persisted = info({ directory: dir, dueAt: Date.now() - 1_000, created: Date.now() - 2_000 }) + + persist(dir, persisted) + yield* wake.adopt(dir) + expect(recorder.calls.map((item) => item.id)).toEqual([persisted.id]) + + // The same id persisted again must fire: a completed fire released its guard. + persist(dir, persisted) + yield* wake.adopt(dir) + expect(recorder.calls.map((item) => item.id)).toEqual([persisted.id, persisted.id]) + }), + ) + + it.effect("accepts a when sooner than the delay minimum", () => + Effect.gen(function* () { + const now = Date.now() + expect(yield* Wakeup.resolve({ when: new Date(now + 1_000).toISOString() }, now)).toBe(now + 1_000) + expect(yield* Wakeup.resolve({ delay: "1s" }, now)).toBe(now + Wakeup.MIN_DELAY_MS) + }), + ) + + it.effect("describes the wake with the scheduled prompt and wake id", () => + Effect.gen(function* () { + const info: Wakeup.Info = { + id: Wakeup.ID.ascending(), + sessionID: session(), + directory: "/tmp/example", + prompt: "inspect the release", + dueAt: Date.now() + 60_000, + created: Date.now(), + } + + const text = Wakeup.text(info) + expect(text).toContain("inspect the release") + expect(text).toContain(info.id) + }), + ) + + it.effect("describes only the clamp that actually applied", () => + Effect.gen(function* () { + const now = Date.now() + + expect(Wakeup.clampNotice({ delay: "1s" }, now + Wakeup.MIN_DELAY_MS, now)).toBe( + `Requested delay: "1s" is under the 10-second minimum and was raised to it.`, + ) + expect(Wakeup.clampNotice({ delay: "10s" }, now + Wakeup.MIN_DELAY_MS, now)).toBeUndefined() + expect(Wakeup.clampNotice({ delay: "30d" }, now + Wakeup.MAX_HORIZON_MS, now)).toContain("7-day horizon") + expect( + Wakeup.clampNotice( + { when: new Date(now + Wakeup.MAX_HORIZON_MS * 2).toISOString() }, + now + Wakeup.MAX_HORIZON_MS, + now, + ), + ).toContain("Requested when:") + expect(Wakeup.clampNotice({ delay: "1h" }, now + 3_600_000, now)).toBeUndefined() + expect(Wakeup.clampNotice({ when: new Date(now + 3_600_000).toISOString() }, now + 3_600_000, now)).toBeUndefined() + }), + ) + + it.live( + "fires a persisted wakeup exactly once after a restart", + () => + Effect.gen(function* () { + const dir = (yield* TestDir).dir + const sessionID = session() + const first: Wakeup.Info[] = [] + const second: Wakeup.Info[] = [] + + yield* Effect.scoped( + Effect.gen(function* () { + const ctx = yield* Layer.build(serviceLayer(dir, fireLayer(first))) + const wake = Context.get(ctx, Wakeup.Service) + yield* wake.schedule({ sessionID, directory: dir, prompt: "resume the task", delay: "10s" }) + }), + ) + + // Let the stored due time pass after the first process released the wakeup. + yield* Effect.sleep("10500 millis") + + yield* Effect.scoped( + Effect.gen(function* () { + const ctx = yield* Layer.build(serviceLayer(dir, fireLayer(second))) + const wake = Context.get(ctx, Wakeup.Service) + yield* wake.adopt(dir) + yield* wake.adopt(dir) + expect(yield* wake.list({ sessionID })).toEqual([]) + }), + ) + + expect(first).toEqual([]) + expect(second.map((info) => info.prompt)).toEqual(["resume the task"]) + }), + 20_000, + ) + + it.live( + "fires an armed wakeup at its due time with the scheduled prompt", + () => + Effect.gen(function* () { + const wake = yield* Wakeup.Service + const recorder = yield* Recorder + const dir = (yield* TestDir).dir + + yield* wake.schedule({ + sessionID: session(), + directory: dir, + prompt: "poll the deploy", + when: new Date(Date.now() + 1200).toISOString(), + }) + + const fired = yield* pollWithTimeout( + Effect.sync(() => recorder.calls[0]), + "armed wakeup never fired", + "8 seconds", + ) + expect(fired.prompt).toBe("poll the deploy") + // A timer fire re-resolves the instance through provide, not in place. + expect(recorder.modes[0]).toEqual({ inPlace: false }) + }), + 20_000, + ) + + it.live( + "never fires a wakeup cancelled before its due time", + () => + Effect.gen(function* () { + const wake = yield* Wakeup.Service + const recorder = yield* Recorder + const dir = (yield* TestDir).dir + + const info = yield* wake.schedule({ + sessionID: session(), + directory: dir, + prompt: "should not fire", + when: new Date(Date.now() + 1200).toISOString(), + }) + yield* wake.cancel(info.id) + + // The sleep is the assertion: it spans the due time the cancelled + // timer would have fired at. + yield* Effect.sleep("1800 millis") + expect(recorder.calls).toEqual([]) + }), + 20_000, + ) + + it.live( + "fires two wakeups due at the same instant", + () => + Effect.gen(function* () { + const wake = yield* Wakeup.Service + const recorder = yield* Recorder + const dir = (yield* TestDir).dir + const when = new Date(Date.now() + 1200).toISOString() + + yield* wake.schedule({ sessionID: session(), directory: dir, prompt: "first wake", when }) + yield* wake.schedule({ sessionID: session(), directory: dir, prompt: "second wake", when }) + + yield* pollWithTimeout( + Effect.sync(() => (recorder.calls.length >= 2 ? recorder.calls : undefined)), + "both wakeups never fired", + "8 seconds", + ) + expect(recorder.calls.map((info) => info.prompt).toSorted()).toEqual(["first wake", "second wake"]) + }), + 20_000, + ) +}) diff --git a/packages/opencode/test/tool/registry.test.ts b/packages/opencode/test/tool/registry.test.ts index e65461e93cf..cb40012cc42 100644 --- a/packages/opencode/test/tool/registry.test.ts +++ b/packages/opencode/test/tool/registry.test.ts @@ -251,6 +251,18 @@ describe("tool.registry", () => { }), ) + // kilocode_change start - the CLI can schedule and cancel its own future wakeups + it.instance("exposes the scheduled wakeup tools", () => + Effect.gen(function* () { + const registry = yield* ToolRegistry.Service + const ids = yield* registry.ids() + + expect(ids).toContain("schedule_wakeup") + expect(ids).toContain("cancel_wakeup") + }), + ) + // kilocode_change end + it.instance("does not expose execute unless code mode is enabled", () => Effect.gen(function* () { const registry = yield* ToolRegistry.Service diff --git a/script/check-opencode-promise-facades.ts b/script/check-opencode-promise-facades.ts index ba889a40dfe..619a52f02f8 100644 --- a/script/check-opencode-promise-facades.ts +++ b/script/check-opencode-promise-facades.ts @@ -79,6 +79,19 @@ const testAllow: Record = { "server/experimental-session-list.test.ts": { count: 2, reason: "Kilo session list integration test" }, "kilocode/server/cloud-session-import.test.ts": { count: 5, reason: "full app cloud import transaction integration" }, "kilocode/server/listener-runtime.test.ts": { count: 4, reason: "listener and AppRuntime integration test" }, + "kilocode/wakeup/wakeup-resume.test.ts": { + count: 11, + reason: + "the wakeup resume integration test schedules through the production Wakeup service and asserts the mock " + + "model receives the scheduled prompt, so it must run the production Fire/resume path " + + "(src/kilocode/wakeup/resume.ts). That path resolves Session and SessionPrompt from the global AppRuntime " + + "because a static layer dependency is impossible: Wakeup.node <- kilocode/tool/registry.ts (via " + + "schedule_wakeup/cancel_wakeup) <- SessionPrompt.node <- ToolRegistry.node, which already depends on Wakeup.node. " + + "The test therefore creates the instance, session, and wakeup through that same global runtime and asserts the " + + "pending list on it; scoped layers cannot express the boundary under test. The paused-session case pauses the " + + "session and reads SessionPrompt.paused through the same runtime to prove resume refuses and logs instead of " + + "dropping the wake.", + }, "tool/recall.test.ts": { count: 11, reason: "existing runtime integration test" }, }