mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-13 07:12:00 +01:00
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 <noreply@paperclip.ing>
This commit is contained in:
1 parent
e601f9d2fc
commit
5dfbb04b85
11 files changed
+233
-27
No files matched your search
@@ -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 |
|
||||
|
||||
@@ -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<PluginEnvironmentValidationResult>;
|
||||
|
||||
/** Required with manifest.workspaceProvider, together with project resource lifecycle. */
|
||||
onWorkspacePrepare?(params: PluginWorkspacePrepareParams): Promise<PluginWorkspacePrepareResult>;
|
||||
onWorkspaceOperation?(params: PluginWorkspaceParams): Promise<PluginWorkspaceResult>;
|
||||
|
||||
/** Provide Runner execution: prepare resources, start Runner, connect via PRP, and manage task lifecycle. */
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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");
|
||||
|
||||
@@ -125,6 +125,34 @@ export const pluginWorkspaceParamsSchema = z
|
||||
/** Finalization receives the originating run/agent context; maintenance remains agent-independent. */
|
||||
export type PluginWorkspaceParams = z.infer<typeof pluginWorkspaceParamsSchema>;
|
||||
|
||||
export const pluginWorkspacePrepareParamsSchema = pluginWorkspaceParamsSchema.extend({
|
||||
phase: z.enum(["submit", "poll"]),
|
||||
}).refine((input) => input.operation.kind === "prepare");
|
||||
export type PluginWorkspacePrepareParams = z.infer<typeof pluginWorkspacePrepareParamsSchema>;
|
||||
|
||||
export type PluginWorkspacePrepareResult =
|
||||
| { kind: "unsupported" }
|
||||
| { kind: "pending"; deadline: string }
|
||||
| Extract<PluginWorkspaceResult, { kind: "prepared" }>;
|
||||
|
||||
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,
|
||||
|
||||
@@ -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) } : {}) };
|
||||
|
||||
@@ -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<WorkspaceWorkers["call"]>(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: [] }],
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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<PluginWorkspaceParams, "deadline">;
|
||||
type WorkspaceWorker = Pick<PluginWorkerManager, "call" | "getWorker">;
|
||||
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<PluginWorkerManager, "call">, 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");
|
||||
}
|
||||
}
|
||||
@@ -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<PluginWorkspaceParams, "deadline"> = {
|
||||
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;
|
||||
|
||||
@@ -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",
|
||||
|
||||
Reference in new issue
Block a user