diff --git a/doc/native-workspace-finalization-recovery.md b/doc/native-workspace-finalization-recovery.md index 1628394d481..d1914ccbe21 100644 --- a/doc/native-workspace-finalization-recovery.md +++ b/doc/native-workspace-finalization-recovery.md @@ -407,6 +407,10 @@ and startup cancellation. Before dispatch, the controller stores a separate and original plugin. The initial cleanup and restart sweep both require that plugin to advertise `environmentStopLease`. Missing capability or failed stop leaves pending cleanup; ordinary release and destroy are never fallback methods. +A typed terminal stop failure (`PluginEnvironmentStopError` with reason +`resource_unrecoverable` or `recovery_failed`) escalates the lease for operator +attention on its first attempt and uses the long retry interval. The lease stays +in pending cleanup. A matching stopped receipt clears the pending intent, leaves the lease released and resumable, and saves `sandboxStopAndRetainReceipt`. Its request identity, diff --git a/doc/plugins/SANDBOX_PROVIDER_CAPABILITIES.md b/doc/plugins/SANDBOX_PROVIDER_CAPABILITIES.md index 5c8eb927d2a..796b487ebbd 100644 --- a/doc/plugins/SANDBOX_PROVIDER_CAPABILITIES.md +++ b/doc/plugins/SANDBOX_PROVIDER_CAPABILITIES.md @@ -213,3 +213,5 @@ reintroduce a concurrency capability when a runtime path enforces it. Native export recovery requires the separately discovered `environmentStopLease` RPC (`onEnvironmentStopLease`). It stops the exact allocation and preserves files regardless of its normal release policy. A failed or unconfirmed stop must throw and must never fall back to deletion. The host uses the recorded plugin ID and defers when that worker does not advertise the hook; a generic release method is insufficient evidence of stop-only support. Normal successful ephemeral release remains destructive after verified copyback. Daytona stop-only preservation disables provider auto-delete and refreshes the provider record to confirm the disabled policy before stopping. An unavailable or unconfirmed policy leaves cleanup pending; it never falls back to stop or delete. The original ephemeral destroy policy applies only after exact accepted-result copyback and commitment. + +A provider can refuse Stop while the allocation is in a provider error state. A stop-only hook may run one bounded provider recovery and then stop. When it cannot stop the allocation, it throws `PluginEnvironmentStopError` with a closed reason. The host treats `resource_unrecoverable` and `recovery_failed` as terminal: it escalates the pending cleanup at once and retries at its slowest interval. Deletion is never a fallback. diff --git a/packages/plugins/sandbox-providers/daytona/README.md b/packages/plugins/sandbox-providers/daytona/README.md index db72ee3b15e..087c1dd581e 100644 --- a/packages/plugins/sandbox-providers/daytona/README.md +++ b/packages/plugins/sandbox-providers/daytona/README.md @@ -36,6 +36,7 @@ Notes: - A session log socket can close while its command still runs. The driver requires a recorded command exit before returning a completed execution. It reconnects the log stream once, then polls status and log snapshots at most once per second. Replayed output is removed by byte offset, new output continues to reach the host, and the command is never dispatched again. A socket that delivers no output for 15 seconds switches directly to polling, even if the socket stays open or its initial connection never completes. Late callbacks from that socket are ignored. This recovers pending permission requests and cancellation results from saved output without replaying the command. Healthy initial and reconnected streams retain their existing lifetime under the caller’s RPC/run guard. After a clean close or an idle socket, each status or snapshot read gets a fresh operation timeout; successful recovery polling does not impose a new command lifetime. When both stream attempts fail without a clean close, fallback retains its original timeout budget starting after the stream attempts. - The SDK returns full log snapshots, so fallback bandwidth grows with retained output; it has no paged snapshot or stream cancellation API. A recovery-read timeout keeps partial output and explicitly reports whether command exit remains unconfirmed; it does not prove the remote command stopped. Provider/session teardown remains responsible for closing an outstanding socket. A final snapshot reconciles bytes missed by the stream before returning a recorded exit. - A sandbox record can survive the loss of its underlying container. Resume treats it as expired only when a fresh provider read confirms the exact missing-container error for that sandbox and marks it unrecoverable. Unknown errors and failed confirmation reads preserve the lease. The host still requires a verified native-runner backup before replacement. +- Daytona rejects Stop for a sandbox in its `error` state, for example after the sandbox disk fills. The stop-only hook confirms the retention policy first. If Daytona marks the error recoverable, the hook calls Recover once and waits at most 20 seconds for the sandbox to start. Then it stops the sandbox normally. The hook never calls Recover a second time in the same call. An unrecoverable error, a sandbox that is back in `error` after recovery, or an unfinished recovery returns a typed `PluginEnvironmentStopError` instead of the generic stop error. The sandbox is never deleted. - Directory export normally builds a sandbox archive and downloads it. If archive creation exhausts disk space, quota, or the file-size limit, export streams the archive through a raw terminal instead. Each bounded chunk waits for a host write acknowledgement. Import still requires successful tar completion and the existing archive/path validation. The total archive cannot exceed the provider-reported sandbox disk allocation, and each host write must leave a 1 GiB free-space reserve. This fallback writes no sandbox archive or command log, preserves source files, and retains the configured transfer timeout. Terminal cleanup has a separate five-second limit and cannot hold the host archive open. It requires Bash and GNU `tar`, `dd`, and `base64` in the sandbox. It does not repair a provider session that already failed to save its state or increase the sandbox's disk allocation. - Single-file export normally downloads a guarded sandbox copy. If that copy exhausts disk space, quota, or the file-size limit, the provider removes partial copies and streams the source through the same bounded sandbox-user channel. The stream rechecks canonical confinement and regular-file identity, refuses a symlink at open, and publishes host files only after a successful transfer receipt. It never hands a mutable source to the privileged download API. Permission, confinement, and transport failures still fail; a failed stream leaves prior host files intact. This removes the second archive-sized copy needed to download an already captured workspace archive. Creating the workspace archive itself still needs sandbox space. diff --git a/packages/plugins/sandbox-providers/daytona/src/plugin.test.ts b/packages/plugins/sandbox-providers/daytona/src/plugin.test.ts index 2eb882a76c3..e853a9df28d 100644 --- a/packages/plugins/sandbox-providers/daytona/src/plugin.test.ts +++ b/packages/plugins/sandbox-providers/daytona/src/plugin.test.ts @@ -37,7 +37,7 @@ import plugin, { __setDaytonaPluginContextForTest, } from "./plugin.js"; import type { PluginContext } from "@paperclipai/plugin-sdk"; -import { environmentSyncErrorData, readEnvironmentSyncErrorDiagnostic, withEnvironmentSyncErrorCapture, environmentCreationCleanupErrorData, readEnvironmentCreationCleanupError, readEnvironmentAcquisitionDiagnostic } from "@paperclipai/plugin-sdk"; +import { environmentSyncErrorData, readEnvironmentSyncErrorDiagnostic, withEnvironmentSyncErrorCapture, environmentCreationCleanupErrorData, readEnvironmentCreationCleanupError, readEnvironmentAcquisitionDiagnostic, PluginEnvironmentStopError, environmentStopErrorData, readEnvironmentStopFailure } from "@paperclipai/plugin-sdk"; import manifest from "./manifest.js"; import { parseTarVerboseListingLine, splitLinkEntryOnce } from "./file-sync.js"; import * as duplexCommands from "./duplex-command-stream.js"; @@ -2590,6 +2590,95 @@ describe("Daytona sandbox provider plugin", () => { expect(sandbox.delete).not.toHaveBeenCalled(); }); + describe("explicit stop of a sandbox in the provider error state", () => { + const stopParams = (sandboxId: string) => ({ driverKey: "daytona", companyId: "company-1", environmentId: "env-1", + providerLeaseId: sandboxId, config: { reuseLease: false } }); + function erroredSandbox(recoverable: boolean) { + const sandbox = { ...createMockSandbox({ id: "sandbox-disk-full", state: "error", recoverable }), + errorReason: "Error response from daemon: mkdir /var/lib/docker/overlay2/layer: no space left on device" }; + sandbox.stop.mockImplementation(async () => { + if (sandbox.state === "error") throw new Error("Sandbox failed to stop with status: error"); + sandbox.state = "stopped"; + }); + mockGet.mockResolvedValue(sandbox); + return sandbox; + } + async function stopFailure(sandboxId: string) { + const error = await plugin.definition.onEnvironmentStopLease!(stopParams(sandboxId)).then( + () => { throw new Error("stop unexpectedly succeeded"); }, (caught: unknown) => caught); + expect(error).toBeInstanceOf(PluginEnvironmentStopError); + return readEnvironmentStopFailure({ data: environmentStopErrorData(error) }); + } + beforeEach(() => { process.env.DAYTONA_API_KEY = "host-key"; }); + + it("recovers a recoverable sandbox, then stops and retains it", async () => { + const sandbox = erroredSandbox(true); + sandbox.recover.mockImplementation(async () => { sandbox.state = "started"; }); + await expect(plugin.definition.onEnvironmentStopLease!(stopParams(sandbox.id))) + .resolves.toEqual({ providerLeaseId: sandbox.id, state: "stopped" }); + expect(sandbox.recover).toHaveBeenCalledExactlyOnceWith(20); + expect(sandbox.setAutoDeleteInterval.mock.invocationCallOrder[0]).toBeLessThan(sandbox.recover.mock.invocationCallOrder[0]); + expect(sandbox.recover.mock.invocationCallOrder[0]).toBeLessThan(sandbox.stop.mock.invocationCallOrder[0]); + expect(sandbox.stop).toHaveBeenCalledOnce(); + expect(sandbox.state).toBe("stopped"); + for (const method of [sandbox.start, sandbox.delete, sandbox.archive]) expect(method).not.toHaveBeenCalled(); + }); + + it("returns a typed terminal failure for an unrecoverable sandbox without provider lifecycle calls", async () => { + const sandbox = erroredSandbox(false); + expect(await stopFailure(sandbox.id)).toEqual({ reason: "resource_unrecoverable", terminal: true }); + expect(sandbox.setAutoDeleteInterval).toHaveBeenCalledWith(-1); + for (const method of [sandbox.recover, sandbox.stop, sandbox.start, sandbox.delete, sandbox.archive]) { + expect(method).not.toHaveBeenCalled(); + } + }); + + it.each([ + { name: "recovery fails back to the error state", state: "error", recoverable: true, + expected: { reason: "recovery_failed", terminal: true } }, + { name: "the provider marks the sandbox unrecoverable during recovery", state: "error", recoverable: false, + expected: { reason: "resource_unrecoverable", terminal: true } }, + { name: "recovery is still in progress at the deadline", state: "starting", recoverable: true, + expected: { reason: "recovery_incomplete", terminal: false } }, + ])("makes one recovery attempt when $name", async ({ state, recoverable, expected }) => { + const sandbox = erroredSandbox(true); + sandbox.recover.mockRejectedValue(new Error("Sandbox failed to start with status: error")); + sandbox.refreshData.mockImplementation(async () => { + if (sandbox.recover.mock.calls.length > 0) Object.assign(sandbox, { state, recoverable }); + }); + expect(await stopFailure(sandbox.id)).toEqual(expected); + expect(sandbox.recover).toHaveBeenCalledOnce(); + expect(sandbox.stop).not.toHaveBeenCalled(); + expect(sandbox.delete).not.toHaveBeenCalled(); + }); + + it("bounds a recovery call that never settles", async () => { + const sandbox = erroredSandbox(true); + sandbox.recover.mockImplementation(() => new Promise(() => {})); + vi.useFakeTimers(); + try { + const failure = stopFailure(sandbox.id); + await vi.advanceTimersByTimeAsync(25_000); + expect(await failure).toEqual({ reason: "recovery_failed", terminal: true }); + } finally { vi.useRealTimers(); } + expect(sandbox.recover).toHaveBeenCalledOnce(); + expect(sandbox.stop).not.toHaveBeenCalled(); + }); + + it("does not recover again when the recovered sandbox fails again before it stops", async () => { + const sandbox = erroredSandbox(true); + sandbox.recover.mockImplementation(async () => { sandbox.state = "started"; }); + sandbox.stop.mockImplementation(async () => { + sandbox.state = "error"; + throw new Error("Sandbox failed to stop with status: error"); + }); + expect(await stopFailure(sandbox.id)).toEqual({ reason: "recovery_failed", terminal: true }); + expect(sandbox.recover).toHaveBeenCalledOnce(); + expect(sandbox.stop).toHaveBeenCalledOnce(); + expect(sandbox.delete).not.toHaveBeenCalled(); + }); + }); + it.each([true, false])("preserves an archived source only after fresh retention confirmation: %s", async confirmed => { process.env.DAYTONA_API_KEY = "host-key"; const sandbox = createMockSandbox({ id: "sandbox-already-archived", state: "archived" }); diff --git a/packages/plugins/sandbox-providers/daytona/src/plugin.ts b/packages/plugins/sandbox-providers/daytona/src/plugin.ts index 85674cf0bcb..97c0aa2f0d4 100644 --- a/packages/plugins/sandbox-providers/daytona/src/plugin.ts +++ b/packages/plugins/sandbox-providers/daytona/src/plugin.ts @@ -9,7 +9,7 @@ import type { Resources, Sandbox, } from "@daytonaio/sdk"; -import { decodeChannelBytes, definePlugin, NOOP_PLUGIN_TRACER, PluginEnvironmentCreationCleanupError, PluginEnvironmentRunnerIngressError, readEnvironmentCreationCleanupError, withEnvironmentSyncTransferStep, preserveEnvironmentSyncErrorDiagnostic } from "@paperclipai/plugin-sdk"; +import { decodeChannelBytes, definePlugin, NOOP_PLUGIN_TRACER, PluginEnvironmentCreationCleanupError, PluginEnvironmentRunnerIngressError, PluginEnvironmentStopError, readEnvironmentCreationCleanupError, withEnvironmentSyncTransferStep, preserveEnvironmentSyncErrorDiagnostic } from "@paperclipai/plugin-sdk"; import type { PluginContext, PluginEnvironmentCreationCleanup, @@ -2520,6 +2520,34 @@ async function closeDaytonaDuplexChannelsForLease(providerLeaseId: string): Prom await Promise.allSettled(matches.map(closeDaytonaDuplexChannel)); } +const STOP_RECOVERY_TIMEOUT_SECONDS = 20; +const STOPPABLE_AFTER_RECOVERY_STATES: ReadonlySet = new Set(["started", "stopped", "archived"]); + +function erroredAfterRecovery(sandbox: Sandbox): PluginEnvironmentStopError { + return new PluginEnvironmentStopError(sandbox.recoverable === false ? "resource_unrecoverable" : "recovery_failed"); +} + +async function refreshedSandbox(sandbox: Sandbox, timeoutMs: number): Promise { + try { + await withLivenessTimeout("sandbox.refreshData", timeoutMs, () => sandbox.refreshData()); + return true; + } catch { + return false; + } +} + +async function recoverSandboxForStop(sandbox: Sandbox, timeoutSeconds: number, readTimeoutMs: number): Promise { + if (sandbox.recoverable !== true) throw new PluginEnvironmentStopError("resource_unrecoverable"); + try { + await withLivenessTimeout("sandbox.recover", timeoutSeconds * 1_000 + LIVENESS_START_TIMEOUT_MARGIN_MS, + () => sandbox.recover(timeoutSeconds)); + } catch { + if (!await refreshedSandbox(sandbox, readTimeoutMs)) throw new PluginEnvironmentStopError("recovery_incomplete"); + } + if (sandbox.state === "error") throw erroredAfterRecovery(sandbox); + if (!STOPPABLE_AFTER_RECOVERY_STATES.has(sandbox.state ?? "")) throw new PluginEnvironmentStopError("recovery_incomplete"); +} + async function stopLease(params: PluginEnvironmentReleaseLeaseParams): Promise { if (!params.providerLeaseId) throw new Error("Daytona stop requires an exact sandbox identity"); const config = parseDriverConfig(params.config); @@ -2539,10 +2567,20 @@ async function stopLease(params: PluginEnvironmentReleaseLeaseParams): Promise

