diff --git a/README.md b/README.md index e8a5375..e881085 100644 --- a/README.md +++ b/README.md @@ -124,7 +124,7 @@ A dynamic loop mirrors Claude Code's self-paced `/loop`: at the end of each iter - 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. - If several loops in one session are due at once, one iteration is injected and the rest wait for the next idle. -- Failed injections are recorded in the loop's `lastError` and retried after a backoff; they never crash OpenCode. +- Failed injections store only the safe `provider/model request failed` category. They use bounded exponential backoff and are paused after five consecutive failures by default; provider responses and session data are never persisted or logged. - Loops are stopped automatically when their session is deleted, when `max_runs` is reached, or after 7 days (configurable). ## Options @@ -141,6 +141,8 @@ In OpenCode 1, server options use the package-and-options tuple in `opencode.jso "max_loops_per_session": 5, "busy_backoff_seconds": 60, "failure_backoff_seconds": 60, + "max_failure_backoff_seconds": 3600, + "max_consecutive_failures": 5, "max_loop_age_days": 7, "dynamic_max_delay_seconds": 86400, "restricted_agents": ["plan"], @@ -164,6 +166,8 @@ In OpenCode 2, use the plugin object form: "max_loops_per_session": 5, "busy_backoff_seconds": 60, "failure_backoff_seconds": 60, + "max_failure_backoff_seconds": 3600, + "max_consecutive_failures": 5, "max_loop_age_days": 7, "dynamic_max_delay_seconds": 86400, "restricted_agents": ["plan"], @@ -180,7 +184,9 @@ Defaults: - `min_interval_seconds`: `30`; the smallest accepted interval and the lower clamp for dynamic delays. - `max_loops_per_session`: `5` open (active or paused) loops per session. - `busy_backoff_seconds`: `60`; retry delay when an iteration comes due while the session is busy. -- `failure_backoff_seconds`: `60`; retry delay when injecting the iteration prompt fails. +- `failure_backoff_seconds`: `60`; base retry delay when injecting the iteration prompt fails. Consecutive failures double this delay. +- `max_failure_backoff_seconds`: `3600`; upper bound for exponential failure backoff. +- `max_consecutive_failures`: `5`; number of consecutive provider/model failures after which the loop is paused with an inspectable `blockedReason`. - `max_loop_age_days`: `7`; loops stop automatically after this age. Set `0` to disable expiry. - `dynamic_max_delay_seconds`: `86400`; upper clamp for `schedule_next_run` delays. - `restricted_agents`: `["plan"]`; iterations are deferred while the session's last prompt came from one of these agents. @@ -203,7 +209,7 @@ If `XDG_DATA_HOME` is not set, the default is: Set `OPENCODE_LOOP_STATE_PATH` to use a custom file. -The state file is written atomically with owner-only permissions when the host filesystem supports it. Active interval loops are rehydrated and rescheduled when OpenCode restarts. Dynamic loops that were waiting on the agent to schedule their next run cannot recover on their own after a restart and are stopped with an explanatory reason. +The state file is written atomically with owner-only permissions when the host filesystem supports it. Persisted loops are rehydrated only after authoritative session context establishes a renewable session lease, preventing concurrent OpenCode instances from driving the same session. Dynamic loops that were waiting on the agent to schedule their next run cannot recover on their own after a restart and are stopped with an explanatory reason. ## Credits diff --git a/dist/server.js b/dist/server.js index 72bc224..7ca1ea8 100644 --- a/dist/server.js +++ b/dist/server.js @@ -1,9 +1,10 @@ // @bun // src/server.ts +import { randomUUID } from "crypto"; import { z } from "zod"; // src/state.ts -import { chmod, mkdir, readFile, rename, writeFile } from "fs/promises"; +import { chmod, mkdir, readFile, rename, rm, stat, writeFile } from "fs/promises"; import { homedir } from "os"; import { dirname, join } from "path"; import { Data, Effect, Schema } from "effect"; @@ -37,15 +38,31 @@ var LoopSchema = Schema.Struct({ default: () => null }), lastError: Schema.optionalWith(NullableString, { default: () => null }), + consecutiveFailures: Schema.optionalWith(Schema.Number, { default: () => 0 }), + nextRetryAt: Schema.optionalWith(NullableNumber, { default: () => null }), + blockedReason: Schema.optionalWith(NullableString, { default: () => null }), lastReason: Schema.optionalWith(NullableString, { default: () => null }), runCount: Schema.optionalWith(Schema.Number, { default: () => 0 }), maxRuns: Schema.optionalWith(NullableNumber, { default: () => null }), agent: Schema.optionalWith(NullableString, { default: () => null }), - stopReason: Schema.optionalWith(NullableString, { default: () => null }) + stopReason: Schema.optionalWith(NullableString, { default: () => null }), + ownerID: Schema.optionalWith(NullableString, { default: () => null }), + ownerLeaseUntil: Schema.optionalWith(NullableNumber, { default: () => null }), + ownerRevision: Schema.optionalWith(Schema.Number, { default: () => 0 }), + runClaimToken: Schema.optionalWith(Schema.Number, { default: () => 0 }) +}); +var SessionLeaseSchema = Schema.Struct({ + sessionID: Schema.String, + instanceID: Schema.String, + expiresAt: Schema.Number, + revision: Schema.Number }); var StateSchema = Schema.Struct({ version: Schema.Literal(1), - loops: Schema.Record({ key: Schema.String, value: LoopSchema }) + loops: Schema.Record({ key: Schema.String, value: LoopSchema }), + sessionLeases: Schema.optionalWith(Schema.Record({ key: Schema.String, value: SessionLeaseSchema }), { + default: () => ({}) + }) }); function defaultStateFile() { const dataHome = process.env.XDG_DATA_HOME || (process.platform === "win32" && process.env.APPDATA ? process.env.APPDATA : join(homedir(), ".local", "share")); @@ -58,7 +75,7 @@ function now() { return Date.now(); } function emptyState() { - return { version: 1, loops: {} }; + return { version: 1, loops: {}, sessionLeases: {} }; } function isMissingStateFile(error) { return typeof error === "object" && error !== null && error.code === "ENOENT"; @@ -107,7 +124,6 @@ function enqueueMutation(operation) { }); return current; } -var MAX_MUTATION_ATTEMPTS = 5; async function readRawState() { try { return await readFile(statePath(), "utf8"); @@ -117,10 +133,42 @@ async function readRawState() { throw error; } } +var LOCK_STALE_MS = 30000; +var LOCK_RETRY_MS = 10; +function sleep(ms) { + return new Promise((resolve) => setTimeout(resolve, ms)); +} +async function acquireMutationLock() { + const file = statePath(); + const lock = `${file}.lock`; + await mkdir(dirname(file), { recursive: true, mode: 448 }); + for (;; ) { + try { + await mkdir(lock, { mode: 448 }); + return async () => rm(lock, { recursive: true, force: true }); + } catch (error) { + if (error.code !== "EEXIST") + throw error; + try { + const lockStat = await stat(lock); + if (Date.now() - lockStat.mtimeMs > LOCK_STALE_MS) { + const stale = `${lock}.stale.${process.pid}.${Date.now()}`; + await rename(lock, stale); + await rm(stale, { recursive: true, force: true }); + continue; + } + } catch (lockError) { + if (!isMissingStateFile(lockError)) + throw lockError; + } + await sleep(LOCK_RETRY_MS); + } + } +} async function mutate(fn) { return enqueueMutation(async () => { - let lastError; - for (let attempt = 0;attempt < MAX_MUTATION_ATTEMPTS; attempt += 1) { + const release = await acquireMutationLock(); + try { const before = await readRawState(); const result = await Effect.runPromise(Effect.gen(function* () { const state = before == null ? emptyState() : yield* Effect.try({ @@ -133,15 +181,11 @@ async function mutate(fn) { }); return { state, value }; })); - const current = await readRawState(); - if (current !== before) { - lastError = new Error("state file changed by a concurrent writer"); - continue; - } await Effect.runPromise(writeStateEffect(result.state)); return result.value; + } finally { + await release(); } - throw lastError instanceof Error ? lastError : new Error("state mutation failed after concurrent-writer retries"); }); } var INTERVAL_PATTERN = /^(\d+(?:\.\d+)?)\s*(s|sec|secs|second|seconds|m|min|mins|minute|minutes|h|hr|hrs|hour|hours|d|day|days)$/i; @@ -224,7 +268,7 @@ async function createLoop(sessionID, options) { const maxLoops = positiveIntegerOrNull(options.maxLoopsPerSession) ?? DEFAULT_MAX_LOOPS_PER_SESSION; const agent = typeof options.agent === "string" && options.agent.trim() ? options.agent.trim() : null; return mutate((state) => { - const open = Object.values(state.loops).filter((loop2) => loop2.sessionID === sessionID && isOpen(loop2.status)); + const open = Object.values(state.loops).filter((loop) => loop.sessionID === sessionID && isOpen(loop.status)); if (open.length >= maxLoops) { throw new Error(`this session already has ${open.length} open loop(s); stop one before creating another (limit ${maxLoops})`); } @@ -245,11 +289,18 @@ async function createLoop(sessionID, options) { lastRunAt: null, lastResult: null, lastError: null, + consecutiveFailures: 0, + nextRetryAt: null, + blockedReason: null, lastReason: null, runCount: 0, maxRuns, agent, - stopReason: null + stopReason: null, + ownerID: null, + ownerLeaseUntil: null, + ownerRevision: 0, + runClaimToken: 0 }; state.loops[id] = loop; return snapshot(loop); @@ -260,20 +311,104 @@ async function getLoop(loopID) { const loop = state.loops[loopID]; return loop ? snapshot(loop) : null; } -async function claimDueRun(loopID, leaseMs) { +function validLeaseMs(value) { + const leaseMs = positiveIntegerOrNull(Math.round(value)); + if (leaseMs == null) + throw new Error("session lease must be a positive number of milliseconds"); + return leaseMs; +} +async function acquireSessionLease(sessionID, instanceID, leaseMs) { + const duration = validLeaseMs(leaseMs); + return mutate((state) => { + const timestamp = now(); + const current = state.sessionLeases[sessionID]; + if (current && current.instanceID !== instanceID && current.expiresAt > timestamp) + return null; + const lease = { + sessionID, + instanceID, + expiresAt: timestamp + duration, + revision: (current?.revision ?? 0) + 1 + }; + state.sessionLeases[sessionID] = lease; + return { ...lease }; + }); +} +async function renewSessionLease(sessionID, instanceID, revision, leaseMs) { + const duration = validLeaseMs(leaseMs); + return mutate((state) => { + const timestamp = now(); + const current = state.sessionLeases[sessionID]; + if (!current || current.instanceID !== instanceID || current.revision !== revision || current.expiresAt <= timestamp) + return null; + current.expiresAt = timestamp + duration; + current.revision += 1; + return { ...current }; + }); +} +async function ownsSessionLease(sessionID, instanceID, revision) { + const state = await readState(); + const lease = state.sessionLeases[sessionID]; + return Boolean(lease && lease.instanceID === instanceID && lease.revision === revision && lease.expiresAt > now()); +} +async function releaseSessionLease(sessionID, instanceID, revision) { + return mutate((state) => { + const current = state.sessionLeases[sessionID]; + if (!current || current.instanceID !== instanceID || current.revision !== revision) + return false; + delete state.sessionLeases[sessionID]; + return true; + }); +} +async function acquireLoopOwner(loopID, ownerID, leaseMs) { + const lease = positiveIntegerOrNull(Math.round(leaseMs)); + if (!ownerID.trim()) + throw new Error("loop owner id must not be empty"); + if (lease == null) + throw new Error("owner lease must be a positive number of milliseconds"); + return mutate((state) => { + const loop = requireLoop(state, loopID); + const timestamp = now(); + if (loop.status !== "active") + return null; + if (loop.ownerID !== ownerID && loop.ownerLeaseUntil != null && loop.ownerLeaseUntil > timestamp) + return null; + if (loop.ownerID !== ownerID) + loop.ownerRevision += 1; + loop.ownerID = ownerID; + loop.ownerLeaseUntil = timestamp + lease; + loop.updatedAt = timestamp; + return { ownerID, ownerRevision: loop.ownerRevision }; + }); +} +async function claimDueRunOwned(loopID, owner, leaseMs) { const lease = positiveIntegerOrNull(Math.round(leaseMs)); if (lease == null) throw new Error("run claim lease must be a positive number of milliseconds"); return mutate((state) => { const loop = requireLoop(state, loopID); const timestamp = now(); + if (!matchesOwner(loop, owner, timestamp)) + return null; if (loop.status !== "active" || loop.nextRunAt == null || loop.nextRunAt > timestamp) return null; + loop.runClaimToken += 1; loop.nextRunAt = timestamp + lease; loop.updatedAt = timestamp; - return snapshot(loop); + return { ...owner, claimToken: loop.runClaimToken, loop: snapshot(loop) }; }); } +function matchesOwner(loop, owner, timestamp = now()) { + return loop.ownerID === owner.ownerID && loop.ownerRevision === owner.ownerRevision && (loop.ownerLeaseUntil ?? 0) > timestamp; +} +function matchesClaim(loop, claim, timestamp = now()) { + return matchesOwner(loop, claim, timestamp) && loop.runClaimToken === claim.claimToken; +} +async function confirmRunClaim(loopID, claim) { + const state = await readState(); + const loop = state.loops[loopID]; + return loop != null && matchesClaim(loop, claim) ? snapshot(loop) : null; +} async function listLoops(sessionID) { const state = await readState(); return Object.values(state.loops).filter((loop) => sessionID == null || loop.sessionID === sessionID).sort((a, b) => a.createdAt - b.createdAt).map(snapshot); @@ -293,6 +428,7 @@ async function pauseLoop(loopID) { throw new Error(`loop "${loopID}" is ${loop.status}; only active loops can be paused`); loop.status = "paused"; loop.nextRunAt = null; + loop.nextRetryAt = null; loop.stopReason = "paused"; loop.updatedAt = now(); return snapshot(loop); @@ -306,6 +442,9 @@ async function resumeLoop(loopID) { const timestamp = now(); loop.status = "active"; loop.stopReason = null; + loop.blockedReason = null; + loop.consecutiveFailures = 0; + loop.nextRetryAt = null; loop.nextRunAt = loop.mode === "interval" ? timestamp + loop.intervalMs : timestamp; loop.updatedAt = timestamp; return snapshot(loop); @@ -318,11 +457,26 @@ async function stopLoop(loopID, reason) { throw new Error(`loop "${loopID}" is already ${loop.status}`); loop.status = "stopped"; loop.nextRunAt = null; + loop.nextRetryAt = null; loop.stopReason = typeof reason === "string" && reason.trim() ? reason.trim().slice(0, 400) : "stopped"; loop.updatedAt = now(); return snapshot(loop); }); } +async function stopLoopIfUnchanged(loopID, expectedUpdatedAt, reason) { + return mutate((state) => { + const loop = state.loops[loopID]; + if (!loop) + return null; + if (loop.updatedAt !== expectedUpdatedAt || loop.status !== "active" || loop.mode !== "dynamic" || loop.nextRunAt != null) { + return null; + } + loop.status = "stopped"; + loop.stopReason = reason.trim().slice(0, 400) || "stopped"; + loop.updatedAt = now(); + return snapshot(loop); + }); +} async function stopLoopsForSession(sessionID, reason) { return mutate((state) => { const stopped = []; @@ -331,6 +485,7 @@ async function stopLoopsForSession(sessionID, reason) { continue; loop.status = "stopped"; loop.nextRunAt = null; + loop.nextRetryAt = null; loop.stopReason = reason; loop.updatedAt = now(); stopped.push(snapshot(loop)); @@ -360,55 +515,87 @@ async function scheduleNextRun(loopID, delayMs, reason) { throw new Error(`loop "${loopID}" is ${loop.status}; only active loops can be scheduled`); const timestamp = now(); loop.nextRunAt = timestamp + delay; + loop.nextRetryAt = null; loop.lastReason = typeof reason === "string" && reason.trim() ? reason.trim().slice(0, 400) : loop.lastReason; loop.updatedAt = timestamp; return snapshot(loop); }); } -async function recordRunSent(loopID) { +async function recordRunSentClaimed(loopID, claim) { return mutate((state) => { const loop = requireLoop(state, loopID); - if (loop.status !== "active") - return snapshot(loop); + if (!matchesClaim(loop, claim) || loop.status !== "active") + return null; const timestamp = now(); loop.runCount += 1; loop.lastRunAt = timestamp; loop.lastResult = "sent"; loop.lastError = null; + loop.consecutiveFailures = 0; + loop.nextRetryAt = null; + loop.blockedReason = null; loop.updatedAt = timestamp; if (loop.maxRuns != null && loop.runCount >= loop.maxRuns) { loop.status = "completed"; loop.nextRunAt = null; loop.stopReason = `max runs reached (${loop.maxRuns})`; - } else if (loop.mode === "interval") { + } else if (loop.mode === "interval") loop.nextRunAt = timestamp + loop.intervalMs; - } else { + else loop.nextRunAt = null; + return snapshot(loop); + }); +} +async function recordRunFailedClaimed(loopID, claim, _error, retryDelayMs, maxConsecutiveFailures = 5, maxRetryDelayMs = 60 * 60 * 1000) { + return mutate((state) => { + const loop = requireLoop(state, loopID); + if (!matchesClaim(loop, claim) || loop.status !== "active") + return null; + const timestamp = now(); + loop.lastResult = "failed"; + loop.lastError = "provider/model request failed"; + loop.consecutiveFailures += 1; + loop.updatedAt = timestamp; + if (loop.consecutiveFailures >= Math.max(1, Math.round(maxConsecutiveFailures))) { + loop.status = "paused"; + loop.nextRunAt = null; + loop.nextRetryAt = null; + loop.blockedReason = `paused after ${loop.consecutiveFailures} consecutive provider/model failures`; + loop.stopReason = loop.blockedReason; + } else { + const baseDelay = Math.max(1, Math.round(retryDelayMs)); + const cap = Math.max(baseDelay, Math.round(maxRetryDelayMs)); + const delay = Math.min(cap, baseDelay * 2 ** (loop.consecutiveFailures - 1)); + loop.nextRetryAt = timestamp + delay; + loop.nextRunAt = loop.nextRetryAt; + loop.blockedReason = null; } return snapshot(loop); }); } -async function recordRunDeferred(loopID, result, retryDelayMs) { +async function recordRunDeferredClaimed(loopID, claim, result, retryDelayMs) { return mutate((state) => { const loop = requireLoop(state, loopID); - if (loop.status !== "active") - return snapshot(loop); + if (!matchesClaim(loop, claim) || loop.status !== "active") + return null; const timestamp = now(); loop.lastResult = result; + loop.nextRetryAt = null; loop.nextRunAt = timestamp + Math.max(0, Math.round(retryDelayMs)); loop.updatedAt = timestamp; return snapshot(loop); }); } -async function recordRunFailed(loopID, error, retryDelayMs) { +async function stopLoopClaimed(loopID, claim, reason) { return mutate((state) => { const loop = requireLoop(state, loopID); - const timestamp = now(); - loop.lastResult = "failed"; - loop.lastError = error.slice(0, 400); - loop.updatedAt = timestamp; - if (loop.status === "active") - loop.nextRunAt = timestamp + Math.max(0, Math.round(retryDelayMs)); + if (!matchesClaim(loop, claim) || !isOpen(loop.status)) + return null; + loop.status = "stopped"; + loop.nextRunAt = null; + loop.nextRetryAt = null; + loop.stopReason = typeof reason === "string" && reason.trim() ? reason.trim().slice(0, 400) : "stopped"; + loop.updatedAt = now(); return snapshot(loop); }); } @@ -513,9 +700,13 @@ Preserve each loop's id, cadence, instruction, and status in the compacted conte var DEFAULT_COMMAND_NAME = "loop"; var DEFAULT_BUSY_BACKOFF_SECONDS = 60; var DEFAULT_FAILURE_BACKOFF_SECONDS = 60; +var DEFAULT_MAX_FAILURE_BACKOFF_SECONDS = 60 * 60; +var DEFAULT_MAX_CONSECUTIVE_FAILURES = 5; var DEFAULT_MAX_LOOP_AGE_DAYS = 7; var DEFAULT_DYNAMIC_MAX_DELAY_SECONDS = 24 * 60 * 60; var RUN_CLAIM_LEASE_MS = 30000; +var SESSION_LEASE_MS = 90000; +var SESSION_LEASE_HEARTBEAT_MS = 30000; var DEFAULT_RESTRICTED_AGENTS = ["plan"]; var LOOP_SYSTEM_MARKER = "OpenCode loop mode"; function commandNameFromOptions(options) { @@ -565,6 +756,21 @@ function isBusyEvent(event) { const status = event.properties?.status; return event.type === "session.status" && isRecord(status) && status.type === "busy"; } +function enqueueSessionOperation(queues, sessionID, operation) { + const previous = queues.get(sessionID) ?? Promise.resolve(); + const current = previous.then(operation, operation); + const settled = current.then(() => { + return; + }, () => { + return; + }); + queues.set(sessionID, settled); + settled.then(() => { + if (queues.get(sessionID) === settled) + queues.delete(sessionID); + }); + return current; +} async function toolResult(sessionID, extra = {}) { const loops = await listLoops(sessionID); return JSON.stringify({ ...extra, loops, report: formatLoops(loops) }, null, 2); @@ -576,13 +782,20 @@ var server = async ({ client }, options) => { const maxLoopsPerSession = positiveNumberOr(options?.max_loops_per_session, DEFAULT_MAX_LOOPS_PER_SESSION); const busyBackoffMs = positiveNumberOr(options?.busy_backoff_seconds, DEFAULT_BUSY_BACKOFF_SECONDS) * 1000; const failureBackoffMs = positiveNumberOr(options?.failure_backoff_seconds, DEFAULT_FAILURE_BACKOFF_SECONDS) * 1000; + const maxFailureBackoffMs = positiveNumberOr(options?.max_failure_backoff_seconds, DEFAULT_MAX_FAILURE_BACKOFF_SECONDS) * 1000; + const maxConsecutiveFailures = positiveNumberOr(options?.max_consecutive_failures, DEFAULT_MAX_CONSECUTIVE_FAILURES); const maxLoopAgeMs = nonNegativeNumberOr(options?.max_loop_age_days, DEFAULT_MAX_LOOP_AGE_DAYS) * 24 * 60 * 60 * 1000; const dynamicMaxDelaySeconds = positiveNumberOr(options?.dynamic_max_delay_seconds, DEFAULT_DYNAMIC_MAX_DELAY_SECONDS); const restrictedAgents = restrictedAgentSet(options); const timers = new Map; const sendingLoops = new Set; const busySessions = new Set; + const instanceID = randomUUID(); + const loopOwnerID = `server:${process.pid}:${randomUUID()}`; + const ownedSessions = new Map; + const ownershipQueues = new Map; const observedSessions = new Set; + const staleDynamicCandidates = new Map; const lastPromptAgentBySession = new Map; const dynamicPending = new Map; const isRestrictedAgent = (agent) => typeof agent === "string" && restrictedAgents.has(agent.trim().toLowerCase()); @@ -597,11 +810,48 @@ var server = async ({ client }, options) => { clearTimeout(timer); timers.delete(loopID); } - function scheduleTimer(loop) { + async function loseOwnership(sessionID) { + ownedSessions.delete(sessionID); + for (const loop of await activeLoops(sessionID)) + cancelTimer(loop.id); + for (const [loopID, pending] of dynamicPending) { + if (pending.sessionID === sessionID) + dynamicPending.delete(loopID); + } + } + async function acquireOwnership(sessionID) { + return enqueueSessionOperation(ownershipQueues, sessionID, async () => { + observedSessions.add(sessionID); + const lease = await acquireSessionLease(sessionID, instanceID, SESSION_LEASE_MS); + if (!lease) { + await loseOwnership(sessionID); + return false; + } + ownedSessions.set(sessionID, lease); + for (const loop of await activeLoops(sessionID)) + scheduleTimer(loop); + return true; + }); + } + async function renewOwnership(sessionID) { + return enqueueSessionOperation(ownershipQueues, sessionID, async () => { + const lease = ownedSessions.get(sessionID); + if (!lease) + return false; + const renewed = await renewSessionLease(sessionID, instanceID, lease.revision, SESSION_LEASE_MS); + if (!renewed) { + await loseOwnership(sessionID); + return false; + } + ownedSessions.set(sessionID, renewed); + return true; + }); + } + function scheduleTimer(loop, minimumDelayMs = 0) { cancelTimer(loop.id); - if (loop.status !== "active" || loop.nextRunAt == null) + if (loop.status !== "active" || loop.nextRunAt == null || !ownedSessions.has(loop.sessionID)) return; - const delay = Math.max(0, loop.nextRunAt - Date.now()); + const delay = Math.max(minimumDelayMs, loop.nextRunAt - Date.now()); const timer = setTimeout(() => { timers.delete(loop.id); runDue(loop.id); @@ -617,11 +867,8 @@ var server = async ({ client }, options) => { sendingLoops.add(loopID); try { await runDueLocked(loopID); - } catch (error) { - await log("error", "Loop iteration failed unexpectedly", { - loopID, - error: error instanceof Error ? error.message : String(error) - }); + } catch { + await log("error", "Loop scheduler operation failed", { loopID, category: "scheduler" }); } finally { sendingLoops.delete(loopID); } @@ -630,35 +877,62 @@ var server = async ({ client }, options) => { let loop = await getLoop(loopID); if (!loop || loop.status !== "active" || loop.nextRunAt == null) return; + if (!await renewOwnership(loop.sessionID)) + return; + const lease = ownedSessions.get(loop.sessionID); + if (!lease || !await ownsSessionLease(loop.sessionID, instanceID, lease.revision)) { + await loseOwnership(loop.sessionID); + return; + } if (loop.nextRunAt > Date.now()) { scheduleTimer(loop); return; } - const claimed = await claimDueRun(loopID, RUN_CLAIM_LEASE_MS); + if (!observedSessions.has(loop.sessionID)) { + scheduleTimer(loop, Math.min(loop.intervalMs ?? busyBackoffMs, busyBackoffMs)); + await log("info", "Skipping loop because session ownership is unknown", { loopID, category: "ownership" }); + return; + } + const owner = await acquireLoopOwner(loopID, loopOwnerID, RUN_CLAIM_LEASE_MS); + const claimed = owner ? await claimDueRunOwned(loopID, owner, RUN_CLAIM_LEASE_MS) : null; if (!claimed) { loop = await getLoop(loopID); if (loop) scheduleTimer(loop); return; } - loop = claimed; + loop = claimed.loop; if (maxLoopAgeMs > 0 && Date.now() - loop.createdAt >= maxLoopAgeMs) { - await stopLoop(loopID, `expired after ${Math.round(maxLoopAgeMs / 86400000)} days`); + await stopLoopClaimed(loopID, claimed, `expired after ${Math.round(maxLoopAgeMs / 86400000)} days`); return; } if (busySessions.has(loop.sessionID)) { - const deferred = await recordRunDeferred(loopID, "skipped_busy", Math.min(loop.intervalMs ?? busyBackoffMs, busyBackoffMs)); - scheduleTimer(deferred); + const deferred = await recordRunDeferredClaimed(loopID, claimed, "skipped_busy", Math.min(loop.intervalMs ?? busyBackoffMs, busyBackoffMs)); + if (deferred) + scheduleTimer(deferred); return; } if (isRestrictedAgent(lastPromptAgentBySession.get(loop.sessionID))) { - const deferred = await recordRunDeferred(loopID, "skipped_plan", Math.min(loop.intervalMs ?? busyBackoffMs, busyBackoffMs)); - scheduleTimer(deferred); + const deferred = await recordRunDeferredClaimed(loopID, claimed, "skipped_plan", Math.min(loop.intervalMs ?? busyBackoffMs, busyBackoffMs)); + if (deferred) + scheduleTimer(deferred); + return; + } + const injectionLease = ownedSessions.get(loop.sessionID); + if (!injectionLease || !await ownsSessionLease(loop.sessionID, instanceID, injectionLease.revision)) { + await loseOwnership(loop.sessionID); return; } if (loop.mode === "dynamic") { dynamicPending.set(loopID, { sessionID: loop.sessionID, sawBusy: false }); } + if (!await confirmRunClaim(loopID, claimed)) { + dynamicPending.delete(loopID); + const current = await getLoop(loopID); + if (current) + scheduleTimer(current); + return; + } try { await client.session.promptAsync({ path: { id: loop.sessionID }, @@ -669,27 +943,31 @@ var server = async ({ client }, options) => { }); } catch (error) { dynamicPending.delete(loopID); - if (!observedSessions.has(loop.sessionID)) { - await log("info", "Skipping loop for a session this process has not observed", { loopID, sessionID: loop.sessionID }); - return; + const failed = await recordRunFailedClaimed(loopID, claimed, error instanceof Error ? error.message : String(error), failureBackoffMs, maxConsecutiveFailures, maxFailureBackoffMs); + if (failed) + scheduleTimer(failed); + else { + const current = await getLoop(loopID); + if (current) + scheduleTimer(current); } - const failed = await recordRunFailed(loopID, error instanceof Error ? error.message : String(error), failureBackoffMs); - scheduleTimer(failed); - await log("error", "Loop iteration prompt failed", { loopID, error: failed.lastError ?? undefined }); + await log("error", "Loop iteration prompt failed", { loopID, error: failed?.lastError ?? undefined }); return; } busySessions.add(loop.sessionID); observedSessions.add(loop.sessionID); - const sent = await recordRunSent(loopID); + const sent = await recordRunSentClaimed(loopID, claimed); + if (!sent) + return; if (sent.mode !== "dynamic" || sent.status !== "active") dynamicPending.delete(loopID); scheduleTimer(sent); } async function runDueForSession(sessionID) { const loops = await activeLoops(sessionID); - const now2 = Date.now(); + const now = Date.now(); for (const loop of loops) { - if (loop.nextRunAt == null || loop.nextRunAt > now2) + if (loop.nextRunAt == null || loop.nextRunAt > now) continue; await runDue(loop.id); if (busySessions.has(sessionID)) @@ -716,15 +994,30 @@ var server = async ({ client }, options) => { const loops = await activeLoops(); for (const loop of loops) { if (loop.nextRunAt == null) { - if (loop.mode === "dynamic") - await stopLoop(loop.id, "not rescheduled before OpenCode restarted"); + if (loop.mode === "dynamic") { + staleDynamicCandidates.set(loop.id, { sessionID: loop.sessionID, updatedAt: loop.updatedAt }); + await log("info", "Dynamic loop is an orphaned/stale restart candidate pending session ownership", { + loopID: loop.id, + sessionID: loop.sessionID + }); + } continue; } - scheduleTimer(loop); } } - async function requireSessionLoop(loopID, sessionID) { + async function observeSession(sessionID) { observedSessions.add(sessionID); + if (!await acquireOwnership(sessionID)) + return; + for (const [loopID, candidate] of staleDynamicCandidates) { + if (candidate.sessionID !== sessionID) + continue; + staleDynamicCandidates.delete(loopID); + await stopLoopIfUnchanged(loopID, candidate.updatedAt, "not rescheduled before OpenCode restarted"); + } + } + async function requireSessionLoop(loopID, sessionID) { + await observeSession(sessionID); const loop = await getLoop(loopID); if (!loop) throw new Error(`no loop found with id "${loopID}"`); @@ -732,13 +1025,21 @@ var server = async ({ client }, options) => { throw new Error(`loop "${loopID}" belongs to a different session`); return loop; } - await rehydrate().catch((error) => log("error", "Failed to rehydrate loops", { error: error instanceof Error ? error.message : String(error) })); + await rehydrate().catch(() => log("error", "Failed to rehydrate loops", { category: "state" })); + const leaseHeartbeat = setInterval(() => { + for (const sessionID of ownedSessions.keys()) + renewOwnership(sessionID); + }, SESSION_LEASE_HEARTBEAT_MS); + leaseHeartbeat.unref?.(); return { async dispose() { + clearInterval(leaseHeartbeat); for (const timer of timers.values()) clearTimeout(timer); timers.clear(); dynamicPending.clear(); + await Promise.all([...ownedSessions.values()].map((lease) => releaseSessionLease(lease.sessionID, instanceID, lease.revision))); + ownedSessions.clear(); }, async config(config) { if (!registerCommand) @@ -755,7 +1056,7 @@ var server = async ({ client }, options) => { }, async execute(args, context) { const input = args; - observedSessions.add(context.sessionID); + await observeSession(context.sessionID); const dynamic = !input.interval?.trim(); const loop = await createLoop(context.sessionID, { prompt: input.instruction, @@ -777,7 +1078,6 @@ var server = async ({ client }, options) => { description: "List the loops for this OpenCode session, including status, cadence, run counts, and next scheduled run.", args: {}, async execute(_args, context) { - observedSessions.add(context.sessionID); return toolResult(context.sessionID); } }, @@ -868,7 +1168,7 @@ var server = async ({ client }, options) => { description: "Delete stopped and completed loops for this session. Active and paused loops are kept.", args: {}, async execute(_args, context) { - observedSessions.add(context.sessionID); + await observeSession(context.sessionID); const cleared = await clearClosedLoops(context.sessionID); return toolResult(context.sessionID, { cleared }); } @@ -879,7 +1179,7 @@ var server = async ({ client }, options) => { const agent = typeof input?.agent === "string" && input.agent.trim() ? input.agent : isRecord(output.message) && typeof output.message.agent === "string" ? output.message.agent : undefined; if (typeof sessionID !== "string") return; - observedSessions.add(sessionID); + await observeSession(sessionID); if (typeof agent !== "string" || !agent.trim()) return; lastPromptAgentBySession.set(sessionID, agent.trim()); @@ -887,6 +1187,7 @@ var server = async ({ client }, options) => { async "experimental.chat.system.transform"(input, output) { if (typeof input.sessionID !== "string") return; + await observeSession(input.sessionID); const loops = await openLoops(input.sessionID); const reminder = systemReminder(loops); if (!reminder) @@ -901,6 +1202,7 @@ var server = async ({ client }, options) => { ${reminder}`; }, async "experimental.session.compacting"(input, output) { + await observeSession(input.sessionID); const loops = await openLoops(input.sessionID); const context = compactionContext(loops); if (context) @@ -911,7 +1213,7 @@ ${reminder}`; const sessionID = sessionIDFromEvent(typed); if (!sessionID) return; - observedSessions.add(sessionID); + await observeSession(sessionID); if (isBusyEvent(typed)) { busySessions.add(sessionID); for (const pending of dynamicPending.values()) { @@ -963,13 +1265,20 @@ async function setupV2(context) { const maxLoopsPerSession = positiveNumberOr(options.max_loops_per_session, DEFAULT_MAX_LOOPS_PER_SESSION); const busyBackoffMs = positiveNumberOr(options.busy_backoff_seconds, DEFAULT_BUSY_BACKOFF_SECONDS) * 1000; const failureBackoffMs = positiveNumberOr(options.failure_backoff_seconds, DEFAULT_FAILURE_BACKOFF_SECONDS) * 1000; + const maxFailureBackoffMs = positiveNumberOr(options.max_failure_backoff_seconds, DEFAULT_MAX_FAILURE_BACKOFF_SECONDS) * 1000; + const maxConsecutiveFailures = positiveNumberOr(options.max_consecutive_failures, DEFAULT_MAX_CONSECUTIVE_FAILURES); const maxLoopAgeMs = nonNegativeNumberOr(options.max_loop_age_days, DEFAULT_MAX_LOOP_AGE_DAYS) * 24 * 60 * 60 * 1000; const dynamicMaxDelaySeconds = positiveNumberOr(options.dynamic_max_delay_seconds, DEFAULT_DYNAMIC_MAX_DELAY_SECONDS); const restrictedAgents = restrictedAgentSet(options); const timers = new Map; const sendingLoops = new Set; const busySessions = new Set; + const instanceID = randomUUID(); + const loopOwnerID = `server-v2:${process.pid}:${randomUUID()}`; + const ownedSessions = new Map; + const ownershipQueues = new Map; const observedSessions = new Set; + const staleDynamicCandidates = new Map; const lastPromptAgentBySession = new Map; const dynamicPending = new Map; const registrations = []; @@ -996,11 +1305,48 @@ async function setupV2(context) { clearTimeout(timer); timers.delete(loopID); } - function scheduleTimer(loop) { + async function loseOwnership(sessionID) { + ownedSessions.delete(sessionID); + for (const loop of await activeLoops(sessionID)) + cancelTimer(loop.id); + for (const [loopID, pending] of dynamicPending) { + if (pending.sessionID === sessionID) + dynamicPending.delete(loopID); + } + } + async function acquireOwnership(sessionID) { + return enqueueSessionOperation(ownershipQueues, sessionID, async () => { + observedSessions.add(sessionID); + const lease = await acquireSessionLease(sessionID, instanceID, SESSION_LEASE_MS); + if (!lease) { + await loseOwnership(sessionID); + return false; + } + ownedSessions.set(sessionID, lease); + for (const loop of await activeLoops(sessionID)) + scheduleTimer(loop); + return true; + }); + } + async function renewOwnership(sessionID) { + return enqueueSessionOperation(ownershipQueues, sessionID, async () => { + const lease = ownedSessions.get(sessionID); + if (!lease) + return false; + const renewed = await renewSessionLease(sessionID, instanceID, lease.revision, SESSION_LEASE_MS); + if (!renewed) { + await loseOwnership(sessionID); + return false; + } + ownedSessions.set(sessionID, renewed); + return true; + }); + } + function scheduleTimer(loop, minimumDelayMs = 0) { cancelTimer(loop.id); - if (loop.status !== "active" || loop.nextRunAt == null) + if (loop.status !== "active" || loop.nextRunAt == null || !ownedSessions.has(loop.sessionID)) return; - const delay = Math.max(0, loop.nextRunAt - Date.now()); + const delay = Math.max(minimumDelayMs, loop.nextRunAt - Date.now()); const timer = setTimeout(() => { timers.delete(loop.id); runDue(loop.id); @@ -1016,11 +1362,8 @@ async function setupV2(context) { sendingLoops.add(loopID); try { await runDueLocked(loopID); - } catch (error) { - v2Log("error", "Loop iteration failed unexpectedly", { - loopID, - error: error instanceof Error ? error.message : String(error) - }); + } catch { + v2Log("error", "Loop scheduler operation failed", { loopID, category: "scheduler" }); } finally { sendingLoops.delete(loopID); } @@ -1029,35 +1372,62 @@ async function setupV2(context) { let loop = await getLoop(loopID); if (!loop || loop.status !== "active" || loop.nextRunAt == null) return; + if (!await renewOwnership(loop.sessionID)) + return; + const lease = ownedSessions.get(loop.sessionID); + if (!lease || !await ownsSessionLease(loop.sessionID, instanceID, lease.revision)) { + await loseOwnership(loop.sessionID); + return; + } if (loop.nextRunAt > Date.now()) { scheduleTimer(loop); return; } - const claimed = await claimDueRun(loopID, RUN_CLAIM_LEASE_MS); + if (!observedSessions.has(loop.sessionID)) { + scheduleTimer(loop, Math.min(loop.intervalMs ?? busyBackoffMs, busyBackoffMs)); + v2Log("info", "Skipping loop because session ownership is unknown", { loopID, category: "ownership" }); + return; + } + const owner = await acquireLoopOwner(loopID, loopOwnerID, RUN_CLAIM_LEASE_MS); + const claimed = owner ? await claimDueRunOwned(loopID, owner, RUN_CLAIM_LEASE_MS) : null; if (!claimed) { loop = await getLoop(loopID); if (loop) scheduleTimer(loop); return; } - loop = claimed; + loop = claimed.loop; if (maxLoopAgeMs > 0 && Date.now() - loop.createdAt >= maxLoopAgeMs) { - await stopLoop(loopID, `expired after ${Math.round(maxLoopAgeMs / 86400000)} days`); + await stopLoopClaimed(loopID, claimed, `expired after ${Math.round(maxLoopAgeMs / 86400000)} days`); return; } if (await isSessionBusy(loop.sessionID)) { - const deferred = await recordRunDeferred(loopID, "skipped_busy", Math.min(loop.intervalMs ?? busyBackoffMs, busyBackoffMs)); - scheduleTimer(deferred); + const deferred = await recordRunDeferredClaimed(loopID, claimed, "skipped_busy", Math.min(loop.intervalMs ?? busyBackoffMs, busyBackoffMs)); + if (deferred) + scheduleTimer(deferred); return; } if (isRestrictedAgent(lastPromptAgentBySession.get(loop.sessionID))) { - const deferred = await recordRunDeferred(loopID, "skipped_plan", Math.min(loop.intervalMs ?? busyBackoffMs, busyBackoffMs)); - scheduleTimer(deferred); + const deferred = await recordRunDeferredClaimed(loopID, claimed, "skipped_plan", Math.min(loop.intervalMs ?? busyBackoffMs, busyBackoffMs)); + if (deferred) + scheduleTimer(deferred); + return; + } + const injectionLease = ownedSessions.get(loop.sessionID); + if (!injectionLease || !await ownsSessionLease(loop.sessionID, instanceID, injectionLease.revision)) { + await loseOwnership(loop.sessionID); return; } if (loop.mode === "dynamic") { dynamicPending.set(loopID, { sessionID: loop.sessionID, sawBusy: false }); } + if (!await confirmRunClaim(loopID, claimed)) { + dynamicPending.delete(loopID); + const current = await getLoop(loopID); + if (current) + scheduleTimer(current); + return; + } try { await context.session.prompt({ sessionID: loop.sessionID, @@ -1066,27 +1436,31 @@ async function setupV2(context) { }); } catch (error) { dynamicPending.delete(loopID); - if (!observedSessions.has(loop.sessionID)) { - v2Log("info", "Skipping loop for a session this process has not observed", { loopID, sessionID: loop.sessionID }); - return; + const failed = await recordRunFailedClaimed(loopID, claimed, error instanceof Error ? error.message : String(error), failureBackoffMs, maxConsecutiveFailures, maxFailureBackoffMs); + if (failed) + scheduleTimer(failed); + else { + const current = await getLoop(loopID); + if (current) + scheduleTimer(current); } - const failed = await recordRunFailed(loopID, error instanceof Error ? error.message : String(error), failureBackoffMs); - scheduleTimer(failed); - v2Log("error", "Loop iteration prompt failed", { loopID, error: failed.lastError ?? undefined }); + v2Log("error", "Loop iteration prompt failed", { loopID, error: failed?.lastError ?? undefined }); return; } busySessions.add(loop.sessionID); observedSessions.add(loop.sessionID); - const sent = await recordRunSent(loopID); + const sent = await recordRunSentClaimed(loopID, claimed); + if (!sent) + return; if (sent.mode !== "dynamic" || sent.status !== "active") dynamicPending.delete(loopID); scheduleTimer(sent); } async function runDueForSession(sessionID) { const loops = await activeLoops(sessionID); - const now2 = Date.now(); + const now = Date.now(); for (const loop of loops) { - if (loop.nextRunAt == null || loop.nextRunAt > now2) + if (loop.nextRunAt == null || loop.nextRunAt > now) continue; await runDue(loop.id); if (busySessions.has(sessionID)) @@ -1113,15 +1487,30 @@ async function setupV2(context) { const loops = await activeLoops(); for (const loop of loops) { if (loop.nextRunAt == null) { - if (loop.mode === "dynamic") - await stopLoop(loop.id, "not rescheduled before OpenCode restarted"); + if (loop.mode === "dynamic") { + staleDynamicCandidates.set(loop.id, { sessionID: loop.sessionID, updatedAt: loop.updatedAt }); + v2Log("info", "Dynamic loop is an orphaned/stale restart candidate pending session ownership", { + loopID: loop.id, + sessionID: loop.sessionID + }); + } continue; } - scheduleTimer(loop); } } - async function requireSessionLoop(loopID, sessionID) { + async function observeSession(sessionID) { observedSessions.add(sessionID); + if (!await acquireOwnership(sessionID)) + return; + for (const [loopID, candidate] of staleDynamicCandidates) { + if (candidate.sessionID !== sessionID) + continue; + staleDynamicCandidates.delete(loopID); + await stopLoopIfUnchanged(loopID, candidate.updatedAt, "not rescheduled before OpenCode restarted"); + } + } + async function requireSessionLoop(loopID, sessionID) { + await observeSession(sessionID); const loop = await getLoop(loopID); if (!loop) throw new Error(`no loop found with id "${loopID}"`); @@ -1134,7 +1523,7 @@ async function setupV2(context) { const sessionID = typeof data.sessionID === "string" ? data.sessionID : undefined; if (!sessionID) return; - observedSessions.add(sessionID); + await observeSession(sessionID); switch (event.type) { case "session.status": { const status = data.status; @@ -1187,10 +1576,12 @@ async function setupV2(context) { maxLoopsPerSession, dynamicMaxDelaySeconds, observedSessions, + observeSession, dynamicPending, scheduleTimer, cancelTimer, - requireSessionLoop + requireSessionLoop, + acquireOwnership }; if (registerCommand) { registrations.push(await context.command.transform((draft) => { @@ -1217,6 +1608,7 @@ async function setupV2(context) { draft.add(tool); })); registrations.push(await context.session.hook("context", async (sessionContext) => { + await observeSession(sessionContext.sessionID); const loops = await openLoops(sessionContext.sessionID); const reminder = systemReminder(loops); if (!reminder) @@ -1225,7 +1617,12 @@ async function setupV2(context) { return; sessionContext.system.push({ type: "text", text: reminder }); })); - await rehydrate().catch((error) => v2Log("error", "Failed to rehydrate loops", { error: error instanceof Error ? error.message : String(error) })); + await rehydrate().catch(() => v2Log("error", "Failed to rehydrate loops", { category: "state" })); + const leaseHeartbeat = setInterval(() => { + for (const sessionID of ownedSessions.keys()) + renewOwnership(sessionID); + }, SESSION_LEASE_HEARTBEAT_MS); + leaseHeartbeat.unref?.(); const abortController = new AbortController; let eventIterator; const consumer = (async () => { @@ -1239,20 +1636,21 @@ async function setupV2(context) { break; await handleV2Event(value); } - } catch (error) { + } catch { if (!abortController.signal.aborted) - v2Log("error", "V2 event consumer stopped", { - error: error instanceof Error ? error.message : String(error) - }); + v2Log("error", "V2 event consumer stopped", { category: "event" }); } })(); return async () => { + clearInterval(leaseHeartbeat); abortController.abort(); for (const timer of timers.values()) clearTimeout(timer); timers.clear(); dynamicPending.clear(); sendingLoops.clear(); + await Promise.all([...ownedSessions.values()].map((lease) => releaseSessionLease(lease.sessionID, instanceID, lease.revision))); + ownedSessions.clear(); for (const registration of registrations) await registration.dispose(); const termination = Promise.allSettled([consumer, eventIterator?.return?.()]); @@ -1284,7 +1682,7 @@ function loopToolsV2(services) { options: { codemode: false }, execute: async (args, toolContext) => { const input = args; - services.observedSessions.add(toolContext.sessionID); + await services.observeSession(toolContext.sessionID); const dynamic = !input.interval?.trim(); const loop = await createLoop(toolContext.sessionID, { prompt: input.instruction, @@ -1308,7 +1706,6 @@ function loopToolsV2(services) { input: v2ObjectSchema({}), options: { codemode: false }, execute: async (_args, toolContext) => { - services.observedSessions.add(toolContext.sessionID); return { content: await toolResult(toolContext.sessionID) }; } }, @@ -1419,7 +1816,7 @@ function loopToolsV2(services) { input: v2ObjectSchema({}), options: { codemode: false }, execute: async (_args, toolContext) => { - services.observedSessions.add(toolContext.sessionID); + await services.observeSession(toolContext.sessionID); const cleared = await clearClosedLoops(toolContext.sessionID); return { content: await toolResult(toolContext.sessionID, { cleared }) }; } diff --git a/src/server.ts b/src/server.ts index 69dd28d..489c3bf 100644 --- a/src/server.ts +++ b/src/server.ts @@ -2,28 +2,38 @@ import type { Config, Plugin } from "@opencode-ai/plugin" import type * as PluginV2 from "@opencode-ai/plugin-v2" import type { Info as ToolV2Info } from "@opencode-ai/plugin-v2/promise/tool" import type { Tool as ToolSchema } from "@opencode-ai/schema/tool" +import { randomUUID } from "node:crypto" import { z } from "zod" import { DEFAULT_MAX_LOOPS_PER_SESSION, DEFAULT_MIN_INTERVAL_SECONDS, MAX_PROMPT_CHARS, activeLoops, - claimDueRun, + acquireLoopOwner, + acquireSessionLease, + claimDueRunOwned, clearClosedLoops, + confirmRunClaim, createLoop, formatLoops, getLoop, listLoops, openLoops, + ownsSessionLease, parseInterval, pauseLoop, resumeLoop, scheduleNextRun, stopLoop, + stopLoopClaimed, + stopLoopIfUnchanged, stopLoopsForSession, - recordRunDeferred, - recordRunFailed, - recordRunSent, + recordRunDeferredClaimed, + recordRunFailedClaimed, + recordRunSentClaimed, + releaseSessionLease, + renewSessionLease, + type SessionLease, type LoopSnapshot, } from "./state" import { compactionContext, iterationPrompt, loopCommandTemplate, systemReminder } from "./prompts" @@ -35,6 +45,8 @@ type Options = { max_loops_per_session?: number busy_backoff_seconds?: number failure_backoff_seconds?: number + max_failure_backoff_seconds?: number + max_consecutive_failures?: number max_loop_age_days?: number dynamic_max_delay_seconds?: number restricted_agents?: string[] @@ -43,9 +55,13 @@ type Options = { const DEFAULT_COMMAND_NAME = "loop" const DEFAULT_BUSY_BACKOFF_SECONDS = 60 const DEFAULT_FAILURE_BACKOFF_SECONDS = 60 +const DEFAULT_MAX_FAILURE_BACKOFF_SECONDS = 60 * 60 +const DEFAULT_MAX_CONSECUTIVE_FAILURES = 5 const DEFAULT_MAX_LOOP_AGE_DAYS = 7 const DEFAULT_DYNAMIC_MAX_DELAY_SECONDS = 24 * 60 * 60 const RUN_CLAIM_LEASE_MS = 30_000 +const SESSION_LEASE_MS = 90_000 +const SESSION_LEASE_HEARTBEAT_MS = 30_000 const DEFAULT_RESTRICTED_AGENTS = ["plan"] const LOOP_SYSTEM_MARKER = "OpenCode loop mode" @@ -100,6 +116,21 @@ function isBusyEvent(event: { type?: string; properties?: Record( + queues: Map>, + sessionID: string, + operation: () => Promise, +) { + const previous = queues.get(sessionID) ?? Promise.resolve() + const current = previous.then(operation, operation) + const settled = current.then(() => undefined, () => undefined) + queues.set(sessionID, settled) + void settled.then(() => { + if (queues.get(sessionID) === settled) queues.delete(sessionID) + }) + return current +} + async function toolResult(sessionID: string, extra: Record = {}) { const loops = await listLoops(sessionID) return JSON.stringify({ ...extra, loops, report: formatLoops(loops) }, null, 2) @@ -112,6 +143,9 @@ const server: Plugin = async ({ client }, options?: Options) => { const maxLoopsPerSession = positiveNumberOr(options?.max_loops_per_session, DEFAULT_MAX_LOOPS_PER_SESSION) const busyBackoffMs = positiveNumberOr(options?.busy_backoff_seconds, DEFAULT_BUSY_BACKOFF_SECONDS) * 1000 const failureBackoffMs = positiveNumberOr(options?.failure_backoff_seconds, DEFAULT_FAILURE_BACKOFF_SECONDS) * 1000 + const maxFailureBackoffMs = + positiveNumberOr(options?.max_failure_backoff_seconds, DEFAULT_MAX_FAILURE_BACKOFF_SECONDS) * 1000 + const maxConsecutiveFailures = positiveNumberOr(options?.max_consecutive_failures, DEFAULT_MAX_CONSECUTIVE_FAILURES) const maxLoopAgeMs = nonNegativeNumberOr(options?.max_loop_age_days, DEFAULT_MAX_LOOP_AGE_DAYS) * 24 * 60 * 60 * 1000 const dynamicMaxDelaySeconds = positiveNumberOr(options?.dynamic_max_delay_seconds, DEFAULT_DYNAMIC_MAX_DELAY_SECONDS) const restrictedAgents = restrictedAgentSet(options) @@ -119,10 +153,18 @@ const server: Plugin = async ({ client }, options?: Options) => { const timers = new Map>() const sendingLoops = new Set() const busySessions = new Set() + const instanceID = randomUUID() + const loopOwnerID = `server:${process.pid}:${randomUUID()}` + const ownedSessions = new Map() + const ownershipQueues = new Map>() // Sessions this process has seen through events, prompts, or tool calls. Used // 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() + // 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. + const staleDynamicCandidates = new Map() const lastPromptAgentBySession = new Map() // Dynamic loops whose latest injected (or creating) turn has not yet gone idle: // if that turn ends without schedule_next_run or stop_loop, the loop ends. @@ -143,10 +185,46 @@ const server: Plugin = async ({ client }, options?: Options) => { timers.delete(loopID) } - function scheduleTimer(loop: LoopSnapshot) { + async function loseOwnership(sessionID: string) { + ownedSessions.delete(sessionID) + for (const loop of await activeLoops(sessionID)) cancelTimer(loop.id) + for (const [loopID, pending] of dynamicPending) { + if (pending.sessionID === sessionID) dynamicPending.delete(loopID) + } + } + + async function acquireOwnership(sessionID: string) { + return enqueueSessionOperation(ownershipQueues, sessionID, async () => { + observedSessions.add(sessionID) + const lease = await acquireSessionLease(sessionID, instanceID, SESSION_LEASE_MS) + if (!lease) { + await loseOwnership(sessionID) + return false + } + ownedSessions.set(sessionID, lease) + for (const loop of await activeLoops(sessionID)) scheduleTimer(loop) + return true + }) + } + + async function renewOwnership(sessionID: string) { + return enqueueSessionOperation(ownershipQueues, sessionID, async () => { + const lease = ownedSessions.get(sessionID) + if (!lease) return false + const renewed = await renewSessionLease(sessionID, instanceID, lease.revision, SESSION_LEASE_MS) + if (!renewed) { + await loseOwnership(sessionID) + return false + } + ownedSessions.set(sessionID, renewed) + return true + }) + } + + function scheduleTimer(loop: LoopSnapshot, minimumDelayMs = 0) { cancelTimer(loop.id) - if (loop.status !== "active" || loop.nextRunAt == null) return - const delay = Math.max(0, loop.nextRunAt - Date.now()) + if (loop.status !== "active" || loop.nextRunAt == null || !ownedSessions.has(loop.sessionID)) return + const delay = Math.max(minimumDelayMs, loop.nextRunAt - Date.now()) const timer = setTimeout(() => { timers.delete(loop.id) void runDue(loop.id) @@ -161,11 +239,8 @@ const server: Plugin = async ({ client }, options?: Options) => { sendingLoops.add(loopID) try { await runDueLocked(loopID) - } catch (error) { - await log("error", "Loop iteration failed unexpectedly", { - loopID, - error: error instanceof Error ? error.message : String(error), - }) + } catch { + await log("error", "Loop scheduler operation failed", { loopID, category: "scheduler" }) } finally { sendingLoops.delete(loopID) } @@ -174,29 +249,50 @@ const server: Plugin = async ({ client }, options?: Options) => { async function runDueLocked(loopID: string) { let loop = await getLoop(loopID) if (!loop || loop.status !== "active" || loop.nextRunAt == null) return + if (!(await renewOwnership(loop.sessionID))) return + const lease = ownedSessions.get(loop.sessionID) + if (!lease || !(await ownsSessionLease(loop.sessionID, instanceID, lease.revision))) { + await loseOwnership(loop.sessionID) + return + } if (loop.nextRunAt > Date.now()) { scheduleTimer(loop) return } - const claimed = await claimDueRun(loopID, RUN_CLAIM_LEASE_MS) + if (!observedSessions.has(loop.sessionID)) { + // 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 + // allow the scheduler to take ownership instead of stranding the loop. + scheduleTimer(loop, Math.min(loop.intervalMs ?? busyBackoffMs, busyBackoffMs)) + await log("info", "Skipping loop because session ownership is unknown", { loopID, category: "ownership" }) + return + } + const owner = await acquireLoopOwner(loopID, loopOwnerID, RUN_CLAIM_LEASE_MS) + const claimed = owner ? await claimDueRunOwned(loopID, owner, RUN_CLAIM_LEASE_MS) : null if (!claimed) { loop = await getLoop(loopID) if (loop) scheduleTimer(loop) return } - loop = claimed + loop = claimed.loop if (maxLoopAgeMs > 0 && Date.now() - loop.createdAt >= maxLoopAgeMs) { - await stopLoop(loopID, `expired after ${Math.round(maxLoopAgeMs / 86_400_000)} days`) + await stopLoopClaimed(loopID, claimed, `expired after ${Math.round(maxLoopAgeMs / 86_400_000)} days`) return } if (busySessions.has(loop.sessionID)) { - const deferred = await recordRunDeferred(loopID, "skipped_busy", Math.min(loop.intervalMs ?? busyBackoffMs, busyBackoffMs)) - scheduleTimer(deferred) + const deferred = await recordRunDeferredClaimed(loopID, claimed, "skipped_busy", Math.min(loop.intervalMs ?? busyBackoffMs, busyBackoffMs)) + if (deferred) scheduleTimer(deferred) return } if (isRestrictedAgent(lastPromptAgentBySession.get(loop.sessionID))) { - const deferred = await recordRunDeferred(loopID, "skipped_plan", Math.min(loop.intervalMs ?? busyBackoffMs, busyBackoffMs)) - scheduleTimer(deferred) + const deferred = await recordRunDeferredClaimed(loopID, claimed, "skipped_plan", Math.min(loop.intervalMs ?? busyBackoffMs, busyBackoffMs)) + if (deferred) scheduleTimer(deferred) + return + } + const injectionLease = ownedSessions.get(loop.sessionID) + if (!injectionLease || !(await ownsSessionLease(loop.sessionID, instanceID, injectionLease.revision))) { + await loseOwnership(loop.sessionID) return } // Register before injecting: the injected turn's busy event can arrive while @@ -206,6 +302,12 @@ const server: Plugin = async ({ client }, options?: Options) => { if (loop.mode === "dynamic") { dynamicPending.set(loopID, { sessionID: loop.sessionID, sawBusy: false }) } + if (!(await confirmRunClaim(loopID, claimed))) { + dynamicPending.delete(loopID) + const current = await getLoop(loopID) + if (current) scheduleTimer(current) + return + } try { await client.session.promptAsync({ path: { id: loop.sessionID }, @@ -216,20 +318,26 @@ const server: Plugin = async ({ client }, options?: Options) => { }) } catch (error) { dynamicPending.delete(loopID) - if (!observedSessions.has(loop.sessionID)) { - // Likely a session owned by another OpenCode process sharing the state - // file: leave its record alone and stop driving it from this process. - await log("info", "Skipping loop for a session this process has not observed", { loopID, sessionID: loop.sessionID }) - return + const failed = await recordRunFailedClaimed( + loopID, + claimed, + error instanceof Error ? error.message : String(error), + failureBackoffMs, + maxConsecutiveFailures, + maxFailureBackoffMs, + ) + if (failed) scheduleTimer(failed) + else { + const current = await getLoop(loopID) + if (current) scheduleTimer(current) } - const failed = await recordRunFailed(loopID, error instanceof Error ? error.message : String(error), failureBackoffMs) - scheduleTimer(failed) - await log("error", "Loop iteration prompt failed", { loopID, error: failed.lastError ?? undefined }) + await log("error", "Loop iteration prompt failed", { loopID, error: failed?.lastError ?? undefined }) return } busySessions.add(loop.sessionID) observedSessions.add(loop.sessionID) - const sent = await recordRunSent(loopID) + const sent = await recordRunSentClaimed(loopID, claimed) + if (!sent) return if (sent.mode !== "dynamic" || sent.status !== "active") dynamicPending.delete(loopID) scheduleTimer(sent) } @@ -258,34 +366,58 @@ const server: Plugin = async ({ client }, options?: Options) => { } async function rehydrate() { + // Persisted loops are only observed here. A fresh process must receive an + // authoritative lifecycle signal or an explicit tool invocation before it + // may acquire the session lease and arm timers. const loops = await activeLoops() for (const loop of loops) { if (loop.nextRunAt == null) { - // A dynamic loop whose scheduling turn died with the previous process cannot recover on its own. - if (loop.mode === "dynamic") await stopLoop(loop.id, "not rescheduled before OpenCode restarted") + if (loop.mode === "dynamic") { + staleDynamicCandidates.set(loop.id, { sessionID: loop.sessionID, updatedAt: loop.updatedAt }) + await log("info", "Dynamic loop is an orphaned/stale restart candidate pending session ownership", { + loopID: loop.id, + sessionID: loop.sessionID, + }) + } continue } - scheduleTimer(loop) } } - async function requireSessionLoop(loopID: string, sessionID: string) { + async function observeSession(sessionID: string) { observedSessions.add(sessionID) + if (!(await acquireOwnership(sessionID))) return + for (const [loopID, candidate] of staleDynamicCandidates) { + if (candidate.sessionID !== sessionID) continue + staleDynamicCandidates.delete(loopID) + await stopLoopIfUnchanged(loopID, candidate.updatedAt, "not rescheduled before OpenCode restarted") + } + } + + async function requireSessionLoop(loopID: string, sessionID: string) { + await observeSession(sessionID) const loop = await getLoop(loopID) if (!loop) throw new Error(`no loop found with id "${loopID}"`) if (loop.sessionID !== sessionID) throw new Error(`loop "${loopID}" belongs to a different session`) return loop } - await rehydrate().catch((error) => - log("error", "Failed to rehydrate loops", { error: error instanceof Error ? error.message : String(error) }), - ) + await rehydrate().catch(() => log("error", "Failed to rehydrate loops", { category: "state" })) + const leaseHeartbeat = setInterval(() => { + for (const sessionID of ownedSessions.keys()) void renewOwnership(sessionID) + }, SESSION_LEASE_HEARTBEAT_MS) + leaseHeartbeat.unref?.() return { async dispose() { + clearInterval(leaseHeartbeat) for (const timer of timers.values()) clearTimeout(timer) timers.clear() dynamicPending.clear() + await Promise.all( + [...ownedSessions.values()].map((lease) => releaseSessionLease(lease.sessionID, instanceID, lease.revision)), + ) + ownedSessions.clear() }, async config(config) { if (!registerCommand) return @@ -305,7 +437,7 @@ const server: Plugin = async ({ client }, options?: Options) => { }, async execute(args, context) { const input = args as { instruction: string; interval?: string; max_runs?: number } - observedSessions.add(context.sessionID) + await observeSession(context.sessionID) const dynamic = !input.interval?.trim() const loop = await createLoop(context.sessionID, { prompt: input.instruction, @@ -327,7 +459,6 @@ const server: Plugin = async ({ client }, options?: Options) => { description: "List the loops for this OpenCode session, including status, cadence, run counts, and next scheduled run.", args: {}, async execute(_args, context) { - observedSessions.add(context.sessionID) return toolResult(context.sessionID) }, }, @@ -420,7 +551,7 @@ const server: Plugin = async ({ client }, options?: Options) => { description: "Delete stopped and completed loops for this session. Active and paused loops are kept.", args: {}, async execute(_args, context) { - observedSessions.add(context.sessionID) + await observeSession(context.sessionID) const cleared = await clearClosedLoops(context.sessionID) return toolResult(context.sessionID, { cleared }) }, @@ -440,12 +571,13 @@ const server: Plugin = async ({ client }, options?: Options) => { ? output.message.agent : undefined if (typeof sessionID !== "string") return - observedSessions.add(sessionID) + await observeSession(sessionID) if (typeof agent !== "string" || !agent.trim()) return lastPromptAgentBySession.set(sessionID, agent.trim()) }, async "experimental.chat.system.transform"(input, output) { if (typeof input.sessionID !== "string") return + await observeSession(input.sessionID) const loops = await openLoops(input.sessionID) const reminder = systemReminder(loops) if (!reminder) return @@ -454,6 +586,7 @@ const server: Plugin = async ({ client }, options?: Options) => { else output.system[0] = `${output.system[0]}\n\n${reminder}` }, async "experimental.session.compacting"(input, output) { + await observeSession(input.sessionID) const loops = await openLoops(input.sessionID) const context = compactionContext(loops) if (context) output.context.push(context) @@ -462,7 +595,7 @@ const server: Plugin = async ({ client }, options?: Options) => { const typed = event as { type?: string; properties?: Record } const sessionID = sessionIDFromEvent(typed) if (!sessionID) return - observedSessions.add(sessionID) + await observeSession(sessionID) if (isBusyEvent(typed)) { busySessions.add(sessionID) for (const pending of dynamicPending.values()) { @@ -519,10 +652,12 @@ type LoopServices = { maxLoopsPerSession: number dynamicMaxDelaySeconds: number observedSessions: Set + observeSession: (sessionID: string) => Promise dynamicPending: Map scheduleTimer: (loop: LoopSnapshot) => void cancelTimer: (loopID: string) => void requireSessionLoop: (loopID: string, sessionID: string) => Promise + acquireOwnership: (sessionID: string) => Promise } async function setupV2(context: PluginV2.Plugin.Context): Promise { @@ -533,6 +668,8 @@ async function setupV2(context: PluginV2.Plugin.Context): Promise>() const sendingLoops = new Set() const busySessions = new Set() + const instanceID = randomUUID() + const loopOwnerID = `server-v2:${process.pid}:${randomUUID()}` + const ownedSessions = new Map() + const ownershipQueues = new Map>() // Sessions this process has seen through events, prompts, or tool calls. Used // 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() + // 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() const lastPromptAgentBySession = new Map() // Dynamic loops whose latest injected (or creating) turn has not yet gone idle: // if that turn ends without schedule_next_run or stop_loop, the loop ends. @@ -575,10 +719,46 @@ async function setupV2(context: PluginV2.Plugin.Context): Promise { + observedSessions.add(sessionID) + const lease = await acquireSessionLease(sessionID, instanceID, SESSION_LEASE_MS) + if (!lease) { + await loseOwnership(sessionID) + return false + } + ownedSessions.set(sessionID, lease) + for (const loop of await activeLoops(sessionID)) scheduleTimer(loop) + return true + }) + } + + async function renewOwnership(sessionID: string) { + return enqueueSessionOperation(ownershipQueues, sessionID, async () => { + const lease = ownedSessions.get(sessionID) + if (!lease) return false + const renewed = await renewSessionLease(sessionID, instanceID, lease.revision, SESSION_LEASE_MS) + if (!renewed) { + await loseOwnership(sessionID) + return false + } + ownedSessions.set(sessionID, renewed) + return true + }) + } + + function scheduleTimer(loop: LoopSnapshot, minimumDelayMs = 0) { cancelTimer(loop.id) - if (loop.status !== "active" || loop.nextRunAt == null) return - const delay = Math.max(0, loop.nextRunAt - Date.now()) + if (loop.status !== "active" || loop.nextRunAt == null || !ownedSessions.has(loop.sessionID)) return + const delay = Math.max(minimumDelayMs, loop.nextRunAt - Date.now()) const timer = setTimeout(() => { timers.delete(loop.id) void runDue(loop.id) @@ -593,11 +773,8 @@ async function setupV2(context: PluginV2.Plugin.Context): Promise Date.now()) { scheduleTimer(loop) return } - const claimed = await claimDueRun(loopID, RUN_CLAIM_LEASE_MS) + if (!observedSessions.has(loop.sessionID)) { + // Keep a bounded polling timer while ownership is unknown. The loop stays + // untouched on disk, but can run after any later event or tool call makes + // this process the observed owner of its session. + scheduleTimer(loop, Math.min(loop.intervalMs ?? busyBackoffMs, busyBackoffMs)) + v2Log("info", "Skipping loop because session ownership is unknown", { loopID, category: "ownership" }) + return + } + const owner = await acquireLoopOwner(loopID, loopOwnerID, RUN_CLAIM_LEASE_MS) + const claimed = owner ? await claimDueRunOwned(loopID, owner, RUN_CLAIM_LEASE_MS) : null if (!claimed) { loop = await getLoop(loopID) if (loop) scheduleTimer(loop) return } - loop = claimed + loop = claimed.loop if (maxLoopAgeMs > 0 && Date.now() - loop.createdAt >= maxLoopAgeMs) { - await stopLoop(loopID, `expired after ${Math.round(maxLoopAgeMs / 86_400_000)} days`) + await stopLoopClaimed(loopID, claimed, `expired after ${Math.round(maxLoopAgeMs / 86_400_000)} days`) return } if (await isSessionBusy(loop.sessionID)) { - const deferred = await recordRunDeferred(loopID, "skipped_busy", Math.min(loop.intervalMs ?? busyBackoffMs, busyBackoffMs)) - scheduleTimer(deferred) + const deferred = await recordRunDeferredClaimed(loopID, claimed, "skipped_busy", Math.min(loop.intervalMs ?? busyBackoffMs, busyBackoffMs)) + if (deferred) scheduleTimer(deferred) return } if (isRestrictedAgent(lastPromptAgentBySession.get(loop.sessionID))) { - const deferred = await recordRunDeferred(loopID, "skipped_plan", Math.min(loop.intervalMs ?? busyBackoffMs, busyBackoffMs)) - scheduleTimer(deferred) + const deferred = await recordRunDeferredClaimed(loopID, claimed, "skipped_plan", Math.min(loop.intervalMs ?? busyBackoffMs, busyBackoffMs)) + if (deferred) scheduleTimer(deferred) + return + } + const injectionLease = ownedSessions.get(loop.sessionID) + if (!injectionLease || !(await ownsSessionLease(loop.sessionID, instanceID, injectionLease.revision))) { + await loseOwnership(loop.sessionID) return } // Register before injecting: the injected turn's busy event can arrive while @@ -638,6 +835,12 @@ async function setupV2(context: PluginV2.Plugin.Context): Promise { + await observeSession(sessionContext.sessionID) const loops = await openLoops(sessionContext.sessionID) const reminder = systemReminder(loops) if (!reminder) return @@ -810,9 +1036,11 @@ async function setupV2(context: PluginV2.Plugin.Context): Promise - v2Log("error", "Failed to rehydrate loops", { error: error instanceof Error ? error.message : String(error) }), - ) + await rehydrate().catch(() => v2Log("error", "Failed to rehydrate loops", { category: "state" })) + const leaseHeartbeat = setInterval(() => { + for (const sessionID of ownedSessions.keys()) void renewOwnership(sessionID) + }, SESSION_LEASE_HEARTBEAT_MS) + leaseHeartbeat.unref?.() const abortController = new AbortController() let eventIterator: AsyncIterator | undefined @@ -826,19 +1054,22 @@ async function setupV2(context: PluginV2.Plugin.Context): Promise { + clearInterval(leaseHeartbeat) abortController.abort() for (const timer of timers.values()) clearTimeout(timer) timers.clear() dynamicPending.clear() sendingLoops.clear() + await Promise.all( + [...ownedSessions.values()].map((lease) => releaseSessionLease(lease.sessionID, instanceID, lease.revision)), + ) + ownedSessions.clear() for (const registration of registrations) await registration.dispose() // Best-effort termination of the event consumer. Never block plugin // unload on a stream that does not close promptly. @@ -876,7 +1107,7 @@ function loopToolsV2(services: LoopServices): ToolV2Info[] { options: { codemode: false }, execute: async (args, toolContext) => { const input = args as { instruction: string; interval?: string; max_runs?: number } - services.observedSessions.add(toolContext.sessionID) + await services.observeSession(toolContext.sessionID) const dynamic = !input.interval?.trim() const loop = await createLoop(toolContext.sessionID, { prompt: input.instruction, @@ -900,7 +1131,6 @@ function loopToolsV2(services: LoopServices): ToolV2Info[] { input: v2ObjectSchema({}), options: { codemode: false }, execute: async (_args, toolContext) => { - services.observedSessions.add(toolContext.sessionID) return { content: await toolResult(toolContext.sessionID) } }, }, @@ -1028,7 +1258,7 @@ function loopToolsV2(services: LoopServices): ToolV2Info[] { input: v2ObjectSchema({}), options: { codemode: false }, execute: async (_args, toolContext) => { - services.observedSessions.add(toolContext.sessionID) + await services.observeSession(toolContext.sessionID) const cleared = await clearClosedLoops(toolContext.sessionID) return { content: await toolResult(toolContext.sessionID, { cleared }) } }, diff --git a/src/state.ts b/src/state.ts index 9c74675..d71e364 100644 --- a/src/state.ts +++ b/src/state.ts @@ -1,4 +1,4 @@ -import { chmod, mkdir, readFile, rename, writeFile } from "node:fs/promises" +import { chmod, mkdir, readFile, rename, rm, stat, writeFile } from "node:fs/promises" import { homedir } from "node:os" import { dirname, join } from "node:path" import { Data, Effect, Schema } from "effect" @@ -7,6 +7,13 @@ export type LoopStatus = "active" | "paused" | "stopped" | "completed" export type LoopMode = "interval" | "dynamic" export type LoopRunResult = "sent" | "skipped_busy" | "skipped_plan" | "failed" +export type SessionLease = { + sessionID: string + instanceID: string + expiresAt: number + revision: number +} + export type CreateLoopOptions = { prompt: string intervalMs?: number | null @@ -29,16 +36,27 @@ export type Loop = { lastRunAt: number | null lastResult: LoopRunResult | null lastError: string | null + consecutiveFailures: number + nextRetryAt: number | null + blockedReason: string | null lastReason: string | null runCount: number maxRuns: number | null agent: string | null stopReason: string | null + ownerID: string | null + ownerLeaseUntil: number | null + ownerRevision: number + runClaimToken: number } +export type LoopOwnerLease = { ownerID: string; ownerRevision: number } +export type RunClaim = LoopOwnerLease & { claimToken: number; loop: LoopSnapshot } + type State = { version: 1 loops: Record + sessionLeases: Record } class StateReadError extends Data.TaggedError("StateReadError")<{ @@ -75,15 +93,31 @@ const LoopSchema = Schema.Struct({ default: () => null, }), lastError: Schema.optionalWith(NullableString, { default: () => null }), + consecutiveFailures: Schema.optionalWith(Schema.Number, { default: () => 0 }), + nextRetryAt: Schema.optionalWith(NullableNumber, { default: () => null }), + blockedReason: Schema.optionalWith(NullableString, { default: () => null }), lastReason: Schema.optionalWith(NullableString, { default: () => null }), runCount: Schema.optionalWith(Schema.Number, { default: () => 0 }), maxRuns: Schema.optionalWith(NullableNumber, { default: () => null }), agent: Schema.optionalWith(NullableString, { default: () => null }), stopReason: Schema.optionalWith(NullableString, { default: () => null }), + ownerID: Schema.optionalWith(NullableString, { default: () => null }), + ownerLeaseUntil: Schema.optionalWith(NullableNumber, { default: () => null }), + ownerRevision: Schema.optionalWith(Schema.Number, { default: () => 0 }), + runClaimToken: Schema.optionalWith(Schema.Number, { default: () => 0 }), +}) +const SessionLeaseSchema = Schema.Struct({ + sessionID: Schema.String, + instanceID: Schema.String, + expiresAt: Schema.Number, + revision: Schema.Number, }) const StateSchema = Schema.Struct({ version: Schema.Literal(1), loops: Schema.Record({ key: Schema.String, value: LoopSchema }), + sessionLeases: Schema.optionalWith(Schema.Record({ key: Schema.String, value: SessionLeaseSchema }), { + default: () => ({}), + }), }) export type LoopSnapshot = Loop & { @@ -106,7 +140,7 @@ function now() { } function emptyState(): State { - return { version: 1, loops: {} } + return { version: 1, loops: {}, sessionLeases: {} } } function isMissingStateFile(error: unknown) { @@ -171,8 +205,6 @@ function enqueueMutation(operation: () => Promise) { return current } -const MAX_MUTATION_ATTEMPTS = 5 - async function readRawState() { try { return await readFile(statePath(), "utf8") @@ -182,13 +214,46 @@ async function readRawState() { } } +const LOCK_STALE_MS = 30_000 +const LOCK_RETRY_MS = 10 + +function sleep(ms: number) { + return new Promise((resolve) => setTimeout(resolve, ms)) +} + +async function acquireMutationLock() { + const file = statePath() + const lock = `${file}.lock` + await mkdir(dirname(file), { recursive: true, mode: 0o700 }) + for (;;) { + try { + await mkdir(lock, { mode: 0o700 }) + return async () => rm(lock, { recursive: true, force: true }) + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== "EEXIST") throw error + try { + const lockStat = await stat(lock) + if (Date.now() - lockStat.mtimeMs > LOCK_STALE_MS) { + const stale = `${lock}.stale.${process.pid}.${Date.now()}` + await rename(lock, stale) + await rm(stale, { recursive: true, force: true }) + continue + } + } catch (lockError) { + if (!isMissingStateFile(lockError)) throw lockError + } + await sleep(LOCK_RETRY_MS) + } + } +} + async function mutate(fn: (state: State) => T | Promise) { - // The promise queue serializes mutations within this process; the raw-content - // compare before writing detects concurrent writers in other OpenCode - // processes sharing the state file and retries on top of their changes. return enqueueMutation(async () => { - let lastError: unknown - for (let attempt = 0; attempt < MAX_MUTATION_ATTEMPTS; attempt += 1) { + // mkdir is atomic across processes. Holding this lock across the read, + // predicate, and atomic rename makes conditional mutations true + // compare-and-swap operations for every plugin context using this state. + const release = await acquireMutationLock() + try { const before = await readRawState() const result = await Effect.runPromise( Effect.gen(function* () { @@ -206,15 +271,11 @@ async function mutate(fn: (state: State) => T | Promise) { return { state, value } }), ) - const current = await readRawState() - if (current !== before) { - lastError = new Error("state file changed by a concurrent writer") - continue - } await Effect.runPromise(writeStateEffect(result.state)) return result.value + } finally { + await release() } - throw lastError instanceof Error ? lastError : new Error("state mutation failed after concurrent-writer retries") }) } @@ -319,11 +380,18 @@ export async function createLoop(sessionID: string, options: CreateLoopOptions) lastRunAt: null, lastResult: null, lastError: null, + consecutiveFailures: 0, + nextRetryAt: null, + blockedReason: null, lastReason: null, runCount: 0, maxRuns, agent, stopReason: null, + ownerID: null, + ownerLeaseUntil: null, + ownerRevision: 0, + runClaimToken: 0, } state.loops[id] = loop return snapshot(loop) @@ -336,6 +404,60 @@ export async function getLoop(loopID: string) { return loop ? snapshot(loop) : null } +function validLeaseMs(value: number) { + const leaseMs = positiveIntegerOrNull(Math.round(value)) + if (leaseMs == null) throw new Error("session lease must be a positive number of milliseconds") + return leaseMs +} + +/** Atomically acquires an absent/expired lease, or renews this instance's lease. */ +export async function acquireSessionLease(sessionID: string, instanceID: string, leaseMs: number) { + const duration = validLeaseMs(leaseMs) + return mutate((state) => { + const timestamp = now() + const current = state.sessionLeases[sessionID] + if (current && current.instanceID !== instanceID && current.expiresAt > timestamp) return null + const lease: SessionLease = { + sessionID, + instanceID, + expiresAt: timestamp + duration, + revision: (current?.revision ?? 0) + 1, + } + state.sessionLeases[sessionID] = lease + return { ...lease } + }) +} + +/** Renews only the exact lease revision previously returned to this instance. */ +export async function renewSessionLease(sessionID: string, instanceID: string, revision: number, leaseMs: number) { + const duration = validLeaseMs(leaseMs) + return mutate((state) => { + const timestamp = now() + const current = state.sessionLeases[sessionID] + if (!current || current.instanceID !== instanceID || current.revision !== revision || current.expiresAt <= timestamp) return null + current.expiresAt = timestamp + duration + current.revision += 1 + return { ...current } + }) +} + +/** Checks persisted ownership without extending it. */ +export async function ownsSessionLease(sessionID: string, instanceID: string, revision: number) { + const state = await readState() + const lease = state.sessionLeases[sessionID] + return Boolean(lease && lease.instanceID === instanceID && lease.revision === revision && lease.expiresAt > now()) +} + +/** Releases only the exact lease held by this instance; stale owners are no-ops. */ +export async function releaseSessionLease(sessionID: string, instanceID: string, revision: number) { + return mutate((state) => { + const current = state.sessionLeases[sessionID] + if (!current || current.instanceID !== instanceID || current.revision !== revision) return false + delete state.sessionLeases[sessionID] + return true + }) +} + export async function claimDueRun(loopID: string, leaseMs: number) { const lease = positiveIntegerOrNull(Math.round(leaseMs)) if (lease == null) throw new Error("run claim lease must be a positive number of milliseconds") @@ -349,6 +471,54 @@ export async function claimDueRun(loopID: string, leaseMs: number) { }) } +/** Acquire or renew the scheduler-owner lease without changing the run schedule. */ +export async function acquireLoopOwner(loopID: string, ownerID: string, leaseMs: number) { + const lease = positiveIntegerOrNull(Math.round(leaseMs)) + if (!ownerID.trim()) throw new Error("loop owner id must not be empty") + if (lease == null) throw new Error("owner lease must be a positive number of milliseconds") + return mutate((state) => { + const loop = requireLoop(state, loopID) + const timestamp = now() + if (loop.status !== "active") return null + if (loop.ownerID !== ownerID && loop.ownerLeaseUntil != null && loop.ownerLeaseUntil > timestamp) return null + if (loop.ownerID !== ownerID) loop.ownerRevision += 1 + loop.ownerID = ownerID + loop.ownerLeaseUntil = timestamp + lease + loop.updatedAt = timestamp + return { ownerID, ownerRevision: loop.ownerRevision } satisfies LoopOwnerLease + }) +} + +/** Claim a due run only while the supplied fencing lease is still current. */ +export async function claimDueRunOwned(loopID: string, owner: LoopOwnerLease, leaseMs: number): Promise { + const lease = positiveIntegerOrNull(Math.round(leaseMs)) + if (lease == null) throw new Error("run claim lease must be a positive number of milliseconds") + return mutate((state) => { + const loop = requireLoop(state, loopID) + const timestamp = now() + if (!matchesOwner(loop, owner, timestamp)) return null + if (loop.status !== "active" || loop.nextRunAt == null || loop.nextRunAt > timestamp) return null + loop.runClaimToken += 1 + loop.nextRunAt = timestamp + lease + loop.updatedAt = timestamp + return { ...owner, claimToken: loop.runClaimToken, loop: snapshot(loop) } + }) +} + +function matchesOwner(loop: Loop, owner: LoopOwnerLease, timestamp = now()) { + return loop.ownerID === owner.ownerID && loop.ownerRevision === owner.ownerRevision && (loop.ownerLeaseUntil ?? 0) > timestamp +} + +function matchesClaim(loop: Loop, claim: RunClaim, timestamp = now()) { + return matchesOwner(loop, claim, timestamp) && loop.runClaimToken === claim.claimToken +} + +export async function confirmRunClaim(loopID: string, claim: RunClaim) { + const state = await readState() + const loop = state.loops[loopID] + return loop != null && matchesClaim(loop, claim) ? snapshot(loop) : null +} + export async function listLoops(sessionID?: string) { const state = await readState() return Object.values(state.loops) @@ -373,6 +543,7 @@ export async function pauseLoop(loopID: string) { if (loop.status !== "active") throw new Error(`loop "${loopID}" is ${loop.status}; only active loops can be paused`) loop.status = "paused" loop.nextRunAt = null + loop.nextRetryAt = null loop.stopReason = "paused" loop.updatedAt = now() return snapshot(loop) @@ -386,6 +557,9 @@ export async function resumeLoop(loopID: string) { const timestamp = now() loop.status = "active" loop.stopReason = null + loop.blockedReason = null + loop.consecutiveFailures = 0 + loop.nextRetryAt = null loop.nextRunAt = loop.mode === "interval" ? timestamp + loop.intervalMs! : timestamp loop.updatedAt = timestamp return snapshot(loop) @@ -398,12 +572,32 @@ export async function stopLoop(loopID: string, reason?: string | null) { if (!isOpen(loop.status)) throw new Error(`loop "${loopID}" is already ${loop.status}`) loop.status = "stopped" loop.nextRunAt = null + loop.nextRetryAt = null loop.stopReason = typeof reason === "string" && reason.trim() ? reason.trim().slice(0, 400) : "stopped" loop.updatedAt = now() return snapshot(loop) }) } +export async function stopLoopIfUnchanged(loopID: string, expectedUpdatedAt: number, reason: string) { + return mutate((state) => { + const loop = state.loops[loopID] + if (!loop) return null + if ( + loop.updatedAt !== expectedUpdatedAt || + loop.status !== "active" || + loop.mode !== "dynamic" || + loop.nextRunAt != null + ) { + return null + } + loop.status = "stopped" + loop.stopReason = reason.trim().slice(0, 400) || "stopped" + loop.updatedAt = now() + return snapshot(loop) + }) +} + export async function stopLoopsForSession(sessionID: string, reason: string) { return mutate((state) => { const stopped: LoopSnapshot[] = [] @@ -411,6 +605,7 @@ export async function stopLoopsForSession(sessionID: string, reason: string) { if (loop.sessionID !== sessionID || !isOpen(loop.status)) continue loop.status = "stopped" loop.nextRunAt = null + loop.nextRetryAt = null loop.stopReason = reason loop.updatedAt = now() stopped.push(snapshot(loop)) @@ -439,6 +634,7 @@ export async function scheduleNextRun(loopID: string, delayMs: number, reason?: if (loop.status !== "active") throw new Error(`loop "${loopID}" is ${loop.status}; only active loops can be scheduled`) const timestamp = now() loop.nextRunAt = timestamp + delay + loop.nextRetryAt = null loop.lastReason = typeof reason === "string" && reason.trim() ? reason.trim().slice(0, 400) : loop.lastReason loop.updatedAt = timestamp return snapshot(loop) @@ -454,6 +650,9 @@ export async function recordRunSent(loopID: string) { loop.lastRunAt = timestamp loop.lastResult = "sent" loop.lastError = null + loop.consecutiveFailures = 0 + loop.nextRetryAt = null + loop.blockedReason = null loop.updatedAt = timestamp if (loop.maxRuns != null && loop.runCount >= loop.maxRuns) { loop.status = "completed" @@ -468,26 +667,143 @@ export async function recordRunSent(loopID: string) { }) } +export async function recordRunSentClaimed(loopID: string, claim: RunClaim) { + return mutate((state) => { + const loop = requireLoop(state, loopID) + if (!matchesClaim(loop, claim) || loop.status !== "active") return null + const timestamp = now() + loop.runCount += 1 + loop.lastRunAt = timestamp + loop.lastResult = "sent" + loop.lastError = null + loop.consecutiveFailures = 0 + loop.nextRetryAt = null + loop.blockedReason = null + loop.updatedAt = timestamp + if (loop.maxRuns != null && loop.runCount >= loop.maxRuns) { + loop.status = "completed" + loop.nextRunAt = null + loop.stopReason = `max runs reached (${loop.maxRuns})` + } else if (loop.mode === "interval") loop.nextRunAt = timestamp + loop.intervalMs! + else loop.nextRunAt = null + return snapshot(loop) + }) +} + export async function recordRunDeferred(loopID: string, result: "skipped_busy" | "skipped_plan", retryDelayMs: number) { return mutate((state) => { const loop = requireLoop(state, loopID) if (loop.status !== "active") return snapshot(loop) const timestamp = now() loop.lastResult = result + loop.nextRetryAt = null loop.nextRunAt = timestamp + Math.max(0, Math.round(retryDelayMs)) loop.updatedAt = timestamp return snapshot(loop) }) } -export async function recordRunFailed(loopID: string, error: string, retryDelayMs: number) { +/** + * Records a provider/model injection failure. Retries use exponential backoff, + * capped by maxRetryDelayMs; after maxConsecutiveFailures the loop is paused. + * `error` is deliberately reduced to a fixed category so provider responses, + * credentials, prompts, and session data can never enter persisted state. + */ +export async function recordRunFailed( + loopID: string, + _error: string, + retryDelayMs: number, + maxConsecutiveFailures = 5, + maxRetryDelayMs = 60 * 60 * 1000, +) { + return mutate((state) => { + const loop = requireLoop(state, loopID) + const timestamp = now() + loop.lastResult = "failed" + loop.lastError = "provider/model request failed" + loop.consecutiveFailures += 1 + loop.updatedAt = timestamp + if (loop.status === "active") { + if (loop.consecutiveFailures >= Math.max(1, Math.round(maxConsecutiveFailures))) { + loop.status = "paused" + loop.nextRunAt = null + loop.nextRetryAt = null + loop.blockedReason = `paused after ${loop.consecutiveFailures} consecutive provider/model failures` + loop.stopReason = loop.blockedReason + } else { + const baseDelay = Math.max(1, Math.round(retryDelayMs)) + const cap = Math.max(baseDelay, Math.round(maxRetryDelayMs)) + const delay = Math.min(cap, baseDelay * 2 ** (loop.consecutiveFailures - 1)) + loop.nextRetryAt = timestamp + delay + loop.nextRunAt = loop.nextRetryAt + loop.blockedReason = null + } + } + return snapshot(loop) + }) +} + +export async function recordRunFailedClaimed( + loopID: string, + claim: RunClaim, + _error: string, + retryDelayMs: number, + maxConsecutiveFailures = 5, + maxRetryDelayMs = 60 * 60 * 1000, +) { return mutate((state) => { const loop = requireLoop(state, loopID) + if (!matchesClaim(loop, claim) || loop.status !== "active") return null const timestamp = now() loop.lastResult = "failed" - loop.lastError = error.slice(0, 400) + loop.lastError = "provider/model request failed" + loop.consecutiveFailures += 1 + loop.updatedAt = timestamp + if (loop.consecutiveFailures >= Math.max(1, Math.round(maxConsecutiveFailures))) { + loop.status = "paused" + loop.nextRunAt = null + loop.nextRetryAt = null + loop.blockedReason = `paused after ${loop.consecutiveFailures} consecutive provider/model failures` + loop.stopReason = loop.blockedReason + } else { + const baseDelay = Math.max(1, Math.round(retryDelayMs)) + const cap = Math.max(baseDelay, Math.round(maxRetryDelayMs)) + const delay = Math.min(cap, baseDelay * 2 ** (loop.consecutiveFailures - 1)) + loop.nextRetryAt = timestamp + delay + loop.nextRunAt = loop.nextRetryAt + loop.blockedReason = null + } + return snapshot(loop) + }) +} + +export async function recordRunDeferredClaimed( + loopID: string, + claim: RunClaim, + result: "skipped_busy" | "skipped_plan", + retryDelayMs: number, +) { + return mutate((state) => { + const loop = requireLoop(state, loopID) + if (!matchesClaim(loop, claim) || loop.status !== "active") return null + const timestamp = now() + loop.lastResult = result + loop.nextRetryAt = null + loop.nextRunAt = timestamp + Math.max(0, Math.round(retryDelayMs)) loop.updatedAt = timestamp - if (loop.status === "active") loop.nextRunAt = timestamp + Math.max(0, Math.round(retryDelayMs)) + return snapshot(loop) + }) +} + +export async function stopLoopClaimed(loopID: string, claim: RunClaim, reason?: string | null) { + return mutate((state) => { + const loop = requireLoop(state, loopID) + if (!matchesClaim(loop, claim) || !isOpen(loop.status)) return null + loop.status = "stopped" + loop.nextRunAt = null + loop.nextRetryAt = null + loop.stopReason = typeof reason === "string" && reason.trim() ? reason.trim().slice(0, 400) : "stopped" + loop.updatedAt = now() return snapshot(loop) }) } diff --git a/test/server-v2.test.ts b/test/server-v2.test.ts index a8f7149..b947436 100644 --- a/test/server-v2.test.ts +++ b/test/server-v2.test.ts @@ -452,7 +452,7 @@ test("V2 session deletion stops the session's loops", async () => { await cleanup() }) -test("V2 setup rehydrates persisted active loops", async () => { +test("V2 rehydrated loops resume after an authoritative idle event", async () => { const first = makeMockContext() const cleanup1 = await plugin.setup(first as never) const created = JSON.parse( @@ -463,6 +463,13 @@ test("V2 setup rehydrates persisted active loops", async () => { const second = makeMockContext() const cleanup2 = await plugin.setup(second as never) + await Bun.sleep(1100) + expect(second.promptCalls).toHaveLength(0) + second.stream.push({ + type: "session.status", + created: Date.now(), + data: { sessionID: "ses_v2", status: { type: "idle" } }, + }) await waitFor(() => second.promptCalls.length >= 1) expect(second.promptCalls[0]?.text).toContain(created.created) await waitFor(async () => (await getLoop(created.created))?.runCount === 1) @@ -471,6 +478,29 @@ test("V2 setup rehydrates persisted active loops", async () => { await cleanup2() }) +test("V2 rehydrate leaves a foreign unscheduled dynamic loop unchanged", async () => { + const owner = makeMockContext() + const cleanupOwner = await plugin.setup(owner as never) + const created = JSON.parse( + contentOf(await loopTool(owner, "create_loop").execute({ instruction: "watch CI" }, toolContext("ses_foreign"))), + ) as { created: string } + const before = await getLoop(created.created) + + const foreign = makeMockContext() + const cleanupForeign = await plugin.setup(foreign as never) + foreign.stream.push({ type: "session.idle", created: Date.now(), data: { sessionID: "ses_foreign" } }) + await sleep(100) + const after = await getLoop(created.created) + + expect(after?.status).toBe(before?.status) + expect(after?.stopReason).toBe(before?.stopReason) + expect(after?.updatedAt).toBe(before?.updatedAt) + foreign.stream.end() + await cleanupForeign() + owner.stream.end() + await cleanupOwner() +}) + test("V2 concurrent plugin contexts claim a persisted run only once", async () => { const creator = makeMockContext() const cleanupCreator = await plugin.setup(creator as never) @@ -482,6 +512,11 @@ test("V2 concurrent plugin contexts claim a persisted run only once", async () = const second = makeMockContext() const cleanupFirst = await plugin.setup(first as never) const cleanupSecond = await plugin.setup(second as never) + first.stream.push({ type: "session.idle", created: Date.now(), data: { sessionID: "ses_v2" } }) + second.stream.push({ type: "session.idle", created: Date.now(), data: { sessionID: "ses_v2" } }) + + first.stream.push({ type: "session.status", created: Date.now(), data: { sessionID: "ses_v2", status: { type: "idle" } } }) + second.stream.push({ type: "session.status", created: Date.now(), data: { sessionID: "ses_v2", status: { type: "idle" } } }) await waitFor(() => first.promptCalls.length + second.promptCalls.length >= 1) await sleep(100) diff --git a/test/server.test.ts b/test/server.test.ts index 3437773..0f84304 100644 --- a/test/server.test.ts +++ b/test/server.test.ts @@ -203,7 +203,7 @@ test("schedule_next_run rejects fixed-interval loops", async () => { await hooks.dispose?.() }) -test("failed injections record the error and stay active for retry", async () => { +test("failed injections record a safe category and stay active for retry", async () => { const sent: SentPrompt[] = [] const hooks = await makeServer(sent, {}, { failPrompts: true }) const created = JSON.parse( @@ -213,7 +213,7 @@ test("failed injections record the error and stay active for retry", async () => await sleep(1300) const loop = await getLoop(created.created) expect(loop?.lastResult).toBe("failed") - expect(loop?.lastError).toContain("prompt rejected") + expect(loop?.lastError).toBe("provider/model request failed") expect(loop?.status).toBe("active") await hooks.dispose?.() }) @@ -278,7 +278,7 @@ test("system transform merges a loop reminder for sessions with open loops", asy await hooks.dispose?.() }) -test("rehydrates persisted active loops on startup", async () => { +test("rehydrated loops resume after an authoritative idle event", async () => { const sent: SentPrompt[] = [] const first = await makeServer(sent) const created = JSON.parse( @@ -287,8 +287,31 @@ test("rehydrates persisted active loops on startup", async () => { await first.dispose?.() const second = await makeServer(sent) + await tool(second, "list_loops").execute({}, { sessionID: "ses_1" }) await sleep(1300) + expect(sent).toHaveLength(0) + await second.event?.({ event: { type: "session.idle", properties: { sessionID: "ses_1" } } } as never) + await sleep(1100) expect(sent.length).toBeGreaterThanOrEqual(1) expect(sent[0]?.text).toContain(created.created) await second.dispose?.() }) + +test("V1 rehydrate leaves a foreign unscheduled dynamic loop unchanged", async () => { + const first = await makeServer([]) + const created = JSON.parse( + await tool(first, "create_loop").execute({ instruction: "watch CI" }, { sessionID: "ses_foreign" }), + ) as { created: string } + const before = await getLoop(created.created) + + const foreign = await makeServer([]) + await foreign.event?.({ event: { type: "session.idle", properties: { sessionID: "ses_foreign" } } } as never) + await sleep(100) + const after = await getLoop(created.created) + + expect(after?.status).toBe(before?.status) + expect(after?.stopReason).toBe(before?.stopReason) + expect(after?.updatedAt).toBe(before?.updatedAt) + await foreign.dispose?.() + await first.dispose?.() +}) diff --git a/test/state.test.ts b/test/state.test.ts index fb259ec..4143fb8 100644 --- a/test/state.test.ts +++ b/test/state.test.ts @@ -3,7 +3,11 @@ import { mkdtemp, readFile, rm, stat, writeFile } from "node:fs/promises" import { join } from "node:path" import { tmpdir } from "node:os" import { + acquireLoopOwner, + acquireSessionLease, + claimDueRunOwned, clearClosedLoops, + confirmRunClaim, createLoop, formatInterval, formatLoops, @@ -12,12 +16,17 @@ import { activeLoops, parseInterval, pauseLoop, + ownsSessionLease, recordRunDeferred, + recordRunSentClaimed, recordRunFailed, recordRunSent, + releaseSessionLease, + renewSessionLease, resumeLoop, scheduleNextRun, stopLoop, + stopLoopIfUnchanged, stopLoopsForSession, } from "../src/state" @@ -57,6 +66,44 @@ test("formats intervals back to compact strings", () => { expect(formatInterval(null)).toBe("dynamic") }) +test("atomically acquires, revises, validates, and releases session leases", async () => { + const first = await acquireSessionLease("ses_1", "instance-a", 60_000) + expect(first?.revision).toBe(1) + expect(await acquireSessionLease("ses_1", "instance-b", 60_000)).toBeNull() + expect(await ownsSessionLease("ses_1", "instance-a", first!.revision)).toBe(true) + + const renewed = await renewSessionLease("ses_1", "instance-a", first!.revision, 60_000) + expect(renewed?.revision).toBe(2) + expect(await ownsSessionLease("ses_1", "instance-a", first!.revision)).toBe(false) + expect(await releaseSessionLease("ses_1", "instance-a", first!.revision)).toBe(false) + expect(await releaseSessionLease("ses_1", "instance-a", renewed!.revision)).toBe(true) + expect((await acquireSessionLease("ses_1", "instance-b", 60_000))?.instanceID).toBe("instance-b") +}) + +test("only one process can acquire a session lease", async () => { + const worker = join(dir, "acquire.ts") + await writeFile( + worker, + `import { acquireSessionLease } from ${JSON.stringify(join(process.cwd(), "src/state.ts"))}\n` + + `const lease = await acquireSessionLease("ses_shared", process.argv[2]!, 60_000)\n` + + `await Bun.write(Bun.stdout, lease ? "acquired\\n" : "denied\\n")\n` + + `process.exit(0)\n`, + ) + const results = await Promise.all( + Array.from({ length: 8 }, async (_, index) => { + const child = Bun.spawn([globalThis.process.execPath, worker, `instance-${index}`], { + env: { ...Bun.env, OPENCODE_LOOP_STATE_PATH: join(dir, "shared.json") }, + stdout: "pipe", + }) + const output = await new Response(child.stdout).text() + expect(await child.exited).toBe(0) + return output.trim() + }), + ) + expect(results.filter((result) => result === "acquired")).toHaveLength(1) + expect(results.filter((result) => result === "denied")).toHaveLength(7) +}, 15_000) + test("creates, lists, pauses, resumes, and stops a loop", async () => { const created = await createLoop("ses_1", { prompt: "check the deploy", intervalMs: 600_000 }) expect(created.status).toBe("active") @@ -110,6 +157,40 @@ test("dynamic loops schedule one run at a time", async () => { expect(sent.runCount).toBe(1) }) +test("conditional dynamic cleanup ignores missing and changed loops", async () => { + const removed = await createLoop("ses_1", { prompt: "removed", mode: "dynamic" }) + await stopLoop(removed.id) + await clearClosedLoops("ses_1") + expect(await stopLoopIfUnchanged(removed.id, removed.updatedAt, "stale after restart")).toBeNull() + + const scheduled = await createLoop("ses_1", { prompt: "scheduled", mode: "dynamic" }) + await scheduleNextRun(scheduled.id, 60_000, "still owned") + expect(await stopLoopIfUnchanged(scheduled.id, scheduled.updatedAt, "stale after restart")).toBeNull() + expect((await getLoop(scheduled.id))?.status).toBe("active") +}) + +test("serializes state mutations across processes", async () => { + const script = ` + import { createLoop } from "./src/state.ts" + await Promise.all(Array.from({ length: 12 }, (_, index) => + createLoop("session_" + process.pid + "_" + index, { prompt: "tick", intervalMs: 60000 }) + )) + ` + const options = { + cwd: join(import.meta.dir, ".."), + env: { ...process.env }, + stdout: "pipe" as const, + stderr: "pipe" as const, + } + const children = [ + Bun.spawn([process.execPath, "-e", script], options), + Bun.spawn([process.execPath, "-e", script], options), + ] + const exits = await Promise.all(children.map((child) => child.exited)) + expect(exits).toEqual([0, 0]) + expect(await listLoops()).toHaveLength(24) +}) + test("recordRunSent does not resurrect stopped or paused loops", async () => { const created = await createLoop("ses_1", { prompt: "tick", intervalMs: 60_000 }) await stopLoop(created.id, "done") @@ -128,8 +209,58 @@ test("records deferred and failed runs with retry times", async () => { const failed = await recordRunFailed(created.id, "network exploded", 5000) expect(failed.lastResult).toBe("failed") - expect(failed.lastError).toBe("network exploded") + expect(failed.lastError).toBe("provider/model request failed") + expect(failed.consecutiveFailures).toBe(1) + expect(failed.nextRetryAt).toBe(failed.nextRunAt) expect(failed.status).toBe("active") + + const second = await recordRunFailed(created.id, "secret provider response", 5000) + expect(second.nextRetryAt! - second.updatedAt).toBe(10_000) + const recovered = await recordRunSent(created.id) + expect(recovered.consecutiveFailures).toBe(0) + expect(recovered.nextRetryAt).toBeNull() + expect(recovered.lastError).toBeNull() +}) + +test("pauses loops after the bounded consecutive failure threshold", async () => { + const created = await createLoop("ses_1", { prompt: "tick", intervalMs: 600_000 }) + await recordRunFailed(created.id, "sensitive one", 1000, 2, 1500) + const blocked = await recordRunFailed(created.id, "sensitive two", 1000, 2, 1500) + expect(blocked.status).toBe("paused") + expect(blocked.nextRunAt).toBeNull() + expect(blocked.nextRetryAt).toBeNull() + expect(blocked.blockedReason).toContain("2 consecutive") + expect(blocked.lastError).not.toContain("sensitive") +}) + +test("fences a claimed run when its owner lease is replaced", async () => { + const created = await createLoop("ses_1", { prompt: "tick", intervalMs: 60_000 }) + await scheduleNextRun(created.id, 1) + await Bun.sleep(5) + + const ownerA = await acquireLoopOwner(created.id, "owner-a", 1000) + expect(ownerA).not.toBeNull() + const claimA = await claimDueRunOwned(created.id, ownerA!, 30_000) + expect(claimA).not.toBeNull() + expect(await confirmRunClaim(created.id, claimA!)).not.toBeNull() + + await Bun.sleep(1200) + const ownerB = await acquireLoopOwner(created.id, "owner-b", 30_000) + expect(ownerB?.ownerRevision).toBeGreaterThan(ownerA!.ownerRevision) + expect(await confirmRunClaim(created.id, claimA!)).toBeNull() + expect(await recordRunSentClaimed(created.id, claimA!)).toBeNull() + expect((await getLoop(created.id))?.runCount).toBe(0) +}) + +test("owner-aware claims reject a non-current lease", async () => { + const created = await createLoop("ses_1", { prompt: "tick", intervalMs: 60_000 }) + await scheduleNextRun(created.id, 1) + await Bun.sleep(5) + const owner = await acquireLoopOwner(created.id, "owner-a", 30_000) + expect(owner).not.toBeNull() + expect( + await claimDueRunOwned(created.id, { ownerID: "owner-a", ownerRevision: owner!.ownerRevision + 1 }, 30_000), + ).toBeNull() }) test("enforces the per-session open loop limit", async () => { @@ -194,11 +325,15 @@ test("decodes persisted state with optional fields omitted", async () => { expect(loop?.lastResult).toBeNull() expect(loop?.maxRuns).toBeNull() expect(loop?.agent).toBeNull() + expect(loop?.consecutiveFailures).toBe(0) + expect(loop?.nextRetryAt).toBeNull() + expect(loop?.blockedReason).toBeNull() }) test("writes state with owner-only file permissions", async () => { await createLoop("ses_1", { prompt: "tick", intervalMs: 60_000 }) const mode = (await stat(process.env.OPENCODE_LOOP_STATE_PATH!)).mode & 0o777 + if (process.platform === "win32") return expect(mode).toBe(0o600) })