diff --git a/doc/observability.md b/doc/observability.md index e9f5e842ce9..0014bacddb1 100644 --- a/doc/observability.md +++ b/doc/observability.md @@ -2214,7 +2214,7 @@ expired and cross-invocation context cannot select another call's parent. | --- | --- | --- | --- | | Dispatch / receipt / return | `execution.plugin_dispatch` → `plugin.worker` | Host invocation; worker monotonic durations anchored to its epoch start | Outer minus worker time is an unattributed dispatch/scheduling remainder. It is not network latency. | | Repeated host reads | `plugin.host.identity.read`, `config.read`, `progress.read`, `progress.write` | Worker invocation, one span per awaited call, bounded by receipt budget | Round trip and host service together. Span count is call count; overlapping reads are not additive. | -| Plugin coordination | `plugin.storage.queue`, `storage.execute`, `runtime.queue`, `runtime.prepare`, `runtime.reconcile`, `runtime.verify`, `workspace.prepare` | Invoking plugin, monotonic durations | The plugin keeps its locking and authorization owners. `plugin.reused` is boolean. | +| Plugin coordination | `plugin.storage.queue`, `storage.execute`, `runtime.queue`, `runtime.prepare`, `runtime.reconcile`, `runtime.verify`, `workspace.prepare`, `workspace.failure` | Invoking plugin, monotonic durations | The plugin keeps its locking and authorization owners. `plugin.reused` is boolean. `plugin.failure_reason` on `workspace.failure` is a closed workspace operation failure reason or `unknown`. | | Completed remote lifecycle | `plugin.remote.lifecycle`, linked to the host-owned observer | Bounded prior monotonic durations, up to ten minutes old | Independent lifecycle trace; never a child of a later polling request. The worker cannot supply link IDs. | | Remote operation receipt | `plugin.remote.request` → `plugin.remote.operation` | Remote monotonic offsets, checked against local round trip and parent bounds | Closed operation classes cover mount refresh, guest setup, process execution, scripts and cleanup. No remote wall-clock subtraction. | | Retained/cold transport | `runner.transport.attach`, `resume`, `cold_start`, `prepare`, `registration`, `provider_identity` | Controller process, current active context on every operation | A retained transport must not capture the preceding run's trace. These parents overlap their children. | diff --git a/doc/plugins/PLUGIN_RUNNER_EXECUTION.md b/doc/plugins/PLUGIN_RUNNER_EXECUTION.md index b4c75d8a229..e5aba121a67 100644 --- a/doc/plugins/PLUGIN_RUNNER_EXECUTION.md +++ b/doc/plugins/PLUGIN_RUNNER_EXECUTION.md @@ -231,6 +231,41 @@ 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. +### Workspace operation failures + +Core separates a failed `workspaceOperation` call by what it knows about the +outcome: + +| Failure | Run error | Core behavior | +| --- | --- | --- | +| The worker was not running, so no request was sent | `Workspace storage provider is not running; retry the same workspace` | A run that has not started provider work is retried as `plugin_worker_unavailable`. | +| The call timed out, or the worker exited with the call pending | A fixed timeout or stopped-provider message that says to retry the same workspace | The operation may have run, so the same workspace and operation identity must be retried. | +| The provider threw an error with a known reason | The fixed message Core keeps for that reason | Only transient reasons tell the user to retry. | +| 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` +property. The reason must come from `WORKSPACE_OPERATION_FAILURE_REASONS`: + +| Reason | Retrying the same workspace can help | +| --- | --- | +| `unsupported_operation`, `configuration_unavailable`, `credential_unavailable`, `storage_grant_unavailable`, `binding_mismatch`, `runtime_connection_invalid` | No | +| `storage_pending`, `runtime_unavailable`, `remote_incomplete`, `context_changed`, `deadline_exceeded` | Yes | + +```ts +import { PluginWorkspaceOperationError } from "@paperclipai/plugin-sdk"; + +throw new PluginWorkspaceOperationError("storage_grant_unavailable", "Project mount grants unavailable"); +``` + +The SDK sends only the reason. Core never reads the provider's message, so a +message cannot reach the run, the API, or Core logs. The run keeps its +`setup_failed` error code and records `resultJson.workspaceProviderFailure` with +the failure `kind`, `reason`, and `retryable` flag. The +`heartbeat execution setup failed` log carries the same fields. A provider may +also record a `plugin.workspace.failure` span whose `plugin.failure_reason` is a +reason from this set or `unknown`. + ## Warm task retention A driver declares `warmTaskVersion: 1`, implements `onEnvironmentResumeLease`, diff --git a/packages/plugins/sdk/src/index.ts b/packages/plugins/sdk/src/index.ts index d8399427968..2a655ae5603 100644 --- a/packages/plugins/sdk/src/index.ts +++ b/packages/plugins/sdk/src/index.ts @@ -466,6 +466,8 @@ export { PluginEnvironmentRunnerIngressError, environmentRunnerIngressErrorData, export type { PluginEnvironmentRunnerIngressFailure } from "./environment-runner-ingress-error.js"; export { ENVIRONMENT_TASK_ERROR_CODES, PluginEnvironmentTaskError, environmentTaskErrorCodeData, readEnvironmentTaskErrorCode } from "./environment-task-error.js"; export type { PluginEnvironmentTaskErrorCode } from "./environment-task-error.js"; +export { WORKSPACE_OPERATION_FAILURE_REASONS, PluginWorkspaceOperationError, workspaceOperationErrorData, readWorkspaceOperationFailureReason } from "./workspace-operation-error.js"; +export type { PluginWorkspaceOperationFailureReason } from "./workspace-operation-error.js"; export { PluginEnvironmentStopError, environmentStopErrorData, readEnvironmentStopFailure } from "./environment-stop-error.js"; export type { PluginEnvironmentStopFailure, PluginEnvironmentStopFailureReason } from "./environment-stop-error.js"; export { environmentResumeErrorData, readEnvironmentResumeErrorDiagnostic } from "./environment-resume-error.js"; diff --git a/packages/plugins/sdk/src/worker-rpc-host.ts b/packages/plugins/sdk/src/worker-rpc-host.ts index 706fe9da497..91dbec177a4 100644 --- a/packages/plugins/sdk/src/worker-rpc-host.ts +++ b/packages/plugins/sdk/src/worker-rpc-host.ts @@ -10,6 +10,7 @@ import { environmentRunnerIngressErrorData } from "./environment-runner-ingress- import { environmentStopErrorData } from "./environment-stop-error.js"; import { environmentResumeErrorData } from "./environment-resume-error.js"; import { environmentTaskErrorCodeData, environmentTaskErrorData } from "./environment-task-error.js"; +import { workspaceOperationErrorData } from "./workspace-operation-error.js"; import { createInvocationTracer } from "./invocation-tracer.js"; import { createPluginIdleDrain } from "./idle-drain.js"; /** @@ -1655,7 +1656,8 @@ export function startWorkerRpcHost(options: WorkerRpcHostOptions): WorkerRpcHost : method === "environmentRunnerIngressEndpoint" ? environmentRunnerIngressErrorData(err) : method === "environmentStopLease" ? environmentStopErrorData(err) : method === "environmentResumeLease" ? environmentResumeErrorData(err) - : method === "environmentTask" ? taskError ?? environmentTaskErrorCodeData(err) : undefined)); + : method === "environmentTask" ? taskError ?? environmentTaskErrorCodeData(err) + : method === "workspaceOperation" ? workspaceOperationErrorData(err) : undefined)); } finally { done?.(); } diff --git a/packages/plugins/sdk/src/workspace-operation-error.ts b/packages/plugins/sdk/src/workspace-operation-error.ts new file mode 100644 index 00000000000..2cd247d002d --- /dev/null +++ b/packages/plugins/sdk/src/workspace-operation-error.ts @@ -0,0 +1,78 @@ +/** + * Why a workspace provider refused or failed a `workspaceOperation`. The set is + * closed: Core decides the user-facing message and whether retrying the same + * workspace can help, so a provider never sends text or values across the bridge. + */ +export const WORKSPACE_OPERATION_FAILURE_REASONS = [ + /** The provider does not accept this operation or request shape. */ + "unsupported_operation", + /** The provider's own settings are missing or invalid. */ + "configuration_unavailable", + /** A credential the operation needs (for example the project's Git credential) is unusable. */ + "credential_unavailable", + /** Saved storage access grants are missing or no longer match the project. */ + "storage_grant_unavailable", + /** The saved workspace binding does not match this workspace or generation. */ + "binding_mismatch", + /** Project storage is still being provisioned or reconciled. */ + "storage_pending", + /** The selected runtime is not ready. */ + "runtime_unavailable", + /** The selected runtime's saved connection is invalid. */ + "runtime_connection_invalid", + /** A remote storage or runtime request failed or returned an unusable result. */ + "remote_incomplete", + /** The runtime, grants or binding changed while the operation ran. */ + "context_changed", + /** The operation deadline passed. */ + "deadline_exceeded", +] as const; +export type PluginWorkspaceOperationFailureReason = typeof WORKSPACE_OPERATION_FAILURE_REASONS[number]; + +const SCHEMA = "paperclip/workspace-operation-error/v1"; + +function ownField(value: unknown, name: string): unknown { + if (!value || typeof value !== "object") return undefined; + try { return Object.getOwnPropertyDescriptor(value, name)?.value; } + catch { return undefined; } +} + +function validateReason(value: unknown): PluginWorkspaceOperationFailureReason | null { + return WORKSPACE_OPERATION_FAILURE_REASONS.includes(value as PluginWorkspaceOperationFailureReason) + ? value as PluginWorkspaceOperationFailureReason : null; +} + +/** + * Throw from `onWorkspaceOperation` to tell Core why the operation failed. Any + * error with an own `workspaceFailureReason` from the closed set is forwarded the + * same way; the message stays local to the worker's RPC error text. + */ +export class PluginWorkspaceOperationError extends Error { + readonly workspaceFailureReason: PluginWorkspaceOperationFailureReason; + + constructor(reason: PluginWorkspaceOperationFailureReason, message = "Workspace operation failed") { + const safe = validateReason(reason); + if (!safe) throw new TypeError("Unknown workspace operation failure reason"); + super(message); + this.name = "PluginWorkspaceOperationError"; + this.workspaceFailureReason = safe; + } +} + +/** Worker side: copy only the closed reason, never provider text or identities. */ +export function workspaceOperationErrorData(error: unknown): { schema: string; reason: PluginWorkspaceOperationFailureReason } | undefined { + const reason = validateReason(ownField(error, "workspaceFailureReason")); + return reason ? { schema: SCHEMA, reason } : undefined; +} + +/** Host side: validate the envelope again after the worker RPC, through wrapper causes. */ +export function readWorkspaceOperationFailureReason(error: unknown): PluginWorkspaceOperationFailureReason | null { + for (let current = error, depth = 0; current && depth < 5; current = ownField(current, "cause"), depth++) { + const data = ownField(current, "data"); + if (ownField(data, "schema") === SCHEMA) { + const reason = validateReason(ownField(data, "reason")); + if (reason) return reason; + } + } + return null; +} diff --git a/packages/plugins/sdk/tests/workspace-operation-error.test.ts b/packages/plugins/sdk/tests/workspace-operation-error.test.ts new file mode 100644 index 00000000000..f807646aad3 --- /dev/null +++ b/packages/plugins/sdk/tests/workspace-operation-error.test.ts @@ -0,0 +1,84 @@ +import { createInterface } from "node:readline"; +import { PassThrough } from "node:stream"; +import { describe, expect, it } from "vitest"; +import { definePlugin } from "../src/define-plugin.js"; +import { createRequest, isJsonRpcResponse, JsonRpcCallError, parseMessage, serializeMessage, type JsonRpcResponse } from "../src/protocol.js"; +import { startWorkerRpcHost } from "../src/worker-rpc-host.js"; +import { PluginWorkspaceOperationError, readWorkspaceOperationFailureReason, workspaceOperationErrorData } from "../src/workspace-operation-error.js"; + +const id = (n: number) => `10000000-0000-4000-8000-${String(n).padStart(12, "0")}`; +const params = { + pluginId: id(1), companyId: id(2), projectId: id(3), executionWorkspaceId: id(4), agentId: id(9), + operationId: "operation-1", generation: 0, deadline: "2099-01-01T00:00:00.000Z", runId: null, + operation: { kind: "prepare", source: { id: id(5), revision: "source-v1", repoUrl: null, repoRef: null, defaultRef: null }, + mode: "shared_workspace", strategy: "project_primary", branchName: null, baseRef: null, existing: null }, +}; + +async function invoke(method: string, error: unknown) { + const stdin = new PassThrough(); + const stdout = new PassThrough(); + const lines = createInterface({ input: stdout }); + const worker = startWorkerRpcHost({ stdin, stdout, plugin: definePlugin({ + async setup() {}, + async onWorkspaceOperation() { throw error; }, + async onEnvironmentProbe() { throw error; }, + }) }); + try { + const response = new Promise((resolve) => { + lines.on("line", (line) => { + const message = parseMessage(line); + if (isJsonRpcResponse(message)) resolve(message); + }); + }); + stdin.write(serializeMessage(createRequest(method, params, "workspace-test"))); + return await response; + } finally { + worker.stop(); lines.close(); stdin.destroy(); stdout.destroy(); + } +} + +describe("workspace operation failure reasons", () => { + it("carries the closed reason across the real workspaceOperation RPC", async () => { + const response = await invoke("workspaceOperation", new PluginWorkspaceOperationError("storage_grant_unavailable", "Project mount grants unavailable")); + if (!("error" in response) || !response.error) throw new Error("Expected RPC error"); + expect(response.error.data).toEqual({ schema: "paperclip/workspace-operation-error/v1", reason: "storage_grant_unavailable" }); + const callError = new JsonRpcCallError(response.error); + expect(readWorkspaceOperationFailureReason(callError)).toBe("storage_grant_unavailable"); + expect(readWorkspaceOperationFailureReason(new Error("wrapped", { cause: new Error("again", { cause: callError }) }))) + .toBe("storage_grant_unavailable"); + }); + + it("forwards a provider error's own closed reason without the SDK class", async () => { + // A provider built against an older SDK can carry the reason before it can import the class. + const response = await invoke("workspaceOperation", + Object.assign(new Error("Invalid runtime computer connection"), { workspaceFailureReason: "runtime_connection_invalid" })); + if (!("error" in response) || !response.error) throw new Error("Expected RPC error"); + expect(readWorkspaceOperationFailureReason(new JsonRpcCallError(response.error))).toBe("runtime_connection_invalid"); + }); + + it("never derives a reason from message text or forwards an unknown reason", async () => { + for (const error of [new Error("configuration_unavailable"), + Object.assign(new Error("private"), { workspaceFailureReason: "token=secret" })]) { + const response = await invoke("workspaceOperation", error); + expect("error" in response && response.error).not.toHaveProperty("data"); + } + }); + + it("only attaches the envelope to workspaceOperation calls", async () => { + const response = await invoke("environmentProbe", new PluginWorkspaceOperationError("binding_mismatch")); + expect("error" in response && response.error).not.toHaveProperty("data"); + }); + + it("refuses reasons outside the closed set", () => { + expect(() => new PluginWorkspaceOperationError("private" as never)).toThrow(TypeError); + expect(workspaceOperationErrorData(new Error("plain"))).toBeUndefined(); + // Only an own property counts; an inherited one is not the error's reason. + expect(workspaceOperationErrorData(Object.create({ workspaceFailureReason: "binding_mismatch" }))).toBeUndefined(); + }); + + it.each([null, {}, { data: { schema: "other", reason: "binding_mismatch" } }, + { data: { schema: "paperclip/workspace-operation-error/v1", reason: "private" } }])( + "rejects malformed envelopes %j", (value) => { + expect(readWorkspaceOperationFailureReason(value)).toBeNull(); + }); +}); diff --git a/server/src/__tests__/plugin-host-services-span.test.ts b/server/src/__tests__/plugin-host-services-span.test.ts index a91e6c4e626..81b03f40970 100644 --- a/server/src/__tests__/plugin-host-services-span.test.ts +++ b/server/src/__tests__/plugin-host-services-span.test.ts @@ -351,6 +351,18 @@ describe("nested worker timing receipts",()=>{ expect(clampProviderSpanAttributes({ "paperclip.sandbox.startup.state": "private-state" })).toEqual({}); }); + it("admits a workspace failure span with only a closed failure reason", () => { + mockRecordSpan.mockReset(); + const now = Date.now(); + recordWorkerProviderSpan({ name: "plugin.workspace.failure", startTimeMs: now - 1, endTimeMs: now, status: { code: 2 }, + attributes: { "plugin.failure_reason": "storage_grant_unavailable", message: "Project mount grants unavailable" } }, { traceparent: VALID_TRACEPARENT }); + expect(mockRecordSpan.mock.calls[0][0]).toMatchObject({ name: "plugin.workspace.failure", attributes: { "plugin.failure_reason": "storage_grant_unavailable" } }); + expect(JSON.stringify(mockRecordSpan.mock.calls)).not.toContain("mount grants"); + expect(clampProviderSpanAttributes({ "plugin.failure_reason": "unknown" })).toEqual({ "plugin.failure_reason": "unknown" }); + expect(clampProviderSpanAttributes({ "plugin.failure_reason": "token=private" })).toEqual({}); + expect(clampProviderSpanAttributes({ "plugin.failure_reason": 1 })).toEqual({}); + }); + it("admits every closed lease phase and rejects dynamic phase names", () => { mockRecordSpan.mockReset(); const names = ["lease.wait_for_teardown", "lease.lookup", "lease.refresh", "lease.channel_cleanup", diff --git a/server/src/__tests__/plugin-workspaces.integration.test.ts b/server/src/__tests__/plugin-workspaces.integration.test.ts index b72381db25f..c0724796738 100644 --- a/server/src/__tests__/plugin-workspaces.integration.test.ts +++ b/server/src/__tests__/plugin-workspaces.integration.test.ts @@ -20,6 +20,8 @@ import { type WorkspaceWorkers, } from "../services/plugin-workspaces.js"; import { executionWorkspaceService } from "../services/execution-workspaces.js"; +import { WorkspaceProviderFailureError } from "../services/workspace-provider-failure.js"; +import { findPluginWorkerUnavailableError, PluginWorkerUnavailableError } from "../services/plugin-worker-unavailable-error.js"; import { pluginRegistryService } from "../services/plugin-registry.js"; import { projectService } from "../services/projects.js"; import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js"; @@ -45,6 +47,27 @@ const support = await getEmbeddedPostgresTestSupport(); }); const fixture = (companyId?: string) => createPluginWorkspaceFixture(db, companyId); + it.each([ + ["missing", undefined], + ["stopped", { status: "stopped", supportedMethods: [] }], + ["starting", { status: "starting", supportedMethods: [] }], + ] as const)("retries preparation when the ready provider's worker is %s, before sending a request", async (_state, worker) => { + const f = await fixture(); + const workers = { getWorker: () => worker, call: f.workers.call.bind(f.workers) } as unknown as WorkspaceWorkers; + const failure = await preparePluginWorkspace(db, workers, f.input).then(() => null, (error: unknown) => error); + expect(failure).toBeInstanceOf(WorkspaceProviderFailureError); + expect((failure as WorkspaceProviderFailureError).summary()).toEqual({ kind: "worker_unavailable", reason: null, retryable: true }); + expect(findPluginWorkerUnavailableError(failure)).toBeInstanceOf(PluginWorkerUnavailableError); + expect(f.calls).toHaveLength(0); + }); + it("keeps a running worker without workspace operations an unavailable provider", async () => { + const f = await fixture(); + const workers = { getWorker: () => ({ status: "running", supportedMethods: [] }), call: f.workers.call.bind(f.workers) } as unknown as WorkspaceWorkers; + const failure = await preparePluginWorkspace(db, workers, f.input).then(() => null, (error: unknown) => error); + expect(failure).not.toBeInstanceOf(WorkspaceProviderFailureError); + expect((failure as Error).message).toBe("The workspace storage plugin is unavailable"); + expect(f.calls).toHaveLength(0); + }); it.each([1, undefined])("gates recovery hints by provider capability (%s)", async (version) => { const f = await fixture(); const [plugin] = await db.select().from(plugins).where(eq(plugins.id, f.pluginId)); diff --git a/server/src/__tests__/workspace-provider-failure.test.ts b/server/src/__tests__/workspace-provider-failure.test.ts new file mode 100644 index 00000000000..ab2c6d3a867 --- /dev/null +++ b/server/src/__tests__/workspace-provider-failure.test.ts @@ -0,0 +1,111 @@ +import { createInterface } from "node:readline"; +import { PassThrough } from "node:stream"; +import { describe, expect, it } from "vitest"; +import { + createRequest, + definePlugin, + isJsonRpcResponse, + JsonRpcCallError, + parseMessage, + PLUGIN_RPC_ERROR_CODES, + PluginWorkspaceOperationError, + serializeMessage, + startWorkerRpcHost, + WORKSPACE_OPERATION_FAILURE_REASONS, + type JsonRpcResponse, +} from "@paperclipai/plugin-sdk"; +import { HttpError } from "../errors.js"; +import { findPluginWorkerUnavailableError, PluginWorkerUnavailableError } from "../services/plugin-worker-unavailable-error.js"; +import { findWorkspaceProviderFailure, workspaceProviderCallFailure } from "../services/workspace-provider-failure.js"; + +const id = (n: number) => `10000000-0000-4000-8000-${String(n).padStart(12, "0")}`; +const params = { + pluginId: id(1), companyId: id(2), projectId: id(3), executionWorkspaceId: id(4), agentId: id(9), + operationId: "operation-1", generation: 0, deadline: "2099-01-01T00:00:00.000Z", runId: null, + operation: { kind: "prepare", source: { id: id(5), revision: "source-v1", repoUrl: null, repoRef: null, defaultRef: null }, + mode: "shared_workspace", strategy: "project_primary", branchName: null, baseRef: null, existing: null }, +}; + +/** The error Core receives when a real worker's `onWorkspaceOperation` throws. */ +async function providerRejection(error: unknown): Promise { + const stdin = new PassThrough(); + const stdout = new PassThrough(); + const lines = createInterface({ input: stdout }); + const worker = startWorkerRpcHost({ stdin, stdout, plugin: definePlugin({ + async setup() {}, + async onWorkspaceOperation() { throw error; }, + }) }); + try { + const response = await new Promise((resolve) => { + lines.on("line", (line) => { + const message = parseMessage(line); + if (isJsonRpcResponse(message)) resolve(message); + }); + stdin.write(serializeMessage(createRequest("workspaceOperation", params, "workspace-test"))); + }); + if (!("error" in response) || !response.error) throw new Error("Expected RPC error"); + return new JsonRpcCallError(response.error); + } finally { + worker.stop(); lines.close(); stdin.destroy(); stdout.destroy(); + } +} + +describe("workspace provider call failures", () => { + it("tells transport failures apart and keeps the retry hint only where the outcome is unknown", () => { + const refusal = new PluginWorkerUnavailableError({ pluginId: "plugin-1", workerStatus: null, message: "worker is not running" }); + const cases = [ + [refusal, "worker_unavailable", "Workspace storage provider is not running; retry the same workspace"], + [new JsonRpcCallError({ code: PLUGIN_RPC_ERROR_CODES.TIMEOUT, message: "timed out after 310000ms" }), "timeout", + "Workspace storage operation timed out; retry the same workspace"], + [new JsonRpcCallError({ code: PLUGIN_RPC_ERROR_CODES.WORKER_UNAVAILABLE, message: "Worker process exited (code=1)" }), "worker_exited", + "Workspace storage provider stopped during the operation; retry the same workspace"], + [new Error("Plugin is held for idle sleep"), "transport", "Workspace storage operation is incomplete; retry the same workspace"], + ] as const; + for (const [error, kind, message] of cases) { + const failure = workspaceProviderCallFailure(error); + expect(failure).toBeInstanceOf(HttpError); + expect(failure).toMatchObject({ status: 409, message, kind, reason: null, retryable: true, + details: { code: "workspace_provider_failed", kind, reason: null, retryable: true } }); + } + // No request reached the worker, so setup's bounded bootstrap retry still recognizes the refusal. + expect(findPluginWorkerUnavailableError(workspaceProviderCallFailure(refusal))).toBe(refusal); + expect(workspaceProviderCallFailure(cases[1][0]).cause).toBeUndefined(); + }); + + it("passes a provider's closed reason through and says when retrying cannot help", async () => { + const rejection = await providerRejection(new PluginWorkspaceOperationError("storage_grant_unavailable", "Project mount grants unavailable")); + expect(rejection.code).toBe(PLUGIN_RPC_ERROR_CODES.WORKER_ERROR); + const failure = workspaceProviderCallFailure(rejection); + expect(failure.summary()).toEqual({ kind: "provider", reason: "storage_grant_unavailable", retryable: false }); + expect(failure.message).toBe("Project storage access is unavailable or out of date; reconcile the project's storage before retrying"); + expect(failure.message).not.toContain("retry the same workspace"); + // The provider's message and RPC error never become the run's error or cause. + expect(failure.message).not.toContain("mount grants"); + expect(failure.cause).toBeUndefined(); + }); + + it("gives every reason a fixed message and a retry decision", () => { + const transient = new Set(["storage_pending", "runtime_unavailable", "remote_incomplete", "context_changed", "deadline_exceeded"]); + for (const reason of WORKSPACE_OPERATION_FAILURE_REASONS) { + const failure = workspaceProviderCallFailure(new JsonRpcCallError({ code: PLUGIN_RPC_ERROR_CODES.WORKER_ERROR, message: "private provider text", + data: { schema: "paperclip/workspace-operation-error/v1", reason } })); + expect(failure.summary()).toEqual({ kind: "provider", reason, retryable: transient.has(reason) }); + expect(failure.message).not.toContain("private"); + expect(failure.message.includes("retry the same workspace")).toBe(transient.has(reason)); + } + }); + + it("treats a provider error without a known reason as a reported failure, never parsing its text", async () => { + for (const error of [new Error("configuration_unavailable"), Object.assign(new Error("x"), { workspaceFailureReason: "private" })]) { + const failure = workspaceProviderCallFailure(await providerRejection(error)); + expect(failure.summary()).toEqual({ kind: "provider", reason: null, retryable: true }); + expect(failure.message).toBe("Workspace storage provider reported a failure; retry the same workspace"); + } + }); + + it("is found through setup's error wrappers", () => { + const failure = workspaceProviderCallFailure(new Error("lost")); + expect(findWorkspaceProviderFailure(new Error("setup failed", { cause: failure }))).toBe(failure); + expect(findWorkspaceProviderFailure(new Error("other"))).toBeNull(); + }); +}); diff --git a/server/src/services/heartbeat/run-completion.test.ts b/server/src/services/heartbeat/run-completion.test.ts index 524f9c0fc8b..2d96868fcbe 100644 --- a/server/src/services/heartbeat/run-completion.test.ts +++ b/server/src/services/heartbeat/run-completion.test.ts @@ -30,8 +30,10 @@ import { RemoteWorkspaceOperationError, RemoteWorkspaceOperationTimeoutError, Re import { EnvironmentRunError } from "../environment-run-orchestrator.js"; import { ReusableSandboxResumeError } from "../reusable-sandbox-resume-error.js"; import { PluginWorkerUnavailableError } from "../plugin-worker-unavailable-error.js"; -import { environmentTaskErrorCodeData, JsonRpcCallError, PluginEnvironmentTaskError } from "@paperclipai/plugin-sdk"; +import { environmentTaskErrorCodeData, JsonRpcCallError, PluginEnvironmentTaskError, PluginWorkspaceOperationError, workspaceOperationErrorData } from "@paperclipai/plugin-sdk"; import { RUNNER_REVISION_MISMATCH_RETRY } from "../runner-revision-mismatch.js"; +import { workspaceProviderCallFailure } from "../workspace-provider-failure.js"; +import { logger } from "../../middleware/logger.js"; import { unprocessable } from "../../errors.js"; import { createHeartbeatLifecycle, type HeartbeatLifecycleDependencies } from "./run-lifecycle.js"; import { createHeartbeatRunPreparation, ConfigurationIncompleteFailure, classifyManagedProjectAuthFailure } from "./run-preparation.js"; @@ -1000,6 +1002,36 @@ describe.skipIf(!support.supported && !process.env.PAPERCLIP_TEST_DATABASE_URL)( expect(f.deps.scheduleInteractionContinuationInfrastructureRetryIfEligible).not.toHaveBeenCalled(); }); + it("records a workspace provider's reason on the run and in the setup failure log without retrying", async () => { + const f = await fixture(); + const logged = vi.spyOn(logger, "error"); + onTestFinished(() => logged.mockRestore()); + const rejection = new JsonRpcCallError({ code: -32002, message: "Invalid runtime computer connection", + data: workspaceOperationErrorData(new PluginWorkspaceOperationError("runtime_connection_invalid")) }); + await f.completion.failRunSetup(f.setup(workspaceProviderCallFailure(rejection))); + const failed = await read(f.run.id); + expect(failed).toMatchObject({ status: "failed", errorCode: "setup_failed", + error: "The selected runtime's saved connection is invalid; reprovision the runtime before retrying", + resultJson: { workspaceProviderFailure: { kind: "provider", reason: "runtime_connection_invalid", retryable: false } } }); + expect(JSON.stringify(failed)).not.toContain("Invalid runtime computer connection"); + expect(f.deps.scheduleBoundedRetryForRun).not.toHaveBeenCalled(); + const setupLog = logged.mock.calls.find(call => call[1] === "heartbeat execution setup failed"); + expect(setupLog?.[0]).toMatchObject({ runId: f.run.id, errorCode: "setup_failed", + workspaceProviderFailure: { kind: "provider", reason: "runtime_connection_invalid", retryable: false } }); + }); + + it("retries a workspace operation the stopped provider worker never received", async () => { + const f = await fixture(); + const refusal = new PluginWorkerUnavailableError({ pluginId: "plugin-1", workerStatus: "stopped", + message: 'Cannot call "workspaceOperation" — worker for "plugin-1" is stopped' }); + f.deps.scheduleBoundedRetryForRun.mockResolvedValueOnce({ outcome: "not_scheduled", reason: "Fixture scheduler boundary", issueId: null }); + await f.completion.failRunSetup(f.setup(workspaceProviderCallFailure(refusal))); + expect(await read(f.run.id)).toMatchObject({ status: "failed", errorCode: "plugin_worker_unavailable", + error: "Workspace storage provider is not running; retry the same workspace", + resultJson: { workspaceProviderFailure: { kind: "worker_unavailable", reason: null, retryable: true } } }); + expect(f.deps.scheduleBoundedRetryForRun).toHaveBeenCalledWith(expect.objectContaining({ id: f.run.id }), f.agent); + }); + it("keeps a found project authentication conflict a configuration gap that blocks for the board", async () => { const f = await fixture(); const conflict = unprocessable("This agent's managed AI connection conflicts with project authentication settings in .claude/settings.json.", diff --git a/server/src/services/heartbeat/run-completion.ts b/server/src/services/heartbeat/run-completion.ts index 00622710994..76780c4e1ae 100644 --- a/server/src/services/heartbeat/run-completion.ts +++ b/server/src/services/heartbeat/run-completion.ts @@ -13,6 +13,7 @@ import { RemoteWorkspacePreparationTimeoutError, REMOTE_WORKSPACE_RESTORE_FAILED import { escalateRemoteWorkspaceRestoreFailure } from "../remote-workspace-restore-recovery.js"; import { findReusableSandboxResumeError } from "../reusable-sandbox-resume-error.js"; import { findPluginWorkerUnavailableError, PLUGIN_WORKER_UNAVAILABLE_CODE } from "../plugin-worker-unavailable-error.js"; +import { findWorkspaceProviderFailure } from "../workspace-provider-failure.js"; import { isRunnerRevisionMismatch, RUNNER_REVISION_MISMATCH_CODE, RUNNER_REVISION_MISMATCH_RETRY } from "../runner-revision-mismatch.js"; import { type RunSessionOutcome, resolveNextSessionState, normalizeUsageTotals, resolveCacheAdjustedCostUsd, resolveLedgerBiller, resolveLedgerCostStatus, normalizeLedgerBillingType } from "./run-state.js"; import { readGitConnectionFailure } from "../git-connection-failure.js"; @@ -1396,8 +1397,11 @@ export function createHeartbeatRunCompletion(db: Db, dependencies: HeartbeatRunC recordedResponsibleUserDenialCode ?? nonRetryablePreflightCode ?? "setup_failed"; + // Bounded kind and reason only: a workspace provider's own text never crosses into Core. + const workspaceProviderFailure = findWorkspaceProviderFailure(outerErr)?.summary() ?? null; logger.error( - { err: identityRedactor.redact({ message: outerErr instanceof Error ? outerErr.message : String(outerErr), stack: outerErr instanceof Error ? outerErr.stack : undefined }), runId }, + { err: identityRedactor.redact({ message: outerErr instanceof Error ? outerErr.message : String(outerErr), stack: outerErr instanceof Error ? outerErr.stack : undefined }), runId, + errorCode: setupFailureErrorCode, ...(workspaceProviderFailure ? { workspaceProviderFailure } : {}) }, "heartbeat execution setup failed", ); const setupFailureAgent = await getAgent(run.agentId).catch(() => null); @@ -1434,6 +1438,7 @@ export function createHeartbeatRunCompletion(db: Db, dependencies: HeartbeatRunC ...(providerFailure ? { providerFailure: { ...providerFailure, message: redactCurrentUserText(identityRedactor.redact(providerFailure.message), currentUserRedactionOptions) } } : {}), ...(connectionFailure ? { connectionFailure } : {}), + ...(workspaceProviderFailure ? { workspaceProviderFailure } : {}), // A lease lost during setup surfaces as the thrown abort reason, or as // the signal's reason when a setup step threw a different error after it. ...controllerLeaseLossEvidence(outerErr), diff --git a/server/src/services/plugin-host-services.ts b/server/src/services/plugin-host-services.ts index eeff010abbe..1c7030036ea 100644 --- a/server/src/services/plugin-host-services.ts +++ b/server/src/services/plugin-host-services.ts @@ -93,6 +93,7 @@ import { } from "./authorization.js"; import { redactEventPayload, sanitizeRecord } from "../redaction.js"; import type { WorkerHostCallContext } from "@paperclipai/plugin-sdk"; +import { WORKSPACE_OPERATION_FAILURE_REASONS } from "@paperclipai/plugin-sdk"; import { normalizeProviderFamily, SANDBOX_STARTUP_SPAN_ATTRS, @@ -541,7 +542,7 @@ const KNOWN_PROVIDER_SPAN_NAMES: ReadonlySet = new Set([ "lease.start", "lease.recover", "lease.verify", "lease.workspace", "lease.sentinel", "lease.shell", "lease.create", "lease.expiry", "lease.cleanup", "plugin.worker", "plugin.host.identity.read", "plugin.host.config.read", "plugin.host.progress.read", "plugin.host.progress.write", - "plugin.workspace.prepare", "plugin.storage.queue", "plugin.storage.execute", + "plugin.workspace.prepare", "plugin.workspace.failure", "plugin.storage.queue", "plugin.storage.execute", "plugin.runtime.queue", "plugin.runtime.prepare", "plugin.runtime.reconcile", "plugin.runtime.verify", "plugin.remote.request", "plugin.remote.operation", "plugin.remote.lifecycle", ]); @@ -568,9 +569,12 @@ const PROVIDER_SPAN_ATTR_ALLOWLIST: ReadonlySet = new Set([ SPAN_ATTRS.transferGuardCount, SPAN_ATTRS.transferDirection, SPAN_ATTRS.state, - "plugin.operation", "plugin.reused", "plugin.item_count", "plugin.bytes", "plugin.call_count", + "plugin.operation", "plugin.reused", "plugin.item_count", "plugin.bytes", "plugin.call_count", "plugin.failure_reason", ]); +/** The closed value set for `plugin.failure_reason`: a workspace operation failure reason, or `unknown`. */ +const KNOWN_PLUGIN_FAILURE_REASONS: ReadonlySet = new Set([...WORKSPACE_OPERATION_FAILURE_REASONS, "unknown"]); + /** The subset of allowed keys that carry a finite number. */ const PROVIDER_SPAN_NUMERIC_ATTRS: ReadonlySet = new Set([ SPAN_ATTRS.packWallMs, @@ -608,6 +612,10 @@ export function clampProviderSpanAttributes( continue; } if (key === "plugin.reused") { if (typeof value === "boolean") clamped[key] = value; continue; } + if (key === "plugin.failure_reason") { + if (typeof value === "string" && KNOWN_PLUGIN_FAILURE_REASONS.has(value)) clamped[key] = value; + continue; + } if (key === SPAN_ATTRS.provider) { clamped[key] = normalizeProviderFamily(typeof value === "string" ? value : undefined); continue; diff --git a/server/src/services/plugin-workspaces.ts b/server/src/services/plugin-workspaces.ts index 0792926670e..34674cbc0b9 100644 --- a/server/src/services/plugin-workspaces.ts +++ b/server/src/services/plugin-workspaces.ts @@ -28,6 +28,7 @@ import { import type { PluginWorkerManager } from "./plugin-worker-manager.js"; import { notifyTerminalWorkspaceWork } from "./terminal-workspace-work.js"; import { withExecutionWorkspaceLifecycleLock } from "./execution-workspace-lifecycle-lock.js"; +import { workspaceProviderCallFailure, workspaceWorkerRefusal } from "./workspace-provider-failure.js"; import { assertTaskWorkspaceExecutionAccess, assertTaskWorkspaceSourceProjectAccess, @@ -42,6 +43,8 @@ async function provider( workers: WorkspaceWorkers | undefined, companyId: string, pluginId: string, + /** Before a call, a ready provider whose worker is not running is a retryable refusal: no request is sent. */ + beforeCall = false, ) { const [plugin] = await db .select() @@ -56,15 +59,17 @@ async function provider( eq(pluginCompanySettings.companyId, companyId), ), ); + const worker = workers?.getWorker(pluginId); + const refusal = beforeCall && workers && plugin?.status === "ready" && settings?.enabled !== false && + isWorkspaceProvider(plugin.manifestJson) ? workspaceWorkerRefusal(pluginId, worker) : null; + if (refusal) throw refusal; if ( !workers || !plugin || plugin.status !== "ready" || settings?.enabled === false || !isWorkspaceProvider(plugin.manifestJson) || - !workers - .getWorker(pluginId) - ?.supportedMethods?.includes("workspaceOperation") + !worker?.supportedMethods?.includes("workspaceOperation") ) { throw conflict("The workspace storage plugin is unavailable"); } @@ -97,7 +102,7 @@ export async function callWorkspaceProvider( const pluginId = binding?.pluginId ?? input.pluginId; if (!pluginId || !workspace.projectId) throw conflict("Workspace storage ownership is missing"); - const selectedProvider = await provider(db, workers, workspace.companyId, pluginId); + const selectedProvider = await provider(db, workers, workspace.companyId, pluginId, true); if (input.operation.kind === "prepare" && input.operation.fresh && selectedProvider.manifestJson.workspaceProvider?.freshWorkspaceVersion !== 1) throw conflict("Fresh workspace preparation is unavailable"); if (input.operation.kind === "finalize" && @@ -147,10 +152,10 @@ export async function callWorkspaceProvider( params, operation.kind === "finalize" ? 40_000 : 310_000, ); - } catch { - throw conflict( - "Workspace storage operation is incomplete; retry the same workspace", - ); + } 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. + throw workspaceProviderCallFailure(error); } // Recheck enablement and ownership after an external effect, including an ambiguous retry. await provider(db, workers, workspace.companyId, pluginId); @@ -269,12 +274,7 @@ export async function preparePluginWorkspace( throw conflict( "An existing local workspace cannot change storage providers", ); - const selectedProvider = await provider( - db, - workers, - input.companyId, - pluginId, - ); + const selectedProvider = await provider(db, workers, input.companyId, pluginId, true); const declaration = selectedProvider.manifestJson.workspaceProvider!; if (input.fresh && declaration.freshWorkspaceVersion !== 1) throw conflict("A fresh attempt is unavailable for this workspace storage provider. The existing workspace is retained.", { reason: "fresh_workspace_storage_unsupported" }); diff --git a/server/src/services/workspace-provider-failure.ts b/server/src/services/workspace-provider-failure.ts new file mode 100644 index 00000000000..8db7319a0f1 --- /dev/null +++ b/server/src/services/workspace-provider-failure.ts @@ -0,0 +1,108 @@ +import { + JsonRpcCallError, + PLUGIN_RPC_ERROR_CODES, + readWorkspaceOperationFailureReason, + type PluginWorkspaceOperationFailureReason, +} from "@paperclipai/plugin-sdk"; +import { HttpError } from "../errors.js"; +import { findPluginWorkerUnavailableError, PluginWorkerUnavailableError } from "./plugin-worker-unavailable-error.js"; + +/** + * How a `workspaceOperation` call failed. + * - `worker_unavailable`: the worker was not running, so no request was sent. + * - `timeout`: no reply arrived in time; the operation may still have run. + * - `worker_exited`: the worker stopped with the call pending; the operation may have run. + * - `transport`: any other failure before a reply; the outcome is unknown. + * - `provider`: the provider replied with an error. + */ +export type WorkspaceProviderFailureKind = "worker_unavailable" | "timeout" | "worker_exited" | "transport" | "provider"; + +const INCOMPLETE = "Workspace storage operation is incomplete; retry the same workspace"; + +/** Core owns every message. Provider text and values never reach the run, the log or the API. */ +const REASONS: Record = { + unsupported_operation: { retryable: false, + message: "Workspace storage provider does not support this workspace operation" }, + configuration_unavailable: { retryable: false, + message: "Workspace storage provider configuration is unavailable; an operator must fix it before retrying" }, + credential_unavailable: { retryable: false, + message: "Workspace storage provider cannot use the project's Git credential; reconnect it before retrying" }, + storage_grant_unavailable: { retryable: false, + message: "Project storage access is unavailable or out of date; reconcile the project's storage before retrying" }, + binding_mismatch: { retryable: false, + message: "Saved workspace storage no longer matches this workspace; retrying the same workspace will not help" }, + storage_pending: { retryable: true, + message: "Project storage is still being prepared; retry the same workspace shortly" }, + runtime_unavailable: { retryable: true, + message: "The selected runtime is not ready; retry the same workspace once it is ready" }, + runtime_connection_invalid: { retryable: false, + message: "The selected runtime's saved connection is invalid; reprovision the runtime before retrying" }, + remote_incomplete: { retryable: true, message: INCOMPLETE }, + context_changed: { retryable: true, + message: "Workspace storage changed during the operation; retry the same workspace" }, + deadline_exceeded: { retryable: true, + message: "Workspace storage operation expired; retry the same workspace" }, +}; + +/** A failed `workspaceOperation` call, with a bounded kind and reason. */ +export class WorkspaceProviderFailureError extends HttpError { + readonly kind: WorkspaceProviderFailureKind; + readonly reason: PluginWorkspaceOperationFailureReason | null; + readonly retryable: boolean; + + constructor(input: { kind: WorkspaceProviderFailureKind; reason?: PluginWorkspaceOperationFailureReason | null; + retryable: boolean; message: string; cause?: unknown }) { + const reason = input.reason ?? null; + super(409, input.message, { code: "workspace_provider_failed", kind: input.kind, reason, retryable: input.retryable }); + this.name = "WorkspaceProviderFailureError"; + this.kind = input.kind; + this.reason = reason; + this.retryable = input.retryable; + if (input.cause !== undefined) this.cause = input.cause; + } + + /** Bounded fields for run results and logs. */ + summary() { + return { kind: this.kind, reason: this.reason, retryable: this.retryable }; + } +} + +/** + * Classify a rejected `workspaceOperation` call. A provider error never passes + * its message or data through; only a reason from the SDK's closed set does. + */ +export function workspaceProviderCallFailure(error: unknown): WorkspaceProviderFailureError { + const unavailable = findPluginWorkerUnavailableError(error); + // Keep the refusal as the cause so setup can retry it: no request reached the provider. + if (unavailable) return new WorkspaceProviderFailureError({ kind: "worker_unavailable", retryable: true, + message: "Workspace storage provider is not running; retry the same workspace", cause: unavailable }); + if (!(error instanceof JsonRpcCallError)) return new WorkspaceProviderFailureError({ kind: "transport", retryable: true, message: INCOMPLETE }); + if (error.code === PLUGIN_RPC_ERROR_CODES.TIMEOUT) return new WorkspaceProviderFailureError({ kind: "timeout", retryable: true, + message: "Workspace storage operation timed out; retry the same workspace" }); + if (error.code === PLUGIN_RPC_ERROR_CODES.WORKER_UNAVAILABLE) return new WorkspaceProviderFailureError({ kind: "worker_exited", retryable: true, + message: "Workspace storage provider stopped during the operation; retry the same workspace" }); + const reason = readWorkspaceOperationFailureReason(error); + if (!reason) return new WorkspaceProviderFailureError({ kind: "provider", retryable: true, + message: "Workspace storage provider reported a failure; retry the same workspace" }); + return new WorkspaceProviderFailureError({ kind: "provider", reason, ...REASONS[reason] }); +} + +/** + * Before a call to a ready provider: a worker that is missing, or not running and without + * `workspaceOperation`, refuses the call retryably. A restart removes or replaces the worker + * while the plugin stays ready, and its methods are unknown until it runs. A running worker + * without the method is an unsupported provider, so this returns null for it. + */ +export function workspaceWorkerRefusal(pluginId: string, worker: { status?: string; supportedMethods?: readonly string[] } | undefined) { + if (worker?.supportedMethods?.includes("workspaceOperation") || worker?.status === "running") return null; + return workspaceProviderCallFailure(new PluginWorkerUnavailableError({ pluginId, workerStatus: worker?.status ?? null, + message: `Cannot call "workspaceOperation" — worker for "${pluginId}" is ${worker?.status ?? "not registered"}` })); +} + +/** Find a provider call failure through setup's error wrappers. */ +export function findWorkspaceProviderFailure(error: unknown): WorkspaceProviderFailureError | null { + for (let cause = error, depth = 0; cause && depth < 4; cause = (cause as { cause?: unknown }).cause, depth++) { + if (cause instanceof WorkspaceProviderFailureError) return cause; + } + return null; +}