From 6a932bb979a93bf79f9dd77f7a6b3ddfc72927f6 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Thu, 29 Oct 2026 11:41:01 -0500 Subject: [PATCH] fix(daytona): preserve workspace exports under disk pressure (#18172) ## Thinking Path > - Paperclip manages AI agents and preserves their work when execution stops. > - Remote workspace recovery must save the retained files before another executor can start. > - Daytona directory export first writes a complete tar on the source sandbox. > - When that disk is full, each preservation retry fails while writing the same temporary archive. > - This change streams an archive after a storage-capacity failure, without another full copy on the sandbox. > - Existing source ownership, archive validation, and task admission gates remain in force. ## Linked Issues or Issue Description Related: #14402 established safe workspace export and retained-source recovery. This PR handles the remaining temporary-archive storage requirement. **What happened?** A sandbox ran out of disk space. The provider could no longer save its state, and workspace preservation then failed with tar `No space left on device` or `Wrote only ... bytes`. Repeating the export could not resolve that failure because it required free space on the same disk. **Expected behavior** Export retained files without needing a second full copy on the source disk. Reject incomplete transfers and unsafe archives. Preserve the source and newer host files if export fails. **Steps to reproduce** Run the capacity regression with real tar and a file-size limit. The original export fails during archive creation. The fixed export transfers the same bytes through a real PTY. The credential-gated live test uses a real Daytona sandbox and `/dev/full` as the archive sink to reproduce ENOSPC without filling a provider disk. **Paperclip version or commit** Reproduced on `27be859044`; fix rebased onto `6d59f5ef7e`. **Deployment mode** Source checkout on macOS arm64; live provider verification on a disposable Ubuntu 24.04 Daytona sandbox with SDK 0.203.0. ## What Changed - Fall back only after archive storage-capacity errors. Keep the ordinary bulk download path. - Transfer 40 KiB archive chunks over a raw PTY. Wait for each host write acknowledgement and a final receipt acknowledgement. Send long scripts only after raw-mode readiness. - Require successful tar exit and complete framing before the existing host archive validator can import files. Keep the bounded safe-link retry and exclude rules. - Bound total archive bytes by the provider-reported disk allocation, preserve a 1 GiB host free-space reserve, and bound terminal cleanup to five seconds while closing the host file independently. - Add real-tar/PTY regressions, stream failure tests, a credential-gated provider test, and recovery documentation. ## Verification - Red before production changes: the capacity regression failed on the real tar write with exit 153. Green afterward: unchanged bytes, file mode, confined links, and source files survive. A tar producer that fails after sending bytes cannot overwrite newer host files. - 43 focused tests pass. Missing receipts, nonzero exits, transport loss, invalid base64, oversized frames, stalled transfers, and late connection cleanup fail closed. - Live Daytona test passes: stop/start, actual tar ENOSPC, matching file hashes and modes, unsafe-link filtering, intact source, then another stop/start and export with a later saved file. The disposable sandbox was deleted. Initial default-sandbox creation hit account quota; the successful test used a 1 GiB image. Large output bursts exposed a real socket closure, corrected by bounded chunks and verified in the successful run. - Repository `pnpm -r typecheck` and `pnpm build` pass. `node scripts/check-maps.mjs` passes. Standalone plugin typecheck passes. - Standalone plugin build and final typecheck pass. The final broad plugin run passed 482 tests, skipped seven credential-gated cases, and hit one existing 30-second archive-listing timeout. Earlier runs hit both large-listing cases. Both timeouts reproduce with unchanged `file-sync.ts` from master `6d59f5ef7e`, with 22 other archive-listing cases passing on both base and fixed source. Parallel, serial, and GNU-tar local runs retain those timeouts. The serial local `pnpm test:run` was stopped after 36 minutes once the complete current-head CI test suite passed; it is not counted as a full local pass. GitHub Actions on production-code head `9ffbd9e56e283a01c22f005c022b8f6838cf0143` passed typecheck, all general and serialized test shards, Runner verification, build, E2E, and canary dry run. - Review fixes have passed the live Daytona test again. Sparse-file budget enforcement, rejection before host writes, low host reserve, and stalled provider cleanup are covered by the 43 focused tests. Greptile is 5/5 and security scanning passes on `9ffbd9e56e283a01c22f005c022b8f6838cf0143`; both review threads are resolved. - Test portability: `d2280108f5` verifies GNU tool versions and Python before running the real-PTY regression. All 43 focused tests pass with prerequisites installed; restricting PATH to the macOS system tools cleanly skips the three integration cases. Standalone plugin typecheck passes. Current-head CI passes typecheck, build, all test shards, Runner verification, E2E, and canary dry run: https://github.com/paperclipai/paperclip/actions/runs/40910736373. Greptile and Paperclip Review both assess the final head 5/5, security checks pass, and no review threads remain unresolved. - Paperclip Review assessed the final head 5/5 with no actionable defects after the test-portability fix. Remaining accepted tradeoffs are recorded under Risks. - This is live provider-boundary acceptance. It does not prove a full task Retry journey, repair lost provider session state, or recover the original deployed task. ## Risks - Streaming requires Bash and GNU tar, dd, and base64. Missing tools or transfer failure reject export and retain the recovery obligation. - Acknowledged terminal transfer is slower than bulk download, especially across high-latency links. Very large exports can still reach the configured deadline. The ordinary bulk path is unchanged. - The fallback checks host free space before every 40 KiB write; this deliberately favors a strict reserve check over reducing local syscalls. It carries archive-download error context but does not yet emit a dedicated transfer span. Provider cleanup is bounded; a kernel-stalled local filesystem operation can still delay completion and file close. - This change does not increase sandbox capacity or reconstruct provider state that failed to persist. The plugin must reach the target installation before recovery can use it. No merge, deployment, or original-task replay is included. ## Model Used OpenAI GPT-6 through Codex, with reasoning, repository inspection, code editing, shell execution, and automated testing. The exact serving variant and context-window size are not exposed in this session. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip --- maps/agent-access-map/recovery.md | 5 + maps/feature-map/recovery.md | 8 + .../sandbox-providers/daytona/README.md | 9 ++ .../daytona/src/archive-stream.test.ts | 89 +++++++++++ .../daytona/src/archive-stream.ts | 139 ++++++++++++++++++ .../src/file-sync-capacity.live.test.ts | 74 ++++++++++ .../daytona/src/file-sync-capacity.test.ts | 133 +++++++++++++++++ .../daytona/src/file-sync.ts | 69 ++++++--- 8 files changed, 502 insertions(+), 24 deletions(-) create mode 100644 packages/plugins/sandbox-providers/daytona/src/archive-stream.test.ts create mode 100644 packages/plugins/sandbox-providers/daytona/src/archive-stream.ts create mode 100644 packages/plugins/sandbox-providers/daytona/src/file-sync-capacity.live.test.ts create mode 100644 packages/plugins/sandbox-providers/daytona/src/file-sync-capacity.test.ts diff --git a/maps/agent-access-map/recovery.md b/maps/agent-access-map/recovery.md index 3456b5c5793..2f9bd8214e7 100644 --- a/maps/agent-access-map/recovery.md +++ b/maps/agent-access-map/recovery.md @@ -11,6 +11,11 @@ The server also retires an unowned failed sandbox retry after its linked explici Retry completes. This only registers exact-source preservation; it neither changes the current task nor supplies stop or saved-file proof. +Daytona's directory-export capacity fallback stays inside this server-owned file +preservation path. It changes the archive transport, grants no new agent entry +point, and still requires the caller's exact stopped-source and workspace fences. +Provider transfer tests do not establish task admission or agent authority. + Remote process probes remain server-owned. Confirmed exit requires `ESRCH` or a zombie with the saved generation. Unknown adopted-process ownership rejects the probe and grants no permission to replace a runner, signal a different process, diff --git a/maps/feature-map/recovery.md b/maps/feature-map/recovery.md index 90126dacd6c..54a1bfc2843 100644 --- a/maps/feature-map/recovery.md +++ b/maps/feature-map/recovery.md @@ -39,6 +39,14 @@ Agent access: [companion](../agent-access-map/recovery.md). workspace link when the destination is disabled. Local PostgreSQL and real filesystem regressions cover native and legacy recovery and one-time delivery; hosted provider recovery remains separate acceptance. + Daytona directory export can stream its archive when the sandbox cannot store + a temporary tar. The transfer remains bounded and validates archive completion + and paths before import. The [capacity regression](../../packages/plugins/sandbox-providers/daytona/src/file-sync-capacity.test.ts) + exercises a real local PTY and file-size limit. The credential-gated + [provider test](../../packages/plugins/sandbox-providers/daytona/src/file-sync-capacity.live.test.ts) + covers stop/start, an ENOSPC archive sink, safe-link filtering, and a second + export. Neither test proves the original task's continuation or repairs lost + provider session state. Native finalization deadline failures also retain the exact prepared source for the same stop-and-save workflow before Retry admits another executor. Retry reconstructs missing preservation for historical failures only from an diff --git a/packages/plugins/sandbox-providers/daytona/README.md b/packages/plugins/sandbox-providers/daytona/README.md index 87dcb168832..993b58f3c73 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. +- 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. ## Local development @@ -49,6 +50,14 @@ pnpm typecheck These commands assume the repo root has already been installed once so the local `@paperclipai/plugin-sdk` workspace package is available to the compiler during development. +The capacity regression uses Python 3 for a real local PTY, Bash, and GNU core +utilities (`gtar`, `gdd`, and `gbase64` on macOS). It skips when those prerequisites +are unavailable or their version checks do not match. With `DAYTONA_API_KEY` set, run +`pnpm test src/file-sync-capacity.live.test.ts` to exercise an actual stopped and +resumed Daytona sandbox, a tar write to `/dev/full`, validated export, and a +second export after reconnect. The test creates and deletes one disposable +1 GiB Ubuntu sandbox; it does not fill a disk or modify existing sandboxes. + ## Package layout - `src/manifest.ts` declares the sandbox-provider driver metadata diff --git a/packages/plugins/sandbox-providers/daytona/src/archive-stream.test.ts b/packages/plugins/sandbox-providers/daytona/src/archive-stream.test.ts new file mode 100644 index 00000000000..0ea70cd1f36 --- /dev/null +++ b/packages/plugins/sandbox-providers/daytona/src/archive-stream.test.ts @@ -0,0 +1,89 @@ +import { promises as fs } from "node:fs"; +import os from "node:os"; +import path from "node:path"; +import { open, statfs } from "node:fs/promises"; +import { afterEach, expect, it, vi } from "vitest"; +import type { DuplexChannelSession } from "./duplex-command-stream.js"; + +const { openSession } = vi.hoisted(() => ({ openSession: vi.fn() })); +vi.mock("./duplex-command-stream.js", () => ({ openDaytonaDuplexChannelSession: openSession })); +vi.mock("node:fs/promises", async importOriginal => { + const actual = await importOriginal(); + return { ...actual, open: vi.fn(actual.open), statfs: vi.fn(actual.statfs) }; +}); +import { streamSandboxArchive } from "./archive-stream.js"; + +const roots: string[] = []; +afterEach(async () => { + vi.resetAllMocks(); + vi.useRealTimers(); + await Promise.all(roots.splice(0).map(root => fs.rm(root, { recursive: true, force: true }))); +}); + +it.each(["missing receipt", "tar failure", "transport loss", "invalid base64", "oversized frame", "timeout", "host reserve", "archive budget"])("rejects %s and closes the transfer", async fault => { + const root = await fs.mkdtemp(path.join(os.tmpdir(), "archive-stream-")); + roots.push(root); + const close = vi.fn(async () => {}); + if (fault === "host reserve") vi.mocked(statfs).mockResolvedValue({ bavail: 0, bsize: 4096 } as never); + openSession.mockImplementation(async (_process, command: string[]) => { + const marker = command.join(" ").match(/paperclip-archive-[a-f0-9-]+/)![0]; + let listener: (bytes: Uint8Array) => void; + let end!: (exit: { exitCode: number | null; transportClosed: boolean }) => void; + const completion = new Promise<{ exitCode: number | null; transportClosed: boolean }>(resolve => { end = resolve; }); + let writes = 0; + const emit = (text: string) => listener(Buffer.from(text)); + return { + onData: (callback: typeof listener) => { listener = callback; emit(`${marker}:ready\n`); }, + write: () => { + queueMicrotask(() => { + if (++writes === 1) { + emit(`${marker}:start\n`); + if (fault === "timeout") return; + if (fault === "invalid base64") emit("data:not!base64\n"); + else if (fault === "oversized frame") emit(`data:${"a".repeat(6 * 1024 * 1024)}`); + else emit(`data:${Buffer.from("partial archive").toString("base64")}\n`); + } else { + if (fault === "tar failure") emit(`${marker}:done\n`); + end({ exitCode: fault === "tar failure" ? 2 : fault === "transport loss" ? null : 0, transportClosed: fault === "transport loss" }); + } + }); + }, + wait: () => completion, + kill: () => {}, + close, + } satisfies DuplexChannelSession; + }); + await expect(streamSandboxArchive({ process: {} as never, cwd: "/workspace", archiveScript: "tar -cf - .", destination: path.join(root, "archive.tar"), maxBytes: fault === "archive budget" ? 8 : 1024, timeoutSeconds: fault === "timeout" ? 0.02 : 5 })).rejects.toThrow(/archive stream/i); + if (fault === "host reserve" || fault === "archive budget") expect((await fs.stat(path.join(root, "archive.tar"))).size).toBe(0); + expect(close).toHaveBeenCalledOnce(); +}); + +it("closes the host file and bounds a stalled provider teardown", async () => { + const root = await fs.mkdtemp(path.join(os.tmpdir(), "archive-stream-")); + roots.push(root); + vi.useFakeTimers(); + const close = vi.fn(() => new Promise(() => {})); + openSession.mockResolvedValue({ onData: vi.fn(), write: vi.fn(), wait: () => new Promise(() => {}), kill: vi.fn(), close }); + const transfer = streamSandboxArchive({ process: {} as never, cwd: "/workspace", archiveScript: "tar -cf - .", destination: path.join(root, "archive.tar"), maxBytes: 1024, timeoutSeconds: 0.02 }); + const rejected = expect(transfer).rejects.toThrow("timed out"); + await vi.waitFor(() => expect(openSession).toHaveBeenCalledOnce()); + await vi.advanceTimersByTimeAsync(20); + const file = await vi.mocked(open).mock.results[0].value; + await vi.waitFor(async () => { await expect(file.stat()).rejects.toMatchObject({ code: "EBADF" }); }); + await vi.advanceTimersByTimeAsync(5_000); + await rejected; + expect(close).toHaveBeenCalledOnce(); +}); + +it("closes a session that connects after the deadline", async () => { + const root = await fs.mkdtemp(path.join(os.tmpdir(), "archive-stream-")); + roots.push(root); + let connect!: (session: DuplexChannelSession) => void; + openSession.mockReturnValue(new Promise(resolve => { connect = resolve; })); + await expect(streamSandboxArchive({ process: {} as never, cwd: "/workspace", archiveScript: "tar -cf - .", destination: path.join(root, "archive.tar"), maxBytes: 1024, timeoutSeconds: 0.02 })).rejects.toThrow("timed out"); + const close = vi.fn(async () => {}); + const onData = vi.fn(); + connect({ close, onData, write: vi.fn(), wait: vi.fn(), kill: vi.fn() }); + await vi.waitFor(() => expect(close).toHaveBeenCalledOnce()); + expect(onData).not.toHaveBeenCalled(); +}); diff --git a/packages/plugins/sandbox-providers/daytona/src/archive-stream.ts b/packages/plugins/sandbox-providers/daytona/src/archive-stream.ts new file mode 100644 index 00000000000..9659a63b2a9 --- /dev/null +++ b/packages/plugins/sandbox-providers/daytona/src/archive-stream.ts @@ -0,0 +1,139 @@ +import { randomUUID } from "node:crypto"; +import { open, statfs } from "node:fs/promises"; +import { openDaytonaDuplexChannelSession, type DaytonaPtyProcess, type DuplexChannelSession } from "./duplex-command-stream.js"; + +// Keep encoded bursts below the provider's 64 KiB PTY message limit and wait +// for each acknowledgement so output cannot overrun its socket queue. +const CHUNK_BYTES = 40 * 1024; +const MAX_FRAME_BYTES = Math.ceil(CHUNK_BYTES / 3) * 4 + 128; +const HOST_FREE_RESERVE_BYTES = 1024 ** 3; +const CLEANUP_TIMEOUT_MS = 5_000; +const quote = (value: string) => `'${value.replace(/'/g, `'"'"'`)}'`; + +async function closeArchiveSession(session: DuplexChannelSession): Promise { + let timer: ReturnType | undefined; + try { + await Promise.race([ + session.close().catch(() => undefined), + new Promise(resolve => { timer = setTimeout(resolve, CLEANUP_TIMEOUT_MS); }), + ]); + } finally { + clearTimeout(timer); + } +} + +/** No remote archive or session log: one chunk in memory, acknowledged only + * after its host write. The ordinary archive validator still owns import. */ +export async function streamSandboxArchive(input: { + process: DaytonaPtyProcess; + cwd: string; + archiveScript: string; + destination: string; + timeoutSeconds: number; + maxBytes: number; +}): Promise { + if (!Number.isSafeInteger(input.maxBytes) || input.maxBytes <= 0) { + throw new Error("Daytona archive stream requires a verified sandbox disk budget"); + } + const marker = `paperclip-archive-${randomUUID()}`; + const ready = `${marker}:ready`; + const start = `${marker}:start`; + const done = `${marker}:done`; + // fd 3 retains the PTY input while tar owns the pipeline's stdin. pipefail + // preserves tar failure even when the last encoder completes successfully. + const sender = [ + "while :; do", + `chunk=$(dd bs=${CHUNK_BYTES} count=1 iflag=fullblock status=none | base64 -w 0) || exit 1;`, + '[ -n "$chunk" ] || break;', + 'printf "data:%s\\n" "$chunk";', + `IFS= read -r ack <&3 && [ "$ack" = ${quote(marker)} ] || exit 1;`, + "done", + ].join(" "); + // Duplicate the original input before constructing the tar pipeline. + const command = `exec 3<&0; printf '\\n%s\\n' ${quote(start)}; { ${input.archiveScript}; } | ( ${sender} ); `; + const launch = `${command}status=$?; [ "$status" -eq 0 ] || exit "$status"; ` + + `printf '%s\\n' ${quote(done)}; IFS= read -r ack <&3 && [ "$ack" = ${quote(done)} ]`; + const file = await open(input.destination, "w", 0o600); + let session: DuplexChannelSession | undefined; + let stopped = false; + let started = false; + let launched = false; + let complete = false; + let writing = false; + let buffer = ""; + let receivedBytes = 0; + let pending = Promise.resolve(); + let rejectFailure!: (error: Error) => void; + const failure = new Promise((_, reject) => { rejectFailure = reject; }); + const fail = (message: string) => rejectFailure(new Error(message)); + const timer = setTimeout(() => fail("Daytona archive stream timed out"), input.timeoutSeconds * 1000); + const transfer = async () => { + session = await openDaytonaDuplexChannelSession(input.process, + ["bash", "--noprofile", "--norc", "-o", "pipefail", "-c", + `printf '\\n%s\\n' ${quote(ready)}; IFS= read -r script && eval "$(printf %s \"$script\" | base64 -d)"`], + { cwd: input.cwd, diagnosticsPath: "/dev/null" }); + if (stopped) { await closeArchiveSession(session); return; } + session.onData(bytes => { + if (stopped) return; + buffer += Buffer.from(bytes).toString("utf8"); + if (buffer.length > MAX_FRAME_BYTES) { fail("Daytona archive stream exceeded its frame limit"); return; } + for (let newline = buffer.indexOf("\n"); newline >= 0; newline = buffer.indexOf("\n")) { + const line = buffer.slice(0, newline).replace(/\r$/, ""); + buffer = buffer.slice(newline + 1); + // The bootstrap confirms raw mode before sending the potentially long + // archive script, which would exceed a canonical terminal's line cap. + if (!launched && line === ready) { + launched = true; + session!.write(Buffer.from(`${Buffer.from(launch).toString("base64")}\n`)); + continue; + } + if (!started) { if (line === start) started = true; continue; } + if (line === done && !complete && !writing) { + complete = true; + session!.write(Buffer.from(`${done}\n`)); + continue; + } + if (!line || complete || writing || !line.startsWith("data:")) { fail("Invalid Daytona archive stream frame"); return; } + const encoded = line.slice(5); + const data = Buffer.from(encoded, "base64"); + if (!data.length || data.length > CHUNK_BYTES || data.toString("base64") !== encoded) { + fail("Invalid Daytona archive stream payload"); return; + } + writing = true; + pending = pending.then(async () => { + if (stopped) return; + if (receivedBytes + data.length > input.maxBytes) { + throw new Error("Daytona archive stream exceeds the sandbox disk budget"); + } + const available = await statfs(input.destination); + if (available.bavail * available.bsize - data.length < HOST_FREE_RESERVE_BYTES) { + throw new Error("Daytona archive stream would exhaust the host free-space reserve"); + } + if (stopped) return; + await file.writeFile(data); + receivedBytes += data.length; + writing = false; + session!.write(Buffer.from(`${marker}\n`)); + }).catch(error => { rejectFailure(error); }); + } + }); + const exit = await session.wait(); + await pending; + if (exit.exitCode !== 0 || exit.transportClosed || !complete || buffer.trim()) { + throw new Error(`Daytona archive stream ended without a successful archive receipt (exit=${exit.exitCode ?? "unknown"}, started=${started}, complete=${complete}, pendingBytes=${buffer.length})`); + } + }; + try { + await Promise.race([transfer(), failure]); + } finally { + stopped = true; + clearTimeout(timer); + const cleanup = session ? closeArchiveSession(session) : Promise.resolve(); + try { + await pending; + } finally { + await file.close(); + await cleanup; + } + } +} diff --git a/packages/plugins/sandbox-providers/daytona/src/file-sync-capacity.live.test.ts b/packages/plugins/sandbox-providers/daytona/src/file-sync-capacity.live.test.ts new file mode 100644 index 00000000000..c688e3f2401 --- /dev/null +++ b/packages/plugins/sandbox-providers/daytona/src/file-sync-capacity.live.test.ts @@ -0,0 +1,74 @@ +import { createHash, randomUUID } from "node:crypto"; +import { promises as fs } from "node:fs"; +import os from "node:os"; +import path from "node:path"; +import { expect, it, vi } from "vitest"; + +vi.mock("./plugin.js", async () => { + const { NOOP_PLUGIN_TRACER } = await import("@paperclipai/plugin-sdk"); + return { getPluginTracer: () => NOOP_PLUGIN_TRACER }; +}); +import { performSyncOut } from "./file-sync.js"; + +// Fault only the archive's output sink. /dev/full gives real GNU tar ENOSPC +// without exhausting a shared provider host or changing the retained source. +it.skipIf(!process.env.DAYTONA_API_KEY)("preserves a stopped Daytona workspace after archive ENOSPC", async () => { + const { Daytona } = await import("@daytonaio/sdk"); + const client = new Daytona({ apiKey: process.env.DAYTONA_API_KEY }); + const sandbox = await client.create({ + name: `archive-capacity-${randomUUID()}`, image: "ubuntu:24.04", user: "root", + resources: { cpu: 1, memory: 1, disk: 1 }, autoStopInterval: 5, + }); + const output = await fs.mkdtemp(path.join(os.tmpdir(), "daytona-capacity-live-")); + const remote = "/tmp/archive-capacity-proof"; + try { + const setup = await sandbox.process.executeCommand( + `mkdir -p ${remote}/ignored && head -c 5242880 /dev/urandom > ${remote}/saved.bin && ` + + `chmod 600 ${remote}/saved.bin && printf retained > ${remote}/.hidden && ` + + `printf omitted > ${remote}/ignored/output && ln -s .hidden ${remote}/safe-link && ` + + `ln -s /etc/passwd ${remote}/unsafe-link && sha256sum ${remote}/saved.bin`, + ); + expect(setup.exitCode).toBe(0); + const sourceDigest = setup.result.trim().split(/\s/)[0]; + // Recovery resumes the same stopped source before calling syncOut. + await sandbox.stop(); + await sandbox.start(); + const execute = sandbox.process.executeCommand.bind(sandbox.process); + const failures: string[] = []; + sandbox.process.executeCommand = async (...args: Parameters) => { + if (args[0].includes("tar -c")) args[0] = args[0].replace(/\/tmp\/archive-capacity-proof\/\.paperclip-upload-[\w-]+\.tar/g, "/dev/full"); + const result = await execute(...args); + if (result.exitCode !== 0) failures.push(result.result); + return result; + }; + let archiveRecoveries = 0; + const sync = () => performSyncOut({ + sandbox, remoteDir: remote, timeoutSeconds: 120, + onArchiveRecovery: () => { archiveRecoveries += 1; }, + operations: [{ operationId: "preserve", files: [{ sourcePath: remote, targetPath: output, kind: "directory", exclude: ["ignored"] }] }], + }); + await sync(); + expect(failures.join("\n")).toMatch(/No space left on device|Wrote only \d+ of \d+ bytes/); + expect(archiveRecoveries).toBe(1); + const saved = await fs.readFile(path.join(output, "saved.bin")); + expect(createHash("sha256").update(saved).digest("hex")).toBe(sourceDigest); + expect((await fs.stat(path.join(output, "saved.bin"))).mode & 0o777).toBe(0o600); + expect(await fs.readFile(path.join(output, "safe-link"), "utf8")).toBe("retained"); + expect(await fs.readdir(output)).toEqual(expect.arrayContaining([".hidden", "saved.bin", "safe-link"])); + expect(await fs.readdir(output)).not.toEqual(expect.arrayContaining(["unsafe-link"])); + expect(await fs.readdir(output)).not.toEqual(expect.arrayContaining(["ignored"])); + const retained = await execute(`sha256sum ${remote}/saved.bin && readlink ${remote}/unsafe-link && find ${remote} -name '.paperclip-upload*'`); + expect(retained.result.trim().split("\n")).toEqual([`${sourceDigest} ${remote}/saved.bin`, "/etc/passwd"]); + // Reconnect to the retained workspace and export a later saved result. + await sandbox.stop(); + await sandbox.start(); + expect((await execute(`printf second > ${remote}/next.txt`)).exitCode).toBe(0); + await sync(); + expect(await fs.readFile(path.join(output, "next.txt"), "utf8")).toBe("second"); + expect(createHash("sha256").update(await fs.readFile(path.join(output, "saved.bin"))).digest("hex")).toBe(sourceDigest); + console.log("[archive-capacity-live] real Daytona stop/start, ENOSPC, bounded export, safe-link recovery, and second export passed"); + } finally { + await sandbox.delete(); + await fs.rm(output, { recursive: true, force: true }); + } +}, 300_000); diff --git a/packages/plugins/sandbox-providers/daytona/src/file-sync-capacity.test.ts b/packages/plugins/sandbox-providers/daytona/src/file-sync-capacity.test.ts new file mode 100644 index 00000000000..fbb792298b1 --- /dev/null +++ b/packages/plugins/sandbox-providers/daytona/src/file-sync-capacity.test.ts @@ -0,0 +1,133 @@ +import { spawn, spawnSync } from "node:child_process"; +import { promises as fs } from "node:fs"; +import os from "node:os"; +import path from "node:path"; +import { afterEach, expect, it, vi } from "vitest"; +import type { DaytonaPtyCreateOptions } from "./login-pty.js"; + +vi.mock("./plugin.js", async () => { + const { NOOP_PLUGIN_TRACER } = await import("@paperclipai/plugin-sdk"); + return { getPluginTracer: () => NOOP_PLUGIN_TRACER }; +}); +import { performSyncOut } from "./file-sync.js"; + +const roots: string[] = []; +afterEach(async () => { await Promise.all(roots.splice(0).map(root => fs.rm(root, { recursive: true, force: true }))); }); +const executable = (names: string[], version: string) => names.map(name => { + const resolved = spawnSync("/bin/sh", ["-c", 'command -v "$1"', "sh", name], { encoding: "utf8" }).stdout?.trim(); + return resolved && spawnSync(resolved, ["--version"], { encoding: "utf8" }).stdout?.includes(version) ? resolved : undefined; +}).find(Boolean); +// Match the provider's GNU tools. Hosts without these or Python's PTY relay +// skip this integration regression, like the nearby real-GNU-tar tests. +const requiredTools = { + tar: executable(["gtar", "tar"], "GNU tar"), + dd: executable(["gdd", "dd"], "coreutils"), + base64: executable(["gbase64", "base64"], "coreutils"), + bash: executable(["bash"], "GNU bash"), + python3: executable(["python3"], "Python 3."), +}; +const ptyRelay = ` +import os, pty, select, signal, sys +pid, fd = pty.fork() +if pid == 0: + os.execlp("bash", "bash", "--noprofile", "--norc") +def stop(*args): + try: os.kill(pid, signal.SIGKILL) + except ProcessLookupError: pass + sys.exit(1) +signal.signal(signal.SIGTERM, stop) +while True: + ready, _, _ = select.select([0, fd], [], []) + for src in ready: + try: data = os.read(src, 65536) + except OSError: data = b"" + if not data: + _, status = os.waitpid(pid, 0) + sys.exit(os.waitstatus_to_exitcode(status)) + dst = fd if src == 0 else 1 + while data: + data = data[os.write(dst, data):] +`; + +it.skipIf(Object.values(requiredTools).some(tool => !tool)).each(["success", "producer failure", "sparse archive over budget"])("preserves source bytes under archive capacity failure: %s", async scenario => { + const producerFails = scenario === "producer failure"; + const sparseOverBudget = scenario === "sparse archive over budget"; + const root = await fs.mkdtemp(path.join(os.tmpdir(), "daytona-capacity-")); + roots.push(root); + const remote = path.join(root, "remote"); + const output = path.join(root, "output"); + const bin = path.join(root, "bin"); + await fs.mkdir(remote); + await fs.mkdir(bin); + for (const name of ["dd", "base64", "bash"] as const) { + await fs.symlink(requiredTools[name]!, path.join(bin, name)); + } + const tar = requiredTools.tar!; + if (producerFails) { + await fs.writeFile(path.join(bin, "tar"), `#!/bin/sh\n'${tar}' "$@"\nstatus=$?\ncase " $* " in *" -f - "*) exit 23 ;; esac\nexit "$status"\n`, { mode: 0o755 }); + } else { + await fs.symlink(tar, path.join(bin, "tar")); + } + if (producerFails || sparseOverBudget) { + await fs.mkdir(output); + await fs.writeFile(path.join(output, "result.bin"), "newer host work"); + } + const env = { ...process.env, PATH: `${bin}:${process.env.PATH}` }; + const bytes = Buffer.alloc(5 * 1024 * 1024, sparseOverBudget ? 0 : "saved work\n"); + await fs.writeFile(path.join(remote, "result.bin"), sparseOverBudget ? "" : bytes, { mode: 0o600 }); + if (sparseOverBudget) await fs.truncate(path.join(remote, "result.bin"), bytes.length); + await fs.symlink("/etc/passwd", path.join(remote, "unsafe-link")); + await fs.symlink("result.bin", path.join(remote, "safe-link")); + const failures: string[] = []; + const sandbox = { + // A miniature provider-reported quota makes the sparse-file case bounded. + disk: sparseOverBudget ? 1 / 4096 : 1, + process: { + executeCommand: async (command: string) => { + const result = spawnSync("bash", ["-c", 'ulimit -f 128; exec sh -c "$1"', "bash", command], { env, encoding: "utf8" }); + if (result.status !== 0) failures.push(result.stderr); + return { exitCode: result.status ?? 1, result: result.stdout + result.stderr }; + }, + createPty: async (options: DaytonaPtyCreateOptions) => { + // A real PTY exercises raw-mode startup, echo suppression and the + // canonical line limit, including the longer safe-link retry script. + const child = spawn(requiredTools.python3!, ["-c", ptyRelay], { env, cwd: options.cwd }); + child.stdout.on("data", options.onData); + const completion = new Promise<{ exitCode: number }>(resolve => child.once("exit", code => resolve({ exitCode: code ?? 1 }))); + return { + waitForConnection: async () => {}, + sendInput: async (data: string | Uint8Array) => { child.stdin.write(Buffer.from(data)); }, + wait: () => completion, + kill: async () => { child.kill(); }, + disconnect: async () => { child.kill(); }, + }; + }, + }, + fs: { + downloadFiles: async (requests: Array<{ source: string; destination: string }>) => { + for (const request of requests) await fs.copyFile(request.source, request.destination); + return requests.map(({ source }) => ({ source })); + }, + deleteFile: (file: string) => fs.rm(file, { force: true }), + }, + }; + const sync = performSyncOut({ sandbox: sandbox as never, remoteDir: remote, timeoutSeconds: 30, + operations: [{ operationId: "export", files: [{ sourcePath: remote, targetPath: output, kind: "directory" }] }], + }); + if (producerFails || sparseOverBudget) { + await expect(sync).rejects.toThrow(sparseOverBudget ? "sandbox disk budget" : "without a successful archive receipt"); + expect(await fs.readFile(path.join(output, "result.bin"), "utf8")).toBe("newer host work"); + expect(await fs.readdir(output)).toEqual(["result.bin"]); + expect((await fs.readFile(path.join(remote, "result.bin"))).equals(bytes)).toBe(true); + return; + } + await sync; + expect(failures.join("\n")).toMatch(/File\s?size limit exceeded/i); + expect((await fs.readFile(path.join(output, "result.bin"))).equals(bytes)).toBe(true); + expect((await fs.readFile(path.join(remote, "result.bin"))).equals(bytes)).toBe(true); + expect((await fs.stat(path.join(output, "result.bin"))).mode & 0o777).toBe(0o600); + expect(await fs.readlink(path.join(output, "safe-link"))).toBe("result.bin"); + expect(await fs.readdir(output)).not.toContain("unsafe-link"); + expect(await fs.readlink(path.join(remote, "unsafe-link"))).toBe("/etc/passwd"); + expect((await fs.readdir(remote)).filter(name => name.startsWith(".paperclip-upload"))).toEqual([]); +}, 40_000); diff --git a/packages/plugins/sandbox-providers/daytona/src/file-sync.ts b/packages/plugins/sandbox-providers/daytona/src/file-sync.ts index c155bf1b86d..96824bac4fa 100644 --- a/packages/plugins/sandbox-providers/daytona/src/file-sync.ts +++ b/packages/plugins/sandbox-providers/daytona/src/file-sync.ts @@ -16,6 +16,7 @@ import type { } from "@paperclipai/plugin-sdk"; import { recordEnvironmentSyncError, withEnvironmentSyncTransferStep } from "@paperclipai/plugin-sdk"; import { getPluginTracer } from "./plugin.js"; +import { streamSandboxArchive } from "./archive-stream.js"; const execFileAsync = promisify(execFile); @@ -1297,13 +1298,13 @@ async function syncOutDirectoryMapping(input: { // Tar the source in-sandbox (naming top-level entries so no "." self-entry is // embedded), reproducing the `followSymlinks` → `-h` mapping, then stream the // single archive back over the native bulk channel. - const tarScript = [ + const tarScript = (archive: string | null) => [ `cd ${shellQuote(mapping.sourcePath)}`, "set -- *", 'if [ "$#" -eq 1 ] && [ "$1" = "*" ] && [ ! -e "$1" ] && [ ! -L "$1" ]; then set --; fi', 'for entry in .[!.]* ..?*; do [ -e "$entry" ] || [ -L "$entry" ] || continue; set -- "$@" "$entry"; done', - `if [ "$#" -eq 0 ]; then dd if=/dev/zero of=${shellQuote(remoteTar)} bs=1024 count=1; ` + - `else tar -c --no-xattrs ${mapping.followSymlinks ? "-h " : ""}${excludeFlags} -f ${shellQuote(remoteTar)} -- "$@"; fi`, + `if [ "$#" -eq 0 ]; then dd if=/dev/zero ${archive ? `of=${shellQuote(archive)} ` : ""}bs=1024 count=1; ` + + `else tar -c --no-xattrs ${mapping.followSymlinks ? "-h " : ""}${excludeFlags} -f ${shellQuote(archive ?? "-")} -- "$@"; fi`, ].join(" && "); // Rebuild the archive with files, directories, and relative links whose // resolved targets remain inside this mapping. Nothing is deleted or @@ -1322,35 +1323,55 @@ async function syncOutDirectoryMapping(input: { ` case "$_pc_real/" in "$_pc_root"/*) printf '%s\\0' "$_pc_link" ;; esac;`, 'done', ].join("\n"); - const confinedEntriesScript = [ + const confinedEntriesScript = (archive: string | null) => [ `cd ${shellQuote(mapping.sourcePath)}`, - `find . -mindepth 1 \\( ${prunePaths} \\) -prune -o -type l -exec sh -c ${shellQuote(filterLinks)} sh {} + -o \\( -type f -o -type d \\) -print0 > ${shellQuote(remoteList)}`, - `tar -c --no-xattrs --hard-dereference --no-recursion --null ${excludeFlags} -f ${shellQuote(remoteTar)} -T ${shellQuote(remoteList)}`, + `find . -mindepth 1 \\( ${prunePaths} \\) -prune -o -type l -exec sh -c ${shellQuote(filterLinks)} sh {} + -o \\( -type f -o -type d \\) -print0 ${archive ? `> ${shellQuote(remoteList)} && ` : "| "}` + + `tar -c --no-xattrs --hard-dereference --no-recursion --null ${excludeFlags} -f ${shellQuote(archive ?? "-")} -T ${shellQuote(archive ? remoteList : "-")}`, ].join(" && "); const localTar = path.join(tmp, "sync-out.tar"); let bytesTransferred = 0; + let streaming = false; try { for (let attempt = 0; attempt < 2; attempt += 1) { - await withEnvironmentSyncTransferStep("archive_create", () => - assertSandboxCommandOk(sandbox, `sh -c ${shellQuote(attempt === 0 ? tarScript : confinedEntriesScript)}`, timeoutSeconds, "syncOut tar")); + const script = attempt === 0 ? tarScript : confinedEntriesScript; + if (!streaming) { + try { + await withEnvironmentSyncTransferStep("archive_create", () => + assertSandboxCommandOk(sandbox, `sh -c ${shellQuote(script(remoteTar))}`, timeoutSeconds, "syncOut tar")); + } catch (error) { + // GNU tar can report either ENOSPC/EDQUOT or a short write when + // archive storage is exhausted. Preserve all source files and use + // a bounded, disk-free transfer instead of retrying that write. + if (!(error instanceof Error) || !/No space left on device|Disk quota exceeded|File\s?size limit exceeded|Wrote only \d+ of \d+ bytes/i.test(error.message)) throw error; + await sandbox.fs.deleteFile(remoteTar).catch(() => undefined); + streaming = true; + } + } guardRoundTrips += 1; - // `transfer` span: the real byte download — `sandbox.fs.downloadFiles`. - const responses = await withEnvironmentSyncTransferStep("archive_download", () => withProviderSpan({ - name: "transfer", - wallMsAttr: SPAN_ATTR.transferWallMs, - attributes: { - [SPAN_ATTR.transferGuardCount]: guardRoundTrips, - [SPAN_ATTR.transferDirection]: "outbound", - }, - run: () => - sandbox.fs.downloadFiles([{ source: remoteTar, destination: localTar }], timeoutSeconds), - })); - const response = responses.find((entry) => entry.source === remoteTar) ?? responses[0]; - if (!response || response.error) { - throw recordEnvironmentSyncError(new Error( - `Daytona syncOut directory download failed for ${mapping.sourcePath}: ${response?.error ?? "no response returned"}`, - ), { transferStep: "archive_download", transferFailureKind: response ? "download_failed" : "download_missing" }); + if (streaming) { + await withEnvironmentSyncTransferStep("archive_download", () => streamSandboxArchive({ + process: sandbox.process, cwd: remoteDir, archiveScript: script(null), destination: localTar, timeoutSeconds, + maxBytes: Math.floor(sandbox.disk * 1024 ** 3), + })); + } else { + // `transfer` span: the real byte download — `sandbox.fs.downloadFiles`. + const responses = await withEnvironmentSyncTransferStep("archive_download", () => withProviderSpan({ + name: "transfer", + wallMsAttr: SPAN_ATTR.transferWallMs, + attributes: { + [SPAN_ATTR.transferGuardCount]: guardRoundTrips, + [SPAN_ATTR.transferDirection]: "outbound", + }, + run: () => + sandbox.fs.downloadFiles([{ source: remoteTar, destination: localTar }], timeoutSeconds), + })); + const response = responses.find((entry) => entry.source === remoteTar) ?? responses[0]; + if (!response || response.error) { + throw recordEnvironmentSyncError(new Error( + `Daytona syncOut directory download failed for ${mapping.sourcePath}: ${response?.error ?? "no response returned"}`, + ), { transferStep: "archive_download", transferFailureKind: response ? "download_failed" : "download_missing" }); + } } bytesTransferred += await withEnvironmentSyncTransferStep("archive_download", async () => (await fs.stat(localTar)).size); try {