mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-30 13:20:54 +01:00
feat(adapter-utils): fast single-stream sandbox sync + byte-level progress
Replace the 32 KB serial-printf base64 chunk loop in the command-managed writeFile with a single `base64 -d` process fed via stdin (temp file + atomic rename), collapsing thousands of serial round-trips into one. A bounded large-chunk stdin fallback covers payloads too big for one stdin string. Stream readFile downloads through the runner's onLog byte counter, decoding incrementally instead of buffering one big string. Make writeFile/readFile the byte-level source of truth via an optional per-call progress hook, and have prepareSandboxManagedRuntime own the phase labels (workspace + per-asset upload, workspace restore), direction, and throttled log lines via the shared runtime-progress reporter. Tests: O(1) upload process for a multi-MB payload with exact (binary-safe) bytes; streamed download reports monotonic progress to the total; upload and restore emit throttled, labeled, directional percentage lines.
This commit is contained in:
1 parent
29c3d15507
commit
639e1715d9
4 files changed
+382
-84
No files matched your search
@@ -1,15 +1,79 @@
|
||||
import { mkdir, mkdtemp, readFile, rm, writeFile } from "node:fs/promises";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { execFile as execFileCallback } from "node:child_process";
|
||||
import { execFile as execFileCallback, spawn } from "node:child_process";
|
||||
import { promisify } from "node:util";
|
||||
import { afterEach, describe, expect, it } from "vitest";
|
||||
|
||||
import { prepareCommandManagedRuntime } from "./command-managed-runtime.js";
|
||||
import {
|
||||
createCommandManagedRuntimeClient,
|
||||
prepareCommandManagedRuntime,
|
||||
type CommandManagedRuntimeRunner,
|
||||
} from "./command-managed-runtime.js";
|
||||
import type { RunProcessResult } from "./server-utils.js";
|
||||
|
||||
const execFile = promisify(execFileCallback);
|
||||
|
||||
interface SpawnRunnerHandle {
|
||||
runner: CommandManagedRuntimeRunner;
|
||||
calls: Array<{ command: string; args?: string[]; cwd?: string; stdin?: string }>;
|
||||
}
|
||||
|
||||
// A runner that actually executes the shell scripts (piping stdin through a real
|
||||
// pipe so multi-MB payloads work) and replays stdout through onLog in several
|
||||
// chunks so the streaming readFile byte-counter is exercised.
|
||||
function makeSpawnRunner(): SpawnRunnerHandle {
|
||||
const calls: Array<{ command: string; args?: string[]; cwd?: string; stdin?: string }> = [];
|
||||
const runner: CommandManagedRuntimeRunner = {
|
||||
execute: async (input) =>
|
||||
await new Promise<RunProcessResult>((resolve) => {
|
||||
calls.push({ command: input.command, args: input.args, cwd: input.cwd, stdin: input.stdin });
|
||||
const startedAt = new Date().toISOString();
|
||||
const command =
|
||||
input.command === "sh" ? "/bin/sh" : input.command === "bash" ? "/bin/bash" : input.command;
|
||||
const child = spawn(command, input.args ?? [], {
|
||||
cwd: input.cwd,
|
||||
env: { ...process.env, ...input.env },
|
||||
});
|
||||
let stdout = "";
|
||||
let stderr = "";
|
||||
child.stdout.on("data", (chunk) => {
|
||||
stdout += chunk.toString("utf8");
|
||||
});
|
||||
child.stderr.on("data", (chunk) => {
|
||||
stderr += chunk.toString("utf8");
|
||||
});
|
||||
child.on("error", () => {
|
||||
resolve({ exitCode: 127, signal: null, timedOut: false, stdout, stderr, pid: null, startedAt });
|
||||
});
|
||||
child.on("close", async (code) => {
|
||||
if (input.onLog && stdout.length > 0) {
|
||||
const chunkSize = Math.max(1, Math.ceil(stdout.length / 4));
|
||||
for (let offset = 0; offset < stdout.length; offset += chunkSize) {
|
||||
await input.onLog("stdout", stdout.slice(offset, offset + chunkSize));
|
||||
}
|
||||
}
|
||||
resolve({
|
||||
exitCode: code ?? 0,
|
||||
signal: null,
|
||||
timedOut: false,
|
||||
stdout,
|
||||
stderr,
|
||||
pid: child.pid ?? null,
|
||||
startedAt,
|
||||
});
|
||||
});
|
||||
if (input.stdin != null) child.stdin.write(input.stdin);
|
||||
child.stdin.end();
|
||||
}),
|
||||
};
|
||||
return { runner, calls };
|
||||
}
|
||||
|
||||
function toArrayBuffer(buffer: Buffer): ArrayBuffer {
|
||||
return Uint8Array.from(buffer).buffer;
|
||||
}
|
||||
|
||||
describe("command managed runtime", () => {
|
||||
const cleanupDirs: string[] = [];
|
||||
|
||||
@@ -117,7 +181,9 @@ describe("command managed runtime", () => {
|
||||
await expect(readFile(path.join(remoteWorkspaceDir, "README.md"), "utf8")).resolves.toBe("local workspace\n");
|
||||
await expect(readFile(path.join(remoteWorkspaceDir, ".paperclip-runtime", "state.json"), "utf8")).rejects
|
||||
.toMatchObject({ code: "ENOENT" });
|
||||
expect(calls.every((call) => call.stdin == null)).toBe(true);
|
||||
// The single-stream upload pipes the tarball through exactly one stdin-backed
|
||||
// process (the speed fix); nothing else streams stdin.
|
||||
expect(calls.filter((call) => call.stdin != null).length).toBe(1);
|
||||
|
||||
await mkdir(path.join(remoteWorkspaceDir, ".paperclip-runtime"), { recursive: true });
|
||||
await writeFile(path.join(remoteWorkspaceDir, "README.md"), "remote workspace\n", "utf8");
|
||||
@@ -129,7 +195,9 @@ describe("command managed runtime", () => {
|
||||
.toBe("{\"keep\":true}\n");
|
||||
await expect(readFile(path.join(localWorkspaceDir, ".paperclip-runtime", "remote-state.json"), "utf8")).rejects
|
||||
.toMatchObject({ code: "ENOENT" });
|
||||
expect(calls.every((call) => call.stdin == null)).toBe(true);
|
||||
// Restore streams the download through `base64`/onLog (no stdin), so the only
|
||||
// stdin-backed call remains the single upload from prepare.
|
||||
expect(calls.filter((call) => call.stdin != null).length).toBe(1);
|
||||
});
|
||||
|
||||
it("runs setup commands from a stable root cwd when staging into a nested remote workspace dir", async () => {
|
||||
@@ -143,64 +211,7 @@ describe("command managed runtime", () => {
|
||||
await mkdir(remoteBaseDir, { recursive: true });
|
||||
await writeFile(path.join(localWorkspaceDir, "README.md"), "local workspace\n", "utf8");
|
||||
|
||||
const calls: Array<{
|
||||
command: string;
|
||||
args?: string[];
|
||||
cwd?: string;
|
||||
env?: Record<string, string>;
|
||||
stdin?: string;
|
||||
timeoutMs?: number;
|
||||
}> = [];
|
||||
const runner = {
|
||||
execute: async (input: {
|
||||
command: string;
|
||||
args?: string[];
|
||||
cwd?: string;
|
||||
env?: Record<string, string>;
|
||||
stdin?: string;
|
||||
timeoutMs?: number;
|
||||
}): Promise<RunProcessResult> => {
|
||||
calls.push({ ...input });
|
||||
const startedAt = new Date().toISOString();
|
||||
try {
|
||||
const result = await execFile(input.command === "sh" ? "/bin/sh" : input.command, input.args ?? [], {
|
||||
cwd: input.cwd,
|
||||
env: {
|
||||
...process.env,
|
||||
...input.env,
|
||||
},
|
||||
maxBuffer: 32 * 1024 * 1024,
|
||||
timeout: input.timeoutMs,
|
||||
});
|
||||
return {
|
||||
exitCode: 0,
|
||||
signal: null,
|
||||
timedOut: false,
|
||||
stdout: result.stdout,
|
||||
stderr: result.stderr,
|
||||
pid: null,
|
||||
startedAt,
|
||||
};
|
||||
} catch (error) {
|
||||
const err = error as NodeJS.ErrnoException & {
|
||||
stdout?: string;
|
||||
stderr?: string;
|
||||
code?: string | number | null;
|
||||
signal?: NodeJS.Signals | null;
|
||||
killed?: boolean;
|
||||
};
|
||||
return {
|
||||
exitCode: typeof err.code === "number" ? err.code : null,
|
||||
signal: err.signal ?? null,
|
||||
timedOut: Boolean(err.killed && input.timeoutMs),
|
||||
stdout: err.stdout ?? "",
|
||||
stderr: err.stderr ?? "",
|
||||
pid: null,
|
||||
startedAt,
|
||||
};
|
||||
}
|
||||
},
|
||||
};
|
||||
const { runner, calls } = makeSpawnRunner();
|
||||
|
||||
await prepareCommandManagedRuntime({
|
||||
runner,
|
||||
@@ -217,4 +228,68 @@ describe("command managed runtime", () => {
|
||||
expect(calls.every((call) => call.cwd === "/")).toBe(true);
|
||||
await expect(readFile(path.join(remoteWorkspaceDir, "README.md"), "utf8")).resolves.toBe("local workspace\n");
|
||||
});
|
||||
|
||||
it("uploads a multi-MB payload in a single process and preserves exact bytes", async () => {
|
||||
const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-command-write-"));
|
||||
cleanupDirs.push(rootDir);
|
||||
const remotePath = path.join(rootDir, "nested", "payload.bin");
|
||||
|
||||
// ~3 MB of every byte value so the test catches any non-binary-safe handling.
|
||||
const payload = Buffer.alloc(3 * 1024 * 1024);
|
||||
for (let i = 0; i < payload.length; i++) payload[i] = i % 256;
|
||||
|
||||
const { runner, calls } = makeSpawnRunner();
|
||||
const client = createCommandManagedRuntimeClient({ runner, commandCwd: "/", timeoutMs: 30_000 });
|
||||
|
||||
const progress: Array<{ done: number; total: number | null }> = [];
|
||||
await client.writeFile(remotePath, toArrayBuffer(payload), {
|
||||
onProgress: (done, total) => {
|
||||
progress.push({ done, total });
|
||||
},
|
||||
});
|
||||
|
||||
// Exactly one upload process: O(1) round-trips regardless of payload size.
|
||||
expect(calls.length).toBe(1);
|
||||
expect(calls[0].stdin).toBeTypeOf("string");
|
||||
|
||||
const written = await readFile(remotePath);
|
||||
expect(written.equals(payload)).toBe(true);
|
||||
|
||||
// Progress is monotonically non-decreasing and reaches the total.
|
||||
expect(progress.length).toBeGreaterThan(0);
|
||||
for (let i = 1; i < progress.length; i++) {
|
||||
expect(progress[i].done).toBeGreaterThanOrEqual(progress[i - 1].done);
|
||||
}
|
||||
expect(progress.at(-1)).toEqual({ done: payload.length, total: payload.length });
|
||||
});
|
||||
|
||||
it("streams a download through onLog and reports monotonic byte progress to the total", async () => {
|
||||
const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-command-read-"));
|
||||
cleanupDirs.push(rootDir);
|
||||
const remotePath = path.join(rootDir, "download.bin");
|
||||
|
||||
const payload = Buffer.alloc(512 * 1024);
|
||||
for (let i = 0; i < payload.length; i++) payload[i] = (i * 7) % 256;
|
||||
await writeFile(remotePath, payload);
|
||||
|
||||
const { runner } = makeSpawnRunner();
|
||||
const client = createCommandManagedRuntimeClient({ runner, commandCwd: "/", timeoutMs: 30_000 });
|
||||
|
||||
const progress: Array<{ done: number; total: number | null }> = [];
|
||||
const bytes = await client.readFile(remotePath, {
|
||||
onProgress: (done, total) => {
|
||||
progress.push({ done, total });
|
||||
},
|
||||
});
|
||||
|
||||
expect(Buffer.from(bytes as ArrayBuffer).equals(payload)).toBe(true);
|
||||
|
||||
// The runner replays stdout in several chunks, so progress arrives in steps.
|
||||
expect(progress.length).toBeGreaterThan(1);
|
||||
for (let i = 1; i < progress.length; i++) {
|
||||
expect(progress[i].done).toBeGreaterThanOrEqual(progress[i - 1].done);
|
||||
}
|
||||
expect(progress.every((entry) => entry.total === payload.length)).toBe(true);
|
||||
expect(progress.at(-1)?.done).toBe(payload.length);
|
||||
});
|
||||
});
|
||||
@@ -41,7 +41,14 @@ function mergeRuntimeExcludes(entries: string[] | undefined): string[] {
|
||||
return [...new Set([".paperclip-runtime", ...(entries ?? [])])];
|
||||
}
|
||||
|
||||
const REMOTE_WRITE_BASE64_CHUNK_SIZE = 32 * 1024;
|
||||
// Largest base64 body we hand to the runner as a single stdin string. Normal
|
||||
// multi-MB workspace/asset tarballs stay well under this and upload in one
|
||||
// round-trip; anything larger uses the bounded chunked-append fallback so a
|
||||
// runaway stdin string can't blow the runner/provider RPC limits.
|
||||
const REMOTE_WRITE_SINGLE_STREAM_MAX_BASE64_BYTES = 96 * 1024 * 1024;
|
||||
// Fallback chunk size (base64 bytes). Kept a multiple of 4 so each chunk is a
|
||||
// self-contained base64 unit that decodes cleanly on its own.
|
||||
const REMOTE_WRITE_FALLBACK_BASE64_CHUNK_SIZE = 4 * 1024 * 1024;
|
||||
|
||||
function toBuffer(bytes: Buffer | Uint8Array | ArrayBuffer): Buffer {
|
||||
if (Buffer.isBuffer(bytes)) return bytes;
|
||||
@@ -63,13 +70,21 @@ export function createCommandManagedRuntimeClient(input: {
|
||||
shellCommand?: "bash" | "sh" | null;
|
||||
}): SandboxManagedRuntimeClient {
|
||||
const shellCommand = preferredShellForSandbox(input.shellCommand);
|
||||
const runShell = async (script: string, opts: { stdin?: string; timeoutMs?: number } = {}) => {
|
||||
const runShell = async (
|
||||
script: string,
|
||||
opts: {
|
||||
stdin?: string;
|
||||
timeoutMs?: number;
|
||||
onLog?: (stream: "stdout" | "stderr", chunk: string) => Promise<void>;
|
||||
} = {},
|
||||
) => {
|
||||
const result = await input.runner.execute({
|
||||
command: shellCommand,
|
||||
args: shellCommandArgs(script),
|
||||
cwd: input.commandCwd,
|
||||
stdin: opts.stdin,
|
||||
timeoutMs: opts.timeoutMs ?? input.timeoutMs,
|
||||
onLog: opts.onLog,
|
||||
});
|
||||
requireSuccessfulResult(result, script);
|
||||
return result;
|
||||
@@ -79,25 +94,96 @@ export function createCommandManagedRuntimeClient(input: {
|
||||
makeDir: async (remotePath) => {
|
||||
await runShell(`mkdir -p ${shellQuote(remotePath)}`);
|
||||
},
|
||||
writeFile: async (remotePath, bytes) => {
|
||||
const body = toBuffer(bytes).toString("base64");
|
||||
writeFile: async (remotePath, bytes, options) => {
|
||||
const buffer = toBuffer(bytes);
|
||||
const total = buffer.byteLength;
|
||||
const body = buffer.toString("base64");
|
||||
const remoteDir = path.posix.dirname(remotePath);
|
||||
const remoteTempPath = `${remotePath}.paperclip-upload.b64`;
|
||||
const remoteTempPath = `${remotePath}.paperclip-upload`;
|
||||
|
||||
await runShell(
|
||||
`mkdir -p ${shellQuote(remoteDir)} && rm -f ${shellQuote(remoteTempPath)} && : > ${shellQuote(remoteTempPath)}`,
|
||||
);
|
||||
for (let offset = 0; offset < body.length; offset += REMOTE_WRITE_BASE64_CHUNK_SIZE) {
|
||||
const chunk = body.slice(offset, offset + REMOTE_WRITE_BASE64_CHUNK_SIZE);
|
||||
await runShell(`printf '%s' ${shellQuote(chunk)} >> ${shellQuote(remoteTempPath)}`);
|
||||
// Primary path: a single round-trip. Stream the entire base64 body to one
|
||||
// `base64 -d` process via stdin, decode straight into a temp file, then
|
||||
// atomically rename into place. This replaces the previous loop that did
|
||||
// one `printf >> tmpfile` shell round-trip per 32 KB — thousands of serial
|
||||
// processes for a large workspace — with exactly one process.
|
||||
if (body.length <= REMOTE_WRITE_SINGLE_STREAM_MAX_BASE64_BYTES) {
|
||||
await options?.onProgress?.(0, total);
|
||||
await runShell(
|
||||
`mkdir -p ${shellQuote(remoteDir)} && ` +
|
||||
`base64 -d > ${shellQuote(remoteTempPath)} && ` +
|
||||
`mv -f ${shellQuote(remoteTempPath)} ${shellQuote(remotePath)}`,
|
||||
{ stdin: body },
|
||||
);
|
||||
await options?.onProgress?.(total, total);
|
||||
return;
|
||||
}
|
||||
|
||||
// Bounded fallback for payloads too large to hand the runner as one stdin
|
||||
// string: append the base64 body to a remote temp file in large chunks
|
||||
// (orders of magnitude fewer round-trips than the old 32 KB loop), decoding
|
||||
// each self-contained chunk on arrival and emitting progress per write,
|
||||
// then atomically rename into place.
|
||||
await runShell(
|
||||
`base64 -d < ${shellQuote(remoteTempPath)} > ${shellQuote(remotePath)} && rm -f ${shellQuote(remoteTempPath)}`,
|
||||
`mkdir -p ${shellQuote(remoteDir)} && ` +
|
||||
`rm -f ${shellQuote(remoteTempPath)} && : > ${shellQuote(remoteTempPath)}`,
|
||||
);
|
||||
for (let offset = 0; offset < body.length; offset += REMOTE_WRITE_FALLBACK_BASE64_CHUNK_SIZE) {
|
||||
const chunk = body.slice(offset, offset + REMOTE_WRITE_FALLBACK_BASE64_CHUNK_SIZE);
|
||||
await runShell(`base64 -d >> ${shellQuote(remoteTempPath)}`, { stdin: chunk });
|
||||
const decodedSoFar = Math.min(total, Math.floor(((offset + chunk.length) * 3) / 4));
|
||||
await options?.onProgress?.(decodedSoFar, total);
|
||||
}
|
||||
await runShell(`mv -f ${shellQuote(remoteTempPath)} ${shellQuote(remotePath)}`);
|
||||
await options?.onProgress?.(total, total);
|
||||
},
|
||||
readFile: async (remotePath) => {
|
||||
const result = await runShell(`base64 < ${shellQuote(remotePath)}`);
|
||||
return Buffer.from(result.stdout.replace(/\s+/g, ""), "base64");
|
||||
readFile: async (remotePath, options) => {
|
||||
// Decoded file size up front so download progress can be a percentage.
|
||||
// Only paid when a progress hook is attached.
|
||||
let totalBytes: number | null = null;
|
||||
if (options?.onProgress) {
|
||||
const sizeResult = await runShell(`wc -c < ${shellQuote(remotePath)}`);
|
||||
const parsed = Number.parseInt(sizeResult.stdout.trim(), 10);
|
||||
totalBytes = Number.isFinite(parsed) && parsed >= 0 ? parsed : null;
|
||||
}
|
||||
|
||||
// Stream the remote `base64` stdout through a byte counter, decoding each
|
||||
// 4-char-aligned slice incrementally rather than buffering the whole
|
||||
// response as one string. Falls back to the buffered result when the
|
||||
// runner does not surface incremental stdout.
|
||||
const decodedChunks: Buffer[] = [];
|
||||
let b64Remainder = "";
|
||||
let decodedSoFar = 0;
|
||||
let streamed = false;
|
||||
const result = await runShell(`base64 < ${shellQuote(remotePath)}`, {
|
||||
onLog: async (stream, chunk) => {
|
||||
if (stream !== "stdout") return;
|
||||
streamed = true;
|
||||
const data = b64Remainder + chunk.replace(/\s+/g, "");
|
||||
const alignedLen = data.length - (data.length % 4);
|
||||
if (alignedLen > 0) {
|
||||
const decoded = Buffer.from(data.slice(0, alignedLen), "base64");
|
||||
decodedChunks.push(decoded);
|
||||
decodedSoFar += decoded.byteLength;
|
||||
}
|
||||
b64Remainder = data.slice(alignedLen);
|
||||
if (options?.onProgress) {
|
||||
const done = totalBytes != null ? Math.min(totalBytes, decodedSoFar) : decodedSoFar;
|
||||
await options.onProgress(done, totalBytes);
|
||||
}
|
||||
},
|
||||
});
|
||||
|
||||
let out: Buffer;
|
||||
if (streamed) {
|
||||
if (b64Remainder.length > 0) {
|
||||
decodedChunks.push(Buffer.from(b64Remainder, "base64"));
|
||||
}
|
||||
out = Buffer.concat(decodedChunks);
|
||||
} else {
|
||||
out = Buffer.from(result.stdout.replace(/\s+/g, ""), "base64");
|
||||
}
|
||||
await options?.onProgress?.(out.byteLength, totalBytes ?? out.byteLength);
|
||||
return out;
|
||||
},
|
||||
listFiles: async (remotePath) => {
|
||||
const result = await runShell(
|
||||
|
||||
@@ -197,6 +197,89 @@ describe("sandbox managed runtime", () => {
|
||||
await expect(readFile(path.join(remoteWorkspaceDir, "src", "main.ts"), "utf8")).resolves.toBe("x\n");
|
||||
});
|
||||
|
||||
it("emits throttled, labeled upload and restore progress with direction and percentages", async () => {
|
||||
const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-sandbox-progress-"));
|
||||
cleanupDirs.push(rootDir);
|
||||
const localWorkspaceDir = path.join(rootDir, "local-workspace");
|
||||
const remoteWorkspaceDir = path.join(rootDir, "remote-workspace");
|
||||
const localAssetsDir = path.join(rootDir, "local-assets");
|
||||
await mkdir(localWorkspaceDir, { recursive: true });
|
||||
await mkdir(localAssetsDir, { recursive: true });
|
||||
await writeFile(path.join(localWorkspaceDir, "README.md"), "workspace\n", "utf8");
|
||||
await writeFile(path.join(localAssetsDir, "skill.md"), "skill\n", "utf8");
|
||||
|
||||
// Drive byte progress in 100 fine (1%) increments so the throttle has many
|
||||
// chances to emit; the reporter must collapse them to ~one line per 10% step.
|
||||
const driveProgress = async (
|
||||
total: number,
|
||||
onProgress: ((done: number, total: number | null) => void | Promise<void>) | undefined,
|
||||
) => {
|
||||
if (!onProgress) return;
|
||||
for (let i = 1; i <= 100; i++) {
|
||||
await onProgress(Math.floor((total * i) / 100), total);
|
||||
}
|
||||
};
|
||||
|
||||
const client: SandboxManagedRuntimeClient = {
|
||||
makeDir: async (remotePath) => {
|
||||
await mkdir(remotePath, { recursive: true });
|
||||
},
|
||||
writeFile: async (remotePath, bytes, options) => {
|
||||
await mkdir(path.dirname(remotePath), { recursive: true });
|
||||
const buffer = Buffer.from(bytes);
|
||||
await writeFile(remotePath, buffer);
|
||||
await driveProgress(buffer.byteLength, options?.onProgress);
|
||||
},
|
||||
readFile: async (remotePath, options) => {
|
||||
const buffer = await readFile(remotePath);
|
||||
await driveProgress(buffer.byteLength, options?.onProgress);
|
||||
return buffer;
|
||||
},
|
||||
listFiles: async () => [],
|
||||
remove: async (remotePath) => {
|
||||
await rm(remotePath, { recursive: true, force: true });
|
||||
},
|
||||
run: async (command) => {
|
||||
await execFile("sh", ["-c", command], { maxBuffer: 32 * 1024 * 1024 });
|
||||
},
|
||||
};
|
||||
|
||||
const lines: string[] = [];
|
||||
const prepared = await prepareSandboxManagedRuntime({
|
||||
spec: {
|
||||
transport: "sandbox",
|
||||
provider: "test",
|
||||
sandboxId: "sandbox-1",
|
||||
remoteCwd: remoteWorkspaceDir,
|
||||
timeoutMs: 30_000,
|
||||
apiKey: null,
|
||||
},
|
||||
adapterKey: "test-adapter",
|
||||
client,
|
||||
workspaceLocalDir: localWorkspaceDir,
|
||||
assets: [{ key: "skills", localDir: localAssetsDir }],
|
||||
onProgress: (line) => {
|
||||
lines.push(line);
|
||||
},
|
||||
});
|
||||
|
||||
const uploadWorkspaceLines = lines.filter((line) => line.includes("Syncing workspace to sandbox"));
|
||||
const uploadAssetLines = lines.filter((line) => line.includes("Syncing skills to sandbox"));
|
||||
expect(uploadWorkspaceLines.length).toBeGreaterThan(0);
|
||||
expect(uploadAssetLines.length).toBeGreaterThan(0);
|
||||
// 100 reported increments must be throttled to at most ~one line per 10% step.
|
||||
expect(uploadWorkspaceLines.length).toBeLessThanOrEqual(11);
|
||||
// Reaches 100% and shows the MB breakdown.
|
||||
expect(uploadWorkspaceLines.some((line) => line.includes("100%"))).toBe(true);
|
||||
expect(uploadWorkspaceLines.every((line) => /\(\d+\.\d\/\d+\.\d MB\)/.test(line))).toBe(true);
|
||||
|
||||
await prepared.restoreWorkspace();
|
||||
const restoreLines = lines.filter((line) => line.includes("Restoring workspace from sandbox"));
|
||||
expect(restoreLines.length).toBeGreaterThan(0);
|
||||
expect(restoreLines.length).toBeLessThanOrEqual(11);
|
||||
expect(restoreLines.some((line) => line.includes("100%"))).toBe(true);
|
||||
});
|
||||
|
||||
it("creates a valid empty workspace tarball when the local workspace is empty", async () => {
|
||||
const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-sandbox-empty-"));
|
||||
cleanupDirs.push(rootDir);
|
||||
|
||||
@@ -4,7 +4,7 @@ import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { promisify } from "node:util";
|
||||
import { captureDirectorySnapshot, mergeDirectoryWithBaseline } from "./workspace-restore-merge.js";
|
||||
import type { RuntimeProgressSink } from "./runtime-progress.js";
|
||||
import { createRuntimeProgressReporter, type RuntimeProgressSink } from "./runtime-progress.js";
|
||||
|
||||
const execFile = promisify(execFileCallback);
|
||||
|
||||
@@ -24,10 +24,23 @@ export interface SandboxManagedRuntimeAsset {
|
||||
exclude?: string[];
|
||||
}
|
||||
|
||||
/**
|
||||
* Per-call byte-level progress hook. `transferredBytes`/`totalBytes` are decoded
|
||||
* file bytes (not the base64 wire size). `totalBytes` is null when the size is
|
||||
* not known up front. The transport is the source of truth for byte counts; the
|
||||
* orchestrator owns the phase label and direction.
|
||||
*/
|
||||
export interface SandboxTransferProgressOptions {
|
||||
onProgress?: (transferredBytes: number, totalBytes: number | null) => void | Promise<void>;
|
||||
}
|
||||
|
||||
export interface SandboxManagedRuntimeClient {
|
||||
makeDir(remotePath: string): Promise<void>;
|
||||
writeFile(remotePath: string, bytes: ArrayBuffer): Promise<void>;
|
||||
readFile(remotePath: string): Promise<Buffer | Uint8Array | ArrayBuffer>;
|
||||
writeFile(remotePath: string, bytes: ArrayBuffer, options?: SandboxTransferProgressOptions): Promise<void>;
|
||||
readFile(
|
||||
remotePath: string,
|
||||
options?: SandboxTransferProgressOptions,
|
||||
): Promise<Buffer | Uint8Array | ArrayBuffer>;
|
||||
listFiles(remotePath: string): Promise<string[]>;
|
||||
remove(remotePath: string): Promise<void>;
|
||||
run(command: string, options: { timeoutMs: number }): Promise<void>;
|
||||
@@ -262,6 +275,36 @@ function tarExcludeFlags(exclude: string[] | undefined): string {
|
||||
return ["._*", ...(exclude ?? [])].map((entry) => `--exclude ${shellQuote(entry)}`).join(" ");
|
||||
}
|
||||
|
||||
// Bridge a single byte-level transfer to the throttled progress reporter. The
|
||||
// transport reports decoded bytes via `options.onProgress`; the reporter turns
|
||||
// them into a throttled, fully-formatted log line. `finish()` emits the terminal
|
||||
// completion line (idempotent) once the transfer returns.
|
||||
function makeTransferProgress(
|
||||
sink: RuntimeProgressSink | undefined,
|
||||
phase: "Syncing" | "Restoring",
|
||||
direction: "to" | "from",
|
||||
label: string,
|
||||
): { options: SandboxTransferProgressOptions | undefined; finish: () => Promise<void> } {
|
||||
if (!sink) return { options: undefined, finish: async () => {} };
|
||||
const reporter = createRuntimeProgressReporter({
|
||||
sink,
|
||||
phase,
|
||||
direction,
|
||||
target: "sandbox",
|
||||
label,
|
||||
});
|
||||
return {
|
||||
options: {
|
||||
onProgress: async (transferredBytes, totalBytes) => {
|
||||
await reporter.report(transferredBytes, totalBytes);
|
||||
},
|
||||
},
|
||||
finish: async () => {
|
||||
await reporter.complete();
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
export async function prepareSandboxManagedRuntime(input: {
|
||||
spec: SandboxRemoteExecutionSpec;
|
||||
adapterKey: string;
|
||||
@@ -291,7 +334,13 @@ export async function prepareSandboxManagedRuntime(input: {
|
||||
const workspaceTarBytes = await fs.readFile(workspaceTarPath);
|
||||
const remoteWorkspaceTar = path.posix.join(runtimeRootDir, "workspace-upload.tar");
|
||||
await input.client.makeDir(runtimeRootDir);
|
||||
await input.client.writeFile(remoteWorkspaceTar, toArrayBuffer(workspaceTarBytes));
|
||||
const workspaceUpload = makeTransferProgress(input.onProgress, "Syncing", "to", "workspace");
|
||||
await input.client.writeFile(
|
||||
remoteWorkspaceTar,
|
||||
toArrayBuffer(workspaceTarBytes),
|
||||
workspaceUpload.options,
|
||||
);
|
||||
await workspaceUpload.finish();
|
||||
const preservedNames = new Set([".paperclip-runtime", ...(input.preserveAbsentOnRestore ?? [])]);
|
||||
const findPreserveArgs = [...preservedNames].map((entry) => `! -name ${shellQuote(entry)}`).join(" ");
|
||||
await input.client.run(
|
||||
@@ -315,7 +364,9 @@ export async function prepareSandboxManagedRuntime(input: {
|
||||
const assetTarBytes = await fs.readFile(assetTarPath);
|
||||
const remoteAssetDir = path.posix.join(runtimeRootDir, asset.key);
|
||||
const remoteAssetTar = path.posix.join(runtimeRootDir, `${asset.key}-upload.tar`);
|
||||
await input.client.writeFile(remoteAssetTar, toArrayBuffer(assetTarBytes));
|
||||
const assetUpload = makeTransferProgress(input.onProgress, "Syncing", "to", asset.key);
|
||||
await input.client.writeFile(remoteAssetTar, toArrayBuffer(assetTarBytes), assetUpload.options);
|
||||
await assetUpload.finish();
|
||||
await input.client.run(
|
||||
`sh -c ${shellQuote(
|
||||
`rm -rf ${shellQuote(remoteAssetDir)} && ` +
|
||||
@@ -338,7 +389,8 @@ export async function prepareSandboxManagedRuntime(input: {
|
||||
workspaceRemoteDir,
|
||||
runtimeRootDir,
|
||||
assetDirs,
|
||||
restoreWorkspace: async (_onProgress?: RuntimeProgressSink) => {
|
||||
restoreWorkspace: async (onProgress?: RuntimeProgressSink) => {
|
||||
const restoreSink = onProgress ?? input.onProgress;
|
||||
await withTempDir("paperclip-sandbox-restore-", async (tempDir) => {
|
||||
const remoteWorkspaceTar = path.posix.join(runtimeRootDir, "workspace-download.tar");
|
||||
await input.client.run(
|
||||
@@ -349,7 +401,9 @@ export async function prepareSandboxManagedRuntime(input: {
|
||||
)}`,
|
||||
{ timeoutMs: input.spec.timeoutMs },
|
||||
);
|
||||
const archiveBytes = await input.client.readFile(remoteWorkspaceTar);
|
||||
const workspaceRestore = makeTransferProgress(restoreSink, "Restoring", "from", "workspace");
|
||||
const archiveBytes = await input.client.readFile(remoteWorkspaceTar, workspaceRestore.options);
|
||||
await workspaceRestore.finish();
|
||||
await input.client.remove(remoteWorkspaceTar).catch(() => undefined);
|
||||
const localArchivePath = path.join(tempDir, "workspace.tar");
|
||||
const extractedDir = path.join(tempDir, "workspace");
|
||||
|
||||
Reference in new issue
Block a user