From 8080620fc9d8c35748835e037890d2acf2197555 Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Thu, 5 Nov 2026 16:09:32 -0800 Subject: [PATCH] fix(workspaces): size archive deadlines and preserve resume budgets (#19364) Give large workspace archives a bounded per-transfer deadline without changing provider settings. Preserve configured post-upload command limits and the saved provider budget when resuming an exact retained sandbox. Verification: 341 focused Core/SDK/adapter tests and 588 Daytona tests passed. Workspace and provider typecheck/build passed. Hosted CI: 54 successful checks, two intended Storybook skips. Greptile 5/5 with no unresolved review threads. The bounded full local run passed all six chat shards and stopped at twenty minutes during general server suites; that full local run remains incomplete. Co-Authored-By: Paperclip --- packages/adapter-utils/README.md | 20 ++++ .../src/command-managed-runtime.test.ts | 14 ++- .../src/command-managed-runtime.ts | 13 +- .../src/sandbox-managed-runtime.ts | 10 +- .../sandbox-providers/daytona/README.md | 6 + .../daytona/src/file-sync.ts | 3 +- .../daytona/src/plugin.test.ts | 57 +++++++++ .../sandbox-providers/daytona/src/plugin.ts | 21 +++- packages/plugins/sdk/src/protocol.ts | 2 + .../environment-sync-negotiation.test.ts | 5 + ...ment-execution-target-capabilities.test.ts | 13 +- .../src/__tests__/environment-runtime.test.ts | 45 ++++++- .../services/environment-execution-target.ts | 6 +- server/src/services/environment-runtime.ts | 15 ++- .../remote-workspace-transfer-timeout.test.ts | 112 ++++++++++++++++++ .../services/remote-workspace-transport.ts | 20 +++- 16 files changed, 334 insertions(+), 28 deletions(-) create mode 100644 server/src/services/remote-workspace-transfer-timeout.test.ts diff --git a/packages/adapter-utils/README.md b/packages/adapter-utils/README.md index b1e77d207d9..0484184c7ef 100644 --- a/packages/adapter-utils/README.md +++ b/packages/adapter-utils/README.md @@ -22,6 +22,26 @@ remote command stopped and does not replay an uncertain write. The process session closes failed input delivery and uses its existing shutdown path; normal execution settlement must still verify termination. +## Native archive transfer deadlines + +Native `syncIn` and `syncOut` calls accept an optional `minimumTimeoutMs` budget. The +server validates this budget and extends the worker RPC deadline. Providers can +honor this field for file transfers without changing their configuration. Daytona +uses the larger of this minimum and its configured timeout. Other providers keep +their existing behavior. The minimum is limited to 30 minutes and does not change +sandbox lifetimes, stored settings or the timeout for other commands. + +The remote-workspace archive path uses the verified capture size for downloads +and the local archive size for uploads. Large archives get one second per MiB +plus 60 seconds, capped at 30 minutes. Small archives keep the existing provider +timeout when that calculation is no greater than five minutes. Archive size, +checksum, confinement and host space checks still apply. + +An exact retained-sandbox resume keeps its recorded provider timeout and RPC +overhead. It uses the normal one-minute fallback only when no timeout was +recorded. A five-minute cold restore must not be cut off by that fallback. +Resume still targets the recorded allocation and does not acquire a replacement. + ## No-remote-git contract The local execution-workspace cwd is the only persistence boundary across diff --git a/packages/adapter-utils/src/command-managed-runtime.test.ts b/packages/adapter-utils/src/command-managed-runtime.test.ts index cc264f0cd35..48075cf5470 100644 --- a/packages/adapter-utils/src/command-managed-runtime.test.ts +++ b/packages/adapter-utils/src/command-managed-runtime.test.ts @@ -3,7 +3,7 @@ import os from "node:os"; import path from "node:path"; import { execFile as execFileCallback, spawn } from "node:child_process"; import { promisify } from "node:util"; -import { afterEach, describe, expect, it } from "vitest"; +import { afterEach, describe, expect, it, vi } from "vitest"; import { createCommandManagedRuntimeClient, @@ -690,6 +690,18 @@ describe("command managed runtime", () => { expect(result.operations[0]).toMatchObject({ operationId: "op-1", filesTransferred: 1 }); }); + it.each(["syncIn", "syncOut"] as const)("forwards a native %s budget without changing subsequent defaults", async method => { + const transfer = vi.fn(async (..._args: unknown[]) => ({ operations: [] })); + const execute = vi.fn(); + const runner: CommandManagedRuntimeRunner = { execute, syncIn: transfer, syncOut: transfer }; + const client = createCommandManagedRuntimeClient({ runner, commandCwd: "/", timeoutMs: 1 }); + const operations: SandboxSyncOperation[] = [{ operationId: "bounded-sync", files: [] }]; + await client[method]!(operations, { minimumTimeoutMs: 900_000 }); + await client[method]!(operations); + expect(transfer.mock.calls).toEqual([[operations, { minimumTimeoutMs: 900_000 }], [operations]]); + expect(execute).not.toHaveBeenCalled(); + }); + it("fallback syncIn tarballs+uploads a directory then runs post-upload commands in order", async () => { const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-syncin-fallback-")); cleanupDirs.push(rootDir); diff --git a/packages/adapter-utils/src/command-managed-runtime.ts b/packages/adapter-utils/src/command-managed-runtime.ts index 501859e8eca..8f152b71d3d 100644 --- a/packages/adapter-utils/src/command-managed-runtime.ts +++ b/packages/adapter-utils/src/command-managed-runtime.ts @@ -12,6 +12,7 @@ import { type SandboxManagedRuntimeClient, type SandboxRemoteExecutionSpec, type SandboxSyncOperation, + type SandboxSyncOptions, type SandboxSyncResult, type WorkspaceDurableSeedPaths, type WorkspaceInboundMode, @@ -116,9 +117,9 @@ export interface CommandManagedRuntimeRunner { * orchestrator either uses the native path for both directions or falls back * to the base64 transport for both. */ - syncIn?(operations: SandboxSyncOperation[]): Promise; + syncIn?(operations: SandboxSyncOperation[], options?: SandboxSyncOptions): Promise; /** Optional native outbound file transfer. See {@link syncIn}. */ - syncOut?(operations: SandboxSyncOperation[]): Promise; + syncOut?(operations: SandboxSyncOperation[], options?: SandboxSyncOptions): Promise; /** * Optional persistent duplex channel. Present only when the sandbox provider's * effective capability grants `duplexCommandStream`. The runner opens one @@ -528,15 +529,17 @@ export function createCommandManagedRuntimeClient(input: { client.allowConcurrentSyncOperations = hasNativeBoth ? input.runner.allowConcurrentSyncOperations === true : true; - client.syncIn = async (operations) => { + client.syncIn = async (operations, options) => { assertPostUploadCommandsConfined(operations); if (hasNativeBoth) { - return await nativeSyncIn!(operations); + return await (options === undefined ? nativeSyncIn!(operations) : nativeSyncIn!(operations, options)); } return await fallbackSyncIn(operations); }; if (hasNativeBoth) { - client.syncOut = (operations) => nativeSyncOut!(operations); + client.syncOut = (operations, options) => options === undefined + ? nativeSyncOut!(operations) + : nativeSyncOut!(operations, options); } return client; diff --git a/packages/adapter-utils/src/sandbox-managed-runtime.ts b/packages/adapter-utils/src/sandbox-managed-runtime.ts index b7f4270292b..d330ee6b820 100644 --- a/packages/adapter-utils/src/sandbox-managed-runtime.ts +++ b/packages/adapter-utils/src/sandbox-managed-runtime.ts @@ -451,6 +451,12 @@ export interface SandboxTransferProgressOptions { onProgress?: (transferredBytes: number, totalBytes: number | null) => void | Promise; } +export const MAX_SANDBOX_SYNC_TIMEOUT_MS = 30 * 60_000; + +export interface SandboxSyncOptions { + minimumTimeoutMs?: number; +} + /** * A single source→target file or directory transfer within a sync operation. * Mirrors the plugin SDK `PluginSyncFileMapping`; kept as a local structural @@ -561,9 +567,9 @@ export interface SandboxManagedRuntimeClient { * orchestrator falls back to the tar + base64 `writeFile`/`run` path so * behavior is byte-identical to a provider that never opted in. */ - syncIn?(operations: SandboxSyncOperation[]): Promise; + syncIn?(operations: SandboxSyncOperation[], options?: SandboxSyncOptions): Promise; /** Optional native outbound transfer. See {@link syncIn}. */ - syncOut?(operations: SandboxSyncOperation[]): Promise; + syncOut?(operations: SandboxSyncOperation[], options?: SandboxSyncOptions): Promise; } /** diff --git a/packages/plugins/sandbox-providers/daytona/README.md b/packages/plugins/sandbox-providers/daytona/README.md index aff28720d73..ac492dc7b3a 100644 --- a/packages/plugins/sandbox-providers/daytona/README.md +++ b/packages/plugins/sandbox-providers/daytona/README.md @@ -40,6 +40,12 @@ Notes: - 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. The sender can send up to sixteen 40 KiB chunks before it waits for their host write acknowledgements. The host writes them in order and rejects a sender that exceeds the pending-frame limit. This bounds queued payloads to 640 KiB while avoiding one network round trip per chunk. 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. +Native file transfers accept an optional `minimumTimeoutMs` from the host. The +minimum must be a positive whole number no greater than 30 minutes. File sync +uses the larger of this minimum and the configured timeout. Sandbox start and +command timeouts keep their configured limits. Without the field, behavior is +unchanged. + ## Local development ```bash diff --git a/packages/plugins/sandbox-providers/daytona/src/file-sync.ts b/packages/plugins/sandbox-providers/daytona/src/file-sync.ts index 8ea41f23363..90108c3a37f 100644 --- a/packages/plugins/sandbox-providers/daytona/src/file-sync.ts +++ b/packages/plugins/sandbox-providers/daytona/src/file-sync.ts @@ -1147,6 +1147,7 @@ export async function performSyncIn(input: { operations: PluginSyncOperation[]; remoteDir: string; timeoutSeconds: number; + commandTimeoutSeconds?: number; }): Promise { const operations: PluginEnvironmentSyncResult["operations"] = []; for (const operation of input.operations) { @@ -1182,7 +1183,7 @@ export async function performSyncIn(input: { sandbox: input.sandbox, commands: operation.postUploadCommands ?? [], remoteDir: input.remoteDir, - timeoutSeconds: input.timeoutSeconds, + timeoutSeconds: input.commandTimeoutSeconds ?? input.timeoutSeconds, }); operations.push({ operationId: operation.operationId, filesTransferred, bytesTransferred }); diff --git a/packages/plugins/sandbox-providers/daytona/src/plugin.test.ts b/packages/plugins/sandbox-providers/daytona/src/plugin.test.ts index 7a45aac70be..e3d412acfe1 100644 --- a/packages/plugins/sandbox-providers/daytona/src/plugin.test.ts +++ b/packages/plugins/sandbox-providers/daytona/src/plugin.test.ts @@ -5909,6 +5909,63 @@ describe("daytona native file-sync hooks", () => { expect(__getDaytonaWritableDirsForTest(params)).toEqual([]); }); + it.each((["syncIn", "syncOut"] as const).flatMap(method => [ + { minimumTimeoutMs: undefined, configured: 300_000, expected: 300 }, + { minimumTimeoutMs: 900_000, configured: 300_000, expected: 900 }, + { minimumTimeoutMs: 900_000, configured: 1_200_000, expected: 1200 }, + ].map(budget => ({ method, ...budget }))))("uses a $method minimum of $minimumTimeoutMs without changing later transfers", async ({ method, minimumTimeoutMs, configured, expected }) => { + mockCreate.mockClear(); + const root = await makeHostDir(); + const local = path.join(root, "archive.pcws"); + await fs.writeFile(local, "bytes"); + const sandbox = createMockSandbox(); + sandbox.fs.downloadFiles.mockImplementation((...args: unknown[]) => fulfilDownload(args)); + mockGet.mockResolvedValue(sandbox); + const params = method === "syncIn" + ? syncInParams({ operationId: "bounded-upload", sourcePath: local, targetPath: `${REMOTE_DIR}/archive.pcws` }) + : syncOutParams({ operationId: "bounded-download", sourcePath: `${REMOTE_DIR}/archive.pcws`, targetPath: local }); + params.config.timeoutMs = configured; + const hook = method === "syncIn" ? plugin.definition.onEnvironmentSyncIn! : plugin.definition.onEnvironmentSyncOut!; + await hook({ ...params, minimumTimeoutMs }); + const transfer = method === "syncIn" ? sandbox.fs.uploadFiles : sandbox.fs.downloadFiles; + expect(transfer.mock.calls[0]?.[1]).toBe(expected); + await hook(params); + expect(transfer.mock.calls[1]?.[1]).toBe(configured / 1000); + expect(params.config.timeoutMs).toBe(configured); + expect(mockCreate).not.toHaveBeenCalled(); + }); + + it.each([ + { minimumTimeoutMs: undefined, transferSeconds: 300 }, + { minimumTimeoutMs: 900_000, transferSeconds: 900 }, + ])("keeps command deadlines separate from a $minimumTimeoutMs transfer minimum", async ({ minimumTimeoutMs, transferSeconds }) => { + const root = await makeHostDir(); + const source = path.join(root, "archive.pcws"); + await fs.writeFile(source, "bytes"); + const sandbox = createMockSandbox(); + mockGet.mockResolvedValue(sandbox); + const params = syncInParams({ operationId: "transfer-with-commands", sourcePath: source, targetPath: `${REMOTE_DIR}/archive.pcws` }); + await plugin.definition.onEnvironmentSyncIn!({ + ...params, minimumTimeoutMs, + operations: [{ ...params.operations[0], postUploadCommands: [ + { command: "check-default-deadline" }, + { command: "check-explicit-deadline", timeoutMs: 2000 }, + ] }], + }); + expect(sandbox.fs.uploadFiles.mock.calls[0]?.[1]).toBe(transferSeconds); + expect(sandbox.process.executeCommand).toHaveBeenCalledWith("check-default-deadline", REMOTE_DIR, undefined, 300); + expect(sandbox.process.executeCommand).toHaveBeenCalledWith("check-explicit-deadline", REMOTE_DIR, undefined, 2); + }); + + it.each(["syncIn", "syncOut"] as const)("rejects an invalid $0 minimum before sandbox access", async method => { + const hook = method === "syncIn" ? plugin.definition.onEnvironmentSyncIn! : plugin.definition.onEnvironmentSyncOut!; + for (const minimumTimeoutMs of [0, -1, 1.5, Number.NaN, Number.POSITIVE_INFINITY, 30 * 60_000 + 1]) { + await expect(hook({ ...syncInParams({ operationId: "invalid-budget", sourcePath: "/host", targetPath: REMOTE_DIR }), + minimumTimeoutMs })).rejects.toThrow("Invalid sandbox sync timeout"); + } + expect(mockGet).not.toHaveBeenCalled(); + }); + it("syncIn coalesces file mappings into one uploadFiles batch to reserved temp destinations, then one batched mv, applying secret mode via setFilePermissions before the rename", async () => { const hostDir = await makeHostDir(); const secretSource = path.join(hostDir, "auth.json"); diff --git a/packages/plugins/sandbox-providers/daytona/src/plugin.ts b/packages/plugins/sandbox-providers/daytona/src/plugin.ts index 488e6480b4b..931dff2d7aa 100644 --- a/packages/plugins/sandbox-providers/daytona/src/plugin.ts +++ b/packages/plugins/sandbox-providers/daytona/src/plugin.ts @@ -476,6 +476,14 @@ function toTimeoutSeconds(timeoutMs: number): number { return Math.max(1, Math.ceil(timeoutMs / 1000)); } +function resolveSyncTimeoutSeconds(config: DaytonaDriverConfig, minimumTimeoutMs: number | undefined): number { + if (minimumTimeoutMs !== undefined && (!Number.isSafeInteger(minimumTimeoutMs) + || minimumTimeoutMs <= 0 || minimumTimeoutMs > 30 * 60_000)) { + throw new RangeError("Invalid sandbox sync timeout"); + } + return toTimeoutSeconds(Math.max(config.timeoutMs, minimumTimeoutMs ?? 0)); +} + function resolveTimeoutMs(paramsTimeoutMs: number | undefined, config: DaytonaDriverConfig): number { return paramsTimeoutMs != null && Number.isFinite(paramsTimeoutMs) && paramsTimeoutMs > 0 ? Math.trunc(paramsTimeoutMs) @@ -3696,14 +3704,14 @@ const plugin = definePlugin(trackDaytonaHooks({ // base64-over-exec fallback. Providers that do not define these keep the // byte-identical fallback. async onEnvironmentSyncIn( - params: PluginEnvironmentSyncInParams, + params: PluginEnvironmentSyncInParams & { minimumTimeoutMs?: number }, ): Promise { if (!params.lease.providerLeaseId) { throw new Error("Daytona syncIn requires a provider lease ID."); } const config = parseDriverConfig(params.config); const remoteDir = resolveSyncRemoteDir(params.lease); - const timeoutSeconds = toTimeoutSeconds(config.timeoutMs); + const timeoutSeconds = resolveSyncTimeoutSeconds(config, params.minimumTimeoutMs); const scope = { driverKey: params.driverKey, companyId: params.companyId, @@ -3716,12 +3724,13 @@ const plugin = definePlugin(trackDaytonaHooks({ sandboxHandleWritableDirs.recordWritableTargets(scope, params.operations); return await withSandboxActivityGate(scope, async () => { const sandbox = await getSandbox(scope, { bypassTeardownGate: true }); - await ensureSandboxStarted(sandbox, timeoutSeconds, scope); + await ensureSandboxStarted(sandbox, toTimeoutSeconds(config.timeoutMs), scope); const result = await performSyncIn({ sandbox, operations: params.operations, remoteDir, timeoutSeconds, + commandTimeoutSeconds: toTimeoutSeconds(config.timeoutMs), }); sandboxHandleCache.markFresh(scope); return result; @@ -3730,14 +3739,14 @@ const plugin = definePlugin(trackDaytonaHooks({ // Opt-in native outbound transfer. See onEnvironmentSyncIn. async onEnvironmentSyncOut( - params: PluginEnvironmentSyncOutParams, + params: PluginEnvironmentSyncOutParams & { minimumTimeoutMs?: number }, ): Promise { if (!params.lease.providerLeaseId) { throw new Error("Daytona syncOut requires a provider lease ID."); } const config = parseDriverConfig(params.config); const remoteDir = resolveSyncRemoteDir(params.lease); - const timeoutSeconds = toTimeoutSeconds(config.timeoutMs); + const timeoutSeconds = resolveSyncTimeoutSeconds(config, params.minimumTimeoutMs); const scope = { driverKey: params.driverKey, companyId: params.companyId, @@ -3749,7 +3758,7 @@ const plugin = definePlugin(trackDaytonaHooks({ return await withSandboxActivityGate(scope, async () => { const sandbox = await withEnvironmentSyncTransferStep("sandbox_access", async () => { const resolved = await getSandbox(scope, { bypassTeardownGate: true }); - await ensureSandboxStarted(resolved, timeoutSeconds, scope); + await ensureSandboxStarted(resolved, toTimeoutSeconds(config.timeoutMs), scope); return resolved; }); const result = await performSyncOut({ diff --git a/packages/plugins/sdk/src/protocol.ts b/packages/plugins/sdk/src/protocol.ts index 5ca50907f2b..3d70c65bd3d 100644 --- a/packages/plugins/sdk/src/protocol.ts +++ b/packages/plugins/sdk/src/protocol.ts @@ -898,11 +898,13 @@ export interface PluginSyncOperation { export interface PluginEnvironmentSyncInParams extends PluginEnvironmentDriverBaseParams { lease: PluginEnvironmentLease; operations: PluginSyncOperation[]; + minimumTimeoutMs?: number; } export interface PluginEnvironmentSyncOutParams extends PluginEnvironmentDriverBaseParams { lease: PluginEnvironmentLease; operations: PluginSyncOperation[]; + minimumTimeoutMs?: number; } /** Per-operation transfer accounting returned by a sync hook, for observability. */ diff --git a/packages/plugins/sdk/tests/environment-sync-negotiation.test.ts b/packages/plugins/sdk/tests/environment-sync-negotiation.test.ts index 383247023ae..5176c47efe2 100644 --- a/packages/plugins/sdk/tests/environment-sync-negotiation.test.ts +++ b/packages/plugins/sdk/tests/environment-sync-negotiation.test.ts @@ -109,11 +109,13 @@ describe("environment sync verb negotiation", () => { it("routes environmentSyncIn/environmentSyncOut to the hooks when defined", async () => { const seen: string[] = []; + const minimums: Array = []; const worker = startTestWorker( definePlugin({ async setup() {}, async onEnvironmentSyncIn(params): Promise { seen.push("in"); + minimums.push(params.minimumTimeoutMs); return { operations: params.operations.map((op) => ({ operationId: op.operationId, @@ -124,6 +126,7 @@ describe("environment sync verb negotiation", () => { }, async onEnvironmentSyncOut(params): Promise { seen.push("out"); + minimums.push(params.minimumTimeoutMs); return { operations: params.operations.map((op) => ({ operationId: op.operationId, @@ -141,6 +144,7 @@ describe("environment sync verb negotiation", () => { companyId: "company", environmentId: "env", config: {}, + minimumTimeoutMs: 900_000, lease: { providerLeaseId: "lease-1" }, }; const inParams: PluginEnvironmentSyncInParams = { @@ -161,6 +165,7 @@ describe("environment sync verb negotiation", () => { const outResult = await worker.callWorker("environmentSyncOut", outParams); expect(outResult.operations[0]).toMatchObject({ operationId: "op-b", filesTransferred: 1 }); expect(seen).toEqual(["in", "out"]); + expect(minimums).toEqual([900_000, 900_000]); } finally { worker.stop(); } diff --git a/server/src/__tests__/environment-execution-target-capabilities.test.ts b/server/src/__tests__/environment-execution-target-capabilities.test.ts index 549943e65e0..677223491a5 100644 --- a/server/src/__tests__/environment-execution-target-capabilities.test.ts +++ b/server/src/__tests__/environment-execution-target-capabilities.test.ts @@ -90,7 +90,7 @@ async function buildSandboxTarget(input: { if (target?.kind !== "remote" || target.transport !== "sandbox") { throw new Error("expected a sandbox target"); } - return { target, execute }; + return { target, execute, environmentRuntime }; } describe("resolveEnvironmentExecutionTarget effective capability snapshot", () => { @@ -268,6 +268,17 @@ describe("effective snapshot gates the sync decision", () => { expect(target.runner?.syncOut).toBeTypeOf("function"); }); + it.each(["syncIn", "syncOut"] as const)("carries a %s deadline only on the requested transfer", async method => { + const { target, environmentRuntime } = await buildSandboxTarget({ snapshot: FULL_GRANT, supportsSync: true }); + const operations = [{ operationId: "archive-deadline", files: [] }]; + await target.runner![method]!(operations, { minimumTimeoutMs: 900_000 }); + await target.runner![method]!(operations); + const calls = vi.mocked(environmentRuntime[method]).mock.calls; + expect(calls[0]?.[0]).toMatchObject({ operations, minimumTimeoutMs: 900_000 }); + expect(calls[1]?.[0]).toMatchObject({ operations }); + expect(calls[1]?.[0]).not.toHaveProperty("minimumTimeoutMs"); + }); + it("omits the native sync hooks when the snapshot removes a sync verb", async () => { // The snapshot verified inbound sync but not outbound sync. The runner // exposes the sync hooks both-or-neither, so it keeps the base64 fallback. diff --git a/server/src/__tests__/environment-runtime.test.ts b/server/src/__tests__/environment-runtime.test.ts index cffa283241f..03abe6b72cf 100644 --- a/server/src/__tests__/environment-runtime.test.ts +++ b/server/src/__tests__/environment-runtime.test.ts @@ -881,6 +881,41 @@ describeEmbeddedPostgres("environmentRuntimeService", () => { return { pluginId, companyId, agentId, environment, runId, executionWorkspaceId, reusableLease }; } + it.each((["syncIn", "syncOut"] as const).flatMap(method => + [1234, 1_200_000].map(timeoutMs => ({ method, timeoutMs }))))("uses a per-call $method deadline (configured $timeoutMs) for the provider and its RPC without changing settings", async ({ method, timeoutMs }) => { + const seeded = await seedReusablePluginSandboxLease(null, { timeoutMs }); + const call = vi.fn(async (..._args: unknown[]) => ({ operations: [] })); + const workerManager = { + isRunning: vi.fn(() => true), call, + getWorker: vi.fn(() => ({ supportedMethods: ["environmentSyncIn", "environmentSyncOut"] })), + } as unknown as PluginWorkerManager; + const runtime = environmentRuntimeService(db, { pluginWorkerManager: workerManager }); + const input = { environment: seeded.environment, lease: seeded.reusableLease, operations: [] }; + await runtime[method]({ ...input, minimumTimeoutMs: 900_000 }); + expect(call.mock.calls[0]?.[1]).toBe(method === "syncIn" ? "environmentSyncIn" : "environmentSyncOut"); + expect(call.mock.calls[0]?.[2]).toMatchObject({ config: { timeoutMs }, minimumTimeoutMs: 900_000 }); + expect(call.mock.calls[0]?.[3]).toBeGreaterThan(Math.max(900_000, timeoutMs)); + await runtime[method](input); + expect(call.mock.calls[1]?.[2]).toMatchObject({ config: { timeoutMs } }); + expect(call.mock.calls[1]?.[2]).not.toHaveProperty("minimumTimeoutMs"); + expect(seeded.environment.config.timeoutMs).toBe(timeoutMs); + expect((await environmentService(db).getById(seeded.environment.id))?.config.timeoutMs).toBe(timeoutMs); + }); + + it.each(["syncIn", "syncOut"] as const)("rejects invalid %s deadlines before a provider call", async method => { + const seeded = await seedReusablePluginSandboxLease(); + const call = vi.fn(); + const runtime = environmentRuntimeService(db, { pluginWorkerManager: { + isRunning: vi.fn(() => true), call, + getWorker: vi.fn(() => ({ supportedMethods: ["environmentSyncIn", "environmentSyncOut"] })), + } as unknown as PluginWorkerManager }); + for (const timeoutMs of [0, -1, Number.NaN, Number.POSITIVE_INFINITY, 30 * 60_000 + 1]) { + await expect(runtime[method]({ environment: seeded.environment, lease: seeded.reusableLease, + operations: [], minimumTimeoutMs: timeoutMs })).rejects.toThrow("Invalid sandbox sync timeout"); + } + expect(call).not.toHaveBeenCalled(); + }); + it.each(["active", "released"] as const)("does not select an ephemeral allocation for warm reuse (%s)", async status => { const seeded = await seedReusablePluginSandboxLease("paperclip_runner"); await db.update(environmentLeases).set({ leasePolicy: "ephemeral", status, @@ -3897,8 +3932,12 @@ describeEmbeddedPostgres("environmentRuntimeService", () => { }); }); - it("resumes only the exact stopped sandbox lifecycle without acquiring or reseeding", async () => { - const seeded = await seedReusablePluginSandboxLease("paperclip_runner"); + it.each([ + { timeoutMs: 300_000, rpcTimeoutMs: 330_000 }, + { timeoutMs: 1234, rpcTimeoutMs: 31_234 }, + { timeoutMs: null, rpcTimeoutMs: 60_000 }, + ])("resumes only the exact stopped sandbox lifecycle within its recorded budget ($timeoutMs)", async ({ timeoutMs, rpcTimeoutMs }) => { + const seeded = await seedReusablePluginSandboxLease("paperclip_runner", { timeoutMs }); await environmentService(db).releaseLease(seeded.reusableLease.id, "released", { cleanupStatus: "success" }); const lease = (await environmentService(db).getLeaseById(seeded.reusableLease.id))!; const call = vi.fn(async (_id: string, method: string, _input?: unknown) => { @@ -3914,7 +3953,7 @@ describeEmbeddedPostgres("environmentRuntimeService", () => { expect(call).toHaveBeenCalledWith(seeded.pluginId, "environmentResumeLease", expect.objectContaining({ companyId: seeded.companyId, environmentId: seeded.environment.id, providerLeaseId: lease.providerLeaseId, leaseMetadata: lease.metadata, - }), expect.any(Number)); + }), rpcTimeoutMs); expect(call.mock.calls[0]?.[2]).not.toHaveProperty("requestedCwd"); expect((await environmentService(db).getLeaseById(lease.id))?.status).toBe("released"); expect(await db.select().from(environmentLeases)).toHaveLength(1); diff --git a/server/src/services/environment-execution-target.ts b/server/src/services/environment-execution-target.ts index a4d3a8c16da..51313404fd8 100644 --- a/server/src/services/environment-execution-target.ts +++ b/server/src/services/environment-execution-target.ts @@ -631,17 +631,19 @@ export async function resolveEnvironmentExecutionTarget(input: { lease: input.lease, }) ? { - syncIn: (operations) => + syncIn: (operations, options) => input.environmentRuntime!.syncIn({ environment: input.environment as Environment, lease: input.lease!, operations, + ...(options?.minimumTimeoutMs === undefined ? {} : { minimumTimeoutMs: options.minimumTimeoutMs }), }), - syncOut: (operations) => + syncOut: (operations, options) => input.environmentRuntime!.syncOut({ environment: input.environment as Environment, lease: input.lease!, operations, + ...(options?.minimumTimeoutMs === undefined ? {} : { minimumTimeoutMs: options.minimumTimeoutMs }), }), } : {}), diff --git a/server/src/services/environment-runtime.ts b/server/src/services/environment-runtime.ts index 50a779cef76..dfd42af573c 100644 --- a/server/src/services/environment-runtime.ts +++ b/server/src/services/environment-runtime.ts @@ -37,6 +37,7 @@ import type { } from "@paperclipai/shared"; import { resolveDeclaredSandboxCapabilities } from "@paperclipai/shared"; import type { EffectiveExecutionCapabilities } from "@paperclipai/adapter-utils/execution-target"; +import { MAX_SANDBOX_SYNC_TIMEOUT_MS } from "@paperclipai/adapter-utils/sandbox-managed-runtime"; import type { RunnerIngressEndpoint } from "@paperclipai/adapter-utils/runner-connectivity"; import type { CommandManagedDuplexChannel, @@ -606,6 +607,7 @@ export interface EnvironmentDriverExecuteInput extends EnvironmentDriverLeaseInp export interface EnvironmentDriverSyncInput extends EnvironmentDriverLeaseInput { operations: PluginSyncOperation[]; + minimumTimeoutMs?: number; } export interface EnvironmentDriverOpenDuplexChannelInput extends EnvironmentDriverLeaseInput { @@ -1551,6 +1553,10 @@ function createSandboxEnvironmentDriver( method: "environmentSyncIn" | "environmentSyncOut", input: EnvironmentDriverSyncInput, ): Promise { + if (input.minimumTimeoutMs !== undefined && (!Number.isSafeInteger(input.minimumTimeoutMs) + || input.minimumTimeoutMs <= 0 || input.minimumTimeoutMs > MAX_SANDBOX_SYNC_TIMEOUT_MS)) { + throw new RangeError("Invalid sandbox sync timeout"); + } if (!input.lease.metadata?.sandboxProviderPlugin || !pluginWorkerManager) { throw new Error("Sandbox driver does not support native file sync for this lease."); } @@ -1565,6 +1571,10 @@ function createSandboxEnvironmentDriver( provider: providerKey, }); const sanitizedConfig = stripSandboxProviderEnvelope(config as SandboxEnvironmentConfig); + const rpcTimeoutMs = input.minimumTimeoutMs === undefined + ? resolvePluginSandboxRpcTimeoutMs(sanitizedConfig) + : Math.max(resolvePluginSandboxRpcTimeoutMs(sanitizedConfig) ?? 0, + resolvePluginSandboxRpcTimeoutMs({ ...sanitizedConfig, timeoutMs: input.minimumTimeoutMs }) ?? 0); return await pluginWorkerManager.call(pluginId, method, { driverKey: providerKey, companyId: input.lease.companyId, @@ -1577,7 +1587,8 @@ function createSandboxEnvironmentDriver( expiresAt: input.lease.expiresAt?.toISOString() ?? null, }, operations: input.operations, - }, resolvePluginSandboxRpcTimeoutMs(sanitizedConfig)).catch((error: unknown) => { + ...(input.minimumTimeoutMs === undefined ? {} : { minimumTimeoutMs: input.minimumTimeoutMs }), + }, rpcTimeoutMs).catch((error: unknown) => { throw method === "environmentSyncOut" ? preserveEnvironmentSyncOutErrorDiagnostic(error) : error; }); } @@ -2591,7 +2602,7 @@ function createSandboxEnvironmentDriver( issueId: input.lease.issueId, config, providerLeaseId: input.lease.providerLeaseId, leaseMetadata: input.lease.metadata ?? undefined, ...(input.requestedCwd !== undefined ? { requestedCwd: input.requestedCwd } : {}), - }, Math.min(resolvePluginSandboxRpcTimeoutMs(config) ?? 60_000, 60_000)); + }, resolvePluginSandboxRpcTimeoutMs(config) ?? 60_000); if (resumed?.providerLeaseId !== input.lease.providerLeaseId) { throw new Error("The provider did not confirm the exact retained sandbox. No replacement was acquired."); } diff --git a/server/src/services/remote-workspace-transfer-timeout.test.ts b/server/src/services/remote-workspace-transfer-timeout.test.ts new file mode 100644 index 00000000000..b242f7d0894 --- /dev/null +++ b/server/src/services/remote-workspace-transfer-timeout.test.ts @@ -0,0 +1,112 @@ +import { createHash, randomUUID } from "node:crypto"; +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { afterEach, expect, it, vi } from "vitest"; +import type { CommandManagedRuntimeRunner } from "@paperclipai/adapter-utils/command-managed-runtime"; +import { captureRemoteWorkspaceArchive, restoreRemoteWorkspaceDirectory, type RemoteWorkspaceTarget } from "./remote-workspace-transport.js"; + +const roots: string[] = []; +const captured: Array>> = []; + +afterEach(async () => { + if (vi.isFakeTimers()) vi.clearAllTimers(); + vi.useRealTimers(); + await Promise.all(captured.splice(0).map(archive => archive.cleanup())); + await Promise.all(roots.splice(0).map(root => fs.rm(root, { recursive: true, force: true }))); +}); + +it.each(["download", "upload"] as const)("preserves a valid large archive when %s takes six minutes", async direction => { + const root = await fs.mkdtemp(path.join(os.tmpdir(), "workspace-transfer-deadline-")); + roots.push(root); + const byteSize = 768 * 1024 ** 2; + const zero = Buffer.alloc(1024 ** 2); + const hash = createHash("sha256"); + for (let count = 0; count < 768; count += 1) hash.update(zero); + const sha256 = hash.digest("hex"); + const source = path.join(root, "saved.pcws"); + if (direction === "upload") { + await fs.writeFile(source, ""); + await fs.truncate(source, byteSize); + } + + let announce!: () => void; + const started = new Promise(resolve => { announce = resolve; }); + let commands = 0; + let usedTimeout = 0; + const execute: CommandManagedRuntimeRunner["execute"] = async () => ({ + stdout: direction === "download" && commands++ === 1 + ? JSON.stringify({ manifestDigest: "a".repeat(64), sha256, byteSize, entries: 1, layout: "captured" }) : "", + stderr: "", exitCode: 0, timedOut: false, signal: null, pid: process.pid, startedAt: new Date().toISOString(), + }); + const transfer: NonNullable = async (operations, options) => { + usedTimeout = options?.minimumTimeoutMs ?? 300_000; + await new Promise((resolve, reject) => { + const failure = setTimeout(() => { + clearTimeout(delivery); + reject(new Error("Daytona archive stream timed out")); + }, usedTimeout); + const delivery = setTimeout(() => { clearTimeout(failure); resolve(); }, 360_000); + announce(); + }); + for (const operation of operations) { + for (const file of operation.files) { + if (direction === "download") { + await fs.writeFile(file.targetPath, "", { flag: "wx" }); + await fs.truncate(file.targetPath, byteSize); + } else { + expect((await fs.stat(file.sourcePath)).size).toBe(byteSize); + } + } + } + return { operations: operations.map(operation => ({ operationId: operation.operationId, + filesTransferred: operation.files.length, bytesTransferred: byteSize })) }; + }; + const remote: RemoteWorkspaceTarget = { + kind: "remote", transport: "sandbox", remoteCwd: root, runner: { execute, syncIn: transfer, syncOut: transfer }, + }; + vi.useFakeTimers(); + const pending = direction === "download" + ? captureRemoteWorkspaceArchive({ target: remote, remoteCwd: root, operationId: randomUUID() }).then(value => { + captured.push(value); + return value; + }) + : restoreRemoteWorkspaceDirectory({ target: remote, remoteCwd: path.join(root, "restored"), archivePath: source, + manifestDigest: "a".repeat(64), restorationIdentity: "b".repeat(64) }); + const settled = pending.then(value => ({ value }), error => ({ error })); + await Promise.race([started, settled.then(outcome => { + if ("error" in outcome) throw outcome.error; + throw new Error("The archive transfer did not wait for its data"); + })]); + await vi.advanceTimersByTimeAsync(360_000); + const outcome = await settled; + if ("error" in outcome) throw outcome.error; + expect(usedTimeout).toBeGreaterThan(360_000); + expect(usedTimeout).toBeLessThanOrEqual(30 * 60_000); + if (direction === "download") { + const archive = outcome.value as Awaited>; + expect(archive.sha256).toBe(sha256); + expect(archive.byteSize).toBe(byteSize); + expect((await fs.stat(archive.archivePath)).size).toBe(byteSize); + } +}, 30_000); + +it.each([ + { byteSize: 1024, timeoutMs: undefined }, + { byteSize: 64 * 1024 ** 3, timeoutMs: 30 * 60_000 }, +])("keeps the archive deadline bounded for $byteSize bytes", async ({ byteSize, timeoutMs }) => { + const transfer = vi.fn>(async () => { + throw new Error("Transfer stopped at the budget probe"); + }); + const remote: RemoteWorkspaceTarget = { kind: "remote", transport: "sandbox", remoteCwd: "/work", + runner: { + execute: async () => ({ stdout: JSON.stringify({ layout: "captured", manifestDigest: "a".repeat(64), + sha256: "b".repeat(64), byteSize, entries: 1 }), stderr: "", exitCode: 0, timedOut: false, + signal: null, pid: process.pid, startedAt: new Date().toISOString() }), + syncIn: transfer, syncOut: transfer, + } }; + await expect(captureRemoteWorkspaceArchive({ target: remote, remoteCwd: "/work", operationId: randomUUID() })).rejects.toThrow(); + expect(transfer).toHaveBeenCalledOnce(); + expect(transfer.mock.calls[0]?.[1]?.minimumTimeoutMs).toBe(timeoutMs); + if (timeoutMs === undefined) expect(transfer.mock.calls[0]).toHaveLength(1); +}); diff --git a/server/src/services/remote-workspace-transport.ts b/server/src/services/remote-workspace-transport.ts index 0e01f8c016a..3e821827f68 100644 --- a/server/src/services/remote-workspace-transport.ts +++ b/server/src/services/remote-workspace-transport.ts @@ -1,6 +1,6 @@ import { createHash, randomUUID } from "node:crypto"; import { createReadStream } from "node:fs"; -import { mkdtemp, rm } from "node:fs/promises"; +import { mkdtemp, rm, stat } from "node:fs/promises"; import os from "node:os"; import path from "node:path"; import type { AdapterExecutionTarget } from "@paperclipai/adapter-utils/execution-target"; @@ -8,9 +8,19 @@ import type { RemoteWorkspaceSource } from "@paperclipai/shared"; import { remoteWorkspaceArchiveScript, remoteWorkspaceCaptureFailureMessage, remoteWorkspaceFailureDetail } from "./remote-workspace-snapshot-archive.js"; import type { GitRemoteAuthProvider } from "./git-credentials.js"; import { classifyWorkspaceRestoreFailure } from "@paperclipai/adapter-utils/workspace-restore-merge"; +import { MAX_SANDBOX_SYNC_TIMEOUT_MS, type SandboxSyncOperation } from "@paperclipai/adapter-utils/sandbox-managed-runtime"; export type RemoteWorkspaceTarget = Extract; +async function transferArchive(target: RemoteWorkspaceTarget, direction: "syncIn" | "syncOut", + operations: SandboxSyncOperation[], byteSize: number) { + if (!Number.isSafeInteger(byteSize) || byteSize <= 0) throw new Error("remote_workspace_archive_size_invalid"); + const timeoutMs = Math.min(MAX_SANDBOX_SYNC_TIMEOUT_MS, 60_000 + Math.ceil(byteSize / 1024 ** 2) * 1_000); + return timeoutMs > 300_000 + ? target.runner![direction]!(operations, { minimumTimeoutMs: timeoutMs }) + : target.runner![direction]!(operations); +} + const checkoutConfiguredRefScript = String.raw`if(input.ref){const remoteRef=input.ref.replace(/^refs\/heads\//,''); const selected=await git(['rev-parse','--verify','--end-of-options',input.ref+'^{commit}']) .catch(()=>git(['rev-parse','--verify','--end-of-options','refs/remotes/origin/'+remoteRef+'^{commit}'])); @@ -437,9 +447,9 @@ export async function captureRemoteWorkspaceArchive(input: { const root = await mkdtemp(path.join(os.tmpdir(), "paperclip-workspace-upload-")); const archivePath = path.join(root, "workspace.pcws"); try { - await input.target.runner!.syncOut!([{ operationId: input.operationId, + await transferArchive(input.target, "syncOut", [{ operationId: input.operationId, files: [{ kind: "file", sourcePath: layout === "captured" ? path.posix.join(staged.directory, "captured", "workspace.pcws") : staged.archive, targetPath: archivePath }], - }]); + }], receipt.byteSize); const hash = createHash("sha256"); let bytes = 0; for await (const chunk of createReadStream(archivePath)) { hash.update(chunk); bytes += chunk.length; } if (hash.digest("hex") !== receipt.sha256 || bytes !== receipt.byteSize) throw new Error("remote_workspace_capture_transfer_mismatch"); @@ -464,9 +474,9 @@ export async function restoreRemoteWorkspaceDirectory(input: { // Stage outside the destination; the restore validator requires an empty target. const parent = path.posix.dirname(input.remoteCwd); const staged = await installArchiveHelper(input.target, parent, operationId); - await input.target.runner!.syncIn!([{ operationId, + await stat(input.archivePath).then(archive => transferArchive(input.target, "syncIn", [{ operationId, files: [{ kind: "file", sourcePath: input.archivePath, targetPath: staged.archive }], - }]).catch((error: unknown) => { noteProviderRefusal("upload", error); throw error; }); + }], archive.size)).catch((error: unknown) => { noteProviderRefusal("upload", error); throw error; }); // Publish the restored folder and its identity together. A lost successful // reply can then adopt the same files without repeating a destructive restore. const restore = String.raw`const fs=require('node:fs/promises'),path=require('node:path'),cp=require('node:child_process'),crypto=require('node:crypto');