From 5dfbb04b856543386900d08329c17e06d73a43ca Mon Sep 17 00:00:00 2001 From: Nicky Leach Date: Wed, 11 Nov 2026 10:33:02 -0800 Subject: [PATCH] Poll asynchronous workspace provider preparation (#20865) ## Thinking Path > - Paperclip starts agent runs through a workspace provider when a project uses remote storage. > - The provider must prepare the workspace before the agent starts. > - One long worker call ties the checkout to its transport. > - Some checkouts take longer than the transport allows. > - This change lets an opted-in provider accept work and report its deadline. > - Core waits for completion through short status calls and stops at the provider deadline. ## Linked Issues or Issue Description **What happened?** A workspace provider can lose its reply when preparation takes longer than a single transport request. This stops run bootstrap before agent execution. **Expected behavior** Core should wait for a provider-confirmed workspace through short requests. The provider's deadline must bound the wait. **Steps to reproduce** Use a workspace provider that takes longer to prepare a checkout than its transport can remain open. Start a run that requires a new checkout. ## What Changed - Add an optional workspace preparation method to the plugin SDK. - Validate pending receipts and keep the original binding checks on completed replies. - Poll only during active preparation. Retry a lost status reply within the advertised deadline. - Preserve the original operation for providers that do not offer asynchronous preparation. - Reject deadline offers that exceed the existing lifecycle lock budget. Honor accepted offers without shortening them. - Checkout verification uses the existing synchronous RPC with a 10-second deadline and a 12-second transport timeout. [Core verification PR #20876](https://github.com/paperclipai/paperclip/pull/20876) ## Verification - All 63 workspace SDK and Core integration tests passed with temporary local PostgreSQL. The tests cover lost status replies, deadline expiry, oversized offers, and synchronous verification with an async-capable worker. - All 58 runner SDK input, evidence, and public type tests passed with the regenerated source pins. The local server typecheck passed. ## Risks - This change activates only when the provider advertises the method. Existing providers keep their current behavior. - Polling uses one-second intervals during active preparation. It stops at completion, failure, or the provider deadline. - No database migration is required. ## Model Used - OpenAI GPT-6 Codex with local code execution and tests. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip --- doc/plugins/PLUGIN_RUNNER_EXECUTION.md | 30 ++++++- packages/plugins/sdk/src/define-plugin.ts | 3 +- packages/plugins/sdk/src/protocol.ts | 4 +- packages/plugins/sdk/src/worker-rpc-host.ts | 10 ++- .../plugins/sdk/src/workspace-provider.ts | 28 +++++++ .../sdk/tests/workspace-provider.test.ts | 13 +++ .../plugin-workspaces.integration.test.ts | 80 +++++++++++++++++-- server/src/services/plugin-worker-manager.ts | 2 +- .../services/plugin-workspace-preparation.ts | 63 +++++++++++++++ server/src/services/plugin-workspaces.ts | 19 ++--- .../runner-e2e/source-sdk-public-inputs.json | 8 +- 11 files changed, 233 insertions(+), 27 deletions(-) create mode 100644 server/src/services/plugin-workspace-preparation.ts diff --git a/doc/plugins/PLUGIN_RUNNER_EXECUTION.md b/doc/plugins/PLUGIN_RUNNER_EXECUTION.md index 3199ca755e3..afc0704453b 100644 --- a/doc/plugins/PLUGIN_RUNNER_EXECUTION.md +++ b/doc/plugins/PLUGIN_RUNNER_EXECUTION.md @@ -175,6 +175,8 @@ It does not qualify a deployed plugin, live account, or production WSS proxy. A storage provider can declare `workspaceProvider.workspaceVerificationVersion: 1`. Before creating a fresh workspace for an admitted Retry, Core may send the original task-owned workspace's `prepare` request with `verifyExisting: true`. +Core sends verification through `workspaceOperation` with a 10-second +deadline and a 12-second RPC timeout. The provider must verify a completed checkout without creating or fetching one and return `prepared` with `created: false`. Core checks prior writer settlement and continues normal preparation on the retained workspace after @@ -261,6 +263,32 @@ Provider upgrades, activation, company re-enablement, and finalization settlemen wake cleanup transactionally. No table, independent timer, or background scan of all remote PR refs is required. +### Asynchronous workspace preparation + +A provider can implement `onWorkspacePrepare` alongside `onWorkspaceOperation`. +Core calls the optional method with `phase: "submit"` and a short submission +deadline. A provider that cannot prepare asynchronously returns +`{ kind: "unsupported" }` before starting work, so Core can use the original +operation. Otherwise the provider returns the completed result or a +`{ kind: "pending", deadline }` receipt. The provider sets and enforces that +operation deadline. Core holds the workspace lifecycle lock during preparation +and rejects offers beyond the previous synchronous RPC budget before polling. +It honors accepted deadlines as supplied. Replayed submission must not duplicate +an active prepare. + +Core sends `phase: "poll"` with the same operation identity and the deadline +from the receipt. The provider returns the same receipt while work is pending +and the usual scoped `prepared` result after success. Core rejects late results +and stops polling when the deadline expires. Each submission and poll has a +short transport bound independent of the operation deadline. A provider can +keep reporting pending while it restores recovery snapshots, under the same +deadline. + +Each invocation has its own scoped host access. Complete scoped host calls +inside the submit or poll that owns them. Detached work can use a remote +operation already authorized by that invocation. It cannot reuse the +invocation to call host services after the reply. + ### Workspace operation failures Core separates a failed `workspaceOperation` call by what it knows about the @@ -274,7 +302,7 @@ outcome: | The provider threw any other error | `Workspace storage provider reported a failure; retry the same workspace` | Unchanged setup failure. | To report a reason, throw `PluginWorkspaceOperationError` from -`onWorkspaceOperation`, or any error with an own `workspaceFailureReason` +`onWorkspaceOperation` or `onWorkspacePrepare`, or any error with an own `workspaceFailureReason` property. The reason must come from `WORKSPACE_OPERATION_FAILURE_REASONS`: | Reason | Retrying the same workspace can help | diff --git a/packages/plugins/sdk/src/define-plugin.ts b/packages/plugins/sdk/src/define-plugin.ts index eb6530c0067..a99e07ce77e 100644 --- a/packages/plugins/sdk/src/define-plugin.ts +++ b/packages/plugins/sdk/src/define-plugin.ts @@ -1,4 +1,4 @@ -import type { PluginWorkspaceParams, PluginWorkspaceResult } from "./workspace-provider.js"; +import type { PluginWorkspaceParams, PluginWorkspacePrepareParams, PluginWorkspacePrepareResult, PluginWorkspaceResult } from "./workspace-provider.js"; import type { PluginEnvironmentTaskParams, PluginEnvironmentTaskResult } from "./environment-tasks.js"; import type { AgentLifecycleRequest, AgentLifecycleResult } from "@paperclipai/shared"; import type { AiConnectionRouterRequest, AiConnectionRouterResult } from "@paperclipai/shared"; @@ -408,6 +408,7 @@ export interface PluginDefinition { ): Promise; /** Required with manifest.workspaceProvider, together with project resource lifecycle. */ + onWorkspacePrepare?(params: PluginWorkspacePrepareParams): Promise; onWorkspaceOperation?(params: PluginWorkspaceParams): Promise; /** Provide Runner execution: prepare resources, start Runner, connect via PRP, and manage task lifecycle. */ diff --git a/packages/plugins/sdk/src/protocol.ts b/packages/plugins/sdk/src/protocol.ts index 95e8af999da..97ee5337227 100644 --- a/packages/plugins/sdk/src/protocol.ts +++ b/packages/plugins/sdk/src/protocol.ts @@ -1,4 +1,4 @@ -import type { PluginWorkspaceParams, PluginWorkspaceResult } from "./workspace-provider.js"; +import type { PluginWorkspaceParams, PluginWorkspacePrepareParams, PluginWorkspacePrepareResult, PluginWorkspaceResult } from "./workspace-provider.js"; import type { PluginEnvironmentTaskParams, PluginEnvironmentTaskResult } from "./environment-tasks.js"; import type { PluginAgentRuntimeState, PluginAgentRuntimeIdentity } from "./types.js"; import type { AgentLifecycleRequest, AgentLifecycleResult } from "@paperclipai/shared"; @@ -1462,6 +1462,7 @@ export interface HostToWorkerMethods { params: PluginEnvironmentValidateConfigParams, result: PluginEnvironmentValidationResult, ]; + workspacePrepare: [params: PluginWorkspacePrepareParams, result: PluginWorkspacePrepareResult]; workspaceOperation: [params: PluginWorkspaceParams, result: PluginWorkspaceResult]; environmentTask: [params: PluginEnvironmentTaskParams, result: PluginEnvironmentTaskResult]; environmentProbe: [ @@ -1586,6 +1587,7 @@ export const HOST_TO_WORKER_OPTIONAL_METHODS: readonly HostToWorkerMethodName[] "handleApiRequest", "agentLifecycle", "resourceLifecycle", + "workspacePrepare", "workspaceOperation", "getData", "performAction", diff --git a/packages/plugins/sdk/src/worker-rpc-host.ts b/packages/plugins/sdk/src/worker-rpc-host.ts index 72f844508f6..3de7748e983 100644 --- a/packages/plugins/sdk/src/worker-rpc-host.ts +++ b/packages/plugins/sdk/src/worker-rpc-host.ts @@ -1,6 +1,6 @@ import { freshWorkspaceSchema } from "./workspace-provider.js"; import { workspaceProviderBindingSchema } from "@paperclipai/shared"; -import { pluginWorkspaceParamsSchema, parseWorkspaceProviderResult } from "./workspace-provider.js"; +import { pluginWorkspaceParamsSchema, pluginWorkspacePrepareParamsSchema, parseWorkspacePrepareResult, parseWorkspaceProviderResult } from "./workspace-provider.js"; import { environmentTaskOperationSchema, parseEnvironmentTaskResult, type PluginEnvironmentTaskParams } from "./environment-tasks.js"; import type { AgentLifecycleRequest, AgentLifecycleResult } from "@paperclipai/shared"; import type { AiConnectionRouterRequest, AiConnectionRouterResult } from "@paperclipai/shared"; @@ -1667,7 +1667,7 @@ export function startWorkerRpcHost(options: WorkerRpcHostOptions): WorkerRpcHost : method === "environmentStopLease" ? environmentStopErrorData(err) : method === "environmentResumeLease" ? environmentResumeErrorData(err) : method === "environmentTask" ? taskError ?? environmentTaskErrorCodeData(err) - : method === "workspaceOperation" ? workspaceOperationErrorData(err) : undefined)); + : method === "workspaceOperation" || method === "workspacePrepare" ? workspaceOperationErrorData(err) : undefined)); } finally { done?.(); } @@ -1725,6 +1725,11 @@ export function startWorkerRpcHost(options: WorkerRpcHostOptions): WorkerRpcHost return handleExecuteTool(params as ExecuteToolParams); case "detectExternalObjects": return handleDetectExternalObjects(params as DetectExternalObjectsParams); + case "workspacePrepare": { + if (!plugin.definition.onWorkspacePrepare) throw methodNotImplemented("workspacePrepare"); + const input = pluginWorkspacePrepareParamsSchema.parse(params); + return parseWorkspacePrepareResult(input, await plugin.definition.onWorkspacePrepare(input)); + } case "workspaceOperation": { if (!plugin.definition.onWorkspaceOperation) throw methodNotImplemented("workspaceOperation"); const input = pluginWorkspaceParamsSchema.parse(params); @@ -1875,6 +1880,7 @@ export function startWorkerRpcHost(options: WorkerRpcHostOptions): WorkerRpcHost if (plugin.definition.onIdleDrain) supportedMethods.push("prepareIdleSleep", "releaseIdleSleep"); if (plugin.definition.onApiRequest) supportedMethods.push("handleApiRequest"); if (plugin.definition.onDetectExternalObjects) supportedMethods.push("detectExternalObjects"); + if (plugin.definition.onWorkspacePrepare) supportedMethods.push("workspacePrepare"); if (plugin.definition.onWorkspaceOperation) supportedMethods.push("workspaceOperation"); if (plugin.definition.onResourceLifecycle) supportedMethods.push("resourceLifecycle"); if (plugin.definition.onAgentLifecycle) supportedMethods.push("agentLifecycle"); diff --git a/packages/plugins/sdk/src/workspace-provider.ts b/packages/plugins/sdk/src/workspace-provider.ts index 5db136efa4b..dea90b92e5b 100644 --- a/packages/plugins/sdk/src/workspace-provider.ts +++ b/packages/plugins/sdk/src/workspace-provider.ts @@ -125,6 +125,34 @@ export const pluginWorkspaceParamsSchema = z /** Finalization receives the originating run/agent context; maintenance remains agent-independent. */ export type PluginWorkspaceParams = z.infer; +export const pluginWorkspacePrepareParamsSchema = pluginWorkspaceParamsSchema.extend({ + phase: z.enum(["submit", "poll"]), +}).refine((input) => input.operation.kind === "prepare"); +export type PluginWorkspacePrepareParams = z.infer; + +export type PluginWorkspacePrepareResult = + | { kind: "unsupported" } + | { kind: "pending"; deadline: string } + | Extract; + +export function parseWorkspacePrepareResult( + params: PluginWorkspacePrepareParams, + value: unknown, +): PluginWorkspacePrepareResult { + const result = z.discriminatedUnion("kind", [ + z.object({ kind: z.literal("unsupported") }).strict(), + z.object({ kind: z.literal("pending"), deadline: z.string().datetime() }).strict(), + ]).safeParse(value); + if (result.success) { + if (result.data.kind === "unsupported" && params.phase !== "submit") + throw new Error("Workspace provider withdrew asynchronous preparation"); + return result.data; + } + const prepared = parseWorkspaceProviderResult(params, value); + if (prepared.kind !== "prepared") throw new Error("Workspace provider returned the wrong preparation result"); + return prepared; +} + export function parseWorkspaceProviderResult( params: PluginWorkspaceParams, value: unknown, diff --git a/packages/plugins/sdk/tests/workspace-provider.test.ts b/packages/plugins/sdk/tests/workspace-provider.test.ts index 9385be933cd..fe8a9501193 100644 --- a/packages/plugins/sdk/tests/workspace-provider.test.ts +++ b/packages/plugins/sdk/tests/workspace-provider.test.ts @@ -2,6 +2,8 @@ import { describe, expect, it } from "vitest"; import { workspaceProviderOperationSchema, parseWorkspaceProviderResult, + parseWorkspacePrepareResult, + pluginWorkspacePrepareParamsSchema, type PluginWorkspaceParams, } from "../src/workspace-provider.js"; import { createWorkspaceProviderHarness } from "../src/testing.js"; @@ -50,6 +52,17 @@ const result = { branchCreatedByRuntime: true, }; describe("workspace provider SDK", () => { + it("validates asynchronous preparation receipts and preserves result scope", () => { + const submit = pluginWorkspacePrepareParamsSchema.parse({ ...params, phase: "submit" }); + const poll = { ...submit, phase: "poll" as const }; + const pending = { kind: "pending", deadline: "2099-01-01T00:01:00Z" }; + expect(parseWorkspacePrepareResult(submit, pending)).toEqual(pending); + expect(parseWorkspacePrepareResult(poll, result)).toEqual(result); + expect(() => parseWorkspacePrepareResult(poll, { kind: "unsupported" })).toThrow(); + expect(() => parseWorkspacePrepareResult(poll, { ...pending, deadline: "whenever" })).toThrow(); + expect(() => parseWorkspacePrepareResult(poll, { ...result, binding: { ...result.binding, projectId: id(20) } })).toThrow(); + }); + it("bounds recovery hints to an exact PR number and commit", () => { for (const kind of ["inspect", "remove"] as const) { const operation = { kind, binding: result.binding, ...(kind === "remove" ? { expectedHeadSha: "a".repeat(40) } : {}) }; diff --git a/server/src/__tests__/plugin-workspaces.integration.test.ts b/server/src/__tests__/plugin-workspaces.integration.test.ts index ae1a45b3fcd..dd05afef90a 100644 --- a/server/src/__tests__/plugin-workspaces.integration.test.ts +++ b/server/src/__tests__/plugin-workspaces.integration.test.ts @@ -1,6 +1,6 @@ import { createPluginWorkspaceFixture } from "./helpers/plugin-workspaces.js"; import { randomUUID } from "node:crypto"; -import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; import { and, eq, sql } from "drizzle-orm"; import { createDb, @@ -76,11 +76,19 @@ const support = await getEmbeddedPostgresTestSupport(); issueId: f.issueId, status: "running", contextSnapshot: { taskRetryRequestId: retryIntentId }, }).returning(); const fresh = { mode: "fresh_isolated_workspace", retryIntentId, sourceRunId: f.runId, retainedSourceIds: [] } as const; - const workers: WorkspaceWorkers = { ...f.workers, call: async (...args) => { + const call = vi.fn(async (...args) => { const result = await f.workers.call(...args) as { kind: string }; if (state === "unconfirmed") throw new Error("checkout incomplete"); return { ...result, created: false }; - } }; + }); + const workers: WorkspaceWorkers = { + getWorker: (pluginId) => { + const worker = f.workers.getWorker(pluginId); + return worker && { ...worker, supportedMethods: ["workspaceOperation", "workspacePrepare"] }; + }, + call, + }; + const verificationStarted = Date.now(); const result = await verifyRetryWorkspace(db, workers, retry, f.issueId, null, { ...fresh, retainedSourceIds: [] }); const verified = state === "ready" || state === "lost_reply"; @@ -88,11 +96,73 @@ const support = await getEmbeddedPostgresTestSupport(); ? { mode: "reuse_issue_workspace", retryIntentId, sourceRunId: f.runId, workspaceId, cwd: null } : fresh); expect(f.calls.slice(1)).toHaveLength(state === "live_writer" || state === "unsupported" ? 0 : 1); - if (f.calls.length > 1) expect(f.calls[1]).toMatchObject({ executionWorkspaceId: workspaceId, - runId: retry.id, operation: { kind: "prepare", verifyExisting: true } }); + if (f.calls.length > 1) { + expect(call).toHaveBeenCalledExactlyOnceWith(f.pluginId, "workspaceOperation", expect.objectContaining({ + executionWorkspaceId: workspaceId, runId: retry.id, operation: expect.objectContaining({ verifyExisting: true }), + }), 12_000); + const deadline = Date.parse(f.calls[1].deadline); + expect(deadline).toBeGreaterThanOrEqual(verificationStarted + 10_000); + expect(deadline).toBeLessThanOrEqual(Date.now() + 10_000); + } const [retained] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, workspaceId)); expect(retained.providerRef).toBe(`checkout:${workspaceId}`); }); + + it("polls bounded requests using the provider's advertised preparation deadline", async () => { + const f = await fixture(); + const deadline = new Date(Date.now() + 30_000).toISOString(); + const phases: string[] = []; + const workers = { + getWorker: () => ({ supportedMethods: ["workspaceOperation", "workspacePrepare"] }), + call: async (pluginId: string, method: string, params: any, timeout: number) => { + expect(method).toBe("workspacePrepare"); + expect(timeout).toBeLessThanOrEqual(15_000); + phases.push(params.phase); + if (params.phase === "submit") return { kind: "pending", deadline }; + expect(params.deadline).toBe(deadline); + if (phases.length === 2) throw new Error("lost status reply"); + const { phase: _phase, ...original } = params; + return f.workers.call(pluginId, "workspaceOperation", original); + }, + } as unknown as WorkspaceWorkers; + expect(await preparePluginWorkspace(db, workers, f.input)).toBeTruthy(); + expect(phases).toEqual(["submit", "poll", "poll"]); + expect(f.calls).toHaveLength(1); + }); + + it("stops preparation at the provider deadline before issuing a poll", async () => { + const f = await fixture(); + const workers = { + getWorker: () => ({ supportedMethods: ["workspaceOperation", "workspacePrepare"] }), + call: async (_pluginId: string, method: string, params: { phase: string }) => { + expect(method).toBe("workspacePrepare"); + expect(params.phase).toBe("submit"); + return { kind: "pending", deadline: new Date(Date.now() - 1).toISOString() }; + }, + } as unknown as WorkspaceWorkers; + await expect(preparePluginWorkspace(db, workers, f.input)).rejects.toMatchObject({ + kind: "provider", reason: "deadline_exceeded", retryable: true, + }); + }); + + it("rejects an oversized provider deadline before it can retain the workspace lifecycle lock", async () => { + const f = await fixture(); + const workers = { + getWorker: () => ({ supportedMethods: ["workspaceOperation", "workspacePrepare"] }), + call: async (_pluginId: string, method: string, params: { phase: string }) => { + expect(method).toBe("workspacePrepare"); + expect(params.phase).toBe("submit"); + return { kind: "pending", deadline: "2099-01-01T00:00:00Z" }; + }, + } as unknown as WorkspaceWorkers; + await expect(preparePluginWorkspace(db, workers, f.input)).rejects.toMatchObject({ + kind: "provider", reason: "unsupported_operation", retryable: false, + }); + const [task] = await db.select().from(issues).where(eq(issues.id, f.issueId)); + expect(task?.executionWorkspaceId).toBeTruthy(); + expect(await preparePluginWorkspace(db, f.workers, { ...f.input, existingWorkspaceId: task!.executionWorkspaceId })).toBeTruthy(); + }); + it.each([ ["missing", undefined], ["stopped", { status: "stopped", supportedMethods: [] }], diff --git a/server/src/services/plugin-worker-manager.ts b/server/src/services/plugin-worker-manager.ts index fb2ac061a0b..ceb257fec44 100644 --- a/server/src/services/plugin-worker-manager.ts +++ b/server/src/services/plugin-worker-manager.ts @@ -1285,7 +1285,7 @@ export function createPluginWorkerHandle( if (method === "resourceLifecycle" && params.resourceType === "project" && readNonEmptyString(params.resourceId)) { return { companyId: directCompanyId, projectGit: { projectId: params.resourceId as string } }; } - if (method === "workspaceOperation" && isRecord(params.operation)) { + if ((method === "workspaceOperation" || method === "workspacePrepare") && isRecord(params.operation)) { const operation = params.operation; const source = operation.kind === "prepare" ? operation.source : operation.binding; const workspaceId = isRecord(source) ? readNonEmptyString(operation.kind === "prepare" ? source.id : source.projectWorkspaceId) : null; diff --git a/server/src/services/plugin-workspace-preparation.ts b/server/src/services/plugin-workspace-preparation.ts new file mode 100644 index 00000000000..895fc6c0e4f --- /dev/null +++ b/server/src/services/plugin-workspace-preparation.ts @@ -0,0 +1,63 @@ +import { setTimeout as delay } from "node:timers/promises"; +import { parseWorkspacePrepareResult, parseWorkspaceProviderResult, type PluginWorkspaceParams } from "@paperclipai/plugin-sdk"; +import type { PluginWorkerManager } from "./plugin-worker-manager.js"; +import { WorkspaceProviderFailureError, workspaceProviderCallFailure } from "./workspace-provider-failure.js"; + +type ProviderRequest = Omit; +type WorkspaceWorker = Pick; +const legacyWorkspaceOperationMs = 300_000; +const legacyRpcGraceMs = 10_000; + +export async function invokeWorkspaceProvider(workers: WorkspaceWorker, params: ProviderRequest) { + const verifyExisting = params.operation.kind === "prepare" && params.operation.verifyExisting; + if (!verifyExisting && params.operation.kind === "prepare" && workers.getWorker(params.pluginId)?.supportedMethods?.includes("workspacePrepare")) { + const result = await prepareWorkspaceWithPolling(workers, params); + if (result) return result; + } + const timeout = verifyExisting ? 10_000 : params.operation.kind === "finalize" ? 30_000 : legacyWorkspaceOperationMs; + const request = { ...params, deadline: new Date(Date.now() + timeout).toISOString() }; + return parseWorkspaceProviderResult(request, await workers.call( + params.pluginId, "workspaceOperation", request, timeout + (verifyExisting ? 2_000 : legacyRpcGraceMs), + )); +} + +async function prepareWorkspaceWithPolling(workers: Pick, params: ProviderRequest) { + const rpcTimeout = 15_000; + const submit = { ...params, phase: "submit" as const, deadline: new Date(Date.now() + rpcTimeout).toISOString() }; + const receipt = parseWorkspacePrepareResult(submit, await workers.call(params.pluginId, "workspacePrepare", submit, rpcTimeout)); + if (receipt.kind === "unsupported") return null; + if (receipt.kind === "prepared") return receipt; + const offeredMs = Date.parse(receipt.deadline) - Date.now(); + if (offeredMs > legacyWorkspaceOperationMs + legacyRpcGraceMs) + throw new WorkspaceProviderFailureError({ + kind: "provider", reason: "unsupported_operation", retryable: false, + message: "Workspace provider offered a deadline outside the supported lifecycle budget", + }); + const expiresAt = performance.now() + offeredMs; + const remaining = () => { + const ms = Math.ceil(expiresAt - performance.now()); + if (ms <= 0) throw new WorkspaceProviderFailureError({ + kind: "provider", reason: "deadline_exceeded", retryable: true, + message: "Workspace storage operation expired; retry the same workspace", + }); + return ms; + }; + const poll = { ...params, phase: "poll" as const, deadline: receipt.deadline }; + for (;;) { + await delay(Math.min(1000, remaining())); + const timeout = Math.min(rpcTimeout, remaining()); + let response: unknown; + try { + response = await workers.call(params.pluginId, "workspacePrepare", poll, timeout); + } catch (error) { + const failure = workspaceProviderCallFailure(error); + if (failure.kind === "transport" || failure.kind === "timeout") continue; + throw failure; + } + remaining(); + const status = parseWorkspacePrepareResult(poll, response); + if (status.kind === "prepared") return status; + if (status.kind === "unsupported" || status.deadline !== receipt.deadline) + throw new Error("Workspace provider changed the preparation deadline"); + } +} diff --git a/server/src/services/plugin-workspaces.ts b/server/src/services/plugin-workspaces.ts index 6faed0e455b..cab1346ed1d 100644 --- a/server/src/services/plugin-workspaces.ts +++ b/server/src/services/plugin-workspaces.ts @@ -17,16 +17,17 @@ import { type WorkspaceProviderBinding, } from "@paperclipai/shared"; import { - parseWorkspaceProviderResult, workspaceProviderOperationSchema, type PluginWorkspaceOperation, type PluginWorkspaceRecoveryPullRequest, type PluginWorkspaceParams, + type PluginWorkspaceResult, type PluginFreshWorkspace, } from "@paperclipai/plugin-sdk"; +import { invokeWorkspaceProvider } from "./plugin-workspace-preparation.js"; import { notifyTerminalWorkspaceWork } from "./terminal-workspace-work.js"; import { withExecutionWorkspaceLifecycleLock } from "./execution-workspace-lifecycle-lock.js"; -import { workspaceProviderCallFailure } from "./workspace-provider-failure.js"; +import { WorkspaceProviderFailureError, workspaceProviderCallFailure } from "./workspace-provider-failure.js"; import { assertTaskWorkspaceExecutionAccess, assertTaskWorkspaceSourceProjectAccess, @@ -87,7 +88,7 @@ export async function callWorkspaceProvider( operation.fresh = await stageProviderFreshWorkspace({ db, companyId: workspace.companyId, issueId: workspace.sourceIssueId, ...operation.fresh, existing: operation.fresh }); } - const params: PluginWorkspaceParams = { + const params: Omit = { pluginId, companyId: workspace.companyId, projectId: workspace.projectId, @@ -102,22 +103,17 @@ export async function callWorkspaceProvider( ) .digest("hex")), generation: operationGeneration, - deadline: new Date(Date.now() + (verifyExisting ? 10_000 : operation.kind === "finalize" ? 30_000 : 300_000)).toISOString(), runId: input.runId ?? null, agentId: input.agentId ?? null, operation, }; - let value: unknown; + let result: PluginWorkspaceResult; try { - value = await workers!.call( - pluginId, - "workspaceOperation", - params, - verifyExisting ? 12_000 : operation.kind === "finalize" ? 40_000 : 310_000, - ); + result = await invokeWorkspaceProvider(workers!, params); } catch (error) { // A worker that never received the call, a lost reply and a provider-reported // failure need different recovery; only a closed provider reason crosses over. + if (error instanceof WorkspaceProviderFailureError) throw error; throw workspaceProviderCallFailure(error); } // Recheck enablement and ownership after an external effect, including an ambiguous retry. @@ -139,7 +135,6 @@ export async function callWorkspaceProvider( ) { throw conflict("Workspace changed during storage preparation"); } - const result = parseWorkspaceProviderResult(params, value); if (result.kind === "prepared" && result.binding.pluginId !== pluginId) throw conflict("Workspace provider ownership changed"); return result; diff --git a/tests/runner-e2e/source-sdk-public-inputs.json b/tests/runner-e2e/source-sdk-public-inputs.json index 2cb39222091..3ef601ea748 100644 --- a/tests/runner-e2e/source-sdk-public-inputs.json +++ b/tests/runner-e2e/source-sdk-public-inputs.json @@ -3,7 +3,7 @@ "package.json": "e704ca28508fa7c887f704fc26603cf8fd623867", "packages/plugins/sdk/package.json": "b7a4936e2803fd0d3b32dd7f17fc1ea86609ead0", "packages/plugins/sdk/src/bundlers.ts": "883bf3d9f2bd42877cfc0c3707cd882a6f6248d0", - "packages/plugins/sdk/src/define-plugin.ts": "eb6530c006722d22f55355ec810745f1524b874a", + "packages/plugins/sdk/src/define-plugin.ts": "a99e07ce77ef72fcf1df1347346aacb8ba12e284", "packages/plugins/sdk/src/dev-cli.ts": "7c1b0e2b6e4b7fd82027cca5321066e473a03d20", "packages/plugins/sdk/src/dev-server.ts": "2eadff81b74ca0706aec53aaf8637797fe4b4093", "packages/plugins/sdk/src/environment-acquire-error.ts": "50a74c3d1f83150f4215d44a763aa8ce4d9469c2", @@ -20,7 +20,7 @@ "packages/plugins/sdk/src/index.ts": "4e573c181198ea98442321cc136341741e856ff7", "packages/plugins/sdk/src/invocation-tracer.ts": "b28ac4aa4bc3e82658abac4a8081ed4508008696", "packages/plugins/sdk/src/protocol.duplex-channel.test.ts": "18255eb3d63038dc459240c111cbcb29fcba25cc", - "packages/plugins/sdk/src/protocol.ts": "95e8af999dabf93b1b0e531edfec4c80daf52b61", + "packages/plugins/sdk/src/protocol.ts": "97ee5337227e3a1ede6801fe5b7586cc6785edad", "packages/plugins/sdk/src/testing.ts": "fa99065efa867d8f8a9372cb4a448b8f9d687883", "packages/plugins/sdk/src/types.ts": "80375e46504f96c7c0c1f156319dfb81275d16cc", "packages/plugins/sdk/src/ui/clipboard.ts": "1f42069c97d731e636a6ad5481dad22d20c0ed63", @@ -29,9 +29,9 @@ "packages/plugins/sdk/src/ui/index.ts": "f7d1cee64a8f0c924bd10a1bd20745a52fe2060e", "packages/plugins/sdk/src/ui/runtime.ts": "998428e9701eba3b762d3417236d32c2d0d5763d", "packages/plugins/sdk/src/ui/types.ts": "5a5644acf6554d2e5e1d000bfd03b97d57001eb2", - "packages/plugins/sdk/src/worker-rpc-host.ts": "72f844508f695737a44d34b01b3949ec3be10aec", + "packages/plugins/sdk/src/worker-rpc-host.ts": "3de7748e983ad1a749a7a7f62f685dab4399cdd1", "packages/plugins/sdk/src/workspace-operation-error.ts": "2cd247d002d83b163239229e631bca4e519e5075", - "packages/plugins/sdk/src/workspace-provider.ts": "5db136efa4bd904044113e341f443e523e336d0d", + "packages/plugins/sdk/src/workspace-provider.ts": "dea90b92e5bf0c05062eb57598c70f370e207cea", "packages/plugins/sdk/tsconfig.json": "d62ebdff94bc4b75dabf1fcad187e1e7a90fc794", "packages/shared/package.json": "6417ef8baef0cb9b836b14f453dc8bc5025709e0", "packages/shared/src/__snapshots__/slack-app-manifest.test.ts.snap": "79020ba3aa6662923f70aa44e2c483844e9901d9",