fix(daytona): send archive chunks in bounded windows (#19339)

## Thinking Path > - Paperclip manages agents and saves their work when execution stops. > - The Daytona plugin saves an archive through a terminal when disk capacity prevents the normal archive or download copy. > - That fallback sent one 40 KiB chunk and waited for its host write acknowledgement. > - Network delay then applied to every small chunk. A large archive could exceed the recovery deadline even while bytes were moving safely. > - This pull request sends a bounded window of chunks and still acknowledges every completed host write. > - The benefit is faster recovery over delayed connections, with bounded memory and the existing size, space and completion checks. ## Linked Issues or Issue Description Refs #19034, which added the capacity fallback. **What happened?** The archive fallback waits for one network round trip after each 40 KiB chunk. A saved workspace archive can be much larger than that chunk. Repeating a timed-out transfer does not remove this throughput limit. **Expected behavior** Allow a small amount of data to travel before waiting for acknowledgements. Keep memory bounded, write bytes in order, and stop after any write or protocol failure. **Steps to reproduce** 1. Run the real Bash archive sender against a host that delays each acknowledgement by 400 ms. 2. Transfer 1 MiB plus a partial final chunk with a four-second deadline. 3. The original implementation times out. The bounded-window implementation completes and preserves the exact source bytes. **Paperclip version or commit** The failing regression uses the original implementation from `6a66c09f7f7`. ## What Changed - Send at most sixteen 40 KiB chunks before collecting their acknowledgements. Drain the final partial window before the completion receipt. - Serialize host writes and reject more than sixteen pending frames. Stop queued writes on the first failure. - Keep per-frame size checks, total archive limits, the 1 GiB host space reserve, the transfer deadline and bounded teardown. - Add a real Bash regression with delayed acknowledgements, plus overflow and failed-write coverage. Update the plugin documentation. - Update the agent-creator test fixture for the required directory membership role introduced by #15183, which otherwise breaks merged CI typechecking. ## Verification - Failing first: the final real-Bash regression fails against the original source with `Daytona archive stream timed out`. It passes with the change. The empty-stream case passes too. - Standalone Daytona package `pnpm test`: 578 passed, 8 skipped. The skips are the existing opt-in live tests. - Standalone package `pnpm typecheck` and `pnpm build`: passed. - `pnpm test src/file-sync-capacity.live.test.ts` with a staging Daytona key: 2 passed on isolated test sandboxes. This covers actual terminal transport, stop/start, archive and file-copy capacity failures, exact saved hashes, safe links, cleanup and a second export. The tests removed their own sandboxes. - The production dependency assumption is bounded: at most 640 KiB of decoded payloads can wait for host writes. The parser also has a bounded encoded buffer. No background work or idle timer is added. - Initial hosted CI failed with TS2741 in `ui/src/lib/agent-creator.test.ts` after #15183 made `membershipRole` required. Rebased onto that change and added the nullable field to the fixture. UI `pnpm typecheck`, all 4 existing creator-label tests, and `pnpm check:token-gates` now pass locally. Hosted CI subsequently passed on the updated head. - `git diff --check` and a local secret and PII scan: passed. - Full repository checks were not repeated locally for this standalone plugin change. Hosted verification at `bf6b3cb6a622a9523c72b86fdfe616555c8f3ef8`: all 53 check runs passed, with the 2 opt-in Storybook checks skipped; both required gates (`ci / verify`, `ci / e2e`) passed. Greptile is 5/5 at this exact head, with no review threads or merge conflicts. The separate non-required `security/snyk (cryppadotta)` status remains pending; it is not claimed as passed. ## Risks - A burst can queue up to sixteen chunks instead of one. The receiver enforces this bound and acknowledges a chunk only after its ordered host write. - The timeout budget is unchanged. A connection that cannot finish within that budget can still fail. This change does not retry a provider turn or change file-recovery authority. - Source confinement, archive validation, successful producer completion and publication rules are unchanged. No schema migration. ## Model Used - OpenAI `gpt-6.1-sol` with xhigh reasoning, in Codex through T3 Code. Tool use included source edits, real Bash and filesystem tests, and isolated Daytona tests. The runtime does not expose its context window. ## 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 <noreply@paperclip.ing>
2026-11-09 07:10:34 +01:00 · 2026-11-05 13:12:11 -08:00
1 parent fb2e07ebdd
commit c3550cfb38
@@ -37,7 +37,7 @@ Notes:
 - The SDK returns full log snapshots, so fallback bandwidth grows with retained output; it has no paged snapshot or stream cancellation API. A recovery-read timeout keeps partial output and explicitly reports whether command exit remains unconfirmed; it does not prove the remote command stopped. Provider/session teardown remains responsible for closing an outstanding socket. A final snapshot reconciles bytes missed by the stream before returning a recorded exit.
 - A sandbox record can survive the loss of its underlying container. Resume treats it as expired only when a fresh provider read confirms the exact missing-container error for that sandbox and marks it unrecoverable. Unknown errors and failed confirmation reads preserve the lease. The host still requires a verified native-runner backup before replacement.
 - Daytona rejects Stop for a sandbox in its `error` state, for example after the sandbox disk fills. The stop-only hook confirms the retention policy first. If Daytona marks the error recoverable, the hook calls Recover once and waits at most 20 seconds for the sandbox to start. Then it stops the sandbox normally. The hook never calls Recover a second time in the same call. An unrecoverable error, a sandbox that is back in `error` after recovery, or an unfinished recovery returns a typed `PluginEnvironmentStopError` instead of the generic stop error. The sandbox is never deleted.
- Directory export normally builds a sandbox archive and downloads it. If archive creation exhausts disk space, quota, or the file-size limit, export streams the archive through a raw terminal instead. Each bounded chunk waits for a host write acknowledgement. Import still requires successful tar completion and the existing archive/path validation. The total archive cannot exceed the provider-reported sandbox disk allocation, and each host write must leave a 1 GiB free-space reserve. This fallback writes no sandbox archive or command log, preserves source files, and retains the configured transfer timeout. Terminal cleanup has a separate five-second limit and cannot hold the host archive open. It requires Bash and GNU `tar`, `dd`, and `base64` in the sandbox. It does not repair a provider session that already failed to save its state or increase the sandbox's disk allocation.
+- 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.

 ## Local development
@@ -0,0 +1,87 @@
+import { spawn, spawnSync } from "node:child_process";
+import { randomBytes } from "node:crypto";
+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 { DuplexChannelSession } from "./duplex-command-stream.js";
+
+const { openSession } = vi.hoisted(() => ({ openSession: vi.fn() }));
+vi.mock("./duplex-command-stream.js", () => ({ openDaytonaDuplexChannelSession: openSession }));
+import { streamSandboxArchive } from "./archive-stream.js";
+
+const roots: string[] = [];
+const timers = new Set<ReturnType<typeof setTimeout>>();
+const children = new Set<ReturnType<typeof spawn>>();
+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);
+const tools = { dd: executable(["gdd", "dd"], "coreutils"), base64: executable(["gbase64", "base64"], "coreutils") };
+
+function kill(child: ReturnType<typeof spawn>) {
+  if (child.exitCode !== null || child.signalCode !== null || !child.pid) return;
+  try { process.kill(-child.pid, "SIGKILL"); } catch (error) {
+    if ((error as NodeJS.ErrnoException).code !== "ESRCH") throw error;
+  }
+}
+
+afterEach(async () => {
+  for (const timer of timers) clearTimeout(timer);
+  timers.clear();
+  for (const child of children) kill(child);
+  children.clear();
+  vi.resetAllMocks();
+  await Promise.all(roots.splice(0).map(root => fs.rm(root, { recursive: true, force: true })));
+});
+
+it.skipIf(process.platform === "win32" || !tools.dd || !tools.base64).each([0, 1024 * 1024 + 17])(
+  "transfers %i bytes within its deadline across delayed acknowledgements", async size => {
+    const root = await fs.mkdtemp(path.join(os.tmpdir(), "archive-credit-window-"));
+    roots.push(root);
+    const bin = path.join(root, "bin");
+    await fs.mkdir(bin);
+    await fs.symlink(tools.dd!, path.join(bin, "dd"));
+    await fs.symlink(tools.base64!, path.join(bin, "base64"));
+    const source = path.join(root, "source.bin");
+    const destination = path.join(root, "saved.bin");
+    const bytes = randomBytes(size);
+    await fs.writeFile(source, bytes);
+    let closed = false;
+    openSession.mockImplementation(async (_process, command: string[]) => {
+      const marker = command.join(" ").match(/paperclip-archive-[a-f0-9-]+/)![0];
+      const child = spawn(command[0], command.slice(1), { cwd: root, detached: true,
+        env: { ...process.env, PATH: `${bin}:${process.env.PATH}` }, stdio: ["pipe", "pipe", "pipe"] });
+      children.add(child);
+      let stderr = "";
+      child.stderr!.on("data", chunk => { stderr += chunk.toString(); });
+      const completion = new Promise<{ exitCode: number | null; transportClosed: boolean }>((resolve, reject) => {
+        child.once("error", reject);
+        child.once("close", code => {
+          if (code && !closed) reject(new Error(stderr || `archive producer exited ${code}`));
+          else resolve({ exitCode: code, transportClosed: false });
+        });
+      });
+      return {
+        onData: callback => { child.stdout!.on("data", callback); },
+        write: data => {
+          if (!data.toString().startsWith(marker)) { child.stdin!.write(data); return; }
+          const timer = setTimeout(() => {
+            timers.delete(timer);
+            if (!closed) child.stdin!.write(data);
+          }, 400);
+          timers.add(timer);
+        },
+        wait: () => completion,
+        kill: () => kill(child),
+        close: async () => { closed = true; kill(child); await completion; },
+      } satisfies DuplexChannelSession;
+    });
+    const quote = (value: string) => `'${value.replace(/'/g, `'"'"'`)}'`;
+    await streamSandboxArchive({ process: {} as never, cwd: root, archiveScript: `cat ${quote(source)}`,
+      destination, timeoutSeconds: 4, maxBytes: size + 1024 });
+    expect(await fs.readFile(destination)).toEqual(bytes);
+    expect(await fs.readFile(source)).toEqual(bytes);
+    expect(closed).toBe(true);
+  }, 15_000,
+);
@@ -87,3 +87,30 @@ it("closes a session that connects after the deadline", async () => {
  await vi.waitFor(() => expect(close).toHaveBeenCalledOnce());
  expect(onData).not.toHaveBeenCalled();
 });
