diff --git a/dist/server.js b/dist/server.js index aacdfed..6ea87d7 100644 --- a/dist/server.js +++ b/dist/server.js @@ -16,9 +16,9 @@ import { z } from "zod"; // src/state.ts import { randomUUID } from "crypto"; -import { chmod, mkdir, link, readFile, rename as rename3, rm, writeFile } from "fs/promises"; +import { chmod, mkdir, link, readFile, readdir, rename as rename3, rm, writeFile } from "fs/promises"; import { homedir } from "os"; -import { dirname, join as join3 } from "path"; +import { basename, dirname, join as join3 } from "path"; // node_modules/effect/dist/esm/Function.js var isFunction = (input) => typeof input === "function"; @@ -23641,9 +23641,14 @@ async function readRawState() { } var LOCK_WAIT_MS = 30000; var LOCK_RETRY_MS = 10; +var TRANSIENT_LOCK_CODES = new Set(["EPERM", "EBUSY", "EACCES", "ENOTEMPTY"]); +var TRANSIENT_LOCK_BUDGET_MS = 5000; function sleep5(ms) { return new Promise((resolve) => setTimeout(resolve, ms)); } +function isTransientLockError(error) { + return typeof error === "object" && error !== null && "code" in error && typeof error.code === "string" && TRANSIENT_LOCK_CODES.has(error.code); +} function processIsAlive(pid) { try { process.kill(pid, 0); @@ -23652,37 +23657,94 @@ function processIsAlive(pid) { return error.code !== "ESRCH"; } } +var pendingLockReleases = new Map; +async function retryLockOperation(operation, acquisitionDeadline = Infinity) { + const deadline = Math.min(acquisitionDeadline, Date.now() + TRANSIENT_LOCK_BUDGET_MS); + for (;; ) { + try { + return await operation(); + } catch (error) { + if (!isTransientLockError(error) || Date.now() >= deadline) + throw error; + await sleep5(LOCK_RETRY_MS); + } + } +} +function processLockRelease(lock, ownerPath, token) { + let running3; + let timer; + const finish = () => { + running3 ??= (async () => { + await retryLockOperation(async () => { + const current = await readFile(lock, "utf8").catch((error) => { + if (isMissingStateFile(error)) + return null; + throw error; + }); + if (current === token) + await rm(lock, { force: true }); + }); + if (pendingLockReleases.get(lock) === finish) + pendingLockReleases.delete(lock); + if (timer) + clearTimeout(timer); + timer = undefined; + await rm(ownerPath, { force: true }).catch(() => { + return; + }); + })().finally(() => { + running3 = undefined; + }); + return running3; + }; + const release = async () => { + try { + await finish(); + } catch { + pendingLockReleases.set(lock, finish); + if (!timer) { + timer = setTimeout(() => { + timer = undefined; + release(); + }, 1000); + timer.unref?.(); + } + } + }; + return release; +} async function acquireProcessLock(lock, deadline) { + await pendingLockReleases.get(lock)?.(); const token = `${process.pid}-${randomUUID()}`; const ownerPath = `${lock}.owner.${token}`; - await writeFile(ownerPath, token, { mode: 384, flag: "wx" }); let acquired = false; try { + await retryLockOperation(async () => { + await rm(ownerPath, { force: true }); + await writeFile(ownerPath, token, { mode: 384, flag: "wx" }); + }, deadline); for (;; ) { + if (Date.now() >= deadline) + throw new Error("timed out waiting for the loop state lock"); try { - await link(ownerPath, lock); + await retryLockOperation(() => link(ownerPath, lock), deadline); acquired = true; - return async () => { - try { - if (await readFile(lock, "utf8").catch(() => null) === token) - await rm(lock, { force: true }); - } finally { - await rm(ownerPath, { force: true }); - } - }; + return processLockRelease(lock, ownerPath, token); } catch (error) { - if (error.code !== "EEXIST") + if (typeof error !== "object" || error === null || !("code" in error) || error.code !== "EEXIST") throw error; try { - const owner = await readFile(lock, "utf8"); + const owner = await retryLockOperation(() => readFile(lock, "utf8"), deadline); const pid = Number(owner.split("-", 1)[0]); if (/^\d+-[0-9a-f-]{36}$/.test(owner) && !processIsAlive(pid)) { const release = await acquireProcessLock(`${lock}.reclaim.${owner}`, deadline); try { - if (await readFile(lock, "utf8").catch(() => null) === owner) { - await rm(lock, { force: true }); - await rm(`${lock}.owner.${owner}`, { force: true }); - } + await retryLockOperation(async () => { + if (await readFile(lock, "utf8") === owner) { + await rm(lock, { force: true }); + await rm(`${lock}.owner.${owner}`, { force: true }); + } + }, deadline); } finally { await release(); } @@ -23698,12 +23760,35 @@ async function acquireProcessLock(lock, deadline) { } } finally { if (!acquired) - await rm(ownerPath, { force: true }); + await rm(ownerPath, { force: true }).catch(() => { + return; + }); } } +async function sweepStaleOwnerFiles(lock) { + try { + const dir = dirname(lock); + const prefix = `${basename(lock)}.owner.`; + const entries2 = await readdir(dir); + for (const entry of entries2) { + if (!entry.startsWith(prefix)) + continue; + const rest = entry.slice(prefix.length); + if (!/^\d+-[0-9a-f-]{36}$/.test(rest)) + continue; + const pid = Number(rest.split("-", 1)[0]); + if (pid === process.pid || processIsAlive(pid)) + continue; + await rm(join3(dir, entry), { force: true }).catch(() => { + return; + }); + } + } catch {} +} async function acquireMutationLock() { const file = statePath(); await mkdir(dirname(file), { recursive: true, mode: 448 }); + await sweepStaleOwnerFiles(`${file}.lock`); return acquireProcessLock(`${file}.lock`, Date.now() + LOCK_WAIT_MS); } async function mutate5(fn2) { @@ -23880,13 +23965,18 @@ async function acquireSessionLease(sessionID, instanceID, leaseMs) { return { ...lease }; }); } -async function renewSessionLease(sessionID, instanceID, revision, leaseMs) { +async function renewSessionLease(sessionID, instanceID, revision, leaseMs, recoverExpired = false) { const duration2 = validLeaseMs(leaseMs); return mutate5((state) => { const timestamp = now2(); const current = state.sessionLeases[sessionID]; - if (!current || current.instanceID !== instanceID || current.revision !== revision || current.expiresAt <= timestamp) + if (!current || current.instanceID !== instanceID || current.revision !== revision) return null; + if (current.expiresAt <= timestamp) { + if (!recoverExpired) + return null; + current.revision += 1; + } current.expiresAt = timestamp + duration2; return { ...current }; }); @@ -24014,12 +24104,12 @@ async function stopLoop(loopID, reason) { return snapshot(loop3); }); } -async function stopLoopIfUnchanged(loopID, expectedUpdatedAt, reason, sessionLease) { +async function stopLoopIfUnchanged(loopID, expectedUpdatedAt, reason, sessionLease, expectedScheduleRevision) { return mutate5((state) => { const loop3 = state.loops[loopID]; if (!loop3 || sessionLease && (sessionLease.sessionID !== loop3.sessionID || !matchesSessionLease(state, sessionLease))) return null; - if (loop3.updatedAt !== expectedUpdatedAt || loop3.status !== "active" || loop3.mode !== "dynamic" || loop3.nextRunAt != null) { + if (loop3.updatedAt !== expectedUpdatedAt || expectedScheduleRevision !== undefined && loop3.scheduleRevision !== expectedScheduleRevision || loop3.status !== "active" || loop3.mode !== "dynamic" || loop3.nextRunAt != null) { return null; } loop3.status = "stopped"; @@ -24374,6 +24464,7 @@ var server = async ({ client, directory }, options) => { const timers = new Map; const sendingLoops = new Set; const busySessions = new Set; + const deletedSessions = new Set; const instanceID = randomUUID2(); const loopOwnerID = `server:${process.pid}:${randomUUID2()}`; const ownedSessions = new Map; @@ -24383,6 +24474,7 @@ var server = async ({ client, directory }, options) => { const staleDynamicCandidates = new Map; const lastPromptAgentBySession = new Map; const dynamicPending = new Map; + const deferredDynamicSettlements = new Map; const isRestrictedAgent = (agent) => typeof agent === "string" && restrictedAgents.has(agent.trim().toLowerCase()); async function log3(level, message, extra) { await client.app?.log?.({ body: { service: "opencode-loop-plugin", level, message, extra } }).catch(() => { @@ -24392,7 +24484,7 @@ var server = async ({ client, directory }, options) => { function cancelTimer(loopID) { const timer = timers.get(loopID); if (timer) - clearTimeout(timer); + clearTimeout(timer.handle); timers.delete(loopID); } async function loseOwnership(sessionID) { @@ -24421,20 +24513,31 @@ var server = async ({ client, directory }, options) => { return false; } } + async function reconcileTimers(sessionID) { + for (const loop3 of await activeLoops(sessionID)) { + const timer = timers.get(loop3.id); + if (!sendingLoops.has(loop3.id) && (!timer || timer.nextRunAt !== loop3.nextRunAt)) + scheduleTimer(loop3); + } + } async function acquireOwnership(sessionID) { return enqueueSessionOperation(ownershipQueues, sessionID, async () => { if (disposed) return false; if (!observedSessions.has(sessionID) && !await keepAliveSessionAvailable(sessionID)) return false; + const current = ownedSessions.get(sessionID); + if (current && current.expiresAt > Date.now() + SESSION_LEASE_HEARTBEAT_MS && await ownsSessionLease(sessionID, instanceID, current.revision)) { + await reconcileTimers(sessionID); + return true; + } const lease = await acquireSessionLease(sessionID, instanceID, SESSION_LEASE_MS); if (!lease) { await loseOwnership(sessionID); return false; } ownedSessions.set(sessionID, lease); - for (const loop3 of await activeLoops(sessionID)) - scheduleTimer(loop3); + await reconcileTimers(sessionID); return true; }); } @@ -24445,18 +24548,19 @@ var server = async ({ client, directory }, options) => { const lease = ownedSessions.get(sessionID); if (!lease) return false; - const renewed = await renewSessionLease(sessionID, instanceID, lease.revision, SESSION_LEASE_MS); + const renewed = await renewSessionLease(sessionID, instanceID, lease.revision, SESSION_LEASE_MS, true); if (!renewed) { await loseOwnership(sessionID); return false; } ownedSessions.set(sessionID, renewed); + await reconcileTimers(sessionID); return true; }); } function scheduleTimer(loop3, minimumDelayMs = 0) { cancelTimer(loop3.id); - if (disposed || loop3.status !== "active" || loop3.nextRunAt == null || !ownedSessions.has(loop3.sessionID) || !observedSessions.has(loop3.sessionID) && !loop3.keepAlive) + if (disposed || loop3.status !== "active" || loop3.nextRunAt == null || deletedSessions.has(loop3.sessionID) || !ownedSessions.has(loop3.sessionID) || !observedSessions.has(loop3.sessionID) && !loop3.keepAlive) return; const delay3 = Math.max(minimumDelayMs, loop3.nextRunAt - Date.now()); const timer = setTimeout(() => { @@ -24466,7 +24570,7 @@ var server = async ({ client, directory }, options) => { const maybeUnref = timer; if (typeof maybeUnref.unref === "function") maybeUnref.unref(); - timers.set(loop3.id, timer); + timers.set(loop3.id, { handle: timer, nextRunAt: loop3.nextRunAt }); } async function runDue(loopID) { if (sendingLoops.has(loopID)) @@ -24478,11 +24582,14 @@ var server = async ({ client, directory }, options) => { await log3("error", "Loop scheduler operation failed", { loopID, category: "scheduler" }); } finally { sendingLoops.delete(loopID); + const pending3 = deferredDynamicSettlements.get(loopID); + if (pending3 && !disposed) + await settleDynamicLoop(loopID, pending3).catch(() => log3("error", "Dynamic loop settlement failed", { loopID, category: "state" })); } } async function runDueLocked(loopID) { let loop3 = await getLoop(loopID); - if (!loop3 || loop3.status !== "active" || loop3.nextRunAt == null) + if (!loop3 || loop3.status !== "active" || loop3.nextRunAt == null || deletedSessions.has(loop3.sessionID)) return; if (!await renewOwnership(loop3.sessionID)) return; @@ -24533,7 +24640,7 @@ var server = async ({ client, directory }, options) => { if (loop3.mode === "dynamic") { dynamicPending.set(loopID, { sessionID: loop3.sessionID, sawBusy: false }); } - if (disposed || !await confirmRunClaim(loopID, claimed)) { + if (!await confirmRunClaim(loopID, claimed) || disposed || deletedSessions.has(loop3.sessionID)) { dynamicPending.delete(loopID); const current = await getLoop(loopID); if (current) @@ -24587,23 +24694,63 @@ var server = async ({ client, directory }, options) => { break; } } - async function settleDynamicLoops(sessionID) { - const lease = ownedSessions.get(sessionID); + async function settleDynamicLoop(loopID, pending3) { + if (dynamicPending.get(loopID) !== pending3 || !pending3.sawBusy) { + if (deferredDynamicSettlements.get(loopID) === pending3) + deferredDynamicSettlements.delete(loopID); + return; + } + const lease = ownedSessions.get(pending3.sessionID); if (!lease) return; - for (const [loopID, pending3] of dynamicPending) { - if (pending3.sessionID !== sessionID || !pending3.sawBusy) - continue; - dynamicPending.delete(loopID); - const loop3 = await getLoop(loopID); - if (!loop3 || loop3.status !== "active" || loop3.mode !== "dynamic") - continue; - if (loop3.nextRunAt != null) - continue; - await stopLoopIfUnchanged(loopID, loop3.updatedAt, "the iteration ended without scheduling the next run", lease).catch(() => { + deferredDynamicSettlements.set(loopID, pending3); + if (sendingLoops.has(loopID)) + return; + const loop3 = await getLoop(loopID); + if (dynamicPending.get(loopID) !== pending3) { + if (deferredDynamicSettlements.get(loopID) === pending3) + deferredDynamicSettlements.delete(loopID); + return; + } + if (sendingLoops.has(loopID)) + return; + if (loop3?.status === "active" && loop3.mode === "dynamic" && loop3.nextRunAt == null) { + const stopped = await stopLoopIfUnchanged(loopID, loop3.updatedAt, "the iteration ended without scheduling the next run", lease, loop3.scheduleRevision); + if (!stopped) return; - }); - await log3("info", "Dynamic loop ended because the turn did not schedule the next run", { loopID }); + } + if (dynamicPending.get(loopID) === pending3) + dynamicPending.delete(loopID); + if (deferredDynamicSettlements.get(loopID) === pending3) + deferredDynamicSettlements.delete(loopID); + } + function rememberTerminalTurn(sessionID) { + for (const [loopID, pending3] of dynamicPending) { + if (pending3.sessionID === sessionID && pending3.sawBusy) + deferredDynamicSettlements.set(loopID, pending3); + } + } + async function settleDynamicLoops(sessionID) { + for (const [loopID, pending3] of dynamicPending) { + if (pending3.sessionID === sessionID) + await settleDynamicLoop(loopID, pending3); + } + } + async function retryDeferredState(sessionID) { + if (deletedSessions.has(sessionID)) { + for (const loop3 of await stopLoopsForSession(sessionID, "session deleted")) { + cancelTimer(loop3.id); + dynamicPending.delete(loop3.id); + deferredDynamicSettlements.delete(loop3.id); + } + deletedSessions.delete(sessionID); + observedSessions.delete(sessionID); + keepAliveSessions.delete(sessionID); + return; + } + for (const [loopID, pending3] of deferredDynamicSettlements) { + if (pending3.sessionID === sessionID) + await settleDynamicLoop(loopID, pending3); } } async function rehydrate() { @@ -24629,6 +24776,10 @@ var server = async ({ client, directory }, options) => { observedSessions.add(sessionID); if (!await acquireOwnership(sessionID)) return; + const wasDeleted = deletedSessions.has(sessionID); + await retryDeferredState(sessionID); + if (wasDeleted) + return; const lease = ownedSessions.get(sessionID); if (!lease) return; @@ -24658,9 +24809,9 @@ var server = async ({ client, directory }, options) => { } await rehydrate().catch(() => log3("error", "Failed to rehydrate loops", { category: "state" })); const leaseHeartbeat = setInterval(() => { - for (const sessionID of new Set([...observedSessions, ...keepAliveSessions])) { + for (const sessionID of new Set([...observedSessions, ...keepAliveSessions, ...deletedSessions])) { const operation = ownedSessions.has(sessionID) ? renewOwnership(sessionID) : observedSessions.has(sessionID) ? observeSession(sessionID) : acquireOwnership(sessionID); - operation.catch(() => log3("error", "Loop lease heartbeat failed", { category: "state" })); + operation.then(() => retryDeferredState(sessionID)).catch(() => log3("error", "Loop lease heartbeat failed", { category: "state" })); } }, SESSION_LEASE_HEARTBEAT_MS); leaseHeartbeat.unref?.(); @@ -24670,9 +24821,11 @@ var server = async ({ client, directory }, options) => { await Promise.all(ownershipQueues.values()); clearInterval(leaseHeartbeat); for (const timer of timers.values()) - clearTimeout(timer); + clearTimeout(timer.handle); timers.clear(); dynamicPending.clear(); + deferredDynamicSettlements.clear(); + deletedSessions.clear(); await Promise.all([...ownedSessions.values()].map((lease) => releaseSessionLease(lease.sessionID, instanceID, lease.revision))); ownedSessions.clear(); }, @@ -24816,63 +24969,80 @@ var server = async ({ client, directory }, options) => { const agent = typeof input?.agent === "string" && input.agent.trim() ? input.agent : isRecord2(output.message) && typeof output.message.agent === "string" ? output.message.agent : undefined; if (typeof sessionID !== "string") return; - await observeSession(sessionID); - if (typeof agent !== "string" || !agent.trim()) + if (typeof agent === "string" && agent.trim()) + lastPromptAgentBySession.set(sessionID, agent.trim()); + try { + await observeSession(sessionID); + } catch { + await log3("error", "Loop session observation failed", { sessionID, category: "state" }); 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; - if (output.system.some((block) => block.includes(LOOP_SYSTEM_MARKER))) - return; - if (output.system.length === 0) - output.system.push(reminder); - else - output.system[0] = `${output.system[0]} + try { + await observeSession(input.sessionID); + const loops = await openLoops(input.sessionID); + const reminder = systemReminder(loops); + if (!reminder) + return; + if (output.system.some((block) => block.includes(LOOP_SYSTEM_MARKER))) + return; + if (output.system.length === 0) + output.system.push(reminder); + else + output.system[0] = `${output.system[0]} ${reminder}`; + } catch { + await log3("error", "Loop system reminder failed", { sessionID: input.sessionID, category: "state" }); + } }, async "experimental.session.compacting"(input, output) { - await observeSession(input.sessionID); - const loops = await openLoops(input.sessionID); - const context4 = compactionContext(loops); - if (context4) - output.context.push(context4); + try { + await observeSession(input.sessionID); + const loops = await openLoops(input.sessionID); + const context4 = compactionContext(loops); + if (context4) + output.context.push(context4); + } catch { + await log3("error", "Loop compaction context failed", { sessionID: input.sessionID, category: "state" }); + } }, async event({ event }) { const typed = event; const sessionID = sessionIDFromEvent(typed); if (!sessionID) return; - await observeSession(sessionID); if (isBusyEvent(typed)) { busySessions.add(sessionID); for (const pending3 of dynamicPending.values()) { if (pending3.sessionID === sessionID) pending3.sawBusy = true; } - return; + } + if (isIdleEvent(typed)) { + busySessions.delete(sessionID); + rememberTerminalTurn(sessionID); } if (typed.type === "session.deleted") { + deletedSessions.add(sessionID); busySessions.delete(sessionID); lastPromptAgentBySession.delete(sessionID); - const stopped = await stopLoopsForSession(sessionID, "session deleted"); - for (const loop3 of stopped) { - cancelTimer(loop3.id); - dynamicPending.delete(loop3.id); - } - return; } - if (isIdleEvent(typed)) { - busySessions.delete(sessionID); - await settleDynamicLoops(sessionID); - await runDueForSession(sessionID); + try { + await observeSession(sessionID); + if (typed.type === "session.deleted") { + await retryDeferredState(sessionID); + return; + } + if (isIdleEvent(typed)) { + await settleDynamicLoops(sessionID); + await runDueForSession(sessionID); + } + } catch { + await log3("error", "Loop event handling failed", { sessionID, category: "state" }); } } }; @@ -24914,6 +25084,7 @@ async function setupV2(context4) { const timers = new Map; const sendingLoops = new Set; const busySessions = new Set; + const deletedSessions = new Set; const instanceID = randomUUID2(); const loopOwnerID = `server-v2:${process.pid}:${randomUUID2()}`; const ownedSessions = new Map; @@ -24966,7 +25137,7 @@ async function setupV2(context4) { function cancelTimer(loopID) { const timer = timers.get(loopID); if (timer) - clearTimeout(timer); + clearTimeout(timer.handle); timers.delete(loopID); } async function loseOwnership(sessionID) { @@ -24987,20 +25158,31 @@ async function setupV2(context4) { return false; } } + async function reconcileTimers(sessionID) { + for (const loop3 of await activeLoops(sessionID)) { + const timer = timers.get(loop3.id); + if (!sendingLoops.has(loop3.id) && (!timer || timer.nextRunAt !== loop3.nextRunAt)) + scheduleTimer(loop3); + } + } async function acquireOwnership(sessionID) { return enqueueSessionOperation(ownershipQueues, sessionID, async () => { if (disposed) return false; if (!observedSessions.has(sessionID) && !await keepAliveSessionAvailable(sessionID)) return false; + const current = ownedSessions.get(sessionID); + if (current && current.expiresAt > Date.now() + SESSION_LEASE_HEARTBEAT_MS && await ownsSessionLease(sessionID, instanceID, current.revision)) { + await reconcileTimers(sessionID); + return true; + } const lease = await acquireSessionLease(sessionID, instanceID, SESSION_LEASE_MS); if (!lease) { await loseOwnership(sessionID); return false; } ownedSessions.set(sessionID, lease); - for (const loop3 of await activeLoops(sessionID)) - scheduleTimer(loop3); + await reconcileTimers(sessionID); return true; }); } @@ -25011,18 +25193,19 @@ async function setupV2(context4) { const lease = ownedSessions.get(sessionID); if (!lease) return false; - const renewed = await renewSessionLease(sessionID, instanceID, lease.revision, SESSION_LEASE_MS); + const renewed = await renewSessionLease(sessionID, instanceID, lease.revision, SESSION_LEASE_MS, true); if (!renewed) { await loseOwnership(sessionID); return false; } ownedSessions.set(sessionID, renewed); + await reconcileTimers(sessionID); return true; }); } function scheduleTimer(loop3, minimumDelayMs = 0) { cancelTimer(loop3.id); - if (disposed || loop3.status !== "active" || loop3.nextRunAt == null || !ownedSessions.has(loop3.sessionID) || shuttingDownSessions.has(loop3.sessionID) || !observedSessions.has(loop3.sessionID) && !loop3.keepAlive) + if (disposed || loop3.status !== "active" || loop3.nextRunAt == null || deletedSessions.has(loop3.sessionID) || !ownedSessions.has(loop3.sessionID) || shuttingDownSessions.has(loop3.sessionID) || !observedSessions.has(loop3.sessionID) && !loop3.keepAlive) return; const delay3 = Math.max(minimumDelayMs, loop3.nextRunAt - Date.now()); const timer = setTimeout(() => { @@ -25032,7 +25215,7 @@ async function setupV2(context4) { const maybeUnref = timer; if (typeof maybeUnref.unref === "function") maybeUnref.unref(); - timers.set(loop3.id, timer); + timers.set(loop3.id, { handle: timer, nextRunAt: loop3.nextRunAt }); } async function runDue(loopID) { if (sendingLoops.has(loopID)) @@ -25045,15 +25228,14 @@ async function setupV2(context4) { } finally { sendingLoops.delete(loopID); const pending3 = deferredDynamicSettlements.get(loopID); - deferredDynamicSettlements.delete(loopID); - if (pending3 && dynamicPending.get(loopID) === pending3 && !disposed) { + if (pending3 && !disposed) { await settleDynamicLoop(loopID, pending3).catch(() => v2Log("error", "Dynamic loop settlement failed", { loopID, category: "scheduler" })); } } } async function runDueLocked(loopID) { let loop3 = await getLoop(loopID); - if (!loop3 || loop3.status !== "active" || loop3.nextRunAt == null || shuttingDownSessions.has(loop3.sessionID)) + if (!loop3 || loop3.status !== "active" || loop3.nextRunAt == null || (shuttingDownSessions.has(loop3.sessionID) || deletedSessions.has(loop3.sessionID))) return; if (!await renewOwnership(loop3.sessionID)) return; @@ -25104,7 +25286,7 @@ async function setupV2(context4) { if (loop3.mode === "dynamic") { dynamicPending.set(loopID, { sessionID: loop3.sessionID, sawBusy: false }); } - if (disposed || shuttingDownSessions.has(loop3.sessionID) || !await confirmRunClaim(loopID, claimed)) { + if (!await confirmRunClaim(loopID, claimed) || disposed || shuttingDownSessions.has(loop3.sessionID) || deletedSessions.has(loop3.sessionID)) { dynamicPending.delete(loopID); const current = await getLoop(loopID); if (current) @@ -25159,21 +25341,42 @@ async function setupV2(context4) { } } async function settleDynamicLoop(loopID, pending3) { - const lease = ownedSessions.get(pending3.sessionID); - if (!lease || dynamicPending.get(loopID) !== pending3 || !pending3.sawBusy) + if (shuttingDownSessions.has(pending3.sessionID)) return; - if (sendingLoops.has(loopID)) { - deferredDynamicSettlements.set(loopID, pending3); + if (dynamicPending.get(loopID) !== pending3 || !pending3.sawBusy) { + if (deferredDynamicSettlements.get(loopID) === pending3) + deferredDynamicSettlements.delete(loopID); return; } - dynamicPending.delete(loopID); + const lease = ownedSessions.get(pending3.sessionID); + if (!lease) + return; + deferredDynamicSettlements.set(loopID, pending3); + if (sendingLoops.has(loopID)) + return; const loop3 = await getLoop(loopID); - if (!loop3 || loop3.status !== "active" || loop3.mode !== "dynamic" || loop3.nextRunAt != null) + if (dynamicPending.get(loopID) !== pending3) { + if (deferredDynamicSettlements.get(loopID) === pending3) + deferredDynamicSettlements.delete(loopID); return; - await stopLoopIfUnchanged(loopID, loop3.updatedAt, "the iteration ended without scheduling the next run", lease).catch(() => { + } + if (sendingLoops.has(loopID) || shuttingDownSessions.has(pending3.sessionID)) return; - }); - v2Log("info", "Dynamic loop ended because the turn did not schedule the next run", { loopID }); + if (loop3?.status === "active" && loop3.mode === "dynamic" && loop3.nextRunAt == null) { + const stopped = await stopLoopIfUnchanged(loopID, loop3.updatedAt, "the iteration ended without scheduling the next run", lease, loop3.scheduleRevision); + if (!stopped) + return; + } + if (dynamicPending.get(loopID) === pending3) + dynamicPending.delete(loopID); + if (deferredDynamicSettlements.get(loopID) === pending3) + deferredDynamicSettlements.delete(loopID); + } + function rememberTerminalTurn(sessionID) { + for (const [loopID, pending3] of dynamicPending) { + if (pending3.sessionID === sessionID && pending3.sawBusy) + deferredDynamicSettlements.set(loopID, pending3); + } } async function settleDynamicLoops(sessionID) { for (const [loopID, pending3] of dynamicPending) { @@ -25181,6 +25384,23 @@ async function setupV2(context4) { await settleDynamicLoop(loopID, pending3); } } + async function retryDeferredState(sessionID) { + if (deletedSessions.has(sessionID)) { + for (const loop3 of await stopLoopsForSession(sessionID, "session deleted")) { + cancelTimer(loop3.id); + dynamicPending.delete(loop3.id); + deferredDynamicSettlements.delete(loop3.id); + } + deletedSessions.delete(sessionID); + observedSessions.delete(sessionID); + keepAliveSessions.delete(sessionID); + return; + } + for (const [loopID, pending3] of deferredDynamicSettlements) { + if (pending3.sessionID === sessionID) + await settleDynamicLoop(loopID, pending3); + } + } async function rehydrate() { const loops = await activeLoops(); for (const loop3 of loops) { @@ -25204,6 +25424,10 @@ async function setupV2(context4) { observedSessions.add(sessionID); if (!await acquireOwnership(sessionID)) return; + const wasDeleted = deletedSessions.has(sessionID); + await retryDeferredState(sessionID); + if (wasDeleted) + return; const lease = ownedSessions.get(sessionID); if (!lease) return; @@ -25250,11 +25474,27 @@ async function setupV2(context4) { shuttingDownSessions.add(sessionID); for (const loop3 of await activeLoops(sessionID)) cancelTimer(loop3.id); - await settleDynamicLoops(sessionID); return; } - if (event.type === "session.execution.started") + const terminal = event.type === "session.idle" || event.type === "session.execution.succeeded" || event.type === "session.execution.failed" || event.type === "session.execution.interrupted" || event.type === "session.status" && isRecord2(data.status) && data.status.type === "idle"; + if (terminal) { + busySessions.delete(sessionID); + rememberTerminalTurn(sessionID); + } + if (event.type === "session.deleted") { + deletedSessions.add(sessionID); + busySessions.delete(sessionID); + lastPromptAgentBySession.delete(sessionID); + } + if (event.type === "session.execution.started") { shuttingDownSessions.delete(sessionID); + markExecutionStarted(sessionID); + } + if ((event.type === "session.agent.selected" || event.type === "session.step.started") && typeof data.agent === "string") { + lastPromptAgentBySession.set(sessionID, data.agent); + } + if (event.type === "session.status" && isRecord2(data.status) && data.status.type === "busy") + markExecutionStarted(sessionID); await observeSession(sessionID); switch (event.type) { case "session.execution.started": { @@ -25285,13 +25525,7 @@ async function setupV2(context4) { return; } case "session.deleted": { - busySessions.delete(sessionID); - lastPromptAgentBySession.delete(sessionID); - const stopped = await stopLoopsForSession(sessionID, "session deleted"); - for (const loop3 of stopped) { - cancelTimer(loop3.id); - dynamicPending.delete(loop3.id); - } + await retryDeferredState(sessionID); return; } case "session.agent.selected": { @@ -25343,30 +25577,38 @@ async function setupV2(context4) { draft.add(tool); })); registrations.push(await context4.session.hook("context", async (sessionContext) => { - await observeSession(sessionContext.sessionID); - const loops = await openLoops(sessionContext.sessionID); - const reminder = systemReminder(loops); - if (!reminder) - return; - if (sessionContext.system.some((part) => part.type === "text" && part.text.includes(LOOP_SYSTEM_MARKER))) - return; - sessionContext.system.push({ type: "text", text: reminder }); + try { + await observeSession(sessionContext.sessionID); + const loops = await openLoops(sessionContext.sessionID); + const reminder = systemReminder(loops); + if (!reminder) + return; + if (sessionContext.system.some((part) => part.type === "text" && part.text.includes(LOOP_SYSTEM_MARKER))) + return; + sessionContext.system.push({ type: "text", text: reminder }); + } catch { + v2Log("error", "Loop context hook failed", { sessionID: sessionContext.sessionID, category: "state" }); + } })); registrations.push(await context4.session.hook("compaction", async (sessionCompaction) => { - await observeSession(sessionCompaction.sessionID); - const loops = await openLoops(sessionCompaction.sessionID); - const contextText = compactionContext(loops); - if (!contextText) - return; - if (sessionCompaction.system.some((part) => part.type === "text" && part.text.includes("OpenCode loop mode is tracking recurring loops for this session across compaction."))) - return; - sessionCompaction.system.push({ type: "text", text: contextText }); + try { + await observeSession(sessionCompaction.sessionID); + const loops = await openLoops(sessionCompaction.sessionID); + const contextText = compactionContext(loops); + if (!contextText) + return; + if (sessionCompaction.system.some((part) => part.type === "text" && part.text.includes("OpenCode loop mode is tracking recurring loops for this session across compaction."))) + return; + sessionCompaction.system.push({ type: "text", text: contextText }); + } catch { + v2Log("error", "Loop compaction hook failed", { sessionID: sessionCompaction.sessionID, category: "state" }); + } })); await rehydrate().catch(() => v2Log("error", "Failed to rehydrate loops", { category: "state" })); const leaseHeartbeat = setInterval(() => { - for (const sessionID of new Set([...observedSessions, ...keepAliveSessions])) { + for (const sessionID of new Set([...observedSessions, ...keepAliveSessions, ...deletedSessions])) { const operation = ownedSessions.has(sessionID) ? renewOwnership(sessionID) : observedSessions.has(sessionID) ? observeSession(sessionID) : acquireOwnership(sessionID); - operation.catch(() => v2Log("error", "Loop lease heartbeat failed", { category: "state" })); + operation.then(() => retryDeferredState(sessionID)).catch(() => v2Log("error", "Loop lease heartbeat failed", { category: "state" })); } }, SESSION_LEASE_HEARTBEAT_MS); leaseHeartbeat.unref?.(); @@ -25380,7 +25622,11 @@ async function setupV2(context4) { const { done: done7, value: value3 } = await iterator.next(); if (done7) break; - await handleV2Event(value3); + try { + await handleV2Event(value3); + } catch { + v2Log("error", "V2 event handling failed", { category: "event" }); + } } } catch { if (!abortController.signal.aborted) @@ -25394,11 +25640,12 @@ async function setupV2(context4) { abortController.abort(); pendingIdleChecks.clear(); for (const timer of timers.values()) - clearTimeout(timer); + clearTimeout(timer.handle); timers.clear(); dynamicPending.clear(); deferredDynamicSettlements.clear(); shuttingDownSessions.clear(); + deletedSessions.clear(); sendingLoops.clear(); await Promise.all([...ownedSessions.values()].map((lease) => releaseSessionLease(lease.sessionID, instanceID, lease.revision))); ownedSessions.clear(); diff --git a/src/server.ts b/src/server.ts index 187a33e..8c10d67 100644 --- a/src/server.ts +++ b/src/server.ts @@ -56,6 +56,7 @@ type Options = { } type PendingDynamicTurn = { sessionID: string; sawBusy: boolean } +type LoopTimer = { handle: ReturnType; nextRunAt: number } const KEEP_ALIVE_DESCRIPTION = "Keep this session scheduled across turns and exempt from age expiry. Requires a fixed interval. Explicit max_runs, session deletion, and the failure-pause threshold still apply. Use it in a background or subagent session whose work outlives the turn that started it: the scheduler keeps re-prompting the session on the cadence, and the loop is never ended automatically by age expiry. You remain responsible for calling stop_loop once the work is resolved or abandoned." @@ -176,9 +177,10 @@ const server: Plugin = async ({ client, directory }, options?: Options) => { const restartFallbackMs = dynamicRestartFallbackMs(options, minIntervalSeconds, dynamicMaxDelaySeconds) let disposed = false - const timers = new Map>() + const timers = new Map() const sendingLoops = new Set() const busySessions = new Set() + const deletedSessions = new Set() const instanceID = randomUUID() const loopOwnerID = `server:${process.pid}:${randomUUID()}` const ownedSessions = new Map() @@ -196,6 +198,7 @@ const server: Plugin = async ({ client, directory }, options?: Options) => { // 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. const dynamicPending = new Map() + const deferredDynamicSettlements = new Map() const isRestrictedAgent = (agent: string | null | undefined) => typeof agent === "string" && restrictedAgents.has(agent.trim().toLowerCase()) @@ -208,7 +211,7 @@ const server: Plugin = async ({ client, directory }, options?: Options) => { function cancelTimer(loopID: string) { const timer = timers.get(loopID) - if (timer) clearTimeout(timer) + if (timer) clearTimeout(timer.handle) timers.delete(loopID) } @@ -240,17 +243,30 @@ const server: Plugin = async ({ client, directory }, options?: Options) => { } } + async function reconcileTimers(sessionID: string) { + for (const loop of await activeLoops(sessionID)) { + const timer = timers.get(loop.id) + if (!sendingLoops.has(loop.id) && (!timer || timer.nextRunAt !== loop.nextRunAt)) scheduleTimer(loop) + } + } + async function acquireOwnership(sessionID: string) { return enqueueSessionOperation(ownershipQueues, sessionID, async () => { if (disposed) return false if (!observedSessions.has(sessionID) && !(await keepAliveSessionAvailable(sessionID))) return false + const current = ownedSessions.get(sessionID) + if (current && current.expiresAt > Date.now() + SESSION_LEASE_HEARTBEAT_MS && + await ownsSessionLease(sessionID, instanceID, current.revision)) { + await reconcileTimers(sessionID) + return true + } 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) + await reconcileTimers(sessionID) return true }) } @@ -260,19 +276,20 @@ const server: Plugin = async ({ client, directory }, options?: Options) => { if (disposed) return false const lease = ownedSessions.get(sessionID) if (!lease) return false - const renewed = await renewSessionLease(sessionID, instanceID, lease.revision, SESSION_LEASE_MS) + const renewed = await renewSessionLease(sessionID, instanceID, lease.revision, SESSION_LEASE_MS, true) if (!renewed) { await loseOwnership(sessionID) return false } ownedSessions.set(sessionID, renewed) + await reconcileTimers(sessionID) return true }) } function scheduleTimer(loop: LoopSnapshot, minimumDelayMs = 0) { cancelTimer(loop.id) - if (disposed || loop.status !== "active" || loop.nextRunAt == null || !ownedSessions.has(loop.sessionID) || + if (disposed || loop.status !== "active" || loop.nextRunAt == null || deletedSessions.has(loop.sessionID) || !ownedSessions.has(loop.sessionID) || (!observedSessions.has(loop.sessionID) && !loop.keepAlive)) return const delay = Math.max(minimumDelayMs, loop.nextRunAt - Date.now()) const timer = setTimeout(() => { @@ -281,7 +298,7 @@ const server: Plugin = async ({ client, directory }, options?: Options) => { }, delay) const maybeUnref = timer as { unref?: () => void } if (typeof maybeUnref.unref === "function") maybeUnref.unref() - timers.set(loop.id, timer) + timers.set(loop.id, { handle: timer, nextRunAt: loop.nextRunAt }) } async function runDue(loopID: string) { @@ -293,12 +310,16 @@ const server: Plugin = async ({ client, directory }, options?: Options) => { await log("error", "Loop scheduler operation failed", { loopID, category: "scheduler" }) } finally { sendingLoops.delete(loopID) + const pending = deferredDynamicSettlements.get(loopID) + if (pending && !disposed) await settleDynamicLoop(loopID, pending).catch(() => + log("error", "Dynamic loop settlement failed", { loopID, category: "state" }), + ) } } async function runDueLocked(loopID: string) { let loop = await getLoop(loopID) - if (!loop || loop.status !== "active" || loop.nextRunAt == null) return + if (!loop || loop.status !== "active" || loop.nextRunAt == null || deletedSessions.has(loop.sessionID)) return if (!(await renewOwnership(loop.sessionID))) return const lease = ownedSessions.get(loop.sessionID) if (!lease || !(await ownsSessionLease(loop.sessionID, instanceID, lease.revision))) { @@ -353,7 +374,7 @@ const server: Plugin = async ({ client, directory }, options?: Options) => { if (loop.mode === "dynamic") { dynamicPending.set(loopID, { sessionID: loop.sessionID, sawBusy: false }) } - if (disposed || !(await confirmRunClaim(loopID, claimed))) { + if (!(await confirmRunClaim(loopID, claimed)) || disposed || deletedSessions.has(loop.sessionID)) { dynamicPending.delete(loopID) const current = await getLoop(loopID) if (current) scheduleTimer(current) @@ -410,19 +431,57 @@ const server: Plugin = async ({ client, directory }, options?: Options) => { } } - async function settleDynamicLoops(sessionID: string) { - const lease = ownedSessions.get(sessionID) + async function settleDynamicLoop(loopID: string, pending: PendingDynamicTurn) { + if (dynamicPending.get(loopID) !== pending || !pending.sawBusy) { + if (deferredDynamicSettlements.get(loopID) === pending) deferredDynamicSettlements.delete(loopID) + return + } + const lease = ownedSessions.get(pending.sessionID) if (!lease) return + deferredDynamicSettlements.set(loopID, pending) + if (sendingLoops.has(loopID)) return + const loop = await getLoop(loopID) + if (dynamicPending.get(loopID) !== pending) { + if (deferredDynamicSettlements.get(loopID) === pending) deferredDynamicSettlements.delete(loopID) + return + } + if (sendingLoops.has(loopID)) return + if (loop?.status === "active" && loop.mode === "dynamic" && loop.nextRunAt == null) { + const stopped = await stopLoopIfUnchanged( + loopID, loop.updatedAt, "the iteration ended without scheduling the next run", lease, loop.scheduleRevision, + ) + if (!stopped) return + } + if (dynamicPending.get(loopID) === pending) dynamicPending.delete(loopID) + if (deferredDynamicSettlements.get(loopID) === pending) deferredDynamicSettlements.delete(loopID) + } + + function rememberTerminalTurn(sessionID: string) { for (const [loopID, pending] of dynamicPending) { - if (pending.sessionID !== sessionID || !pending.sawBusy) continue - dynamicPending.delete(loopID) - const loop = await getLoop(loopID) - if (!loop || loop.status !== "active" || loop.mode !== "dynamic") continue - if (loop.nextRunAt != null) continue - await stopLoopIfUnchanged( - loopID, loop.updatedAt, "the iteration ended without scheduling the next run", lease, - ).catch(() => undefined) - await log("info", "Dynamic loop ended because the turn did not schedule the next run", { loopID }) + if (pending.sessionID === sessionID && pending.sawBusy) deferredDynamicSettlements.set(loopID, pending) + } + } + + async function settleDynamicLoops(sessionID: string) { + for (const [loopID, pending] of dynamicPending) { + if (pending.sessionID === sessionID) await settleDynamicLoop(loopID, pending) + } + } + + async function retryDeferredState(sessionID: string) { + if (deletedSessions.has(sessionID)) { + for (const loop of await stopLoopsForSession(sessionID, "session deleted")) { + cancelTimer(loop.id) + dynamicPending.delete(loop.id) + deferredDynamicSettlements.delete(loop.id) + } + deletedSessions.delete(sessionID) + observedSessions.delete(sessionID) + keepAliveSessions.delete(sessionID) + return + } + for (const [loopID, pending] of deferredDynamicSettlements) { + if (pending.sessionID === sessionID) await settleDynamicLoop(loopID, pending) } } @@ -450,6 +509,9 @@ const server: Plugin = async ({ client, directory }, options?: Options) => { async function observeSession(sessionID: string) { observedSessions.add(sessionID) if (!(await acquireOwnership(sessionID))) return + const wasDeleted = deletedSessions.has(sessionID) + await retryDeferredState(sessionID) + if (wasDeleted) return const lease = ownedSessions.get(sessionID) if (!lease) return for (const [loopID, candidate] of staleDynamicCandidates) { @@ -481,10 +543,10 @@ const server: Plugin = async ({ client, directory }, options?: Options) => { await rehydrate().catch(() => log("error", "Failed to rehydrate loops", { category: "state" })) const leaseHeartbeat = setInterval(() => { - for (const sessionID of new Set([...observedSessions, ...keepAliveSessions])) { + for (const sessionID of new Set([...observedSessions, ...keepAliveSessions, ...deletedSessions])) { const operation = ownedSessions.has(sessionID) ? renewOwnership(sessionID) : observedSessions.has(sessionID) ? observeSession(sessionID) : acquireOwnership(sessionID) - void operation.catch(() => log("error", "Loop lease heartbeat failed", { category: "state" })) + void operation.then(() => retryDeferredState(sessionID)).catch(() => log("error", "Loop lease heartbeat failed", { category: "state" })) } }, SESSION_LEASE_HEARTBEAT_MS) leaseHeartbeat.unref?.() @@ -494,9 +556,11 @@ const server: Plugin = async ({ client, directory }, options?: Options) => { disposed = true await Promise.all(ownershipQueues.values()) clearInterval(leaseHeartbeat) - for (const timer of timers.values()) clearTimeout(timer) + for (const timer of timers.values()) clearTimeout(timer.handle) timers.clear() dynamicPending.clear() + deferredDynamicSettlements.clear() + deletedSessions.clear() await Promise.all( [...ownedSessions.values()].map((lease) => releaseSessionLease(lease.sessionID, instanceID, lease.revision)), ) @@ -656,52 +720,71 @@ const server: Plugin = async ({ client, directory }, options?: Options) => { ? output.message.agent : undefined if (typeof sessionID !== "string") return - await observeSession(sessionID) - if (typeof agent !== "string" || !agent.trim()) return - lastPromptAgentBySession.set(sessionID, agent.trim()) + if (typeof agent === "string" && agent.trim()) lastPromptAgentBySession.set(sessionID, agent.trim()) + try { + await observeSession(sessionID) + } catch { + // Passive hooks must never break normal dialogue: a transient state + // lock failure is retried by the scheduler heartbeat. + await log("error", "Loop session observation failed", { sessionID, category: "state" }) + return + } }, 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 - if (output.system.some((block) => block.includes(LOOP_SYSTEM_MARKER))) return - if (output.system.length === 0) output.system.push(reminder) - else output.system[0] = `${output.system[0]}\n\n${reminder}` + try { + await observeSession(input.sessionID) + const loops = await openLoops(input.sessionID) + const reminder = systemReminder(loops) + if (!reminder) return + if (output.system.some((block) => block.includes(LOOP_SYSTEM_MARKER))) return + if (output.system.length === 0) output.system.push(reminder) + else output.system[0] = `${output.system[0]}\n\n${reminder}` + } catch { + await log("error", "Loop system reminder failed", { sessionID: input.sessionID, category: "state" }) + } }, 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) + try { + await observeSession(input.sessionID) + const loops = await openLoops(input.sessionID) + const context = compactionContext(loops) + if (context) output.context.push(context) + } catch { + await log("error", "Loop compaction context failed", { sessionID: input.sessionID, category: "state" }) + } }, async event({ event }) { const typed = event as { type?: string; properties?: Record } const sessionID = sessionIDFromEvent(typed) if (!sessionID) return - await observeSession(sessionID) if (isBusyEvent(typed)) { busySessions.add(sessionID) for (const pending of dynamicPending.values()) { if (pending.sessionID === sessionID) pending.sawBusy = true } - return + } + if (isIdleEvent(typed)) { + busySessions.delete(sessionID) + rememberTerminalTurn(sessionID) } if (typed.type === "session.deleted") { + deletedSessions.add(sessionID) busySessions.delete(sessionID) lastPromptAgentBySession.delete(sessionID) - const stopped = await stopLoopsForSession(sessionID, "session deleted") - for (const loop of stopped) { - cancelTimer(loop.id) - dynamicPending.delete(loop.id) - } - return } - if (isIdleEvent(typed)) { - busySessions.delete(sessionID) - await settleDynamicLoops(sessionID) - await runDueForSession(sessionID) + try { + await observeSession(sessionID) + if (typed.type === "session.deleted") { + await retryDeferredState(sessionID) + return + } + if (isIdleEvent(typed)) { + await settleDynamicLoops(sessionID) + await runDueForSession(sessionID) + } + } catch { + await log("error", "Loop event handling failed", { sessionID, category: "state" }) } }, } @@ -763,9 +846,10 @@ async function setupV2(context: PluginV2.Plugin.Context): Promise>() + const timers = new Map() const sendingLoops = new Set() const busySessions = new Set() + const deletedSessions = new Set() const instanceID = randomUUID() const loopOwnerID = `server-v2:${process.pid}:${randomUUID()}` const ownedSessions = new Map() @@ -827,7 +911,7 @@ async function setupV2(context: PluginV2.Plugin.Context): Promise { if (disposed) return false if (!observedSessions.has(sessionID) && !(await keepAliveSessionAvailable(sessionID))) return false + const current = ownedSessions.get(sessionID) + if (current && current.expiresAt > Date.now() + SESSION_LEASE_HEARTBEAT_MS && + await ownsSessionLease(sessionID, instanceID, current.revision)) { + await reconcileTimers(sessionID) + return true + } 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) + await reconcileTimers(sessionID) return true }) } @@ -870,19 +967,20 @@ async function setupV2(context: PluginV2.Plugin.Context): Promise void } if (typeof maybeUnref.unref === "function") maybeUnref.unref() - timers.set(loop.id, timer) + timers.set(loop.id, { handle: timer, nextRunAt: loop.nextRunAt }) } async function runDue(loopID: string) { @@ -905,8 +1003,7 @@ async function setupV2(context: PluginV2.Plugin.Context): Promise v2Log("error", "Dynamic loop settlement failed", { loopID, category: "scheduler" }), ) @@ -916,7 +1013,7 @@ async function setupV2(context: PluginV2.Plugin.Context): Promise undefined) - v2Log("info", "Dynamic loop ended because the turn did not schedule the next run", { loopID }) + if (dynamicPending.get(loopID) !== pending) { + if (deferredDynamicSettlements.get(loopID) === pending) deferredDynamicSettlements.delete(loopID) + return + } + if (sendingLoops.has(loopID) || shuttingDownSessions.has(pending.sessionID)) return + if (loop?.status === "active" && loop.mode === "dynamic" && loop.nextRunAt == null) { + const stopped = await stopLoopIfUnchanged( + loopID, loop.updatedAt, "the iteration ended without scheduling the next run", lease, loop.scheduleRevision, + ) + if (!stopped) return + } + if (dynamicPending.get(loopID) === pending) dynamicPending.delete(loopID) + if (deferredDynamicSettlements.get(loopID) === pending) deferredDynamicSettlements.delete(loopID) + } + + function rememberTerminalTurn(sessionID: string) { + for (const [loopID, pending] of dynamicPending) { + if (pending.sessionID === sessionID && pending.sawBusy) deferredDynamicSettlements.set(loopID, pending) + } } async function settleDynamicLoops(sessionID: string) { @@ -1047,6 +1160,23 @@ 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 - if (sessionContext.system.some((part) => part.type === "text" && part.text.includes(LOOP_SYSTEM_MARKER))) return - sessionContext.system.push({ type: "text", text: reminder }) + try { + await observeSession(sessionContext.sessionID) + const loops = await openLoops(sessionContext.sessionID) + const reminder = systemReminder(loops) + if (!reminder) return + if (sessionContext.system.some((part) => part.type === "text" && part.text.includes(LOOP_SYSTEM_MARKER))) return + sessionContext.system.push({ type: "text", text: reminder }) + } catch { + // Passive hooks must never break normal dialogue; the scheduler + // heartbeat retries the observation. + v2Log("error", "Loop context hook failed", { sessionID: sessionContext.sessionID, category: "state" }) + } }), ) registrations.push( await context.session.hook("compaction", async (sessionCompaction) => { - await observeSession(sessionCompaction.sessionID) - const loops = await openLoops(sessionCompaction.sessionID) - const contextText = compactionContext(loops) - if (!contextText) return - if (sessionCompaction.system.some((part) => part.type === "text" && part.text.includes("OpenCode loop mode is tracking recurring loops for this session across compaction."))) return - sessionCompaction.system.push({ type: "text", text: contextText }) + try { + await observeSession(sessionCompaction.sessionID) + const loops = await openLoops(sessionCompaction.sessionID) + const contextText = compactionContext(loops) + if (!contextText) return + if (sessionCompaction.system.some((part) => part.type === "text" && part.text.includes("OpenCode loop mode is tracking recurring loops for this session across compaction."))) return + sessionCompaction.system.push({ type: "text", text: contextText }) + } catch { + v2Log("error", "Loop compaction hook failed", { sessionID: sessionCompaction.sessionID, category: "state" }) + } }), ) await rehydrate().catch(() => v2Log("error", "Failed to rehydrate loops", { category: "state" })) const leaseHeartbeat = setInterval(() => { - for (const sessionID of new Set([...observedSessions, ...keepAliveSessions])) { + for (const sessionID of new Set([...observedSessions, ...keepAliveSessions, ...deletedSessions])) { const operation = ownedSessions.has(sessionID) ? renewOwnership(sessionID) : observedSessions.has(sessionID) ? observeSession(sessionID) : acquireOwnership(sessionID) - void operation.catch(() => v2Log("error", "Loop lease heartbeat failed", { category: "state" })) + void operation.then(() => retryDeferredState(sessionID)).catch(() => v2Log("error", "Loop lease heartbeat failed", { category: "state" })) } }, SESSION_LEASE_HEARTBEAT_MS) leaseHeartbeat.unref?.() @@ -1255,7 +1410,13 @@ async function setupV2(context: PluginV2.Plugin.Context): Promise releaseSessionLease(lease.sessionID, instanceID, lease.revision)), diff --git a/src/state.ts b/src/state.ts index d8da6c4..dbc12aa 100644 --- a/src/state.ts +++ b/src/state.ts @@ -1,7 +1,7 @@ import { randomUUID } from "node:crypto" -import { chmod, mkdir, link, readFile, rename, rm, writeFile } from "node:fs/promises" +import { chmod, mkdir, link, readFile, readdir, rename, rm, writeFile } from "node:fs/promises" import { homedir } from "node:os" -import { dirname, join } from "node:path" +import { basename, dirname, join } from "node:path" import { Data, Effect, Schema } from "effect" export type LoopStatus = "active" | "paused" | "stopped" | "completed" @@ -222,11 +222,28 @@ async function readRawState() { const LOCK_WAIT_MS = 30_000 const LOCK_RETRY_MS = 10 +// Windows reports benign lock handoffs (delete-pending handles, antivirus +// scans) as EPERM/EBUSY/EACCES/ENOTEMPTY on open/link. Retry those briefly; +// anything longer is genuine contention (handled via the EEXIST path) or a +// persistent problem worth surfacing instead of hanging until the lock +// deadline. +const TRANSIENT_LOCK_CODES = new Set(["EPERM", "EBUSY", "EACCES", "ENOTEMPTY"]) +const TRANSIENT_LOCK_BUDGET_MS = 5_000 function sleep(ms: number) { return new Promise((resolve) => setTimeout(resolve, ms)) } +export function isTransientLockError(error: unknown) { + return ( + typeof error === "object" && + error !== null && + "code" in error && + typeof error.code === "string" && + TRANSIENT_LOCK_CODES.has(error.code) + ) +} + function processIsAlive(pid: number) { try { process.kill(pid, 0) @@ -236,37 +253,86 @@ function processIsAlive(pid: number) { } } +const pendingLockReleases = new Map Promise>() + +async function retryLockOperation(operation: () => Promise, acquisitionDeadline = Infinity) { + const deadline = Math.min(acquisitionDeadline, Date.now() + TRANSIENT_LOCK_BUDGET_MS) + for (;;) { + try { + return await operation() + } catch (error) { + if (!isTransientLockError(error) || Date.now() >= deadline) throw error + await sleep(LOCK_RETRY_MS) + } + } +} + +function processLockRelease(lock: string, ownerPath: string, token: string) { + let running: Promise | undefined + let timer: ReturnType | undefined + const finish = () => { + running ??= (async () => { + await retryLockOperation(async () => { + const current = await readFile(lock, "utf8").catch((error: unknown) => { + if (isMissingStateFile(error)) return null + throw error + }) + if (current === token) await rm(lock, { force: true }) + }) + if (pendingLockReleases.get(lock) === finish) pendingLockReleases.delete(lock) + if (timer) clearTimeout(timer) + timer = undefined + await rm(ownerPath, { force: true }).catch(() => undefined) + })().finally(() => { running = undefined }) + return running + } + const release = async () => { + try { + await finish() + } catch { + // Keep the exact token available for recovery even while this pid lives. + pendingLockReleases.set(lock, finish) + if (!timer) { + timer = setTimeout(() => { timer = undefined; void release() }, 1000) + timer.unref?.() + } + } + } + return release +} + async function acquireProcessLock(lock: string, deadline: number): Promise<() => Promise> { + await pendingLockReleases.get(lock)?.() const token = `${process.pid}-${randomUUID()}` const ownerPath = `${lock}.owner.${token}` - await writeFile(ownerPath, token, { mode: 0o600, flag: "wx" }) let acquired = false try { + await retryLockOperation(async () => { + // A failed create may leave a partial file at this attempt's unique path. + await rm(ownerPath, { force: true }) + await writeFile(ownerPath, token, { mode: 0o600, flag: "wx" }) + }, deadline) for (;;) { + if (Date.now() >= deadline) throw new Error("timed out waiting for the loop state lock") try { // Publish the complete identity atomically without requiring symlink privileges. - await link(ownerPath, lock) + await retryLockOperation(() => link(ownerPath, lock), deadline) acquired = true - return async () => { - try { - if (await readFile(lock, "utf8").catch(() => null) === token) await rm(lock, { force: true }) - } finally { - await rm(ownerPath, { force: true }) - } - } + return processLockRelease(lock, ownerPath, token) } catch (error) { - if ((error as NodeJS.ErrnoException).code !== "EEXIST") throw error + if (typeof error !== "object" || error === null || !("code" in error) || error.code !== "EEXIST") throw error try { - const owner = await readFile(lock, "utf8") + const owner = await retryLockOperation(() => readFile(lock, "utf8"), deadline) const pid = Number(owner.split("-", 1)[0]) if (/^\d+-[0-9a-f-]{36}$/.test(owner) && !processIsAlive(pid)) { - // Recover the guard by the same protocol if a previous reclaimer died. const release = await acquireProcessLock(`${lock}.reclaim.${owner}`, deadline) try { - if (await readFile(lock, "utf8").catch(() => null) === owner) { - await rm(lock, { force: true }) - await rm(`${lock}.owner.${owner}`, { force: true }) - } + await retryLockOperation(async () => { + if (await readFile(lock, "utf8") === owner) { + await rm(lock, { force: true }) + await rm(`${lock}.owner.${owner}`, { force: true }) + } + }, deadline) } finally { await release() } @@ -279,13 +345,35 @@ async function acquireProcessLock(lock: string, deadline: number): Promise<() => } } } finally { - if (!acquired) await rm(ownerPath, { force: true }) + if (!acquired) await rm(ownerPath, { force: true }).catch(() => undefined) + } +} + +async function sweepStaleOwnerFiles(lock: string) { + // Contenders that crashed between creating their owner file and linking + // leave orphans no other path removes. Reclaim only dead pids; a live + // contender's file may be mid-link. + try { + const dir = dirname(lock) + const prefix = `${basename(lock)}.owner.` + const entries = await readdir(dir) + for (const entry of entries) { + if (!entry.startsWith(prefix)) continue + const rest = entry.slice(prefix.length) + if (!/^\d+-[0-9a-f-]{36}$/.test(rest)) continue + const pid = Number(rest.split("-", 1)[0]) + if (pid === process.pid || processIsAlive(pid)) continue + await rm(join(dir, entry), { force: true }).catch(() => undefined) + } + } catch { + // Best-effort hygiene; lock acquisition proceeds regardless. } } async function acquireMutationLock() { const file = statePath() await mkdir(dirname(file), { recursive: true, mode: 0o700 }) + await sweepStaleOwnerFiles(`${file}.lock`) return acquireProcessLock(`${file}.lock`, Date.now() + LOCK_WAIT_MS) } @@ -480,12 +568,16 @@ export async function acquireSessionLease(sessionID: string, instanceID: string, } /** Renews only the exact lease revision previously returned to this instance. */ -export async function renewSessionLease(sessionID: string, instanceID: string, revision: number, leaseMs: number) { +export async function renewSessionLease(sessionID: string, instanceID: string, revision: number, leaseMs: number, recoverExpired = false) { 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 + if (!current || current.instanceID !== instanceID || current.revision !== revision) return null + if (current.expiresAt <= timestamp) { + if (!recoverExpired) return null + current.revision += 1 + } current.expiresAt = timestamp + duration return { ...current } }) @@ -646,12 +738,14 @@ export async function stopLoopIfUnchanged( expectedUpdatedAt: number, reason: string, sessionLease?: SessionLease, + expectedScheduleRevision?: number, ) { return mutate((state) => { const loop = state.loops[loopID] if (!loop || (sessionLease && (sessionLease.sessionID !== loop.sessionID || !matchesSessionLease(state, sessionLease)))) return null if ( loop.updatedAt !== expectedUpdatedAt || + (expectedScheduleRevision !== undefined && loop.scheduleRevision !== expectedScheduleRevision) || loop.status !== "active" || loop.mode !== "dynamic" || loop.nextRunAt != null diff --git a/test/lock-retry.test.ts b/test/lock-retry.test.ts new file mode 100644 index 0000000..cf440d4 --- /dev/null +++ b/test/lock-retry.test.ts @@ -0,0 +1,90 @@ +import { expect, test } from "bun:test" +import { mkdtemp, rm, writeFile } from "node:fs/promises" +import { tmpdir } from "node:os" +import { join } from "node:path" +import { pathToFileURL } from "node:url" + +for (const scenario of ["release-read", "release-remove", "release-exhausted", "owner-partial", "late-handoff", "read-budget"]) { + test(`lock protocol recovers safely from ${scenario}`, async () => { + const directory = await mkdtemp(join(tmpdir(), "loop-lock-fault-")) + try { + const file = join(directory, "loops.json") + const script = join(directory, "fault.ts") + await writeFile(script, ` + import { mock } from "bun:test" + import * as filesystem from "node:fs/promises" + const fs = { ...filesystem } + const scenario = ${JSON.stringify(scenario)} + const file = ${JSON.stringify(file)} + const lock = file + ".lock" + const originalNow = Date.now + let clock = originalNow() + let faults = 0 + let links = 0 + let retrying = false + const denied = () => Object.assign(new Error("injected lock failure"), { code: "EPERM" }) + Date.now = () => clock + mock.module("node:fs/promises", () => ({ + ...fs, + writeFile: async (path, value, options) => { + if (scenario === "owner-partial" && String(path).includes(".lock.owner.") && faults++ === 0) { + await fs.writeFile(path, "", options) + throw denied() + } + return fs.writeFile(path, value, options) + }, + link: async (source, target) => { + if (target === lock) { + links++ + if (scenario === "read-budget" || (scenario === "late-handoff" && links === 1)) { + throw Object.assign(new Error("contended"), { code: "EEXIST" }) + } + if (scenario === "late-handoff" && links === 2) { faults++; throw denied() } + } + return fs.link(source, target) + }, + readFile: async (path, options) => { + if (path === lock) { + if (scenario === "late-handoff" && links === 1) { + clock += 6000 + return process.pid + "-00000000-0000-0000-0000-000000000000" + } + if (scenario === "read-budget") { faults++; clock += 1000; throw denied() } + if (scenario === "release-read" && faults++ === 0) throw denied() + if (scenario === "release-exhausted" && !retrying) { faults++; clock += 6000; throw denied() } + } + return fs.readFile(path, options) + }, + rm: async (path, options) => { + if (scenario === "release-remove" && path === lock && faults++ < 2) throw denied() + return fs.rm(path, options) + }, + })) + const state = await import(${JSON.stringify(pathToFileURL(join(import.meta.dir, "../src/state.ts")).href)}) + try { + if (scenario === "read-budget") { + let rejected = false + try { await state.createLoop("ses_fault", { prompt: "tick", intervalMs: 1000 }) } + catch { rejected = true } + if (!rejected || faults > 7) throw new Error("transient read exceeded its 5s budget: " + faults) + } else { + await state.createLoop("ses_fault", { prompt: "first", intervalMs: 1000 }) + retrying = true + await state.createLoop("ses_fault", { prompt: "second", intervalMs: 1000 }) + const loops = await state.listLoops("ses_fault") + if (loops.length !== 2 || faults === 0) throw new Error("mutations did not recover") + } + const leftovers = (await fs.readdir(${JSON.stringify(directory)})).filter(x => x.includes(".lock")) + if (leftovers.length) throw new Error("lock leftovers: " + leftovers.join(",")) + } finally { Date.now = originalNow } + `) + const child = Bun.spawn([process.execPath, script], { + env: { ...process.env, OPENCODE_LOOP_STATE_PATH: file }, stdout: "pipe", stderr: "pipe", + }) + const [error, code] = await Promise.all([new Response(child.stderr).text(), child.exited]) + expect(code, error).toBe(0) + } finally { + await rm(directory, { recursive: true, force: true }) + } + }) +} diff --git a/test/server-v2.test.ts b/test/server-v2.test.ts index 2aff693..1bbfa5d 100644 --- a/test/server-v2.test.ts +++ b/test/server-v2.test.ts @@ -1006,3 +1006,169 @@ test("V2 bounds native wait even when the stable adapter ignores cancellation", expect(mock.promptCalls).toHaveLength(0) expect((await getLoop(loop.id))?.runCount).toBe(0) }) + +test("V2 context hook survives loop state failures", async () => { + const mock = makeMockContext() + const cleanup = await plugin.setup(mock as never) + // A file where the state directory should be makes every state mutation + // fail deterministically on any platform. + const blocker = join(dir, "blocker") + await writeFile(blocker, "not a directory") + process.env.OPENCODE_LOOP_STATE_PATH = join(blocker, "loops.json") + + const contextHook = mock.hooks["context"]! + const sessionContext = { + sessionID: "ses_broken_v2", + agent: "build", + system: [] as Array<{ type: string; text: string }>, + messages: [], + tools: {}, + } + await contextHook(sessionContext) + expect(sessionContext.system).toEqual([]) + await mock.hooks.compaction?.(sessionContext) + expect(sessionContext.system).toEqual([]) + + mock.stream.end() + await cleanup() +}) + +test("V2 event consumer survives a failing event", async () => { + const mock = makeMockContext() + const cleanup = await plugin.setup(mock as never) + const blocker = join(dir, "blocker") + await writeFile(blocker, "not a directory") + process.env.OPENCODE_LOOP_STATE_PATH = join(blocker, "loops.json") + + mock.stream.push({ type: "session.execution.started", created: Date.now(), data: { sessionID: "ses_bad" } }) + await waitFor(() => mock.stream.processed >= 1) + + // The consumer must still be alive to process later events once state + // recovers; without per-event handling the first failure kills it. + process.env.OPENCODE_LOOP_STATE_PATH = join(dir, "loops.json") + const loop = await createLoop("ses_ok", { prompt: "tick", intervalMs: 1 }) + const processed = mock.stream.processed + mock.stream.push({ type: "session.idle", created: Date.now(), data: { sessionID: "ses_ok" } }) + await waitFor(() => mock.stream.processed >= processed + 1) + await waitFor(async () => (await getLoop(loop.id))?.runCount === 1) + expect(mock.promptCalls.some((prompt) => prompt.sessionID === "ses_ok")).toBe(true) + + mock.stream.end() + await cleanup() +}) + +test("V2 retains Plan selection when its event cannot observe state", async () => { + const mock = makeMockContext() + const cleanup = await plugin.setup(mock as never) + const originalPath = process.env.OPENCODE_LOOP_STATE_PATH + try { + const result = JSON.parse(contentOf(await loopTool(mock, "create_loop").execute( + { instruction: "tick", interval: "1s" }, toolContext("ses_plan_failure"), + ))) as { created: string } + const blocker = join(dir, "plan-blocker") + await writeFile(blocker, "not a directory") + process.env.OPENCODE_LOOP_STATE_PATH = join(blocker, "loops.json") + const processed = mock.stream.processed + mock.stream.push({ type: "session.agent.selected", created: Date.now(), data: { sessionID: "ses_plan_failure", agent: "plan" } }) + await waitFor(() => mock.stream.processed > processed) + process.env.OPENCODE_LOOP_STATE_PATH = originalPath + await waitFor(async () => (await getLoop(result.created))?.lastResult === "skipped_plan") + expect(mock.promptCalls).toHaveLength(0) + } finally { + process.env.OPENCODE_LOOP_STATE_PATH = originalPath + mock.stream.end() + await cleanup() + } +}) + +test.each(["session.execution.succeeded", "session.deleted"])("V2 retries the persisted transition after %s fails", async (eventType) => { + const mock = makeMockContext() + const cleanup = await plugin.setup(mock as never) + const originalPath = process.env.OPENCODE_LOOP_STATE_PATH + try { + const result = JSON.parse(contentOf(await loopTool(mock, "create_loop").execute( + { instruction: "tick" }, toolContext("ses_retry_event"), + ))) as { created: string } + const blocker = join(dir, "event-blocker") + await writeFile(blocker, "not a directory") + process.env.OPENCODE_LOOP_STATE_PATH = join(blocker, "loops.json") + const processed = mock.stream.processed + mock.stream.push({ type: eventType, created: Date.now(), data: { sessionID: "ses_retry_event" } }) + await waitFor(() => mock.stream.processed > processed) + process.env.OPENCODE_LOOP_STATE_PATH = originalPath + await mock.hooks.context?.({ sessionID: "ses_retry_event", system: [] }) + expect((await getLoop(result.created))?.status).toBe("stopped") + expect(mock.promptCalls).toHaveLength(0) + } finally { + process.env.OPENCODE_LOOP_STATE_PATH = originalPath + mock.stream.end() + await cleanup() + } +}) + +test("V2 shutdown preserves unscheduled dynamic loops for restart fallback", async () => { + const first = makeMockContext() + const cleanupFirst = await plugin.setup(first as never) + const created = JSON.parse(contentOf(await loopTool(first, "create_loop").execute( + { instruction: "tick" }, toolContext(), + ))) as { created: string } + first.stream.push({ type: "session.execution.interrupted", created: Date.now(), data: { sessionID: "ses_v2", reason: "shutdown" } }) + await waitFor(() => first.stream.processed >= 1) + first.stream.push({ type: "session.status", created: Date.now(), data: { sessionID: "ses_v2", status: { type: "idle" } } }) + first.stream.push({ type: "session.idle", created: Date.now(), data: { sessionID: "ses_v2" } }) + await waitFor(() => first.stream.processed >= 3) + expect((await getLoop(created.created))?.status).toBe("active") + expect((await getLoop(created.created))?.nextRunAt).toBeNull() + first.stream.end() + await cleanupFirst() + const second = makeMockContext({ dynamic_restart_fallback_seconds: 120 }) + const cleanupSecond = await plugin.setup(second as never) + try { + await second.hooks.context?.({ sessionID: "ses_v2", system: [] }) + expect((await getLoop(created.created))?.nextRunAt).toBeGreaterThan(Date.now()) + } finally { + second.stream.end() + await cleanupSecond() + } +}) + +test("V2 recovers an expired local lease before delivering and settling a dynamic turn", async () => { + const mock = makeMockContext() + const cleanup = await plugin.setup(mock as never) + const originalNow = Date.now + try { + const created = JSON.parse(contentOf(await loopTool(mock, "create_loop").execute({ instruction: "tick" }, toolContext()))) as { created: string } + await loopTool(mock, "schedule_next_run").execute({ loop_id: created.created, delay_seconds: 1, reason: "test" }, toolContext()) + const future = Date.now() + 91_000 + Date.now = () => future + await waitFor(async () => (await getLoop(created.created))?.runCount === 1) + mock.stream.push({ type: "session.execution.started", created: Date.now(), data: { sessionID: "ses_v2" } }) + mock.stream.push({ type: "session.execution.succeeded", created: Date.now(), data: { sessionID: "ses_v2" } }) + await waitFor(async () => (await getLoop(created.created))?.status === "stopped") + } finally { + Date.now = originalNow + mock.stream.end() + await cleanup() + } +}) + +test("V2 cached recovery arms an ordinary sibling after its session is observed", async () => { + const first = makeMockContext() + const cleanupFirst = await plugin.setup(first as never) + await loopTool(first, "create_loop").execute({ instruction: "keep alive", interval: "1m", keep_alive: true }, toolContext()) + const ordinary = JSON.parse(contentOf(await loopTool(first, "create_loop").execute( + { instruction: "ordinary", interval: "1s" }, toolContext(), + ))) as { created: string } + first.stream.end() + await cleanupFirst() + const second = makeMockContext() + const cleanupSecond = await plugin.setup(second as never) + try { + expect(second.promptCalls).toHaveLength(0) + await second.hooks.context?.({ sessionID: "ses_v2", system: [] }) + await waitFor(async () => (await getLoop(ordinary.created))?.runCount === 1) + } finally { + second.stream.end() + await cleanupSecond() + } +}) diff --git a/test/server.test.ts b/test/server.test.ts index b12fcf1..43f208c 100644 --- a/test/server.test.ts +++ b/test/server.test.ts @@ -575,3 +575,83 @@ test("V1 keep-alive recovery refuses a session in another directory", async () = expect((await getLoop(loop.id))?.lastResult).toBeNull() await hooks.dispose?.() }) + +test("V1 passive hooks survive loop state failures instead of breaking dialogue", async () => { + const sent: SentPrompt[] = [] + const hooks = await makeServer(sent) + // A file where the state directory should be makes every state mutation + // fail deterministically on any platform. + const blocker = join(dir, "blocker") + await writeFile(blocker, "not a directory") + process.env.OPENCODE_LOOP_STATE_PATH = join(blocker, "loops.json") + const sessionID = "ses_broken_state" + await hooks["chat.message"]?.({ sessionID, agent: "build" } as never, { message: { sessionID } } as never) + const system = { system: [] as string[] } + await hooks["experimental.chat.system.transform"]?.({ sessionID } as never, system as never) + expect(system.system).toEqual([]) + const compacting = { sessionID, context: [] as string[] } + await hooks["experimental.session.compacting"]?.(compacting as never, compacting as never) + expect(compacting.context).toEqual([]) + await hooks.event?.({ event: { type: "session.deleted", properties: { sessionID } } } as never) + expect(sent).toHaveLength(0) + await hooks.dispose?.() +}) + +test("V1 retains Plan safety when passive observation fails", async () => { + const sent: SentPrompt[] = [] + const hooks = await makeServer(sent) + const originalPath = process.env.OPENCODE_LOOP_STATE_PATH + try { + const created = JSON.parse(await tool(hooks, "create_loop").execute( + { instruction: "tick", interval: "1s" }, { sessionID: "ses_plan_failure", agent: "build" }, + )) as { created: string } + const blocker = join(dir, "plan-blocker") + await writeFile(blocker, "not a directory") + process.env.OPENCODE_LOOP_STATE_PATH = join(blocker, "loops.json") + await hooks["chat.message"]?.({ sessionID: "ses_plan_failure", agent: "plan" } as never, { message: {} } as never) + process.env.OPENCODE_LOOP_STATE_PATH = originalPath + await waitFor(async () => (await getLoop(created.created))?.lastResult === "skipped_plan") + expect(sent).toHaveLength(0) + } finally { + process.env.OPENCODE_LOOP_STATE_PATH = originalPath + await hooks.dispose?.() + } +}) + +test.each(["session.idle", "session.deleted"])("V1 retries the persisted transition after %s fails", async (eventType) => { + const hooks = await makeServer([]) + const originalPath = process.env.OPENCODE_LOOP_STATE_PATH + try { + const result = JSON.parse(await tool(hooks, "create_loop").execute( + { instruction: "tick" }, { sessionID: "ses_retry_event" }, + )) as { created: string } + const blocker = join(dir, "event-blocker") + await writeFile(blocker, "not a directory") + process.env.OPENCODE_LOOP_STATE_PATH = join(blocker, "loops.json") + await hooks.event?.({ event: { type: eventType, properties: { sessionID: "ses_retry_event" } } } as never) + process.env.OPENCODE_LOOP_STATE_PATH = originalPath + await hooks["chat.message"]?.({ sessionID: "ses_retry_event", agent: "build" } as never, { message: {} } as never) + expect((await getLoop(result.created))?.status).toBe("stopped") + } finally { + process.env.OPENCODE_LOOP_STATE_PATH = originalPath + await hooks.dispose?.() + } +}) + +test("V1 cached recovery arms an ordinary sibling after its session is observed", async () => { + const first = await makeServer([]) + await tool(first, "create_loop").execute({ instruction: "keep alive", interval: "1m", keep_alive: true }, { sessionID: "ses_rearm" }) + const ordinary = JSON.parse(await tool(first, "create_loop").execute( + { instruction: "ordinary", interval: "1s" }, { sessionID: "ses_rearm" }, + )) as { created: string } + await first.dispose?.() + const sent: SentPrompt[] = [] + const second = await makeServer(sent) + try { + expect(sent).toHaveLength(0) + await second["chat.message"]?.({ sessionID: "ses_rearm", agent: "build" } as never, { message: {} } as never) + await waitFor(async () => (await getLoop(ordinary.created))?.runCount === 1) + } finally { + await second.dispose?.() + } +}) diff --git a/test/state.test.ts b/test/state.test.ts index d44b451..9ece731 100644 --- a/test/state.test.ts +++ b/test/state.test.ts @@ -13,6 +13,7 @@ import { formatInterval, formatLoops, getLoop, + isTransientLockError, listLoops, activeLoops, parseInterval, @@ -507,3 +508,51 @@ test("keep-alive loops cannot delegate pacing to the agent", async () => { /keep-alive loops require a fixed interval/, ) }) + +test("classifies transient file-lock errors for the cross-process lock protocol", () => { + // Windows reports benign lock handoffs (delete-pending handles, antivirus + // scans) as these codes on open/link. They must be retried, never fatal. + for (const code of ["EPERM", "EBUSY", "EACCES", "ENOTEMPTY"]) { + expect(isTransientLockError(Object.assign(new Error("denied"), { code }))).toBe(true) + } + for (const code of ["ENOENT", "EEXIST", "EISDIR", "EINVAL"]) { + expect(isTransientLockError(Object.assign(new Error("other"), { code }))).toBe(false) + } + expect(isTransientLockError(new Error("plain"))).toBe(false) + expect(isTransientLockError(null)).toBe(false) + expect(isTransientLockError("EPERM")).toBe(false) +}) + +test("lease recovery preserves exclusive ownership while fencing expired claims", async () => { + const originalNow = Date.now + let clock = originalNow() + Date.now = () => clock + try { + const lease = (await acquireSessionLease("ses_recover", "owner", 1000))! + clock += 1001 + expect(await renewSessionLease("ses_recover", "owner", lease.revision, 1000)).toBeNull() + const recovered = await renewSessionLease("ses_recover", "owner", lease.revision, 1000, true) + expect(recovered?.revision).toBe(lease.revision + 1) + expect(await ownsSessionLease("ses_recover", "owner", lease.revision)).toBe(false) + clock += 1001 + await acquireSessionLease("ses_recover", "foreign", 1000) + expect(await renewSessionLease("ses_recover", "owner", recovered!.revision, 1000, true)).toBeNull() + } finally { + Date.now = originalNow + } +}) + +test("a same-timestamp schedule revision fences an older dynamic settlement", async () => { + const originalNow = Date.now + const clock = originalNow() + Date.now = () => clock + try { + const loop = await createLoop("ses_revision", { prompt: "tick", mode: "dynamic" }) + await scheduleNextRun(loop.id, 1) + await recordRunSent(loop.id) + expect(await stopLoopIfUnchanged(loop.id, loop.updatedAt, "old turn", undefined, loop.scheduleRevision)).toBeNull() + expect((await getLoop(loop.id))?.status).toBe("active") + } finally { + Date.now = originalNow + } +})