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 <noreply@paperclip.ing>
This commit is contained in:
Devin FoleyandPaperclip authored and GitHub committed 2026-11-05 16:09:32 -08:00
1 parent a20dfbc695
commit 8080620fc9
16 files changed
+334 -28

No files matched your search

+20
View File
@@ -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
@@ -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<CommandManagedRuntimeRunner["execute"]>();
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);
@@ -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<SandboxSyncResult>;
syncIn?(operations: SandboxSyncOperation[], options?: SandboxSyncOptions): Promise<SandboxSyncResult>;
/** Optional native outbound file transfer. See {@link syncIn}. */
syncOut?(operations: SandboxSyncOperation[]): Promise<SandboxSyncResult>;
syncOut?(operations: SandboxSyncOperation[], options?: SandboxSyncOptions): Promise<SandboxSyncResult>;
/**
* 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;
@@ -451,6 +451,12 @@ export interface SandboxTransferProgressOptions {
onProgress?: (transferredBytes: number, totalBytes: number | null) => void | Promise<void>;
}
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<SandboxSyncResult>;
syncIn?(operations: SandboxSyncOperation[], options?: SandboxSyncOptions): Promise<SandboxSyncResult>;
/** Optional native outbound transfer. See {@link syncIn}. */
syncOut?(operations: SandboxSyncOperation[]): Promise<SandboxSyncResult>;
syncOut?(operations: SandboxSyncOperation[], options?: SandboxSyncOptions): Promise<SandboxSyncResult>;
}
/**
@@ -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
@@ -1147,6 +1147,7 @@ export async function performSyncIn(input: {
operations: PluginSyncOperation[];
remoteDir: string;
timeoutSeconds: number;
commandTimeoutSeconds?: number;
}): Promise<PluginEnvironmentSyncResult> {
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 });
@@ -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");
@@ -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<PluginEnvironmentSyncResult> {
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<PluginEnvironmentSyncResult> {
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({
+2
View File
@@ -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. */
@@ -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<number | undefined> = [];
const worker = startTestWorker(
definePlugin({
async setup() {},
async onEnvironmentSyncIn(params): Promise<PluginEnvironmentSyncResult> {
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<PluginEnvironmentSyncResult> {
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<PluginEnvironmentSyncResult>("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();
}
@@ -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.
@@ -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);
@@ -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 }),
}),
}
: {}),
+13 -2
View File
@@ -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<PluginEnvironmentSyncResult> {
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.");
}
@@ -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<Awaited<ReturnType<typeof captureRemoteWorkspaceArchive>>> = [];
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<void>(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<CommandManagedRuntimeRunner["syncOut"]> = async (operations, options) => {
usedTimeout = options?.minimumTimeoutMs ?? 300_000;
await new Promise<void>((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<ReturnType<typeof captureRemoteWorkspaceArchive>>;
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<NonNullable<CommandManagedRuntimeRunner["syncOut"]>>(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);
});
@@ -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<AdapterExecutionTarget, { transport: "sandbox" }>;
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');