+
+it.each(["window overflow", "host write failure"])("stops queued frames after %s", async fault => {
+  const root = await fs.mkdtemp(path.join(os.tmpdir(), "archive-stream-window-"));
+  roots.push(root);
+  const destination = path.join(root, "archive.tar");
+  const file = await fs.open(destination, "w");
+  const writeFile = vi.fn(async () => { throw new Error("injected host write failure"); });
+  if (fault === "host write failure") vi.mocked(open).mockResolvedValue({ writeFile, close: () => file.close() } as never);
+  const close = vi.fn(async () => {});
+  openSession.mockImplementation(async (_process, command: string[]) => {
+    const marker = command.join(" ").match(/paperclip-archive-[a-f0-9-]+/)![0];
+    let emit!: (bytes: Uint8Array) => void;
+    return {
+      onData: callback => { emit = callback; queueMicrotask(() => emit(Buffer.from(`${marker}:ready\n`))); },
+      write: () => { queueMicrotask(() => emit(Buffer.from(`${marker}:start\n${"data:Ynl0ZXM=\n".repeat(fault === "window overflow" ? 17 : 16)}`))); },
+      wait: () => new Promise(() => {}), kill: vi.fn(), close,
+    } satisfies DuplexChannelSession;
+  });
+  try {
+    await expect(streamSandboxArchive({ process: {} as never, cwd: "/workspace", archiveScript: "cat source",
+      destination, maxBytes: 1024, timeoutSeconds: 5 })).rejects.toThrow(
+      fault === "window overflow" ? "pending frame limit" : "injected host write failure",
+    );
+    if (fault === "host write failure") expect(writeFile).toHaveBeenCalledOnce();
+    expect(close).toHaveBeenCalledOnce();
+  } finally { await file.close(); }
+});
@@ -2,10 +2,10 @@ 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 CREDIT_WINDOW = 16;
 const MAX_FRAME_BYTES = Math.ceil(CHUNK_BYTES / 3) * 4 + 128;
