diff --git a/README.md b/README.md index 32494ba..a32eb80 100644 --- a/README.md +++ b/README.md @@ -13,6 +13,7 @@ The OpenCode Loop Plugin adds: - `/loop ` and `/loop ` (dynamic pacing) as an OpenCode command for TUI, desktop, and web. - A server-side scheduler with per-loop timers that injects a synthetic iteration prompt only when the session is idle, with busy backoff. - Dynamic loops where the agent itself picks the delay before each next iteration via `schedule_next_run`, mirroring Claude Code's self-paced `/loop`. +- Keep-alive loops: `create_loop({ keep_alive: true })` keeps a background or subagent session scheduled across turns and exempts it from age expiry. - Agent tools: `create_loop`, `list_loops`, `stop_loop`, `pause_loop`, `resume_loop`, `run_loop`, `schedule_next_run`, and `clear_loops`. - Persistent loop state that survives OpenCode restarts, with atomic writes and owner-only file permissions. - A TUI sidebar with live countdowns and a `Loops` command-palette entry to run, pause, resume, or stop loops. @@ -120,6 +121,26 @@ After creating a loop, the agent immediately performs the first iteration in the A dynamic loop mirrors Claude Code's self-paced `/loop`: at the end of each iteration the agent calls `schedule_next_run` with a delay in seconds and a one-sentence reason ("watching CI run"), or calls `stop_loop` to end the loop. If an iteration ends without doing either, the loop ends — exactly like omitting `ScheduleWakeup` in Claude Code. +### Keeping a background session alive + +A subagent's session normally has no future of its own: the turn that spawned it ends, the parent detaches, and nothing re-enters that session again. `keep_alive` gives it one. + +``` +create_loop( + instruction: "Check whether the migration finished. If it has, report the result and call stop_loop.", + interval: "5m", + keep_alive: true +) +``` + +The session keeps being re-prompted on the cadence and is exempt from age expiry. After an OpenCode restart, the plugin verifies that the session is accessible in the current host location before acquiring its ownership lease and restoring the timer. Sessions in another location or unavailable to this host remain dormant; ordinary loops still require an authoritative session observation. + +Explicit `max_runs`, session deletion, and manual pause/stop still apply. Failed injections retain bounded backoff and pause after `max_consecutive_failures`; inspect the loop and use `resume_loop` after resolving the failure. Call `stop_loop` when the work resolves or is abandoned. + +`keep_alive` requires a fixed interval. Dynamic pacing hands the next run to the agent every turn, so a single missed `schedule_next_run` would end the very future the flag exists to hold open; that combination is rejected when the loop is created. + +Handing a result back to whoever asked for the work stays the agent's job. This flag only keeps the session running long enough to do it. + ### How iterations are scheduled - Iterations only run while the session is idle. If a loop comes due while the session is busy, it is deferred with a short backoff and retried when the session goes idle. diff --git a/dist/server.js b/dist/server.js index 9659473..3b38466 100644 --- a/dist/server.js +++ b/dist/server.js @@ -29,6 +29,7 @@ var LoopSchema = Schema.Struct({ sessionID: Schema.String, prompt: Schema.String, mode: Schema.optionalWith(Schema.Literal("interval", "dynamic"), { default: () => "interval" }), + keepAlive: Schema.optionalWith(Schema.Boolean, { default: () => false }), intervalMs: NullableNumber, status: Schema.Literal("active", "paused", "stopped", "completed"), createdAt: Schema.Number, @@ -298,6 +299,9 @@ function requireLoop(state, loopID) { async function createLoop(sessionID, options) { const prompt = validatePrompt(options.prompt); const mode = options.mode === "dynamic" ? "dynamic" : "interval"; + const keepAlive = options.keepAlive === true; + if (keepAlive && mode !== "interval") + throw new Error("keep-alive loops require a fixed interval"); const intervalMs = mode === "interval" ? positiveIntegerOrNull(options.intervalMs) : null; if (mode === "interval" && intervalMs == null) throw new Error("interval loops require a positive interval"); @@ -318,6 +322,7 @@ async function createLoop(sessionID, options) { sessionID, prompt, mode, + keepAlive, intervalMs, status: "active", createdAt: timestamp, @@ -740,6 +745,7 @@ Preserve each loop's id, cadence, instruction, and status in the compacted conte } // src/server.ts +var KEEP_ALIVE_DESCRIPTION = "Keep this session scheduled across turns and exempt from age expiry. Requires a fixed interval. Explicit max_runs, session deletion, and the failure-pause threshold still apply. Use it in a background or subagent session whose work outlives the turn that started it: the scheduler keeps re-prompting the session on the cadence, and the loop is never ended automatically by age expiry. You remain responsible for calling stop_loop once the work is resolved or abandoned."; var DEFAULT_COMMAND_NAME = "loop"; var DEFAULT_BUSY_BACKOFF_SECONDS = 60; var DEFAULT_FAILURE_BACKOFF_SECONDS = 60; @@ -818,7 +824,7 @@ async function toolResult(sessionID, extra = {}) { const loops = await listLoops(sessionID); return JSON.stringify({ ...extra, loops, report: formatLoops(loops) }, null, 2); } -var server = async ({ client }, options) => { +var server = async ({ client, directory }, options) => { const registerCommand = options?.register_command ?? true; const commandName = commandNameFromOptions(options); const minIntervalSeconds = positiveNumberOr(options?.min_interval_seconds, DEFAULT_MIN_INTERVAL_SECONDS); @@ -839,6 +845,7 @@ var server = async ({ client }, options) => { const ownedSessions = new Map; const ownershipQueues = new Map; const observedSessions = new Set; + const keepAliveSessions = new Set; const staleDynamicCandidates = new Map; const lastPromptAgentBySession = new Map; const dynamicPending = new Map; @@ -863,11 +870,29 @@ var server = async ({ client }, options) => { dynamicPending.delete(loopID); } } + async function recoveredSessionBusy(sessionID) { + try { + const response = await client.session.status(); + const status = response.data?.[sessionID]; + return !response.data || status != null && status.type !== "idle"; + } catch { + return true; + } + } + async function keepAliveSessionAvailable(sessionID) { + try { + const response = await client.session.get({ path: { id: sessionID } }); + return response.data?.id === sessionID && (!directory || response.data.directory === directory); + } catch { + return false; + } + } async function acquireOwnership(sessionID) { return enqueueSessionOperation(ownershipQueues, sessionID, async () => { if (disposed) return false; - observedSessions.add(sessionID); + if (!observedSessions.has(sessionID) && !await keepAliveSessionAvailable(sessionID)) + return false; const lease = await acquireSessionLease(sessionID, instanceID, SESSION_LEASE_MS); if (!lease) { await loseOwnership(sessionID); @@ -897,7 +922,7 @@ var server = async ({ client }, options) => { } function scheduleTimer(loop, minimumDelayMs = 0) { cancelTimer(loop.id); - if (disposed || loop.status !== "active" || loop.nextRunAt == null || !ownedSessions.has(loop.sessionID)) + if (disposed || loop.status !== "active" || loop.nextRunAt == null || !ownedSessions.has(loop.sessionID) || !observedSessions.has(loop.sessionID) && !loop.keepAlive) return; const delay = Math.max(minimumDelayMs, loop.nextRunAt - Date.now()); const timer = setTimeout(() => { @@ -936,7 +961,7 @@ var server = async ({ client }, options) => { scheduleTimer(loop); return; } - if (!observedSessions.has(loop.sessionID)) { + if (!observedSessions.has(loop.sessionID) && !loop.keepAlive) { scheduleTimer(loop, Math.min(loop.intervalMs ?? busyBackoffMs, busyBackoffMs)); await log("info", "Skipping loop because session ownership is unknown", { loopID, category: "ownership" }); return; @@ -950,11 +975,11 @@ var server = async ({ client }, options) => { return; } loop = claimed.loop; - if (maxLoopAgeMs > 0 && Date.now() - loop.createdAt >= maxLoopAgeMs) { + if (!loop.keepAlive && maxLoopAgeMs > 0 && Date.now() - loop.createdAt >= maxLoopAgeMs) { await stopLoopClaimed(loopID, claimed, `expired after ${Math.round(maxLoopAgeMs / 86400000)} days`); return; } - if (busySessions.has(loop.sessionID)) { + if (busySessions.has(loop.sessionID) || loop.keepAlive && !observedSessions.has(loop.sessionID) && await recoveredSessionBusy(loop.sessionID)) { const deferred = await recordRunDeferredClaimed(loopID, claimed, "skipped_busy", Math.min(loop.intervalMs ?? busyBackoffMs, busyBackoffMs)); if (deferred) scheduleTimer(deferred); @@ -1050,6 +1075,8 @@ var server = async ({ client }, options) => { async function rehydrate() { const loops = await activeLoops(); for (const loop of loops) { + if (loop.keepAlive) + keepAliveSessions.add(loop.sessionID); if (loop.nextRunAt == null) { if (loop.mode === "dynamic") { staleDynamicCandidates.set(loop.id, { sessionID: loop.sessionID, updatedAt: loop.updatedAt }); @@ -1061,6 +1088,8 @@ var server = async ({ client }, options) => { continue; } } + for (const sessionID of keepAliveSessions) + await acquireOwnership(sessionID); } async function observeSession(sessionID) { observedSessions.add(sessionID); @@ -1087,8 +1116,8 @@ var server = async ({ client }, options) => { } await rehydrate().catch(() => log("error", "Failed to rehydrate loops", { category: "state" })); const leaseHeartbeat = setInterval(() => { - for (const sessionID of observedSessions) { - const operation = ownedSessions.has(sessionID) ? renewOwnership(sessionID) : observeSession(sessionID); + for (const sessionID of new Set([...observedSessions, ...keepAliveSessions])) { + const operation = ownedSessions.has(sessionID) ? renewOwnership(sessionID) : observedSessions.has(sessionID) ? observeSession(sessionID) : acquireOwnership(sessionID); operation.catch(() => log("error", "Loop lease heartbeat failed", { category: "state" })); } }, SESSION_LEASE_HEARTBEAT_MS); @@ -1116,7 +1145,8 @@ var server = async ({ client }, options) => { args: { instruction: z.string().min(1).max(MAX_PROMPT_CHARS).describe("The instruction to perform on each iteration."), interval: z.string().optional().describe('Fixed cadence like "30s", "10m", "2h", or "1d". Omit for a dynamically paced loop.'), - max_runs: z.number().int().positive().optional().describe("Optional maximum number of iterations before the loop completes.") + max_runs: z.number().int().positive().optional().describe("Optional maximum number of iterations before the loop completes."), + keep_alive: z.boolean().optional().describe(KEEP_ALIVE_DESCRIPTION) }, async execute(args, context) { const input = args; @@ -1128,7 +1158,8 @@ var server = async ({ client }, options) => { intervalMs: dynamic ? null : parseInterval(input.interval, minIntervalSeconds), maxRuns: input.max_runs ?? null, agent: typeof context.agent === "string" ? context.agent : null, - maxLoopsPerSession + maxLoopsPerSession, + keepAlive: input.keep_alive === true }); if (loop.mode === "dynamic") { dynamicPending.set(loop.id, { sessionID: loop.sessionID, sawBusy: true }); @@ -1343,6 +1374,7 @@ async function setupV2(context) { const ownedSessions = new Map; const ownershipQueues = new Map; const observedSessions = new Set; + const keepAliveSessions = new Set; const staleDynamicCandidates = new Map; const lastPromptAgentBySession = new Map; const dynamicPending = new Map; @@ -1379,11 +1411,21 @@ async function setupV2(context) { dynamicPending.delete(loopID); } } + async function keepAliveSessionAvailable(sessionID) { + try { + const session = await context.session.get({ sessionID }); + const location = context.location; + return session.id === sessionID && (!location || session.location.directory === location.directory && session.location.workspaceID === location.workspaceID); + } catch { + return false; + } + } async function acquireOwnership(sessionID) { return enqueueSessionOperation(ownershipQueues, sessionID, async () => { if (disposed) return false; - observedSessions.add(sessionID); + if (!observedSessions.has(sessionID) && !await keepAliveSessionAvailable(sessionID)) + return false; const lease = await acquireSessionLease(sessionID, instanceID, SESSION_LEASE_MS); if (!lease) { await loseOwnership(sessionID); @@ -1413,7 +1455,7 @@ async function setupV2(context) { } function scheduleTimer(loop, minimumDelayMs = 0) { cancelTimer(loop.id); - if (disposed || loop.status !== "active" || loop.nextRunAt == null || !ownedSessions.has(loop.sessionID)) + if (disposed || loop.status !== "active" || loop.nextRunAt == null || !ownedSessions.has(loop.sessionID) || !observedSessions.has(loop.sessionID) && !loop.keepAlive) return; const delay = Math.max(minimumDelayMs, loop.nextRunAt - Date.now()); const timer = setTimeout(() => { @@ -1452,7 +1494,7 @@ async function setupV2(context) { scheduleTimer(loop); return; } - if (!observedSessions.has(loop.sessionID)) { + if (!observedSessions.has(loop.sessionID) && !loop.keepAlive) { scheduleTimer(loop, Math.min(loop.intervalMs ?? busyBackoffMs, busyBackoffMs)); v2Log("info", "Skipping loop because session ownership is unknown", { loopID, category: "ownership" }); return; @@ -1466,7 +1508,7 @@ async function setupV2(context) { return; } loop = claimed.loop; - if (maxLoopAgeMs > 0 && Date.now() - loop.createdAt >= maxLoopAgeMs) { + if (!loop.keepAlive && maxLoopAgeMs > 0 && Date.now() - loop.createdAt >= maxLoopAgeMs) { await stopLoopClaimed(loopID, claimed, `expired after ${Math.round(maxLoopAgeMs / 86400000)} days`); return; } @@ -1564,6 +1606,8 @@ async function setupV2(context) { async function rehydrate() { const loops = await activeLoops(); for (const loop of loops) { + if (loop.keepAlive) + keepAliveSessions.add(loop.sessionID); if (loop.nextRunAt == null) { if (loop.mode === "dynamic") { staleDynamicCandidates.set(loop.id, { sessionID: loop.sessionID, updatedAt: loop.updatedAt }); @@ -1575,6 +1619,8 @@ async function setupV2(context) { continue; } } + for (const sessionID of keepAliveSessions) + await acquireOwnership(sessionID); } async function observeSession(sessionID) { observedSessions.add(sessionID); @@ -1700,8 +1746,8 @@ async function setupV2(context) { })); await rehydrate().catch(() => v2Log("error", "Failed to rehydrate loops", { category: "state" })); const leaseHeartbeat = setInterval(() => { - for (const sessionID of observedSessions) { - const operation = ownedSessions.has(sessionID) ? renewOwnership(sessionID) : observeSession(sessionID); + for (const sessionID of new Set([...observedSessions, ...keepAliveSessions])) { + const operation = ownedSessions.has(sessionID) ? renewOwnership(sessionID) : observedSessions.has(sessionID) ? observeSession(sessionID) : acquireOwnership(sessionID); operation.catch(() => v2Log("error", "Loop lease heartbeat failed", { category: "state" })); } }, SESSION_LEASE_HEARTBEAT_MS); @@ -1762,6 +1808,10 @@ function loopToolsV2(services) { type: "integer", minimum: 1, description: "Optional maximum number of iterations before the loop completes." + }, + keep_alive: { + type: "boolean", + description: KEEP_ALIVE_DESCRIPTION } }, ["instruction"]), options: { codemode: false }, @@ -1775,7 +1825,8 @@ function loopToolsV2(services) { intervalMs: dynamic ? null : parseInterval(input.interval, services.minIntervalSeconds), maxRuns: input.max_runs ?? null, agent: typeof toolContext.agent === "string" ? toolContext.agent : null, - maxLoopsPerSession: services.maxLoopsPerSession + maxLoopsPerSession: services.maxLoopsPerSession, + keepAlive: input.keep_alive === true }); if (loop.mode === "dynamic") { services.dynamicPending.set(loop.id, { sessionID: loop.sessionID, sawBusy: true }); diff --git a/src/server.ts b/src/server.ts index 647543c..1c960cb 100644 --- a/src/server.ts +++ b/src/server.ts @@ -52,6 +52,9 @@ type Options = { restricted_agents?: string[] } +const KEEP_ALIVE_DESCRIPTION = + "Keep this session scheduled across turns and exempt from age expiry. Requires a fixed interval. Explicit max_runs, session deletion, and the failure-pause threshold still apply. Use it in a background or subagent session whose work outlives the turn that started it: the scheduler keeps re-prompting the session on the cadence, and the loop is never ended automatically by age expiry. You remain responsible for calling stop_loop once the work is resolved or abandoned." + const DEFAULT_COMMAND_NAME = "loop" const DEFAULT_BUSY_BACKOFF_SECONDS = 60 const DEFAULT_FAILURE_BACKOFF_SECONDS = 60 @@ -136,7 +139,7 @@ async function toolResult(sessionID: string, extra: Record = {} return JSON.stringify({ ...extra, loops, report: formatLoops(loops) }, null, 2) } -const server: Plugin = async ({ client }, options?: Options) => { +const server: Plugin = async ({ client, directory }, options?: Options) => { const registerCommand = options?.register_command ?? true const commandName = commandNameFromOptions(options) const minIntervalSeconds = positiveNumberOr(options?.min_interval_seconds, DEFAULT_MIN_INTERVAL_SECONDS) @@ -162,6 +165,7 @@ const server: Plugin = async ({ client }, options?: Options) => { // as an ownership proxy so a process sharing the state file with another // OpenCode instance does not mutate loops belonging to foreign sessions. const observedSessions = new Set() + const keepAliveSessions = new Set() // Dynamic records with no next run are only restart candidates until this // process proves that it owns their session. The captured timestamp also // prevents us from stopping a record changed by its actual owner meanwhile. @@ -194,10 +198,30 @@ const server: Plugin = async ({ client }, options?: Options) => { } } + async function recoveredSessionBusy(sessionID: string) { + try { + const response = await client.session.status() + const status = response.data?.[sessionID] + return !response.data || (status != null && status.type !== "idle") + } catch { + // A detached session must be known idle before restoring its heartbeat. + return true + } + } + + async function keepAliveSessionAvailable(sessionID: string) { + try { + const response = await client.session.get({ path: { id: sessionID } }) + return response.data?.id === sessionID && (!directory || response.data.directory === directory) + } catch { + return false + } + } + async function acquireOwnership(sessionID: string) { return enqueueSessionOperation(ownershipQueues, sessionID, async () => { if (disposed) return false - observedSessions.add(sessionID) + if (!observedSessions.has(sessionID) && !(await keepAliveSessionAvailable(sessionID))) return false const lease = await acquireSessionLease(sessionID, instanceID, SESSION_LEASE_MS) if (!lease) { await loseOwnership(sessionID) @@ -226,7 +250,8 @@ const server: Plugin = async ({ client }, options?: Options) => { function scheduleTimer(loop: LoopSnapshot, minimumDelayMs = 0) { cancelTimer(loop.id) - if (disposed || loop.status !== "active" || loop.nextRunAt == null || !ownedSessions.has(loop.sessionID)) return + if (disposed || loop.status !== "active" || loop.nextRunAt == null || !ownedSessions.has(loop.sessionID) || + (!observedSessions.has(loop.sessionID) && !loop.keepAlive)) return const delay = Math.max(minimumDelayMs, loop.nextRunAt - Date.now()) const timer = setTimeout(() => { timers.delete(loop.id) @@ -262,7 +287,7 @@ const server: Plugin = async ({ client }, options?: Options) => { scheduleTimer(loop) return } - if (!observedSessions.has(loop.sessionID)) { + if (!observedSessions.has(loop.sessionID) && !loop.keepAlive) { // Ownership is checked before claiming or prompting so this process never // mutates a loop belonging to an unknown session. Keep polling without // changing persisted state so later observations through any hook/tool @@ -279,11 +304,12 @@ const server: Plugin = async ({ client }, options?: Options) => { return } loop = claimed.loop - if (maxLoopAgeMs > 0 && Date.now() - loop.createdAt >= maxLoopAgeMs) { + if (!loop.keepAlive && maxLoopAgeMs > 0 && Date.now() - loop.createdAt >= maxLoopAgeMs) { await stopLoopClaimed(loopID, claimed, `expired after ${Math.round(maxLoopAgeMs / 86_400_000)} days`) return } - if (busySessions.has(loop.sessionID)) { + if (busySessions.has(loop.sessionID) || + (loop.keepAlive && !observedSessions.has(loop.sessionID) && await recoveredSessionBusy(loop.sessionID))) { const deferred = await recordRunDeferredClaimed(loopID, claimed, "skipped_busy", Math.min(loop.intervalMs ?? busyBackoffMs, busyBackoffMs)) if (deferred) scheduleTimer(deferred) return @@ -384,6 +410,7 @@ const server: Plugin = async ({ client }, options?: Options) => { // may acquire the session lease and arm timers. const loops = await activeLoops() for (const loop of loops) { + if (loop.keepAlive) keepAliveSessions.add(loop.sessionID) if (loop.nextRunAt == null) { if (loop.mode === "dynamic") { staleDynamicCandidates.set(loop.id, { sessionID: loop.sessionID, updatedAt: loop.updatedAt }) @@ -395,6 +422,7 @@ const server: Plugin = async ({ client }, options?: Options) => { continue } } + for (const sessionID of keepAliveSessions) await acquireOwnership(sessionID) } async function observeSession(sessionID: string) { @@ -421,8 +449,9 @@ const server: Plugin = async ({ client }, options?: Options) => { await rehydrate().catch(() => log("error", "Failed to rehydrate loops", { category: "state" })) const leaseHeartbeat = setInterval(() => { - for (const sessionID of observedSessions) { - const operation = ownedSessions.has(sessionID) ? renewOwnership(sessionID) : observeSession(sessionID) + for (const sessionID of new Set([...observedSessions, ...keepAliveSessions])) { + const operation = ownedSessions.has(sessionID) ? renewOwnership(sessionID) : + observedSessions.has(sessionID) ? observeSession(sessionID) : acquireOwnership(sessionID) void operation.catch(() => log("error", "Loop lease heartbeat failed", { category: "state" })) } }, SESSION_LEASE_HEARTBEAT_MS) @@ -456,9 +485,10 @@ const server: Plugin = async ({ client }, options?: Options) => { .optional() .describe('Fixed cadence like "30s", "10m", "2h", or "1d". Omit for a dynamically paced loop.'), max_runs: z.number().int().positive().optional().describe("Optional maximum number of iterations before the loop completes."), + keep_alive: z.boolean().optional().describe(KEEP_ALIVE_DESCRIPTION), }, async execute(args, context) { - const input = args as { instruction: string; interval?: string; max_runs?: number } + const input = args as { instruction: string; interval?: string; max_runs?: number; keep_alive?: boolean } await observeSession(context.sessionID) const dynamic = !input.interval?.trim() const loop = await createLoop(context.sessionID, { @@ -468,6 +498,7 @@ const server: Plugin = async ({ client }, options?: Options) => { maxRuns: input.max_runs ?? null, agent: typeof context.agent === "string" ? context.agent : null, maxLoopsPerSession, + keepAlive: input.keep_alive === true, }) if (loop.mode === "dynamic") { dynamicPending.set(loop.id, { sessionID: loop.sessionID, sawBusy: true }) @@ -708,6 +739,7 @@ async function setupV2(context: PluginV2.Plugin.Context): Promise() + const keepAliveSessions = new Set() // Kept read-only until an event or hook establishes that this context owns // the session and the record is still the exact snapshot seen at startup. const staleDynamicCandidates = new Map() @@ -750,10 +782,21 @@ async function setupV2(context: PluginV2.Plugin.Context): Promise { if (disposed) return false - observedSessions.add(sessionID) + if (!observedSessions.has(sessionID) && !(await keepAliveSessionAvailable(sessionID))) return false const lease = await acquireSessionLease(sessionID, instanceID, SESSION_LEASE_MS) if (!lease) { await loseOwnership(sessionID) @@ -782,7 +825,8 @@ async function setupV2(context: PluginV2.Plugin.Context): Promise { timers.delete(loop.id) @@ -818,7 +862,7 @@ async function setupV2(context: PluginV2.Plugin.Context): Promise 0 && Date.now() - loop.createdAt >= maxLoopAgeMs) { + if (!loop.keepAlive && maxLoopAgeMs > 0 && Date.now() - loop.createdAt >= maxLoopAgeMs) { await stopLoopClaimed(loopID, claimed, `expired after ${Math.round(maxLoopAgeMs / 86_400_000)} days`) return } @@ -934,6 +978,7 @@ async function setupV2(context: PluginV2.Plugin.Context): Promise v2Log("error", "Failed to rehydrate loops", { category: "state" })) const leaseHeartbeat = setInterval(() => { - for (const sessionID of observedSessions) { - const operation = ownedSessions.has(sessionID) ? renewOwnership(sessionID) : observeSession(sessionID) + for (const sessionID of new Set([...observedSessions, ...keepAliveSessions])) { + const operation = ownedSessions.has(sessionID) ? renewOwnership(sessionID) : + observedSessions.has(sessionID) ? observeSession(sessionID) : acquireOwnership(sessionID) void operation.catch(() => v2Log("error", "Loop lease heartbeat failed", { category: "state" })) } }, SESSION_LEASE_HEARTBEAT_MS) @@ -1145,12 +1192,16 @@ function loopToolsV2(services: LoopServices): ToolV2Info[] { minimum: 1, description: "Optional maximum number of iterations before the loop completes.", }, + keep_alive: { + type: "boolean", + description: KEEP_ALIVE_DESCRIPTION, + }, }, ["instruction"], ), options: { codemode: false }, execute: async (args, toolContext) => { - const input = args as { instruction: string; interval?: string; max_runs?: number } + const input = args as { instruction: string; interval?: string; max_runs?: number; keep_alive?: boolean } await services.observeSession(toolContext.sessionID) const dynamic = !input.interval?.trim() const loop = await createLoop(toolContext.sessionID, { @@ -1160,6 +1211,7 @@ function loopToolsV2(services: LoopServices): ToolV2Info[] { maxRuns: input.max_runs ?? null, agent: typeof toolContext.agent === "string" ? toolContext.agent : null, maxLoopsPerSession: services.maxLoopsPerSession, + keepAlive: input.keep_alive === true, }) if (loop.mode === "dynamic") { services.dynamicPending.set(loop.id, { sessionID: loop.sessionID, sawBusy: true }) diff --git a/src/state.ts b/src/state.ts index ba06b41..fab3ad0 100644 --- a/src/state.ts +++ b/src/state.ts @@ -22,6 +22,7 @@ export type CreateLoopOptions = { maxRuns?: number | null agent?: string | null maxLoopsPerSession?: number | null + keepAlive?: boolean } export type Loop = { @@ -29,6 +30,7 @@ export type Loop = { sessionID: string prompt: string mode: LoopMode + keepAlive: boolean intervalMs: number | null status: LoopStatus createdAt: number @@ -85,6 +87,7 @@ const LoopSchema = Schema.Struct({ sessionID: Schema.String, prompt: Schema.String, mode: Schema.optionalWith(Schema.Literal("interval", "dynamic"), { default: () => "interval" as const }), + keepAlive: Schema.optionalWith(Schema.Boolean, { default: () => false }), intervalMs: NullableNumber, status: Schema.Literal("active", "paused", "stopped", "completed"), createdAt: Schema.Number, @@ -393,6 +396,11 @@ function requireLoop(state: State, loopID: string) { export async function createLoop(sessionID: string, options: CreateLoopOptions) { const prompt = validatePrompt(options.prompt) const mode: LoopMode = options.mode === "dynamic" ? "dynamic" : "interval" + const keepAlive = options.keepAlive === true + // A keep-alive future needs a self-sufficient heartbeat. Dynamic pacing puts + // the next run in the agent's hands every turn, so one missed reschedule + // would end the very future the flag promises to hold open. + if (keepAlive && mode !== "interval") throw new Error("keep-alive loops require a fixed interval") const intervalMs = mode === "interval" ? positiveIntegerOrNull(options.intervalMs) : null if (mode === "interval" && intervalMs == null) throw new Error("interval loops require a positive interval") const maxRuns = positiveIntegerOrNull(options.maxRuns) @@ -411,6 +419,7 @@ export async function createLoop(sessionID: string, options: CreateLoopOptions) sessionID, prompt, mode, + keepAlive, intervalMs, status: "active", createdAt: timestamp, diff --git a/test/server-v2.test.ts b/test/server-v2.test.ts index 3a02221..4bc6007 100644 --- a/test/server-v2.test.ts +++ b/test/server-v2.test.ts @@ -1,9 +1,9 @@ import { afterEach, beforeEach, expect, test } from "bun:test" -import { mkdtemp, rm } from "node:fs/promises" +import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises" import { join } from "node:path" import { tmpdir } from "node:os" import plugin from "../src/server" -import { activeLoops, getLoop, listLoops } from "../src/state" +import { activeLoops, createLoop, getLoop, listLoops } from "../src/state" const TOOL_NAMES = [ "clear_loops", @@ -80,7 +80,9 @@ function controlledStream() { type MockContext = { options: Record + location?: { directory: string; workspaceID?: string } activeSessions: Set + unavailableSessions: Set promptCalls: Array<{ sessionID: string delivery?: "steer" | "queue" @@ -97,6 +99,7 @@ type MockContext = { transform: (callback: (draft: ToolDraft) => void) => Promise } session: { + get: (input: { sessionID: string }) => Promise<{ id: string; location?: { directory: string; workspaceID?: string } }> hook: (name: string, callback: (input: unknown) => void) => Promise active: () => Promise> prompt: (input: { @@ -114,6 +117,7 @@ function makeMockContext(options: Record = {}): MockContext { const commands: MockContext["commands"] = [] const hooks: MockContext["hooks"] = {} const activeSessions = new Set() + const unavailableSessions = new Set() const promptCalls: MockContext["promptCalls"] = [] const disposals: string[] = [] const stream = controlledStream() @@ -125,6 +129,7 @@ function makeMockContext(options: Record = {}): MockContext { return { options: { min_interval_seconds: 1, busy_backoff_seconds: 1, failure_backoff_seconds: 1, ...options }, activeSessions, + unavailableSessions, promptCalls, tools, commands, @@ -144,6 +149,10 @@ function makeMockContext(options: Record = {}): MockContext { }, }, session: { + get: async ({ sessionID }) => { + if (unavailableSessions.has(sessionID)) throw new Error("session unavailable") + return { id: sessionID } + }, hook: async (name, callback) => { hooks[name] = callback return registration(`session.hook:${name}`) @@ -576,3 +585,112 @@ test("V2 slow accepted delivery retains run accounting after its owner lease exp await cleanup() } }) + +test("V2 create_loop advertises keep_alive without making it required", async () => { + const mock = makeMockContext() + const cleanup = await plugin.setup(mock as never) + + const input = loopTool(mock, "create_loop").input as { + properties: Record + required: string[] + } + expect(input.properties.keep_alive?.type).toBe("boolean") + expect(input.required).toEqual(["instruction"]) + + mock.stream.end() + await cleanup() +}) + +test("V2 keep-alive recovery exempts age expiry and preserves ordinary sibling dormancy", async () => { + const durable = await createLoop("ses_keep", { prompt: "durable", intervalMs: 1, keepAlive: true }) + const ordinary = await createLoop("ses_keep", { prompt: "ordinary", intervalMs: 1 }) + const statePath = process.env.OPENCODE_LOOP_STATE_PATH! + const state = JSON.parse(await readFile(statePath, "utf8")) as { + loops: Record + } + for (const loop of Object.values(state.loops)) { + loop.createdAt = Date.now() - 400 * 86_400_000 + loop.nextRunAt = Date.now() - 1000 + } + await writeFile(statePath, JSON.stringify(state)) + const mock = makeMockContext() + const cleanup = await plugin.setup(mock as never) + await waitFor(() => mock.promptCalls.length === 1) + expect((await getLoop(durable.id))?.status).toBe("active") + expect((await getLoop(ordinary.id))?.lastResult).toBeNull() + mock.stream.push({ type: "session.idle", data: { sessionID: "ses_keep" } }) + await waitFor(async () => (await getLoop(ordinary.id))?.status === "stopped") + expect((await getLoop(durable.id))?.status).toBe("active") + mock.stream.end() + await cleanup() +}) + +test("V2 keep_alive requires a fixed interval", async () => { + const mock = makeMockContext() + const cleanup = await plugin.setup(mock as never) + + await expect( + loopTool(mock, "create_loop").execute({ instruction: "no cadence", keep_alive: true }, toolContext("ses_bad")), + ).rejects.toThrow(/keep-alive loops require a fixed interval/) + + mock.stream.end() + await cleanup() +}) + + +test("V2 keep-alive recovery leaves inaccessible sessions dormant", async () => { + const loop = await createLoop("ses_missing", { prompt: "tick", intervalMs: 1, keepAlive: true }) + const mock = makeMockContext() + mock.unavailableSessions.add("ses_missing") + const cleanup = await plugin.setup(mock as never) + await sleep(200) + expect(mock.promptCalls).toHaveLength(0) + expect((await getLoop(loop.id))?.lastResult).toBeNull() + mock.stream.end() + await cleanup() +}) + +test("V2 keep-alive recovery retains bounded failure pause and explicit resume", async () => { + const loop = await createLoop("ses_retry", { prompt: "tick", intervalMs: 100, keepAlive: true }) + const mock = makeMockContext({ max_consecutive_failures: 1 }) + const prompt = mock.session.prompt + mock.session.prompt = async () => { throw new Error("provider failed") } + const cleanup = await plugin.setup(mock as never) + await waitFor(async () => (await getLoop(loop.id))?.status === "paused") + mock.session.prompt = prompt + await loopTool(mock, "resume_loop").execute({ loop_id: loop.id }, toolContext("ses_retry")) + await waitFor(async () => (await getLoop(loop.id))?.runCount === 1) + expect((await getLoop(loop.id))?.status).toBe("active") + mock.stream.end() + await cleanup() +}) + + +test("V2 keep-alive recovery refuses another directory or workspace", async () => { + const loop = await createLoop("ses_foreign", { prompt: "tick", intervalMs: 1, keepAlive: true }) + for (const location of [{ directory: "/other" }, { directory: "/current", workspaceID: "other" }]) { + const mock = makeMockContext() + mock.location = { directory: "/current" } + mock.session.get = async ({ sessionID }) => ({ id: sessionID, location }) + const cleanup = await plugin.setup(mock as never) + await sleep(200) + expect(mock.promptCalls).toHaveLength(0) + expect((await getLoop(loop.id))?.lastResult).toBeNull() + mock.stream.end() + await cleanup() + } +}) + +test("V2 concurrent recovery contexts deliver a keep-alive iteration once", async () => { + const loop = await createLoop("ses_shared", { prompt: "tick", intervalMs: 1, keepAlive: true, maxRuns: 1 }) + const first = makeMockContext() + const second = makeMockContext() + const cleanFirst = await plugin.setup(first as never) + const cleanSecond = await plugin.setup(second as never) + await waitFor(async () => (await getLoop(loop.id))?.status === "completed") + expect(first.promptCalls.length + second.promptCalls.length).toBe(1) + first.stream.end() + second.stream.end() + await cleanFirst() + await cleanSecond() +}) diff --git a/test/server.test.ts b/test/server.test.ts index ed31e21..dc5038d 100644 --- a/test/server.test.ts +++ b/test/server.test.ts @@ -1,9 +1,9 @@ import { afterEach, beforeEach, expect, test } from "bun:test" -import { mkdtemp, rm } from "node:fs/promises" +import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises" import { join } from "node:path" import { tmpdir } from "node:os" import serverModule from "../src/server" -import { activeLoops, getLoop, listLoops } from "../src/state" +import { activeLoops, createLoop, getLoop, listLoops } from "../src/state" type Hooks = Awaited> @@ -25,9 +25,14 @@ afterEach(async () => { await rm(dir, { recursive: true, force: true }) }) -function fakeClient(sent: SentPrompt[], options: { failPrompts?: boolean } = {}) { +function fakeClient(sent: SentPrompt[], options: { failPrompts?: boolean; unavailableSessions?: Set } = {}) { return { session: { + status: async () => ({ data: {} }), + get: async (input: { path: { id: string } }) => { + if (options.unavailableSessions?.has(input.path.id)) throw new Error("session unavailable") + return { data: { id: input.path.id } } + }, promptAsync: async (input: { path: { id: string }; body: { agent?: string; parts: { type: string; text: string }[] } }) => { if (options.failPrompts) throw new Error("prompt rejected") sent.push({ @@ -43,7 +48,7 @@ function fakeClient(sent: SentPrompt[], options: { failPrompts?: boolean } = {}) } } -async function makeServer(sent: SentPrompt[], options?: Record, clientOptions?: { failPrompts?: boolean }) { +async function makeServer(sent: SentPrompt[], options?: Record, clientOptions?: { failPrompts?: boolean; unavailableSessions?: Set }) { const hooks = await serverModule.server( { client: fakeClient(sent, clientOptions) } as never, { min_interval_seconds: 1, busy_backoff_seconds: 1, failure_backoff_seconds: 1, ...options } as never, @@ -345,3 +350,199 @@ test("slow accepted delivery retains run accounting after the initial owner leas await hooks.dispose?.() } }) + +test("a keep-alive loop survives age expiry that would stop an ordinary loop", async () => { + const sent: SentPrompt[] = [] + const first = await makeServer(sent) + const durable = JSON.parse( + await tool(first, "create_loop").execute( + { instruction: "hold the future open", interval: "1s", keep_alive: true }, + { sessionID: "ses_keep" }, + ), + ) as { created: string } + const ordinary = JSON.parse( + await tool(first, "create_loop").execute({ instruction: "ordinary", interval: "1s" }, { sessionID: "ses_keep" }), + ) as { created: string } + await first.dispose?.() + + // Backdate both well past the expiry window. + const statePath = process.env.OPENCODE_LOOP_STATE_PATH! + const state = JSON.parse(await readFile(statePath, "utf8")) as { + loops: Record + } + for (const id of [durable.created, ordinary.created]) { + state.loops[id]!.createdAt = Date.now() - 400 * 24 * 60 * 60 * 1000 + state.loops[id]!.nextRunAt = Date.now() - 1000 + } + await writeFile(statePath, JSON.stringify(state)) + + const second = await makeServer(sent, { max_loop_age_days: 7 }) + await idle(second, "ses_keep") + await sleep(600) + + expect((await getLoop(durable.created))?.status).toBe("active") + expect((await getLoop(ordinary.created))?.status).toBe("stopped") + await second.dispose?.() +}) + +test("a keep-alive loop retries in a session this process never observed", async () => { + const sent: SentPrompt[] = [] + const first = await makeServer(sent) + const durable = JSON.parse( + await tool(first, "create_loop").execute( + { instruction: "hold the future open", interval: "1s", keep_alive: true }, + { sessionID: "ses_detached" }, + ), + ) as { created: string } + await first.dispose?.() + + // A fresh process never sees an event for the detached session, and the + // prompt fails: the loop must record the failure and re-arm, not stall. + const second = await makeServer(sent, undefined, { failPrompts: true }) + await sleep(1300) + + const loop = await getLoop(durable.created) + expect(loop?.status).toBe("active") + expect(loop?.lastResult).toBe("failed") + expect(loop?.nextRunAt).not.toBeNull() + await second.dispose?.() +}) + +test("an ordinary loop still defers to the owning process when unobserved", async () => { + const sent: SentPrompt[] = [] + const first = await makeServer(sent) + const ordinary = JSON.parse( + await tool(first, "create_loop").execute({ instruction: "ordinary", interval: "1s" }, { sessionID: "ses_other_proc" }), + ) as { created: string } + await first.dispose?.() + + const second = await makeServer(sent, undefined, { failPrompts: true }) + await sleep(1300) + + // Left alone for the process that owns it: no failure recorded. + expect((await getLoop(ordinary.created))?.lastResult).toBeNull() + await second.dispose?.() +}) + +test("a keep-alive loop survives a restart and keeps firing", async () => { + const sent: SentPrompt[] = [] + const first = await makeServer(sent) + const durable = JSON.parse( + await tool(first, "create_loop").execute( + { instruction: "hold the future open", interval: "1s", keep_alive: true }, + { sessionID: "ses_restart" }, + ), + ) as { created: string } + await first.dispose?.() + + const second = await makeServer(sent) + await sleep(1300) + + expect(sent.map((entry) => entry.sessionID)).toContain("ses_restart") + expect((await getLoop(durable.created))?.status).toBe("active") + await second.dispose?.() +}) + +test("a keep-alive future ends only when the agent stops it", async () => { + const sent: SentPrompt[] = [] + const hooks = await makeServer(sent) + const durable = JSON.parse( + await tool(hooks, "create_loop").execute( + { instruction: "hold the future open", interval: "1s", keep_alive: true }, + { sessionID: "ses_stop" }, + ), + ) as { created: string } + + // Several iterations pass without the agent ever stopping the loop. The + // injected turn marks the session busy, so each iteration needs its idle. + for (let iteration = 0; iteration < 3; iteration += 1) { + await sleep(1200) + await idle(hooks, "ses_stop") + } + expect((await getLoop(durable.created))?.status).toBe("active") + expect((await getLoop(durable.created))?.runCount).toBeGreaterThanOrEqual(3) + + await tool(hooks, "stop_loop").execute({ loop_id: durable.created }, { sessionID: "ses_stop" }) + expect((await getLoop(durable.created))?.status).toBe("stopped") + await hooks.dispose?.() +}) + +test("keep_alive requires a fixed interval", async () => { + const sent: SentPrompt[] = [] + const hooks = await makeServer(sent) + + await expect( + tool(hooks, "create_loop").execute({ instruction: "no cadence", keep_alive: true }, { sessionID: "ses_bad" }), + ).rejects.toThrow(/keep-alive loops require a fixed interval/) + + await hooks.dispose?.() +}) + + +test("keep-alive recovery leaves unavailable sessions dormant", async () => { + const loop = await createLoop("ses_missing", { prompt: "tick", intervalMs: 1, keepAlive: true }) + const sent: SentPrompt[] = [] + const hooks = await makeServer(sent, undefined, { unavailableSessions: new Set(["ses_missing"]) }) + await sleep(200) + expect(sent).toHaveLength(0) + expect((await getLoop(loop.id))?.lastResult).toBeNull() + await hooks.dispose?.() +}) + +test("keep-alive recovery does not wake ordinary siblings without observation", async () => { + const durable = await createLoop("ses_shared", { prompt: "durable", intervalMs: 1, keepAlive: true }) + const ordinary = await createLoop("ses_shared", { prompt: "ordinary", intervalMs: 1 }) + const sent: SentPrompt[] = [] + const hooks = await makeServer(sent) + await sleep(200) + expect(sent).toHaveLength(1) + expect(sent[0]?.text).toContain(durable.id) + expect((await getLoop(ordinary.id))?.lastResult).toBeNull() + await hooks.dispose?.() +}) + +test("keep-alive recovery retains bounded failure pause and explicit resume", async () => { + const loop = await createLoop("ses_retry", { prompt: "tick", intervalMs: 100, keepAlive: true }) + const sent: SentPrompt[] = [] + const failures = { failPrompts: true } + const hooks = await makeServer(sent, { max_consecutive_failures: 1 }, failures) + await sleep(300) + expect((await getLoop(loop.id))?.status).toBe("paused") + failures.failPrompts = false + await tool(hooks, "resume_loop").execute({ loop_id: loop.id }, { sessionID: "ses_retry" }) + await sleep(300) + expect((await getLoop(loop.id))?.status).toBe("active") + expect((await getLoop(loop.id))?.runCount).toBe(1) + await hooks.dispose?.() +}) + + +test("V1 recovered keep-alive defers while the authoritative host session is busy", async () => { + const loop = await createLoop("ses_busy", { prompt: "tick", intervalMs: 100, keepAlive: true }) + const sent: SentPrompt[] = [] + const client = fakeClient(sent) + let busy = true + client.session.status = async () => ({ data: busy ? { ses_busy: { type: "busy" } } : {} }) + const hooks = await serverModule.server({ client } as never, { min_interval_seconds: 1 } as never) + await sleep(300) + expect(sent).toHaveLength(0) + expect((await getLoop(loop.id))?.lastResult).toBe("skipped_busy") + busy = false + await sleep(300) + expect(sent).toHaveLength(1) + expect((await getLoop(loop.id))?.runCount).toBe(1) + await hooks.dispose?.() +}) + + +test("V1 keep-alive recovery refuses a session in another directory", async () => { + const loop = await createLoop("ses_foreign", { prompt: "tick", intervalMs: 1, keepAlive: true }) + const sent: SentPrompt[] = [] + const client = fakeClient(sent) + client.session.get = async ({ path }) => ({ data: { id: path.id, directory: "/other" } }) + const hooks = await serverModule.server({ client, directory: "/current" } as never) + await sleep(200) + expect(sent).toHaveLength(0) + expect((await getLoop(loop.id))?.lastResult).toBeNull() + await hooks.dispose?.() +}) diff --git a/test/state.test.ts b/test/state.test.ts index 0c9b831..6b5a3f3 100644 --- a/test/state.test.ts +++ b/test/state.test.ts @@ -328,6 +328,7 @@ test("decodes persisted state with optional fields omitted", async () => { expect(loop?.lastResult).toBeNull() expect(loop?.maxRuns).toBeNull() expect(loop?.agent).toBeNull() + expect(loop?.keepAlive).toBe(false) expect(loop?.consecutiveFailures).toBe(0) expect(loop?.nextRetryAt).toBeNull() expect(loop?.blockedReason).toBeNull() @@ -471,3 +472,17 @@ test("a dynamic iteration counts its run without overwriting an equal-timestamp Date.now = originalNow } }) + +test("createLoop records keep-alive futures and defaults them off", async () => { + const ordinary = await createLoop("ses_1", { prompt: "tick", intervalMs: 60_000 }) + expect(ordinary.keepAlive).toBe(false) + + const durable = await createLoop("ses_1", { prompt: "tick", intervalMs: 60_000, keepAlive: true }) + expect(durable.keepAlive).toBe(true) +}) + +test("keep-alive loops cannot delegate pacing to the agent", async () => { + await expect(createLoop("ses_1", { prompt: "tick", mode: "dynamic", keepAlive: true })).rejects.toThrow( + /keep-alive loops require a fixed interval/, + ) +})