diff --git a/doc/plugins/PLUGIN_RUNNER_EXECUTION.md b/doc/plugins/PLUGIN_RUNNER_EXECUTION.md index cd05568a2be..22400cd5ddb 100644 --- a/doc/plugins/PLUGIN_RUNNER_EXECUTION.md +++ b/doc/plugins/PLUGIN_RUNNER_EXECUTION.md @@ -344,6 +344,15 @@ unsettled Core tool result, other close error, or continuation of the same session still blocks replacement. A controller restart is not required for this stopped-owner reset path. +Core ends a dispatched task lease with `stop`, `status`, and `complete`. When +that sequence confirms the stop and completes the same task, Core does not also +call `onEnvironmentReleaseLease`. Core still calls it when the release did not +complete a dispatched task, and the cleanup retry calls it after any failure. +When `onEnvironmentResumeLease` returns no provider lease, Core sends `status` +first. It sends `stop` only when that `status` does not report +`executionStopped: true`. A driver must release task-scoped resources in its +`status` or `complete` handling, not only in `onEnvironmentReleaseLease`. + Termination receipts cover all run leases that owned the retired physical task. A stopped task gets a new provider task ID for checkpoint resumption. Inspection precedes in-memory session selection so a replacement cannot inherit the stopped diff --git a/server/src/__tests__/environment-task-runtime.test.ts b/server/src/__tests__/environment-task-runtime.test.ts index 71c830b821c..8007d130478 100644 --- a/server/src/__tests__/environment-task-runtime.test.ts +++ b/server/src/__tests__/environment-task-runtime.test.ts @@ -85,9 +85,25 @@ describe("plugin-provided Runner execution", () => { expect(operations).not.toContain("complete"); release(); const result = await checked; - if (preparationStopped) { expect(result).toEqual({ value: true, error: null }); expect(operations).toEqual(["stop", "status", "complete"]); } + if (preparationStopped) { expect(result).toEqual({ value: "completed", error: null }); expect(operations).toEqual(["stop", "status", "complete"]); } else { expect(result.error?.message).toContain("termination is unconfirmed"); expect(fixture.sets).toContainEqual(expect.objectContaining({ status: "pending_cleanup", cleanupStatus: "failed" })); } }); + it("still stops a startup task first when resume already observed the stop", async () => { + const value = row(); + Object.assign(value.lease.metadata, { runnerTaskDispatched: false, + runnerTaskStartupCleanup: { contextId: "startup", requiresPreparationStop: true } }); + const fixture = database(value), workers = worker(); + fixture.db.select = ((...args: unknown[]) => (database(value).db.select as (...values: unknown[]) => unknown)(...args)) as Db["select"]; + Object.assign(workers, { drainStartup: vi.fn(async () => undefined) }); + const operations: string[] = []; + workers.call.mockImplementation(async (...args: any[]) => { + const operation = args[2].operation.kind; operations.push(operation); + return operation === "status" ? { kind: "status", taskId: "attempt-1", phase: "cancelled", executionStopped: true, preparationStopped: true } + : { kind: "accepted", taskId: "attempt-1" }; + }); + await expect(settleDispatchedEnvironmentTask(fixture.db, workers as never, { companyId: "company", id: leaseId }, { executionStopObserved: true })).resolves.toBe("completed"); + expect(operations).toEqual(["stop", "status", "complete"]); + }); it("negotiates the transient context without reading transport for an older plugin", async () => { const context = new NativeExecutionContext({ binding: { companyId: "company", agentId: "agent", runId: "run" } } as never, 1_000); const workers = worker(); diff --git a/server/src/__tests__/plugin-runner-task.integration.test.ts b/server/src/__tests__/plugin-runner-task.integration.test.ts index 403eba17f2e..fa7cebf6cdd 100644 --- a/server/src/__tests__/plugin-runner-task.integration.test.ts +++ b/server/src/__tests__/plugin-runner-task.integration.test.ts @@ -416,7 +416,7 @@ databaseSuite("Core plugin Runner task lifecycle", () => { return; } const next = await acquire; - expect(operations).toEqual(["stop", "status", "complete", "environmentReleaseLease", "environmentAcquireLease"]); + expect(operations).toEqual(["stop", "status", "complete", "environmentAcquireLease"]); expect(next.lease.id).not.toBe(oldLeaseId); await expect(resolvePluginTaskResume(db, { companyId, leaseId: next.lease.id, priorIdentity })).resolves.toBe("old-task"); await expect(runtime.task({ companyId, leaseId: next.lease.id, operation: { kind: "submit", projectIds: [], bootstrapTicket: "fixture", runner } })).rejects.toThrow("identity mismatch"); diff --git a/server/src/__tests__/plugin-task-credential-writeback.test.ts b/server/src/__tests__/plugin-task-credential-writeback.test.ts index 203e2b35e83..1ae26bb1fff 100644 --- a/server/src/__tests__/plugin-task-credential-writeback.test.ts +++ b/server/src/__tests__/plugin-task-credential-writeback.test.ts @@ -65,11 +65,11 @@ suite("durable plugin task credential handoff", () => { const lease = () => db.select().from(environmentLeases).where(and(eq(environmentLeases.id, leaseId), eq(environmentLeases.companyId, companyId))).then(rows => rows[0]!); return { companyId, agentId, runId, environmentId, leaseId, owner, selection: selection!, managed, workers, calls, lease, failExport(value: boolean) { failExport = value; }, credential: () => service.credential(selection!), - settle: () => settleDispatchedEnvironmentTask(db, workers, { companyId, id: leaseId }) }; + settle: (executionStopObserved = false) => settleDispatchedEnvironmentTask(db, workers, { companyId, id: leaseId }, { executionStopObserved }) }; } it("persists refreshed credentials before completing a stopped task, after controller files disappear", async () => { const f = await fixture(); - expect(await f.settle()).toBe(true); + expect(await f.settle()).toBe("completed"); expect(await f.credential()).toBe(auth("fixture-refreshed", 2021)); expect(f.calls.indexOf("credentials")).toBeLessThan(f.calls.indexOf("complete")); const saved = await f.lease(); @@ -77,6 +77,24 @@ suite("durable plugin task credential handoff", () => { expect(JSON.stringify(saved.metadata)).not.toContain("fixture-refreshed"); expect(JSON.stringify(saved.metadata)).not.toContain(f.managed.home!); }); + it("writes back credentials before completing a task whose stop was already observed, without stopping it again", async () => { + const f = await fixture(); + expect(await f.settle(true)).toBe("completed"); + expect(f.calls).toEqual(["status", "credentials", "complete"]); + expect(await f.credential()).toBe(auth("fixture-refreshed", 2021)); + expect((await f.lease()).metadata?.runnerTaskCredentialHandoff).toMatchObject({ state: "complete" }); + }); + it("leaves an observed-stop task for cleanup when credential writeback fails", async () => { + const f = await fixture(); f.failExport(true); + await expect(f.settle(true)).rejects.toThrow("termination is unconfirmed"); + expect(f.calls).toEqual(["status", "credentials"]); + expect(await f.lease()).toMatchObject({ status: "pending_cleanup", cleanupStatus: "failed" }); + expect((await f.lease()).metadata?.runnerTaskCredentialHandoff).toMatchObject({ state: "pending" }); + f.failExport(false); + expect(await f.settle()).toBe("completed"); + expect(f.calls.slice(2)).toEqual(["stop", "status", "credentials", "complete"]); + expect(await f.credential()).toBe(auth("fixture-refreshed", 2021)); + }); it("keeps a failed transfer pending and recovers through a fresh cleanup after run and environment deletion", async () => { const f = await fixture(); f.failExport(true); await expect(f.settle()).rejects.toThrow("termination is unconfirmed"); @@ -86,7 +104,7 @@ suite("durable plugin task credential handoff", () => { await db.delete(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); await db.delete(environments).where(eq(environments.id, f.environmentId)); f.failExport(false); - expect(await f.settle()).toBe(true); + expect(await f.settle()).toBe("completed"); expect(await f.credential()).toBe(auth("fixture-refreshed", 2021)); expect(f.calls.at(-1)).toBe("complete"); }); @@ -104,7 +122,7 @@ suite("durable plugin task credential handoff", () => { expect((await f.lease()).metadata?.runnerTaskCredentialHandoff).toMatchObject({ state: "pending" }); } finally { await db.execute(sql`drop trigger reject_runner_credential_save on company_secret_versions; drop function reject_runner_credential_save()`); } const receiver = createDb(database.connectionString); - expect(await settleDispatchedEnvironmentTask(receiver, f.workers, { companyId: f.companyId, id: f.leaseId })).toBe(true); + expect(await settleDispatchedEnvironmentTask(receiver, f.workers, { companyId: f.companyId, id: f.leaseId })).toBe("completed"); expect(await f.credential()).toBe(auth("fixture-refreshed", 2021)); expect(f.calls.filter(call => call === "credentials")).toHaveLength(2); expect(f.calls.at(-1)).toBe("complete"); @@ -113,13 +131,13 @@ suite("durable plugin task credential handoff", () => { it("does not overwrite an explicit account replacement", async () => { const f = await fixture(); await secretService(db).rotate(f.managed.credentialWriteback!.secretId, { value: auth("fixture-reconnected", 2022) }, { userId: f.owner }); - expect(await f.settle()).toBe(true); + expect(await f.settle()).toBe("completed"); expect(f.calls).not.toContain("credentials"); expect(await f.credential()).toBe(auth("fixture-reconnected", 2022)); }); it("allows concurrent cleanup to complete the same durable handoff", async () => { const f = await fixture(); - expect(await Promise.all([f.settle(), f.settle()])).toEqual([true, true]); + expect(await Promise.all([f.settle(), f.settle()])).not.toContain(false); expect(await f.credential()).toBe(auth("fixture-refreshed", 2021)); expect((await f.lease()).metadata?.runnerTaskCredentialHandoff).toMatchObject({ state: "complete" }); }); diff --git a/server/src/__tests__/plugin-warm-tasks.integration.test.ts b/server/src/__tests__/plugin-warm-tasks.integration.test.ts index 4bc77c5afdb..da9cc6e3ae5 100644 --- a/server/src/__tests__/plugin-warm-tasks.integration.test.ts +++ b/server/src/__tests__/plugin-warm-tasks.integration.test.ts @@ -7,6 +7,7 @@ import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; import { eq, sql } from "drizzle-orm"; import { agents, companies, createDb, applyPendingMigrations, environmentLeases, environments, heartbeatRuns, issues, plugins, projects, sandboxCreationOperations } from "@paperclipai/db"; import { environmentRuntimeService } from "../services/environment-runtime.js"; +import { environmentService } from "../services/environments.js"; import { recordSandboxCreationResource } from "../services/sandbox-creation-operations.js"; import { resolvePluginTaskResume } from "../services/environment-task-runtime.js"; import type { PluginWorkerManager } from "../services/plugin-worker-manager.js"; @@ -20,8 +21,8 @@ const support=external ? {supported:true} : await getEmbeddedPostgresTestSupport afterAll(async()=>{await stop?.();}); async function fixture(initialProjects:string[]=[],initialExpiry?:string,deadline?:Date){ const companyId=randomUUID(),agentId=randomUUID(),issueId=randomUUID(),environmentId=randomUUID(),pluginId=randomUUID(); - let runId=randomUUID(),allocation=0,stopped=false,outage=false,releaseFails=false; - const calls:string[]=[]; + let runId=randomUUID(),allocation=0,stopped=false,outage=false,releaseFails=false,unavailable=false; + const calls:string[]=[],dispatched:Array<[string,string]>=[],failures=new Set(); let projectIds=[...initialProjects],providerExpiry=initialExpiry; const taskExpiries=new Map(),acquireRequests:any[]=[]; let resumeGate:Promise|undefined,releaseResume:(()=>void)|undefined,resumes=0; @@ -38,12 +39,14 @@ const support=external ? {supported:true} : await getEmbeddedPostgresTestSupport const workers={isRunning:()=>true,getWorker:()=>({supportedMethods:["environmentTask","environmentAcquireLease","environmentResumeLease","environmentReleaseLease"]}), call:async(_id:string,method:string,p:any)=>{ calls.push(method==="environmentTask" ? p.operation.kind : method); + if(method==="environmentTask") dispatched.push([p.operation.kind,p.taskId]); + if(method==="environmentTask" && failures.delete(p.operation.kind)) throw Error(`${p.operation.kind} unavailable`); if(method==="environmentAcquireLease") { const id=`task-${++allocation}`;taskExpiries.set(id,providerExpiry);acquireRequests.push(p); return {providerLeaseId:id,...(providerExpiry ? {expiresAt:providerExpiry}:{}),metadata:{runId:p.runId}}; } if(method==="environmentResumeLease") {if(outage) throw Error("unreachable"); - if(resumeGate){if(++resumes===2)releaseResume!();await resumeGate;}return {providerLeaseId:stopped?null:p.providerLeaseId,expiresAt:taskExpiries.get(p.providerLeaseId),metadata:{runId:p.runId}};} + if(resumeGate){if(++resumes===2)releaseResume!();await resumeGate;}return {providerLeaseId:stopped||unavailable?null:p.providerLeaseId,expiresAt:taskExpiries.get(p.providerLeaseId),metadata:{runId:p.runId}};} if(method==="environmentReleaseLease") {if(releaseFails) throw Error("release unavailable");return {providerLeaseId:p.providerLeaseId,state:"stopped"};} if(p.operation.kind==="status") return {kind:"status",taskId:p.taskId,phase:stopped?"completed":"running",executionStopped:stopped,sessionStateAvailable:stopped}; if(p.operation.kind==="read_state") return {kind:"state",taskId:p.taskId,state:{...p.operation.identity,lifecycle:"ready"}}; @@ -63,7 +66,8 @@ const support=external ? {supported:true} : await getEmbeddedPostgresTestSupport launch:{runnerVersion:"fixture",runnerDigest:`sha256:${"a".repeat(64)}`,maxOutboxBytes:262144,p0ReserveBytes:65536,lifecyclePolicy:{mode:"warm",idleTimeoutMs:300000}}}}); await runtime().releaseRunLeases(runId,"released",undefined,"keep_running"); await db.update(heartbeatRuns).set({status:"succeeded"}).where(eq(heartbeatRuns.id,runId)); - return {companyId,first,identity,calls,runtime,acquire,acquireRequests, + return {companyId,first,identity,calls,dispatched,runtime,acquire,acquireRequests, + fail(kind:string){failures.add(kind);},unavailable(){unavailable=true;}, setExpiry(next?:string){providerExpiry=next;}, async nextRun(){runId=randomUUID();await db.insert(heartbeatRuns).values({id:runId,companyId,agentId,nativeIssueId:issueId,invocationSource:"manual",status:"running"});return runId;}, setProjects(next:string[]){projectIds=next;},stopped(){stopped=true;},failRelease(){releaseFails=true;},outage(){outage=true;}, @@ -109,6 +113,35 @@ const support=external ? {supported:true} : await getEmbeddedPostgresTestSupport expect(f.calls).toContain("complete"); await expect(resolvePluginTaskResume(db,{companyId:f.companyId,leaseId:next.lease.id,priorIdentity:f.identity})).resolves.toBe(f.first.lease.providerLeaseId); }); + it("retires a stopped retained task with one status and complete before allocating its replacement",async()=>{ + const f=await fixture();f.stopped();await f.nextRun(); + const calls=f.calls.length,dispatched=f.dispatched.length; + const next=await f.acquire(); + expect(f.calls.slice(calls)).toEqual(["environmentResumeLease","status","complete","environmentAcquireLease"]); + expect(f.dispatched.slice(dispatched)).toEqual([["status",f.first.lease.providerLeaseId],["complete",f.first.lease.providerLeaseId]]); + expect(next.lease.providerLeaseId).not.toBe(f.first.lease.providerLeaseId); + const [old]=await db.select().from(environmentLeases).where(eq(environmentLeases.id,f.first.lease.id)); + expect(old.status).toBe("released"); + expect(old.metadata).toMatchObject({runnerTaskExecutionStopped:true,runnerTaskDispatched:false,remoteExecutionTermination:{providerLeaseId:f.first.lease.providerLeaseId,state:"stopped"}}); + }); + it("stops a retained task that resume could not attach but that is still running",async()=>{ + const f=await fixture();f.unavailable();await f.nextRun(); + const calls=f.calls.length; + await f.acquire(); + expect(f.calls.slice(calls)).toEqual(["environmentResumeLease","status","stop","status","complete","environmentAcquireLease"]); + }); + it.each([["status",false],["complete",false],["stop",true]] as const)("keeps a stopped retained task for cleanup retry when %s fails",async(kind,running)=>{ + const f=await fixture();if(running) f.unavailable(); else f.stopped();await f.nextRun(); + f.fail(kind); + await expect(f.acquire()).rejects.toThrow("termination is unconfirmed"); + expect(f.calls.filter(call=>call==="environmentAcquireLease")).toHaveLength(1); + expect(f.calls).not.toContain("environmentReleaseLease"); + const pending=(await environmentService(db).getLeaseById(f.first.lease.id))!; + expect(pending).toMatchObject({status:"pending_cleanup",cleanupStatus:"failed"}); + const calls=f.calls.length; + await expect(f.runtime().retryPendingSandboxTeardown({environment:null,lease:pending})).resolves.toEqual({providerLeaseId:f.first.lease.providerLeaseId,state:"stopped"}); + expect(f.calls.slice(calls)).toEqual(["stop","status","complete","environmentReleaseLease"]); + }); it("does not allocate a replacement when the retained computer cannot be inspected",async()=>{ const f=await fixture();f.outage();await f.nextRun(); await expect(f.acquire()).rejects.toThrow("unreachable"); diff --git a/server/src/services/environment-runtime.ts b/server/src/services/environment-runtime.ts index 1e1a5d03de4..76b590c0d8f 100644 --- a/server/src/services/environment-runtime.ts +++ b/server/src/services/environment-runtime.ts @@ -3322,6 +3322,30 @@ function createPluginEnvironmentDriver( }; } + async function releasePluginRunLease(input: EnvironmentDriverReleaseInput, executionStopObserved = false) { + await db.update(environmentLeases).set({ metadata: sql`coalesce(${environmentLeases.metadata}, '{}'::jsonb) || '{"pluginLeaseCleanupMethod":"release"}'::jsonb` }).where(and(eq(environmentLeases.id, input.lease.id), eq(environmentLeases.companyId, input.lease.companyId))); + const settled = await settleDispatchedEnvironmentTask(db, workerManager, input.lease, { executionStopObserved }); + const taskStopped = settled !== false || + (input.lease.metadata?.runnerTaskLease === true && input.lease.metadata?.runnerTaskDispatched !== true && !input.lease.metadata?.runnerTaskStartup && !input.lease.metadata?.runnerTaskStartupCleanup); + if (settled !== "completed") { + const { plugin, driverKey, driverConfig } = await resolvePluginDriverForRelease(input); + await workerManager.call(plugin.id, "environmentReleaseLease", { + driverKey, + companyId: input.lease.companyId, + environmentId: input.environment.id, + issueId: input.lease.issueId, + config: driverConfig, + providerLeaseId: input.lease.providerLeaseId, + leaseMetadata: input.lease.metadata ?? undefined, + }); + } + return await environmentsSvc.releaseLease(input.lease.id, input.status, { + cleanupStatus: "success", + ...(taskStopped ? { remoteExecutionTermination: remoteTerminationReceipt(input.lease, + { providerLeaseId: input.lease.providerLeaseId, state: "stopped" }) } : {}), + }); + } + return { driver: "plugin", @@ -3363,7 +3387,7 @@ function createPluginEnvironmentDriver( providerLeaseId: previous.providerLeaseId, leaseMetadata: previous.metadata ?? undefined, }); if (!resumed?.providerLeaseId) { - await this.releaseRunLease({ environment: input.environment, lease: toEnvironmentLeaseSnapshot(previous), status: "released" }); + await releasePluginRunLease({ environment: input.environment, lease: toEnvironmentLeaseSnapshot(previous), status: "released" }, true); resumed = undefined; } } @@ -3423,25 +3447,8 @@ function createPluginEnvironmentDriver( }); }, - async releaseRunLease(input) { - await db.update(environmentLeases).set({ metadata: sql`coalesce(${environmentLeases.metadata}, '{}'::jsonb) || '{"pluginLeaseCleanupMethod":"release"}'::jsonb` }).where(and(eq(environmentLeases.id, input.lease.id), eq(environmentLeases.companyId, input.lease.companyId))); - const taskStopped = await settleDispatchedEnvironmentTask(db, workerManager, input.lease) || - (input.lease.metadata?.runnerTaskLease === true && input.lease.metadata?.runnerTaskDispatched !== true && !input.lease.metadata?.runnerTaskStartup && !input.lease.metadata?.runnerTaskStartupCleanup); - const { plugin, driverKey, driverConfig } = await resolvePluginDriverForRelease(input); - await workerManager.call(plugin.id, "environmentReleaseLease", { - driverKey, - companyId: input.lease.companyId, - environmentId: input.environment.id, - issueId: input.lease.issueId, - config: driverConfig, - providerLeaseId: input.lease.providerLeaseId, - leaseMetadata: input.lease.metadata ?? undefined, - }); - return await environmentsSvc.releaseLease(input.lease.id, input.status, { - cleanupStatus: "success", - ...(taskStopped ? { remoteExecutionTermination: remoteTerminationReceipt(input.lease, - { providerLeaseId: input.lease.providerLeaseId, state: "stopped" }) } : {}), - }); + releaseRunLease(input) { + return releasePluginRunLease(input); }, async resumeRunLease(input) { diff --git a/server/src/services/environment-task-runtime.ts b/server/src/services/environment-task-runtime.ts index b8b8a449aad..5ad92fad86b 100644 --- a/server/src/services/environment-task-runtime.ts +++ b/server/src/services/environment-task-runtime.ts @@ -493,7 +493,8 @@ export async function executeEnvironmentTask(db: Db, workers: PluginWorkerManage } /** Failure leaves the dispatch marker and provider handle available to cleanup retries. */ -export async function settleDispatchedEnvironmentTask(db: Db, workers: PluginWorkerManager, lease: { id: string; companyId: string }) { +export async function settleDispatchedEnvironmentTask(db: Db, workers: PluginWorkerManager, lease: { id: string; companyId: string }, + options: { executionStopObserved?: boolean } = {}): Promise { const [current] = await db.update(environmentLeases).set({ metadata: sql`coalesce(${environmentLeases.metadata}, '{}'::jsonb) || '{"runnerTaskRetiring":true}'::jsonb`, }).where(and(eq(environmentLeases.id, lease.id), eq(environmentLeases.companyId, lease.companyId))) @@ -503,19 +504,25 @@ export async function settleDispatchedEnvironmentTask(db: Db, workers: PluginWor if (current?.metadata?.runnerTaskDispatched !== true && !startup && !startupCleanup) { if (current?.metadata) await settlePluginTaskCredentialHandoff(db, { companyId: lease.companyId, leaseId: lease.id }, current.metadata, async () => { throw new Error("Undispatched task must not request credentials"); }); - return current?.metadata?.runnerTaskExecutionStopped === true; + return current?.metadata?.runnerTaskExecutionStopped === true ? "stopped" : false; } const scope = { companyId: lease.companyId, leaseId: lease.id }; try { const contextId = (startupCleanup as { contextId?: unknown } | undefined)?.contextId ?? (startup as { contextId?: unknown } | undefined)?.contextId; const draining = typeof contextId === "string" ? workers.drainStartup?.(contextId) : undefined; void draining?.catch(() => undefined); - await executeEnvironmentTask(db, workers, { ...scope, operation: { kind: "stop" } }); - const state = await executeEnvironmentTask(db, workers, { ...scope, operation: { kind: "status" } }); - if (state.kind !== "status" || state.executionStopped !== true || ((startup || startupCleanup?.requiresPreparationStop) && state.preparationStopped !== true)) throw new Error("Plugin Runner termination is unconfirmed"); + const confirmed = (state?: Awaited>) => state?.kind === "status" && state.executionStopped === true && + (!(startup || startupCleanup?.requiresPreparationStop) || state.preparationStopped === true); + let state = options.executionStopObserved === true && !startup && !startupCleanup + ? await executeEnvironmentTask(db, workers, { ...scope, operation: { kind: "status" } }) : undefined; + if (!confirmed(state)) { + await executeEnvironmentTask(db, workers, { ...scope, operation: { kind: "stop" } }); + state = await executeEnvironmentTask(db, workers, { ...scope, operation: { kind: "status" } }); + } + if (!confirmed(state)) throw new Error("Plugin Runner termination is unconfirmed"); await draining; await executeEnvironmentTask(db, workers, { ...scope, operation: { kind: "complete" } }); - return true; + return "completed"; } catch { await writeEnvironmentCleanup(db, tx => tx.update(environmentLeases).set({ status: "pending_cleanup", cleanupStatus: "failed", failureReason: "plugin_runner_task_cleanup_unconfirmed", releasedAt: new Date(), updatedAt: new Date(),