+const MAX_BUFFER_BYTES = MAX_FRAME_BYTES * CREDIT_WINDOW;
 const HOST_FREE_RESERVE_BYTES = 1024 ** 3;
 const CLEANUP_TIMEOUT_MS = 5_000;
 const quote = (value: string) => `'${value.replace(/'/g, `'"'"'`)}'`;
@@ -22,8 +22,6 @@ async function closeArchiveSession(session: DuplexChannelSession): Promise<void>
  }
 }

-/** 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;
@@ -41,13 +39,22 @@ export async function streamSandboxArchive(input: {
  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 receiveAcknowledgements = [
+    'while [ "$outstanding" -gt 0 ]; do',
+    `IFS= read -r ack <&3 && [ "$ack" = ${quote(marker)} ] || exit 1;`,
+    'outstanding=$((outstanding - 1));',
+    "done",
+  ].join(" ");
  const sender = [
+    "outstanding=0;",
    "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",
+    "outstanding=$((outstanding + 1));",
+    `if [ "$outstanding" -eq ${CREDIT_WINDOW} ]; then ${receiveAcknowledgements}; fi;`,
+    "done;",
+    receiveAcknowledgements,
  ].join(" ");
  // Duplicate the original input before constructing the tar pipeline.
  const command = `exec 3<&0; printf '\\n%s\\n' ${quote(start)}; { ${input.archiveScript}; } | ( ${sender} ); `;
@@ -59,7 +66,7 @@ export async function streamSandboxArchive(input: {
  let started = false;
  let launched = false;
  let complete = false;
-  let writing = false;
+  let pendingFrames = 0;
  let buffer = "";
  let receivedBytes = 0;
  let pending = Promise.resolve();
@@ -76,7 +83,7 @@ export async function streamSandboxArchive(input: {
    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; }
+      if (buffer.length > MAX_BUFFER_BYTES) { fail("Daytona archive stream exceeded its buffer 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);
@@ -88,18 +95,19 @@ export async function streamSandboxArchive(input: {
          continue;
        }
        if (!started) { if (line === start) started = true; continue; }
-        if (line === done && !complete && !writing) {
+        if (line === done && !complete && pendingFrames === 0) {
          complete = true;
          session!.write(Buffer.from(`${done}\n`));
          continue;
        }
-        if (!line || complete || writing || !line.startsWith("data:")) { fail("Invalid Daytona archive stream frame"); return; }
+        if (!line || complete || !line.startsWith("data:")) { fail("Invalid Daytona archive stream frame"); return; }
+        if (pendingFrames >= CREDIT_WINDOW) { fail("Daytona archive stream exceeded its pending frame limit"); 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;
+        pendingFrames += 1;
        pending = pending.then(async () => {
          if (stopped) return;
          if (receivedBytes + data.length > input.maxBytes) {
@@ -112,9 +120,9 @@ export async function streamSandboxArchive(input: {
          if (stopped) return;
          await file.writeFile(data);
          receivedBytes += data.length;
-          writing = false;
+          pendingFrames -= 1;
          session!.write(Buffer.from(`${marker}\n`));
-        }).catch(error => { rejectFailure(error); });
+        }).catch(error => { stopped = true; rejectFailure(error); });
      }
    });
    const exit = await session.wait();
@@ -3,7 +3,7 @@ import type { CompanyUserDirectoryEntry } from "../api/access";
 import { humanAgentCreatorUserId, resolveAgentCreatorLabel } from "./agent-creator";

 function member(principalId: string, name: string | null, email: string | null = null): CompanyUserDirectoryEntry {
-  return { principalId, status: "active", user: { id: principalId, name, email, image: null } };
+  return { principalId, status: "active", membershipRole: null, user: { id: principalId, name, email, image: null } };
 }

 describe("agent creator label", () => {