diff --git a/doc/plugins/PLUGIN_RUNNER_EXECUTION.md b/doc/plugins/PLUGIN_RUNNER_EXECUTION.md index e5aba121a67..cd05568a2be 100644 --- a/doc/plugins/PLUGIN_RUNNER_EXECUTION.md +++ b/doc/plugins/PLUGIN_RUNNER_EXECUTION.md @@ -218,6 +218,26 @@ Legacy intents without a recovery hint retain their deterministic ID. A hinted intent without a recorded ID is ambiguous and fails closed. None of these receipts replaces the provider's checks. +A reservation is a workspace whose `prepare` never returned a binding. When its +source task is done or cancelled, no task points at it, and no run for the task +is queued or running, Core archives it and sends `remove` with `binding: null` +and `expectedHeadSha: null` instead of reconciling it with an agent-free +`prepare`. The provider must resolve its own record for that execution +workspace, discard anything the unfinished preparation created, and return +`{ kind: "removed" }`, including when nothing was created. The removal is +persisted and replayed with the same operation ID like any other removal. A +provider that refuses it fails the removal, which backs off like any other +cleanup failure. A reservation that a task still points at, or whose task is +open, keeps the `prepare` reconciliation. + +When terminal cleanup inspection or removal fails, Core records +`metadata.terminalCleanupRetry` on the workspace with the failure count, the +next attempt time, and the failure `kind`, `reason`, and Core's message. Retries +back off from 5 minutes, doubling up to 4 hours. A successful inspection or +removal clears the record, and a reopen, which raises the lifecycle generation, +discards it. Other wakes, including provider upgrades, wait for the next attempt +time. Core keeps the record through workspace metadata updates. + The optional `prepared.destinationRef` records the resolved delivery branch as `refs/heads/`, up to 266 characters including the prefix. Providers should return null when preparation cannot establish an unambiguous branch destination diff --git a/packages/plugins/sdk/src/workspace-provider.ts b/packages/plugins/sdk/src/workspace-provider.ts index e46a3c4e98e..c212c3baffd 100644 --- a/packages/plugins/sdk/src/workspace-provider.ts +++ b/packages/plugins/sdk/src/workspace-provider.ts @@ -65,7 +65,7 @@ export const workspaceProviderOperationSchema = z.discriminatedUnion("kind", [ z .object({ kind: z.literal("remove"), - binding: workspaceProviderBindingSchema, + binding: workspaceProviderBindingSchema.nullable(), expectedHeadSha: sha.nullable(), recoveryPullRequest: workspaceRecoveryPullRequestSchema.optional(), }) diff --git a/packages/plugins/sdk/tests/workspace-provider.test.ts b/packages/plugins/sdk/tests/workspace-provider.test.ts index 74779cfa72c..9385be933cd 100644 --- a/packages/plugins/sdk/tests/workspace-provider.test.ts +++ b/packages/plugins/sdk/tests/workspace-provider.test.ts @@ -61,6 +61,11 @@ describe("workspace provider SDK", () => { } } }); + it("accepts an unbound reservation removal but requires a binding elsewhere", () => { + expect(workspaceProviderOperationSchema.parse({ kind: "remove", binding: null, expectedHeadSha: null })) + .toEqual({ kind: "remove", binding: null, expectedHeadSha: null }); + expect(() => workspaceProviderOperationSchema.parse({ kind: "inspect", binding: null })).toThrow(); + }); it("accepts only canonical optional preparation destinations", () => { for (const destinationRef of [undefined, null, "refs/heads/main", "refs/heads/release/one"]) { expect(parseWorkspaceProviderResult(params, { ...result, ...(destinationRef !== undefined ? { destinationRef } : {}) })) diff --git a/server/src/__tests__/plugin-workspace-terminal-retry.integration.test.ts b/server/src/__tests__/plugin-workspace-terminal-retry.integration.test.ts new file mode 100644 index 00000000000..4d879705541 --- /dev/null +++ b/server/src/__tests__/plugin-workspace-terminal-retry.integration.test.ts @@ -0,0 +1,250 @@ +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { eq, sql } from "drizzle-orm"; +import { createDb, executionWorkspaces, heartbeatRuns, issues } from "@paperclipai/db"; +import { JsonRpcCallError, PLUGIN_RPC_ERROR_CODES } from "@paperclipai/plugin-sdk"; +import { createPluginWorkspaceFixture } from "./helpers/plugin-workspaces.js"; +import { inspectPluginWorkspace, preparePluginWorkspace, removePluginWorkspace, type WorkspaceWorkers } from "../services/plugin-workspaces.js"; +import { executionWorkspaceService } from "../services/execution-workspaces.js"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; + +const support = await getEmbeddedPostgresTestSupport(); +const MINUTE = 60_000; + +function providerRefusal() { + return new JsonRpcCallError({ code: PLUGIN_RPC_ERROR_CODES.WORKER_ERROR, message: "private provider text", + data: { schema: "paperclip/workspace-operation-error/v1", reason: "unsupported_operation" } }); +} + +(support.supported ? describe : describe.skip)("plugin workspace terminal cleanup retry", () => { + let db: ReturnType, stopDb: (() => Promise) | undefined; + beforeAll(async () => { + const started = await startEmbeddedPostgresTestDatabase("plugin-workspace-terminal-retry"); + db = createDb(started.connectionString, { maxConnections: 1 }); + stopDb = started.cleanup; + }); + afterAll(async () => { + await db?.$client.end(); + await stopDb?.(); + }); + + async function setup() { + const f = await createPluginWorkspaceFixture(db); + const failures = { inspect: false, remove: false, unboundRemove: false, reopenFirst: false }; + const workers = { ...f.workers, call: async (...args: Parameters) => { + const params = args[2] as { agentId: string | null; operation: { kind: string; binding?: unknown } }; + if (params.operation.kind === "prepare" && params.agentId === null) { + f.calls.push(params); + throw providerRefusal(); + } + if (params.operation.kind === "remove" && params.operation.binding === null) { + f.calls.push(params); + if (failures.unboundRemove) throw providerRefusal(); + if (failures.remove) throw new Error("lost removal"); + return { kind: "removed" }; + } + if ((params.operation.kind === "inspect" && failures.inspect) || (params.operation.kind === "remove" && failures.remove)) { + f.calls.push(params); + if (failures.reopenFirst) await db.update(executionWorkspaces).set({ metadata: sql`jsonb_set(${executionWorkspaces.metadata}, + '{lifecycleGeneration}', to_jsonb(coalesce((${executionWorkspaces.metadata}->>'lifecycleGeneration')::int, 0) + 1))` }) + .where(eq(executionWorkspaces.id, (params as unknown as { executionWorkspaceId: string }).executionWorkspaceId)); + throw new Error("provider outage"); + } + return f.workers.call(...args); + } } as WorkspaceWorkers; + const read = async (id: string) => (await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, id)))[0]!; + const finishTask = async () => { + await db.update(issues).set({ status: "done", completedAt: new Date() }).where(eq(issues.id, f.issueId)); + await db.update(heartbeatRuns).set({ status: "succeeded" }).where(eq(heartbeatRuns.id, f.runId)); + }; + const abandonReservation = async () => { + f.lose(); + await expect(preparePluginWorkspace(db, workers, f.input)).rejects.toThrow("incomplete"); + const [task] = await db.select().from(issues).where(eq(issues.id, f.issueId)); + const reservation = await read(task.executionWorkspaceId!); + expect(reservation.providerRef).toBeNull(); + await db.update(issues).set({ executionWorkspaceId: null }).where(eq(issues.id, f.issueId)); + const replacement = await preparePluginWorkspace(db, workers, f.input); + expect(replacement!.id).not.toBe(reservation.id); + return reservation; + }; + const callsFor = (id: string, kind: string) => f.calls.filter(call => call.executionWorkspaceId === id && call.operation.kind === kind); + return { f, workers, failures, read, finishTask, abandonReservation, callsFor }; + } + + it("retires an unbound reservation of a finished task through an agent-free remove and archives it", async () => { + const s = await setup(); + const reservation = await s.abandonReservation(); + await s.finishTask(); + s.f.dirty(); + await executionWorkspaceService(db, { pluginWorkerManager: s.workers, workspaceReaperCooldownDays: 0 }).sweepTerminalWorkspaces(100); + expect(s.callsFor(reservation.id, "prepare")).toHaveLength(1); + expect(s.callsFor(reservation.id, "remove")).toEqual([expect.objectContaining({ agentId: null, runId: null, + operation: { kind: "remove", binding: null, expectedHeadSha: null } })]); + const retired = await s.read(reservation.id); + expect(retired).toMatchObject({ status: "archived", cleanupReason: "issue_terminal", + metadata: { workspaceProviderRemoved: true, workspaceProviderRemoval: null } }); + expect(retired.closedAt).not.toBeNull(); + expect(retired.metadata?.lifecycleGeneration).toBe(((reservation.metadata?.lifecycleGeneration as number) ?? 0) + 1); + }); + + it("replays a failed reservation removal with the same identity and backs it off", async () => { + const s = await setup(); + const reservation = await s.abandonReservation(); + await s.finishTask(); + s.f.dirty(); + s.failures.remove = true; + let clock = Date.now(); + const svc = executionWorkspaceService(db, { pluginWorkerManager: s.workers, workspaceReaperCooldownDays: 0, now: () => new Date(clock) }); + await svc.sweepTerminalWorkspaces(100); + const pending = await s.read(reservation.id); + expect(pending).toMatchObject({ status: "archived", + metadata: { workspaceProviderRemoval: { operation: { kind: "remove", binding: null } } } }); + expect(pending.metadata?.terminalCleanupRetry).toMatchObject({ failures: 1, generation: pending.metadata?.lifecycleGeneration, + retryAt: new Date(clock + 5 * MINUTE).toISOString(), failure: { kind: "transport" } }); + const retryAt = Date.parse((pending.metadata?.terminalCleanupRetry as { retryAt: string }).retryAt); + clock = Date.now() + 1; + await svc.sweepTerminalWorkspaces(100); + expect(s.callsFor(reservation.id, "remove")).toHaveLength(1); + s.failures.remove = false; + clock = retryAt; + await svc.sweepTerminalWorkspaces(100); + const removes = s.callsFor(reservation.id, "remove"); + expect(removes).toHaveLength(2); + expect(new Set(removes.map(call => call.operationId)).size).toBe(1); + expect(s.callsFor(reservation.id, "prepare")).toHaveLength(1); + const retired = await s.read(reservation.id); + expect(retired).toMatchObject({ status: "archived", metadata: { workspaceProviderRemoved: true } }); + expect(retired.metadata).not.toHaveProperty("terminalCleanupRetry"); + }); + + it("keeps the reservation a task still points at and records why its reconciliation failed", async () => { + const s = await setup(); + s.f.lose(); + await expect(preparePluginWorkspace(db, s.workers, s.f.input)).rejects.toThrow("incomplete"); + const [task] = await db.select().from(issues).where(eq(issues.id, s.f.issueId)); + await s.finishTask(); + const clock = Date.now(); + await executionWorkspaceService(db, { pluginWorkerManager: s.workers, workspaceReaperCooldownDays: 0, now: () => new Date(clock) }) + .sweepTerminalWorkspaces(100); + expect(s.callsFor(task.executionWorkspaceId!, "remove")).toHaveLength(0); + expect(s.callsFor(task.executionWorkspaceId!, "prepare")).toHaveLength(2); + const kept = await s.read(task.executionWorkspaceId!); + expect(kept.status).not.toBe("archived"); + expect(kept.providerRef).toBeNull(); + expect(kept.metadata?.terminalCleanupRetry).toMatchObject({ failures: 1, retryAt: new Date(clock + 5 * MINUTE).toISOString(), + failure: { kind: "provider", reason: "unsupported_operation", message: expect.not.stringContaining("private") } }); + }); + + it("does not retire a reservation whose task is still open", async () => { + const s = await setup(); + const reservation = await s.abandonReservation(); + await db.update(heartbeatRuns).set({ status: "succeeded" }).where(eq(heartbeatRuns.id, s.f.runId)); + await expect(inspectPluginWorkspace(db, s.workers, reservation)).rejects.toThrow(); + expect(s.callsFor(reservation.id, "remove")).toHaveLength(0); + expect(s.callsFor(reservation.id, "prepare")).toHaveLength(2); + const kept = await s.read(reservation.id); + expect(kept.status).toBe(reservation.status); + expect(kept.metadata?.workspaceProviderRemoval).toBeUndefined(); + }); + + it("backs off and keeps the reason when the provider rejects an unbound removal", async () => { + const s = await setup(); + const reservation = await s.abandonReservation(); + await s.finishTask(); + s.f.dirty(); + s.failures.unboundRemove = true; + let clock = Date.now(); + const svc = executionWorkspaceService(db, { pluginWorkerManager: s.workers, workspaceReaperCooldownDays: 0, now: () => new Date(clock) }); + await svc.sweepTerminalWorkspaces(100); + const refused = await s.read(reservation.id); + expect(refused).toMatchObject({ status: "archived", metadata: { workspaceProviderRemoval: { operation: { binding: null } } } }); + expect(refused.metadata?.workspaceProviderRemoved).not.toBe(true); + expect(refused.metadata?.terminalCleanupRetry).toMatchObject({ failures: 1, retryAt: new Date(clock + 5 * MINUTE).toISOString(), + generation: refused.metadata?.lifecycleGeneration, + failure: { kind: "provider", reason: "unsupported_operation", message: expect.not.stringContaining("private") } }); + clock = Date.now() + 1; + await svc.sweepTerminalWorkspaces(100); + expect(s.callsFor(reservation.id, "remove")).toHaveLength(1); + expect(s.callsFor(reservation.id, "prepare")).toHaveLength(1); + }); + + it("backs off repeated inspection failures, records the reason, and resets after a success", async () => { + const s = await setup(); + const workspace = (await preparePluginWorkspace(db, s.workers, s.f.input))!; + await s.finishTask(); + s.f.dirty(); + s.failures.inspect = true; + let clock = Date.now(); + const svc = executionWorkspaceService(db, { pluginWorkerManager: s.workers, workspaceReaperCooldownDays: 0, now: () => new Date(clock) }); + const retryAfter = async () => { + const retry = (await s.read(workspace.id)).metadata?.terminalCleanupRetry as { failures: number; retryAt: string } | undefined; + return retry && { failures: retry.failures, delay: Date.parse(retry.retryAt) - clock }; + }; + await svc.sweepTerminalWorkspaces(100); + expect(await retryAfter()).toEqual({ failures: 1, delay: 5 * MINUTE }); + expect((await s.read(workspace.id)).metadata?.terminalCleanupRetry).toMatchObject({ + failure: { kind: "transport", reason: null, message: "Workspace storage operation is incomplete; retry the same workspace" } }); + expect((await svc.sweepTerminalWorkspaces(100)).skippedBackoff).toBeGreaterThanOrEqual(1); + expect(s.callsFor(workspace.id, "inspect")).toHaveLength(1); + clock += 5 * MINUTE; + await svc.sweepTerminalWorkspaces(100); + expect(await retryAfter()).toEqual({ failures: 2, delay: 10 * MINUTE }); + clock += 10 * MINUTE; + await svc.sweepTerminalWorkspaces(100); + expect(await retryAfter()).toEqual({ failures: 3, delay: 20 * MINUTE }); + svc.terminalWorkspaceWorkSettled(); + await svc.sweepTerminalWorkspaces(100); + expect(s.callsFor(workspace.id, "inspect")).toHaveLength(3); + const edited = await executionWorkspaceService(db).update(workspace.id, { metadata: { note: "operator edit" } }); + expect(edited!.metadata).toMatchObject({ note: "operator edit", terminalCleanupRetry: { failures: 3 } }); + clock += 20 * MINUTE; + s.failures.inspect = false; + await svc.sweepTerminalWorkspaces(100); + expect(s.callsFor(workspace.id, "inspect")).toHaveLength(4); + expect(await retryAfter()).toBeUndefined(); + s.failures.inspect = true; + clock += MINUTE; + await svc.sweepTerminalWorkspaces(100); + expect(await retryAfter()).toEqual({ failures: 1, delay: 5 * MINUTE }); + }); + + it("drops a cleanup failure when the workspace moves to a new lifecycle before it is recorded", async () => { + const s = await setup(); + const workspace = (await preparePluginWorkspace(db, s.workers, s.f.input))!; + await s.finishTask(); + s.f.dirty(); + s.failures.inspect = true; + s.failures.reopenFirst = true; + await executionWorkspaceService(db, { pluginWorkerManager: s.workers, workspaceReaperCooldownDays: 0 }).sweepTerminalWorkspaces(100); + expect(s.callsFor(workspace.id, "inspect")).toHaveLength(1); + const moved = await s.read(workspace.id); + expect(moved.metadata?.lifecycleGeneration).toBe(((workspace.metadata?.lifecycleGeneration as number) ?? 0) + 1); + expect(moved.metadata).not.toHaveProperty("terminalCleanupRetry"); + }); + + it("backs off a pending storage removal that keeps failing", async () => { + const s = await setup(); + const workspace = (await preparePluginWorkspace(db, s.workers, s.f.input))!; + await s.finishTask(); + s.failures.remove = true; + await expect(removePluginWorkspace(db, s.workers, workspace, "a".repeat(40))).rejects.toThrow(); + await db.update(executionWorkspaces).set({ status: "cleanup_failed", closedAt: new Date() }).where(eq(executionWorkspaces.id, workspace.id)); + let clock = Date.now(); + const svc = executionWorkspaceService(db, { pluginWorkerManager: s.workers, workspaceReaperCooldownDays: 0, now: () => new Date(clock) }); + const delays: number[] = []; + for (let attempt = 0; attempt < 3; attempt++) { + await svc.sweepTerminalWorkspaces(100); + const retry = (await s.read(workspace.id)).metadata?.terminalCleanupRetry as { failures: number; retryAt: string }; + expect(retry.failures).toBe(attempt + 1); + delays.push(Date.parse(retry.retryAt) - clock); + clock = Date.parse(retry.retryAt); + } + expect(delays).toEqual([5 * MINUTE, 10 * MINUTE, 20 * MINUTE]); + expect(s.callsFor(workspace.id, "remove")).toHaveLength(4); + s.failures.remove = false; + await svc.sweepTerminalWorkspaces(100); + const removed = await s.read(workspace.id); + expect(removed).toMatchObject({ status: "archived", metadata: { workspaceProviderRemoved: true } }); + expect(removed.metadata).not.toHaveProperty("terminalCleanupRetry"); + }); +}); diff --git a/server/src/__tests__/plugin-workspaces.integration.test.ts b/server/src/__tests__/plugin-workspaces.integration.test.ts index c0724796738..cd4f49d68a1 100644 --- a/server/src/__tests__/plugin-workspaces.integration.test.ts +++ b/server/src/__tests__/plugin-workspaces.integration.test.ts @@ -439,9 +439,11 @@ const support = await getEmbeddedPostgresTestSupport(); return call(...args); }, } as unknown as WorkspaceWorkers; + let clock = Date.now(); const svc = executionWorkspaceService(db, { pluginWorkerManager: workers, workspaceReaperCooldownDays: 0, + now: () => new Date(clock), }); const sweep = () => svc.sweepTerminalWorkspaces(100); expect(await sweep()).toMatchObject({ cleanupFailed: 1 }); @@ -451,7 +453,10 @@ const support = await getEmbeddedPostgresTestSupport(); .where(eq(executionWorkspaces.id, workspace!.id)); expect(failed.status).toBe("cleanup_failed"); expect(failed.metadata?.workspaceProviderRemoval).toBeUndefined(); - expect(await svc.nextTerminalWorkspaceSweepAt()).not.toBeNull(); + expect(failed.metadata?.terminalCleanupRetry).toMatchObject({ failures: 1, failure: { kind: "transport" }, + retryAt: new Date(clock + 300_000).toISOString() }); + expect(await sweep()).toMatchObject({ skippedBackoff: 1 }); + clock += 300_000; await sweep(); const [removed] = await db .select() @@ -459,6 +464,7 @@ const support = await getEmbeddedPostgresTestSupport(); .where(eq(executionWorkspaces.id, workspace!.id)); expect(removed.metadata?.workspaceProviderRemoved).toBe(true); expect(removed.status).toBe("archived"); + expect(removed.metadata).not.toHaveProperty("terminalCleanupRetry"); }); it("prepares a durable reference without a local checkout or agent allocation", async () => { const f = await fixture(); diff --git a/server/src/services/execution-workspaces.ts b/server/src/services/execution-workspaces.ts index 21dbe1a44d3..62c85a4381d 100644 --- a/server/src/services/execution-workspaces.ts +++ b/server/src/services/execution-workspaces.ts @@ -39,7 +39,11 @@ import { authorizationService, executionWorkspaceReadSqlCondition, projectReadSq import type { SQL } from "drizzle-orm"; import { nativeSessionProviderEvidence } from "./native-runtime/native-session-resume.js"; import { hasWorkspaceRestoreFailure } from "@paperclipai/shared"; -import { notifyTerminalWorkspaceWork, TERMINAL_WORKSPACE_RETRY_MS } from "./terminal-workspace-work.js"; +import { notifyTerminalWorkspaceWork } from "./terminal-workspace-work.js"; +import { + clearTerminalWorkspaceRetry, readTerminalWorkspaceRetry, recordTerminalWorkspaceFailure, terminalWorkspaceFailure, + TERMINAL_WORKSPACE_RETRY_KEY, type TerminalWorkspaceFailure, +} from "./terminal-workspace-retry.js"; import { deleteWorkspaceCompactionTrash, findRegenerableWorkspaceDirs, @@ -296,10 +300,10 @@ function preserveWorkspaceMetadataForPatch( } } } - const freshIntentKeys = ["workspaceProviderFresh", "runnerTaskFreshWorkspace", "runnerTaskWorkspaceOwner"] as const; - if (patch.metadata !== undefined && (patch.metadata !== null || freshIntentKeys.some(key => current.metadata?.[key] !== undefined))) { + const coreOwnedKeys = ["workspaceProviderFresh", "runnerTaskFreshWorkspace", "runnerTaskWorkspaceOwner", TERMINAL_WORKSPACE_RETRY_KEY] as const; + if (patch.metadata !== undefined && (patch.metadata !== null || coreOwnedKeys.some(key => current.metadata?.[key] !== undefined))) { nextPatch.metadata = { ...(nextPatch.metadata ?? patch.metadata) }; - for (const key of freshIntentKeys) { + for (const key of coreOwnedKeys) { if (current.metadata?.[key] !== undefined) nextPatch.metadata[key] = current.metadata[key]; else delete nextPatch.metadata[key]; } @@ -1516,7 +1520,10 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic if (!row) throw notFound("Workspace is unavailable"); let state; try { state = await inspectPluginWorkspace(db, opts.pluginWorkerManager, row, recoveryPullRequest); } - catch { return { statusInspectionSucceeded: false, warnings: ["Workspace storage inspection is unavailable; data was retained."], git: null }; } + catch (error) { + const failure = terminalWorkspaceFailure(error); + return { statusInspectionSucceeded: false, warnings: [`Workspace storage inspection is unavailable; data was retained. ${failure.message}`], git: null, failure }; + } return { statusInspectionSucceeded: true, warnings: [], git: { repoRoot: `/workspaces/${workspace.id}`, workspacePath: `/workspaces/${workspace.id}`, branchName: state.branchName, baseRef: workspace.baseRef, @@ -3614,7 +3621,21 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic compacted: 0, compactionFailed: 0, }; - const readinessByWorkingTree = new Map>(); + const readinessByWorkingTree = new Map>(); + const backOffTerminalWorkspace = async (snapshot: ExecutionWorkspaceRow, failure: TerminalWorkspaceFailure) => { + const [target] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, snapshot.id)); + if (!target) return; + const generation = readExecutionWorkspaceLifecycleGeneration(target.metadata); + const removalFence = (target.metadata?.workspaceProviderRemoval as { generation?: unknown } | null | undefined)?.generation === generation; + if (generation !== readExecutionWorkspaceLifecycleGeneration(snapshot.metadata) && !removalFence) return; + const retry = await recordTerminalWorkspaceFailure(db, target, generation, failure, now()); + terminalWorkspaceDeadlines.set(target.id, Date.parse(retry.retryAt)); + logger.warn({ event: "execution_workspace.terminal_cleanup_retry", executionWorkspaceId: target.id, sourceIssueId: target.sourceIssueId, + failures: retry.failures, retryAt: retry.retryAt, failure: retry.failure }, "terminal workspace cleanup failed; a later sweep retries"); + }; + const settleTerminalWorkspaceRetry = async (target: ExecutionWorkspaceRow) => { + if (target.metadata?.[TERMINAL_WORKSPACE_RETRY_KEY] !== undefined) await clearTerminalWorkspaceRetry(db, target.id); + }; let taskDirectoryBudget = opts.taskDirectoryRetention?.perSweep ?? 8; const expireTaskDirectory = async (workspace: ExecutionWorkspaceRow, notBefore: Date) => { const settings = opts.taskDirectoryRetention!; @@ -3845,6 +3866,12 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic result.skippedCooldown += 1; continue; } + const terminalRetry = readTerminalWorkspaceRetry(workspace.metadata, readExecutionWorkspaceLifecycleGeneration(workspace.metadata)); + if (terminalRetry && Date.parse(terminalRetry.retryAt) > now().getTime()) { + terminalWorkspaceDeadlines.set(workspace.id, Date.parse(terminalRetry.retryAt)); + result.skippedBackoff += 1; + continue; + } if (await workspaceHasActiveRun(workspace)) { if (reopenPending) result.skippedReopened += 1; else result.skippedActiveRun += 1; @@ -3856,10 +3883,10 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic const cleanup = await cleanupTerminalWorkspace(workspace, null, generation); if (cleanup.skippedReopened) result.skippedReopened += 1; else if (!cleanup.cleaned) result.cleanupFailed += 1; - else { result.archived += 1; terminalWorkspaceDeadlines.delete(workspace.id); } + else { result.archived += 1; terminalWorkspaceDeadlines.delete(workspace.id); await settleTerminalWorkspaceRetry(workspace); } } catch (error) { result.cleanupFailed += 1; - terminalWorkspaceDeadlines.set(workspace.id, now().getTime() + TERMINAL_WORKSPACE_RETRY_MS); + await backOffTerminalWorkspace(workspace, terminalWorkspaceFailure(error)); await markTerminalCleanupFailedFenced({ workspaceId: workspace.id, capturedGeneration: generation, cleanupReason: `${ISSUE_TERMINAL_WORKSPACE_CLEANUP_REASON} | ${error instanceof Error ? error.message : String(error)}` }); } @@ -3966,12 +3993,20 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic inspection = inspectWorkspace(toExecutionWorkspace(workspace), recoveryPullRequest); readinessByWorkingTree.set(readinessKey, inspection); } - const { git, statusInspectionSucceeded } = await inspection; + const inspected = await inspection; + const { git, statusInspectionSucceeded } = inspected; if (!statusInspectionSucceeded) { if (remoteWorkspace) terminalWorkspaceDeadlines.delete(workspace.id); + if ("failure" in inspected) await backOffTerminalWorkspace(workspace, inspected.failure); result.skippedUndelivered += 1; continue; } + if (workspace.metadata?.[TERMINAL_WORKSPACE_RETRY_KEY] !== undefined) { + await settleTerminalWorkspaceRetry(workspace); + const metadata = { ...workspace.metadata }; + delete metadata[TERMINAL_WORKSPACE_RETRY_KEY]; + workspace = { ...workspace, metadata }; + } if (deliveryCandidate && (git?.headSha !== deliveryCandidate.headSha || git.branchName !== workspace.branchName || (await workspaceDeliveryCandidate(workspace))?.operationId !== deliveryCandidate.operationId)) { result.skippedUndelivered += 1; @@ -4089,10 +4124,10 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic const cleanup = await cleanupTerminalWorkspace(workspace, assessment.workspaceHeadSha, generation, recoveryPullRequest, verifyDeliveryBeforeRemove); if (cleanup.skippedReopened) result.skippedReopened += 1; else if (!cleanup.cleaned) result.cleanupFailed += 1; - else { result.archived += 1; terminalWorkspaceDeadlines.delete(workspace.id); } + else { result.archived += 1; terminalWorkspaceDeadlines.delete(workspace.id); await settleTerminalWorkspaceRetry(workspace); } } catch (error) { result.cleanupFailed += 1; - terminalWorkspaceDeadlines.set(workspace.id, now().getTime() + TERMINAL_WORKSPACE_RETRY_MS); + await backOffTerminalWorkspace(workspace, terminalWorkspaceFailure(error)); await markTerminalCleanupFailedFenced({ workspaceId: workspace.id, capturedGeneration: generation, cleanupReason: `${ISSUE_TERMINAL_WORKSPACE_CLEANUP_REASON} | ${error instanceof Error ? error.message : String(error)}` }); logger.warn({ err: error, executionWorkspaceId: workspace.id }, "workspace storage cleanup remains pending"); @@ -4225,10 +4260,10 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic const cleanup = await cleanupTerminalWorkspace(archived, assessment.workspaceHeadSha, capturedGeneration, recoveryPullRequest, verifyDeliveryBeforeRemove); if (cleanup.skippedReopened) result.skippedReopened += 1; else if (!cleanup.cleaned) result.cleanupFailed += 1; - else result.archived += 1; + else { result.archived += 1; await settleTerminalWorkspaceRetry(archived); } } catch (error) { result.cleanupFailed += 1; - if (archived.providerType === "plugin" || remoteWorkspace) terminalWorkspaceDeadlines.set(archived.id, now().getTime() + TERMINAL_WORKSPACE_RETRY_MS); + if (archived.providerType === "plugin" || remoteWorkspace) await backOffTerminalWorkspace(archived, terminalWorkspaceFailure(error)); const failure = error instanceof Error ? error.message : String(error); // Mark cleanup_failed only while the row is still closed at the // generation this sweep captured. A reopen that raced after the failure diff --git a/server/src/services/plugin-workspace-reservations.ts b/server/src/services/plugin-workspace-reservations.ts new file mode 100644 index 00000000000..d09c31ac05b --- /dev/null +++ b/server/src/services/plugin-workspace-reservations.ts @@ -0,0 +1,46 @@ +import { createHash } from "node:crypto"; +import { and, eq, isNull, sql } from "drizzle-orm"; +import { executionWorkspaces, heartbeatRuns, issues, type Db } from "@paperclipai/db"; +import { workspaceProviderOperationSchema } from "@paperclipai/plugin-sdk"; +import { logger } from "../middleware/logger.js"; +import { notifyTerminalWorkspaceWork } from "./terminal-workspace-work.js"; + +type Workspace = typeof executionWorkspaces.$inferSelect; + +export async function retireAbandonedPluginWorkspaceReservation(db: Db, workspace: Workspace): Promise { + if (!workspace.sourceIssueId) return null; + const previous = typeof workspace.metadata?.lifecycleGeneration === "number" ? workspace.metadata.lifecycleGeneration : 0; + const generation = previous + 1; + const operation = workspaceProviderOperationSchema.parse({ kind: "remove", binding: null, expectedHeadSha: null }); + const operationId = createHash("sha256").update(JSON.stringify([workspace.id, generation, operation])).digest("hex"); + const abandoned = and( + eq(executionWorkspaces.id, workspace.id), + eq(executionWorkspaces.companyId, workspace.companyId), + isNull(executionWorkspaces.providerRef), + sql`coalesce((${executionWorkspaces.metadata}->>'lifecycleGeneration')::int, 0) = ${previous}`, + sql`EXISTS (SELECT 1 FROM ${issues} WHERE ${issues.companyId} = ${workspace.companyId} + AND ${issues.id} = ${workspace.sourceIssueId} AND ${issues.status} IN ('done', 'cancelled'))`, + sql`NOT EXISTS (SELECT 1 FROM ${issues} WHERE ${issues.companyId} = ${workspace.companyId} + AND ${issues.executionWorkspaceId} = ${workspace.id})`, + sql`NOT EXISTS (SELECT 1 FROM ${heartbeatRuns} WHERE ${heartbeatRuns.companyId} = ${workspace.companyId} + AND ${heartbeatRuns.issueId} = ${workspace.sourceIssueId} AND ${heartbeatRuns.status} IN ('queued', 'running'))`, + ); + const [candidate] = await db.select({ id: executionWorkspaces.id }).from(executionWorkspaces).where(abandoned); + if (!candidate) return null; + const closedAt = new Date(); + return db.transaction(async (tx) => { + await notifyTerminalWorkspaceWork(tx, `workspace:${workspace.id}`); + const [retired] = await tx.update(executionWorkspaces).set({ + status: "archived", + closedAt: workspace.closedAt ?? closedAt, + cleanupEligibleAt: workspace.cleanupEligibleAt ?? closedAt, + cleanupReason: "issue_terminal", + updatedAt: closedAt, + metadata: { ...workspace.metadata, lifecycleGeneration: generation, workspaceProviderRemoval: { generation, operation, operationId } }, + }).where(abandoned).returning(); + if (!retired) return null; + logger.info({ event: "execution_workspace.reservation_retired", executionWorkspaceId: workspace.id, sourceIssueId: workspace.sourceIssueId }, + "Unused workspace storage reservation is being removed"); + return retired; + }); +} diff --git a/server/src/services/plugin-workspaces.ts b/server/src/services/plugin-workspaces.ts index 34674cbc0b9..1a9f8ff7e7e 100644 --- a/server/src/services/plugin-workspaces.ts +++ b/server/src/services/plugin-workspaces.ts @@ -1,6 +1,7 @@ import { notifyStatusCardIssueChange } from "./status-card-notifications.js"; import { effectiveProjectWorkspacePolicy, isWorkspaceProvider } from "./project-workspace-provider.js"; import { releaseProviderRecoveryTransfers, stageProviderFreshWorkspace } from "./provider-fresh-workspace.js"; +import { retireAbandonedPluginWorkspaceReservation } from "./plugin-workspace-reservations.js"; import { randomUUID, createHash } from "node:crypto"; import { and, eq, sql } from "drizzle-orm"; import { @@ -557,8 +558,11 @@ async function reconcilePluginWorkspaceReservation( ), ); if (!workspace) throw conflict("Workspace is unavailable"); - if (readWorkspaceProviderBinding(workspace.metadata)) return workspace; + if (readWorkspaceProviderBinding(workspace.metadata) || workspace.metadata?.workspaceProviderRemoval || + workspace.metadata?.workspaceProviderRemoved === true) return workspace; const owner = workspace.metadata?.workspaceProviderOwner; + const retired = typeof owner === "string" ? await retireAbandonedPluginWorkspaceReservation(db, workspace) : null; + if (retired) return retired; if (typeof owner !== "string" || !workspace.metadata?.workspaceProviderSource) throw conflict("Workspace storage reservation is unavailable"); const operation = workspaceProviderOperationSchema.parse({ @@ -765,8 +769,10 @@ async function finishPendingRemoval( const operationId = createHash("sha256").update(JSON.stringify([workspace.id, pending.generation, parsed.data])).digest("hex"); if (pending.operationId !== undefined && pending.operationId !== operationId) throw conflict("Workspace removal identity does not match its recorded payload"); + const owner = workspace.metadata?.workspaceProviderOwner; const result = await callWorkspaceProvider(db, workers, { workspace, + pluginId: typeof owner === "string" ? owner : undefined, operation: parsed.data, removalOperationId: operationId, }); diff --git a/server/src/services/terminal-workspace-retry.ts b/server/src/services/terminal-workspace-retry.ts new file mode 100644 index 00000000000..2db4a263c02 --- /dev/null +++ b/server/src/services/terminal-workspace-retry.ts @@ -0,0 +1,50 @@ +import { and, eq, sql } from "drizzle-orm"; +import { executionWorkspaces, type Db } from "@paperclipai/db"; +import { HttpError } from "../errors.js"; +import { findWorkspaceProviderFailure } from "./workspace-provider-failure.js"; +import { TERMINAL_WORKSPACE_RETRY_MS } from "./terminal-workspace-work.js"; + +export const TERMINAL_WORKSPACE_RETRY_KEY = "terminalCleanupRetry"; +export const TERMINAL_WORKSPACE_RETRY_MAX_MS = 4 * 60 * 60 * 1000; + +export type TerminalWorkspaceFailure = { kind: string | null; reason: string | null; message: string }; +export type TerminalWorkspaceRetry = { + generation: number; + failures: number; + failedAt: string; + retryAt: string; + failure: TerminalWorkspaceFailure; +}; + +export function terminalWorkspaceFailure(error: unknown): TerminalWorkspaceFailure { + const provider = findWorkspaceProviderFailure(error); + const message = error instanceof HttpError ? error.message : "Workspace storage operation failed"; + return { kind: provider?.kind ?? null, reason: provider?.reason ?? null, message: message.slice(0, 500) }; +} + +export function terminalWorkspaceRetryDelayMs(failures: number) { + return Math.min(TERMINAL_WORKSPACE_RETRY_MS * 2 ** Math.min(Math.max(failures - 1, 0), 16), TERMINAL_WORKSPACE_RETRY_MAX_MS); +} + +export function readTerminalWorkspaceRetry(metadata: Record | null | undefined, generation: number) { + const retry = metadata?.[TERMINAL_WORKSPACE_RETRY_KEY] as Partial | undefined; + return retry && typeof retry === "object" && retry.generation === generation && Number.isInteger(retry.failures) + && typeof retry.failedAt === "string" && typeof retry.retryAt === "string" ? retry as TerminalWorkspaceRetry : null; +} + +export async function recordTerminalWorkspaceFailure(db: Db, workspace: { id: string; metadata: Record | null }, + generation: number, failure: TerminalWorkspaceFailure, now: Date): Promise { + const failures = (readTerminalWorkspaceRetry(workspace.metadata, generation)?.failures ?? 0) + 1; + const retry = { generation, failures, failedAt: now.toISOString(), + retryAt: new Date(now.getTime() + terminalWorkspaceRetryDelayMs(failures)).toISOString(), failure }; + await db.update(executionWorkspaces) + .set({ metadata: sql`coalesce(${executionWorkspaces.metadata}, '{}'::jsonb) || jsonb_build_object(${TERMINAL_WORKSPACE_RETRY_KEY}::text, ${JSON.stringify(retry)}::jsonb)` }) + .where(eq(executionWorkspaces.id, workspace.id)); + return retry; +} + +export async function clearTerminalWorkspaceRetry(db: Db, workspaceId: string) { + await db.update(executionWorkspaces) + .set({ metadata: sql`${executionWorkspaces.metadata} - ${TERMINAL_WORKSPACE_RETRY_KEY}::text` }) + .where(and(eq(executionWorkspaces.id, workspaceId), sql`${executionWorkspaces.metadata} -> ${TERMINAL_WORKSPACE_RETRY_KEY}::text IS NOT NULL`)); +}