sandbox.setAutoDeleteInterval(-1)); await withLivenessTimeout("sandbox.refreshData", deadline, () => sandbox.refreshData()); if (sandbox.autoDeleteInterval !== -1) throw new Error("Daytona retention policy was not confirmed"); + const stopSeconds = Math.min(toTimeoutSeconds(config.timeoutMs), 30); + const recovering = sandbox.state === "error"; + if (recovering) await recoverSandboxForStop(sandbox, Math.min(stopSeconds, STOP_RECOVERY_TIMEOUT_SECONDS), deadline); // An archive already has no live process and retains its filesystem. // Daytona rejects Stop in this state; leave the confirmed archive cold. if (sandbox.state !== "stopped" && sandbox.state !== "archived") { - await terminateAtProvider(scope, "sandbox.stop", () => sandbox.stop(Math.min(toTimeoutSeconds(config.timeoutMs), 30))); + try { + await terminateAtProvider(scope, "sandbox.stop", () => sandbox.stop(stopSeconds)); + } catch (error) { + if (recovering && await refreshedSandbox(sandbox, deadline) && sandbox.state === "error") { + throw erroredAfterRecovery(sandbox); + } + throw error; + } } sandboxHandleSessionStore.clear(scope); await closeDaytonaDuplexChannelsForLease(params.providerLeaseId); diff --git a/packages/plugins/sdk/README.md b/packages/plugins/sdk/README.md index 683aa80464b..c72a1f49798 100644 --- a/packages/plugins/sdk/README.md +++ b/packages/plugins/sdk/README.md @@ -64,6 +64,24 @@ from error text. Unknown and configuration failures keep their normal errors. The envelope alone grants no retry, accounting or cleanup authority; native execution also requires its complete, current controller checkpoint proof. +## Stop failures + +Environment providers can throw `PluginEnvironmentStopError` from +`onEnvironmentStopLease` when they cannot stop a resource that is in a provider +error state. The error takes one closed reason: + +- `resource_unrecoverable`: the provider reports that the resource cannot recover. +- `recovery_failed`: the resource is in an error state again after one recovery attempt. +- `recovery_incomplete`: recovery did not finish before the stop deadline. + +The worker forwards only the reason. It omits messages, provider error text and +causes. The host reads it with `readEnvironmentStopFailure`, which derives the +`terminal` flag from the reason. The first two reasons are terminal: another +automatic attempt cannot succeed without provider or operator action. The host +escalates a terminal failure at once and retries it at its slowest cleanup +interval. It never abandons the pending cleanup. Other stop errors keep their +normal retry behavior. + ## Worker quick start ```ts diff --git a/packages/plugins/sdk/src/environment-stop-error.ts b/packages/plugins/sdk/src/environment-stop-error.ts new file mode 100644 index 00000000000..b0e03f8aa2b --- /dev/null +++ b/packages/plugins/sdk/src/environment-stop-error.ts @@ -0,0 +1,50 @@ +const SCHEMA = "paperclip/environment-stop-error/v1"; +const REASONS = ["resource_unrecoverable", "recovery_failed", "recovery_incomplete"] as const; +const TERMINAL_REASONS: ReadonlySet = new Set(["resource_unrecoverable", "recovery_failed"]); + +export type PluginEnvironmentStopFailureReason = typeof REASONS[number]; + +export interface PluginEnvironmentStopFailure { + reason: PluginEnvironmentStopFailureReason; + terminal: boolean; +} + +const MESSAGES: Record = { + resource_unrecoverable: "The provider reports that the environment resource cannot recover from its error state, so it cannot be stopped.", + recovery_failed: "The environment resource returned to an error state after a recovery attempt, so it cannot be stopped.", + recovery_incomplete: "The environment resource did not finish recovery from its error state before the stop deadline.", +}; + +function field(value: unknown, name: string): unknown { + if (!value || typeof value !== "object") return undefined; + try { return Object.getOwnPropertyDescriptor(value, name)?.value; } + catch { return undefined; } +} + +function validate(value: unknown): PluginEnvironmentStopFailure | null { + const reason = field(value, "reason"); + if (!REASONS.includes(reason as PluginEnvironmentStopFailureReason)) return null; + return { reason: reason as PluginEnvironmentStopFailureReason, terminal: TERMINAL_REASONS.has(reason as string) }; +} + +export class PluginEnvironmentStopError extends Error { + readonly failure: PluginEnvironmentStopFailure | null; + + constructor(reason: PluginEnvironmentStopFailureReason) { + const failure = validate({ reason }); + super(failure ? MESSAGES[failure.reason] : "The environment resource could not be stopped."); + this.name = "PluginEnvironmentStopError"; + this.failure = failure; + } +} + +export function environmentStopErrorData(error: unknown): unknown { + if (!(error instanceof PluginEnvironmentStopError)) return undefined; + const failure = validate(field(error, "failure")); + return failure ? { schema: SCHEMA, failure: { reason: failure.reason } } : undefined; +} + +export function readEnvironmentStopFailure(error: unknown): PluginEnvironmentStopFailure | null { + const data = field(error, "data"); + return field(data, "schema") === SCHEMA ? validate(field(data, "failure")) : null; +} diff --git a/packages/plugins/sdk/src/index.ts b/packages/plugins/sdk/src/index.ts index 8a6c4e97074..2c76614c96b 100644 --- a/packages/plugins/sdk/src/index.ts +++ b/packages/plugins/sdk/src/index.ts @@ -464,6 +464,8 @@ export { preserveEnvironmentSyncErrorDiagnostic, environmentSyncErrorData, readE export type { PluginEnvironmentSyncErrorDiagnostic } from "./environment-sync-error.js"; export { PluginEnvironmentRunnerIngressError, environmentRunnerIngressErrorData, readEnvironmentRunnerIngressFailure } from "./environment-runner-ingress-error.js"; export type { PluginEnvironmentRunnerIngressFailure } from "./environment-runner-ingress-error.js"; +export { PluginEnvironmentStopError, environmentStopErrorData, readEnvironmentStopFailure } from "./environment-stop-error.js"; +export type { PluginEnvironmentStopFailure, PluginEnvironmentStopFailureReason } from "./environment-stop-error.js"; export type { AiConnectionPool, AiConnectionPoolConfig, AiConnectionPoolMember, AiConnectionRouterRequest, AiConnectionRouterResult, AiConnectionRouterSelection } from "@paperclipai/shared"; diff --git a/packages/plugins/sdk/src/worker-rpc-host.ts b/packages/plugins/sdk/src/worker-rpc-host.ts index 53ca756118c..ac15da62a11 100644 --- a/packages/plugins/sdk/src/worker-rpc-host.ts +++ b/packages/plugins/sdk/src/worker-rpc-host.ts @@ -7,6 +7,7 @@ import type { AiConnectionRouterRequest, AiConnectionRouterResult } from "@paper import { environmentCreationCleanupErrorData } from "./environment-creation-cleanup.js"; import { environmentSyncErrorData, withEnvironmentSyncErrorCapture } from "./environment-sync-error.js"; import { environmentRunnerIngressErrorData } from "./environment-runner-ingress-error.js"; +import { environmentStopErrorData } from "./environment-stop-error.js"; import { createInvocationTracer } from "./invocation-tracer.js"; import { createPluginIdleDrain } from "./idle-drain.js"; /** @@ -1646,7 +1647,8 @@ export function startWorkerRpcHost(options: WorkerRpcHostOptions): WorkerRpcHost method === "environmentAcquireLease" || method === "environmentDestroyLease" ? environmentCreationCleanupErrorData(err, method === "environmentAcquireLease") : method === "environmentSyncOut" ? environmentSyncErrorData(err) - : method === "environmentRunnerIngressEndpoint" ? environmentRunnerIngressErrorData(err) : undefined)); + : method === "environmentRunnerIngressEndpoint" ? environmentRunnerIngressErrorData(err) + : method === "environmentStopLease" ? environmentStopErrorData(err) : undefined)); } finally { done?.(); } diff --git a/packages/plugins/sdk/tests/environment-stop-error.test.ts b/packages/plugins/sdk/tests/environment-stop-error.test.ts new file mode 100644 index 00000000000..04aca29c6d8 --- /dev/null +++ b/packages/plugins/sdk/tests/environment-stop-error.test.ts @@ -0,0 +1,39 @@ +import { describe, expect, it } from "vitest"; +import { environmentStopErrorData, PluginEnvironmentStopError, readEnvironmentStopFailure } from "../src/environment-stop-error.js"; + +describe("environment stop failure contract", () => { + it.each([ + ["resource_unrecoverable", true], + ["recovery_failed", true], + ["recovery_incomplete", false], + ] as const)("serializes %s with its fixed terminal class", (reason, terminal) => { + const error = new PluginEnvironmentStopError(reason); + const data = environmentStopErrorData(error); + expect(data).toEqual({ schema: "paperclip/environment-stop-error/v1", failure: { reason } }); + expect(readEnvironmentStopFailure({ data })).toEqual({ reason, terminal }); + expect(error.cause).toBeUndefined(); + expect(error.message).not.toContain(reason); + }); + it("forwards only the SDK error class", () => { + const forged = Object.assign(new Error("private-detail"), { failure: { reason: "recovery_failed", terminal: true } }); + expect(environmentStopErrorData(forged)).toBeUndefined(); + expect(environmentStopErrorData(new Error("private-detail"))).toBeUndefined(); + }); + it.each([{}, null, { reason: "unknown" }, { reason: 1 }, { reason: "recovery_incomplete ", terminal: true }])("rejects unsupported evidence %j", value => { + expect(readEnvironmentStopFailure({ data: { schema: "paperclip/environment-stop-error/v1", failure: value } })).toBeNull(); + }); + it("derives the terminal class from the reason instead of trusting the sender", () => { + expect(readEnvironmentStopFailure({ data: { schema: "paperclip/environment-stop-error/v1", + failure: { reason: "recovery_incomplete", terminal: true } } })).toEqual({ reason: "recovery_incomplete", terminal: false }); + expect(readEnvironmentStopFailure({ data: { schema: "paperclip/environment-runner-ingress-error/v1", + failure: { reason: "recovery_failed" } } })).toBeNull(); + }); + it("does not invoke inherited or accessor fields", () => { + let reads = 0; + const getter = Object.defineProperty({}, "data", { get() { reads++; throw new Error("private"); } }); + expect(readEnvironmentStopFailure(getter)).toBeNull(); + expect(readEnvironmentStopFailure(Object.create({ data: environmentStopErrorData( + new PluginEnvironmentStopError("recovery_failed")) }))).toBeNull(); + expect(reads).toBe(0); + }); +}); diff --git a/packages/plugins/sdk/tests/worker-rpc-host.test.ts b/packages/plugins/sdk/tests/worker-rpc-host.test.ts index aa26cc23eb0..ae409e00ef6 100644 --- a/packages/plugins/sdk/tests/worker-rpc-host.test.ts +++ b/packages/plugins/sdk/tests/worker-rpc-host.test.ts @@ -23,6 +23,7 @@ import { type PluginInvocationContext, } from "../src/protocol.js"; import { isWorkerEntrypoint, startWorkerRpcHost } from "../src/worker-rpc-host.js"; +import { PluginEnvironmentStopError, readEnvironmentStopFailure } from "../src/environment-stop-error.js"; describe("isWorkerEntrypoint", () => { const tempRoots: string[] = []; @@ -355,6 +356,7 @@ describe("worker configChanged cross-tenant guard", () => { reject( Object.assign(new Error(response.error.message), { code: response.error.code, + data: response.error.data, }), ); return; @@ -418,6 +420,21 @@ describe("worker configChanged cross-tenant guard", () => { } finally { worker.stop(); } }); + it.each([true, false])("forwards only a typed stop failure to the host: %s", async typed => { + const worker = makeWorker(definePlugin({ async setup() {}, + async onEnvironmentStopLease() { + throw typed ? new PluginEnvironmentStopError("resource_unrecoverable") : new Error("private provider detail"); + }, + })); + try { + await worker.initialize(); + const error = await worker.callWorker("environmentStopLease", { driverKey: "fixture", companyId: "company", + environmentId: "environment", providerLeaseId: "allocation", config: {} }).catch((caught: unknown) => caught); + expect(readEnvironmentStopFailure(error)).toEqual(typed ? { reason: "resource_unrecoverable", terminal: true } : null); + expect(JSON.stringify((error as { data?: unknown }).data ?? null)).not.toContain("private"); + } finally { worker.stop(); } + }); + it.each([false, true])("advertises read-only inspection only when implemented: %s", async supported => { let inspections = 0, mutations = 0; const worker = makeWorker(definePlugin({ async setup() {}, diff --git a/server/src/__tests__/heartbeat-pending-cleanup-sweep.test.ts b/server/src/__tests__/heartbeat-pending-cleanup-sweep.test.ts index f89ccb90131..7bd1c6fb487 100644 --- a/server/src/__tests__/heartbeat-pending-cleanup-sweep.test.ts +++ b/server/src/__tests__/heartbeat-pending-cleanup-sweep.test.ts @@ -38,6 +38,7 @@ vi.mock("../middleware/logger.js", () => ({ import { logger } from "../middleware/logger.ts"; import { heartbeatService, type HeartbeatEnvironmentRuntime } from "../services/heartbeat.ts"; import { hasPendingRunCleanup } from "../services/execution-blocker.js"; +import { environmentStopErrorData, JsonRpcCallError, PLUGIN_RPC_ERROR_CODES, PluginEnvironmentStopError } from "@paperclipai/plugin-sdk"; const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; @@ -355,6 +356,42 @@ describeEmbeddedPostgres("heartbeat sweepPendingCleanupLeases", () => { expect(Number(saved.metadata?.pendingCleanupRetryAfterMs)).toBeGreaterThan(Date.now() + 29 * 60_000); }); + it.each([ + { reason: "resource_unrecoverable", escalated: true }, + { reason: "recovery_failed", escalated: true }, + { reason: "recovery_incomplete", escalated: false }, + ] as const)("escalates only a terminal provider stop failure on its first attempt: $reason", async ({ reason, escalated }) => { + const { companyId, environmentId } = await seedCompanyAndEnvironment(); + const leaseId = await insertOrphanEphemeralLease({ companyId, environmentId, updatedAt: new Date(Date.now() - 60 * 60_000) }); + const failure = new JsonRpcCallError({ code: PLUGIN_RPC_ERROR_CODES.WORKER_ERROR, message: "private provider detail", + data: environmentStopErrorData(new PluginEnvironmentStopError(reason)) }); + const teardown = vi.fn(async () => { throw failure; }); + const service = heartbeatService(db, { environmentRuntime: { + retryPendingSandboxTeardown: teardown, + } as unknown as HeartbeatEnvironmentRuntime }); + + expect(await service.sweepPendingCleanupLeases()).toEqual({ swept: 1, destroyed: 0, capped: escalated ? 1 : 0 }); + expect(teardown).toHaveBeenCalledOnce(); + const [saved] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, leaseId)); + expect(saved.status).toBe("pending_cleanup"); + expect(saved.cleanupStatus).toBe("failed"); + expect(saved.metadata?.[ATTEMPTS_KEY]).toBe(1); + expect(saved.metadata?.[CAP_WARNED_KEY] === true).toBe(escalated); + const retryAfterMs = Number(saved.metadata?.pendingCleanupRetryAfterMs); + if (escalated) expect(retryAfterMs).toBeGreaterThan(Date.now() + 29 * 60_000); + else expect(retryAfterMs).toBeLessThan(Date.now() + 60_000); + + const warnings = vi.mocked(logger.warn).mock.calls; + expect(warnings.find(call => call[1] === "pending_cleanup lease retry failed")?.[0]).toMatchObject({ + leaseId, attempts: 1, providerStopFailure: reason, + errorKind: escalated ? "provider_stop_terminal" : "destroy_failed", + }); + const attention = warnings.filter(call => String(call[1]).startsWith("environment lease needs operator attention")); + expect(attention).toHaveLength(escalated ? 1 : 0); + if (escalated) expect(attention[0]?.[0]).toMatchObject({ leaseId, providerStopFailure: reason }); + expect(JSON.stringify(warnings)).not.toContain("private provider detail"); + }); + it("retries after a restart and cooldown when the provider recovers", async () => { const { companyId, environmentId } = await seedCompanyAndEnvironment(); const leaseId = await insertOrphanEphemeralLease({ companyId, environmentId, diff --git a/server/src/services/heartbeat/recovery.ts b/server/src/services/heartbeat/recovery.ts index d6ccf33bdc6..61aca9060f2 100644 --- a/server/src/services/heartbeat/recovery.ts +++ b/server/src/services/heartbeat/recovery.ts @@ -32,6 +32,7 @@ import { revokeExpiredLegacyControllerWithState, } from "../legacy-controller-lease.js"; import { remoteTerminationReceipt } from "../remote-execution-termination.js"; +import { readEnvironmentStopFailure, type PluginEnvironmentStopFailure } from "@paperclipai/plugin-sdk"; import { runUsedConversationAdapter } from "../conversation-continuation.js"; import { legacyExecutionNeedsReconciliationWithEvidence, terminalizeLegacyExecution } from "../legacy-execution-recovery.js"; import { reconcileRetainedWorkspaces } from "../workspace-recovery.js"; @@ -198,10 +199,13 @@ const ORPHANED_ACTIVE_LEASE_SWEEP_PAGE_SIZE = 20; // A provider or plugin destroy rejection can carry a bearer credential, a // signed URL, or provider response detail in its name, code, message, cause, or // stack. The exception fields cross the server boundary, so they are not a -// trusted enum. The pending_cleanup sweep logs never read the exception. Each -// catch site logs a constant, locally generated `errorKind` instead. +// trusted enum. The pending_cleanup sweep logs never read these fields. Each +// catch site logs a constant, locally generated `errorKind` instead, plus at most +// the closed stop-failure reason that `readEnvironmentStopFailure` validates. const PENDING_CLEANUP_RETRY_ERROR_KIND = "destroy_failed"; +const PENDING_CLEANUP_TERMINAL_STOP_ERROR_KIND = "provider_stop_terminal"; + const PENDING_CLEANUP_SWEEP_ERROR_KIND = "sweep_failed"; const ORPHANED_ACTIVE_LEASE_SWEEP_ERROR_KIND = "orphaned_active_lease_sweep_failed"; @@ -1438,10 +1442,10 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD // Atomically claim the one-time cap warning for a lease. The update only // matches when the lease is still pending_cleanup, its stored attempt count is - // at or above the cap, and it has not yet carried the warned flag. Two - // concurrent sweeps that both reach the cap race here, but only one update - // sets the flag and returns a row. The loser skips the warning. This keeps the - // warning to one log line per lease. + // at or above the cap or the provider reported a terminal stop failure, and it + // has not yet carried the warned flag. Two concurrent sweeps that both reach + // the cap race here, but only one update sets the flag and returns a row. The + // loser skips the warning. This keeps the warning to one log line per lease. // // The status and cap predicates are the last line of defense. They stop a warn // flag write to a lease that left pending_cleanup or dropped below the cap @@ -1449,6 +1453,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD // `jsonb_set`, so a concurrent write to an unrelated metadata key survives. async function claimPendingCleanupCapWarning( leaseId: string, + terminalProviderFailure = false, ): Promise { const now = new Date(); const claimed = await db @@ -1461,7 +1466,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD and( eq(environmentLeases.id, leaseId), eq(environmentLeases.status, "pending_cleanup"), - sql`${pendingCleanupAttemptsSql()} >= ${PENDING_CLEANUP_SWEEP_ATTEMPT_CAP}`, + terminalProviderFailure ? undefined : sql`${pendingCleanupAttemptsSql()} >= ${PENDING_CLEANUP_SWEEP_ATTEMPT_CAP}`, sql`${pendingCleanupCapWarnedSql()} = false`, ), ) @@ -1757,6 +1762,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD pendingCleanupAttemptsInFlight.add(row.id); lease.metadata = { ...lease.metadata, pendingCleanupAttemptId: claimed }; let activeRenewal: Promise | null = null; + let providerStopFailure: PluginEnvironmentStopFailure | null = null; const renewal = setInterval(() => { if (activeRenewal) return; activeRenewal = db.update(environmentLeases).set({ @@ -1803,7 +1809,8 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD destroyed += 1; } } - } catch { + } catch (error) { + providerStopFailure = readEnvironmentStopFailure(error); // The recorded-data teardown throws on failure, so revert the lease to // pending_cleanup for a later sweep. The claimed attempt still counts // for backoff, so requests stay bounded. The `destroyRunLease` @@ -1817,10 +1824,13 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD }); } // Log a constant errorKind only. The exception can carry a credential in - // its name, code, message, cause, or stack, so the sweep never reads it. + // its name, code, message, cause, or stack, so the sweep reads only the + // closed, validated provider stop-failure reason from it. logger.warn( { - errorKind: PENDING_CLEANUP_RETRY_ERROR_KIND, + errorKind: providerStopFailure?.terminal + ? PENDING_CLEANUP_TERMINAL_STOP_ERROR_KIND : PENDING_CLEANUP_RETRY_ERROR_KIND, + ...(providerStopFailure ? { providerStopFailure: providerStopFailure.reason } : {}), leaseId: row.id, environmentId: row.environmentId, attempts: attempts + 1, @@ -1833,15 +1843,17 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD // Late renewals are attempt-fenced and never write the retry cooldown. pendingCleanupAttemptsInFlight.delete(row.id); } - if (attempts + 1 >= PENDING_CLEANUP_SWEEP_ATTEMPT_CAP) { + const escalated = attempts + 1 >= PENDING_CLEANUP_SWEEP_ATTEMPT_CAP || providerStopFailure?.terminal === true; + if (escalated) { capped += 1; // Warn once, then continue automatic cleanup with backoff. The atomic claim // keeps the warning to one log line even when two sweeps overlap. if (metadata[PENDING_CLEANUP_CAP_WARNED_METADATA_KEY] !== true) { - const warned = await claimPendingCleanupCapWarning(row.id); + const warned = await claimPendingCleanupCapWarning(row.id, providerStopFailure?.terminal === true); if (warned) { logger.warn( - { leaseId: row.id, environmentId: row.environmentId, attempts }, + { leaseId: row.id, environmentId: row.environmentId, attempts, + ...(providerStopFailure?.terminal ? { providerStopFailure: providerStopFailure.reason } : {}) }, "environment lease needs operator attention; automatic cleanup continues with backoff", ); } @@ -1852,7 +1864,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD await writeEnvironmentCleanup(db, tx => tx.update(environmentLeases).set({ metadata: sql`${pendingCleanupMetadataObjectSql()} || ${JSON.stringify({ pendingCleanupInFlight: false, - pendingCleanupRetryAfterMs: Date.now() + (attempts + 1 >= PENDING_CLEANUP_SWEEP_ATTEMPT_CAP + pendingCleanupRetryAfterMs: Date.now() + (escalated ? 30 * 60_000 : Math.min(30 * 60_000, Math.max(30_000, backoffMs))), })}::jsonb`, }).where(and(eq(environmentLeases.id, lease.id),