feat(adapter-utils): add observable sandbox sync progress (#8395)
## Thinking Path > - Paperclip is the control plane for running AI-agent companies, so long-running remote work needs to stay observable to human operators. > - Cloud / sandbox agents are an active roadmap area, and their workspace sync path is part of the runtime substrate every remote coding run depends on. > - In the sandbox and SSH execution-target flows, Paperclip logged that sync had started, then often went silent for the full transfer window. > - That made large remote syncs feel stalled and also hid a real performance problem in the command-managed sandbox upload path. > - The first part of this pull request threads a throttled progress-reporting surface through the adapter execution-target stack so sync and restore work can emit meaningful updates. > - The second part fixes the command-managed sandbox transport itself: it removes the old serial 32KB append bottleneck, but also falls back away from the single-stream path when a provider-backed sandbox runner cannot surface mid-flight stdin progress. > - The result is that sandbox and SSH transfers are both faster and more observable, including the live Daytona-style sandbox case that previously only emitted `0%` and `100%`. ## Linked Issues or Issue Description No public GitHub issue exists for this bug, so it is described inline below following the bug report template. ### What happened - Remote sandbox and SSH workspace syncs could spend a long time transferring data while only logging a start line (`Syncing workspace and runtime assets to sandbox environment`) and, at best, a terminal line. - In the command-managed sandbox path, the original upload implementation also paid a large performance cost by appending base64 data in many small sequential remote writes (thousands of serial 32KB round-trips on a large workspace). - After the initial transport rewrite, live provider-backed sandbox runs still only emitted `0%` and `100%` because the single-stream stdin RPC buffered progress until completion. ### Expected behavior - Long-running sandbox and SSH syncs should periodically report how much of the transfer is complete (a percentage and/or MB transferred) so an operator can tell the run is healthy and making progress rather than stuck. - The main sandbox upload path should not be artificially slow. - A transfer that fails partway should leave an explicit failure marker in the log rather than a dangling intermediate percentage. ### Steps to reproduce 1. Run an agent against a sandbox (command-managed) or SSH (remote-managed) execution target with a non-trivial workspace. 2. Watch the run log during the workspace/runtime asset sync phase. 3. Observe that the log shows the sync start line and then stays silent for the full transfer (live provider-backed sandbox runs only show `0%` then `100%`). ### Paperclip version or commit - Branch `PAPA-825-provide-status-updates-when-syncing-sandboxes` off `master`. ### Deployment mode - Self-hosted / local instance using sandbox (command-managed) and SSH (remote-managed) execution targets, including provider-backed sandbox runners. ## What Changed - Added shared throttled runtime progress reporting and threaded `onProgress` through the adapter execution-target surface and adapter `execute.ts` entrypoints. - Added sync and restore progress reporting for the command-managed sandbox path and the SSH/remote-managed path, including git import/export progress where totals are known. - Reworked command-managed sandbox transfer behavior so uploads use the faster single-stream path when appropriate, but fall back to chunked progress-emitting writes when the runner cannot expose mid-stream stdin progress. - Marked provider-backed environment sandbox runners as not supporting single-stream stdin progress so live sandbox runs emit meaningful intermediate updates instead of only `0%` and `100%`. - Emit an explicit terminal failure marker (`failed at NN% (x/y MB)`) when an SSH/tar transfer rejects, so a failed sync no longer leaves a dangling intermediate percentage in the log. - Run the SSH sync/restore size estimate (local directory walk / remote `du` probe) concurrently with the transfer instead of awaiting it before opening the pipe, so progress instrumentation no longer adds startup latency proportional to workspace file count. - Added and extended focused regression coverage for runtime progress throttling and the new failure marker, command-managed sandbox transfers, sandbox orchestration, SSH transfer progress, and environment execution-target wiring. ## Verification - `pnpm exec vitest run packages/adapter-utils/src/runtime-progress.test.ts packages/adapter-utils/src/ssh-fixture.test.ts packages/adapter-utils/src/command-managed-runtime.test.ts packages/adapter-utils/src/sandbox-managed-runtime.test.ts` - `pnpm exec vitest run packages/adapter-utils/src/command-managed-runtime.test.ts server/src/__tests__/environment-execution-target.test.ts` - `npx tsc --noEmit` for `packages/adapter-utils` ## Risks - The provider-backed sandbox fallback now prefers chunked command-managed writes when progress hooks are active, so small-to-medium uploads may trade some raw throughput for observable intermediate progress on runtimes that cannot surface true mid-stream stdin progress. - Progress percentages on tar-based transfers still depend on estimates in some cases, so operators may briefly see MB-only lines before the estimate resolves, then near-final clamping before the terminal `100%` line. - This PR changes shared execution-target behavior used by multiple adapters, so regressions would most likely appear in remote runtime setup/teardown flows rather than in a single adapter. ## Model Used - Initial implementation: OpenAI GPT-5.4 via Codex local agent (`codex_local`), high reasoning mode. - Observability follow-ups (failure marker, concurrent size estimate, added tests): Claude Opus 4.8 via Claude Code (`claude_local`). ## 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] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [ ] If this change affects the UI, I have included before/after screenshots - [ ] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [ ] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
Devin Foley
@@ -1,15 +1,80 @@
|
||||
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(options: { supportsSingleStreamStdinProgress?: boolean } = {}): SpawnRunnerHandle {
|
||||
const calls: Array<{ command: string; args?: string[]; cwd?: string; stdin?: string }> = [];
|
||||
const runner: CommandManagedRuntimeRunner = {
|
||||
supportsSingleStreamStdinProgress: options.supportsSingleStreamStdinProgress,
|
||||
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 +182,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 +196,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 +212,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 +229,99 @@ 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({ supportsSingleStreamStdinProgress: true });
|
||||
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("falls back to chunked upload progress when the runner cannot report mid-stream stdin progress", async () => {
|
||||
const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-command-write-fallback-"));
|
||||
cleanupDirs.push(rootDir);
|
||||
const remotePath = path.join(rootDir, "nested", "payload.bin");
|
||||
|
||||
const payload = Buffer.alloc(12 * 1024 * 1024);
|
||||
for (let i = 0; i < payload.length; i++) payload[i] = i % 256;
|
||||
|
||||
const { runner, calls } = makeSpawnRunner({ supportsSingleStreamStdinProgress: false });
|
||||
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 });
|
||||
},
|
||||
});
|
||||
|
||||
const written = await readFile(remotePath);
|
||||
expect(written.equals(payload)).toBe(true);
|
||||
|
||||
// Provider-backed sandbox runners cannot surface mid-flight progress for a
|
||||
// single stdin RPC, so we intentionally use several large append commands.
|
||||
expect(calls.length).toBeGreaterThan(2);
|
||||
expect(progress.length).toBeGreaterThan(2);
|
||||
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);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -8,8 +8,16 @@ import {
|
||||
} from "./sandbox-managed-runtime.js";
|
||||
import { preferredShellForSandbox, shellCommandArgs } from "./sandbox-shell.js";
|
||||
import type { RunProcessResult } from "./server-utils.js";
|
||||
import type { RuntimeProgressSink } from "./runtime-progress.js";
|
||||
|
||||
export interface CommandManagedRuntimeRunner {
|
||||
/**
|
||||
* True only when `execute({ stdin })` can surface useful in-flight progress
|
||||
* for a single stdin-backed command. Provider-backed sandbox runners usually
|
||||
* complete the entire RPC before returning, so they should leave this false
|
||||
* and let the caller choose a chunked upload path when progress is requested.
|
||||
*/
|
||||
supportsSingleStreamStdinProgress?: boolean;
|
||||
execute(input: {
|
||||
command: string;
|
||||
args?: string[];
|
||||
@@ -40,7 +48,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;
|
||||
@@ -62,13 +77,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;
|
||||
@@ -78,25 +101,101 @@ 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`;
|
||||
const canUseSingleStreamProgressPath =
|
||||
!options?.onProgress || input.runner.supportsSingleStreamStdinProgress === true;
|
||||
|
||||
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 &&
|
||||
canUseSingleStreamProgressPath
|
||||
) {
|
||||
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(
|
||||
@@ -146,6 +245,9 @@ export async function prepareCommandManagedRuntime(input: {
|
||||
installCommand?: string | null;
|
||||
/** When provided alongside `installCommand`, skip the install if `command -v <detectCommand>` succeeds. */
|
||||
detectCommand?: string | null;
|
||||
// Upload progress sink. Forwarded to prepareSandboxManagedRuntime; the child
|
||||
// task wires it into the byte-counting writeFile/readFile transport.
|
||||
onProgress?: RuntimeProgressSink;
|
||||
}): Promise<PreparedSandboxManagedRuntime> {
|
||||
const timeoutMs = input.spec.timeoutMs && input.spec.timeoutMs > 0 ? input.spec.timeoutMs : 300_000;
|
||||
const workspaceRemoteDir = input.workspaceRemoteDir ?? input.spec.remoteCwd;
|
||||
@@ -193,6 +295,7 @@ export async function prepareCommandManagedRuntime(input: {
|
||||
workspaceExclude: mergeRuntimeExcludes(input.workspaceExclude),
|
||||
preserveAbsentOnRestore: input.preserveAbsentOnRestore,
|
||||
assets: input.assets,
|
||||
onProgress: input.onProgress,
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -227,5 +330,6 @@ export async function prepareCommandManagedRuntime(input: {
|
||||
workspaceExclude: mergeRuntimeExcludes(input.workspaceExclude),
|
||||
preserveAbsentOnRestore: input.preserveAbsentOnRestore,
|
||||
assets: input.assets,
|
||||
onProgress: input.onProgress,
|
||||
});
|
||||
}
|
||||
|
||||
@@ -28,6 +28,9 @@ import {
|
||||
} from "./server-utils.js";
|
||||
import { sanitizeRemoteExecutionEnv } from "./remote-execution-env.js";
|
||||
import { preferredShellForSandbox, shellCommandArgs } from "./sandbox-shell.js";
|
||||
import type { RuntimeProgressSink } from "./runtime-progress.js";
|
||||
|
||||
export type { RuntimeProgressSink } from "./runtime-progress.js";
|
||||
|
||||
export interface AdapterLocalExecutionTarget {
|
||||
kind: "local";
|
||||
@@ -70,7 +73,7 @@ export interface PreparedAdapterExecutionTargetRuntime {
|
||||
workspaceRemoteDir: string | null;
|
||||
runtimeRootDir: string | null;
|
||||
assetDirs: Record<string, string>;
|
||||
restoreWorkspace(): Promise<void>;
|
||||
restoreWorkspace(onProgress?: RuntimeProgressSink): Promise<void>;
|
||||
}
|
||||
|
||||
export interface AdapterExecutionTargetProcessOptions {
|
||||
@@ -930,6 +933,11 @@ export async function prepareAdapterExecutionTargetRuntime(input: {
|
||||
installCommand?: string | null;
|
||||
/** When provided alongside `installCommand`, skip the install if the binary is already on PATH. */
|
||||
detectCommand?: string | null;
|
||||
// Optional progress sink for the workspace/asset upload. The returned
|
||||
// `restoreWorkspace(onProgress?)` accepts its own sink for teardown. Both are
|
||||
// forwarded down to the transport so the sandbox/SSH children can attach byte
|
||||
// counters without further changes here.
|
||||
onProgress?: RuntimeProgressSink;
|
||||
}): Promise<PreparedAdapterExecutionTargetRuntime> {
|
||||
const target = input.target ?? { kind: "local" as const };
|
||||
if (target.kind === "local") {
|
||||
@@ -950,6 +958,7 @@ export async function prepareAdapterExecutionTargetRuntime(input: {
|
||||
workspaceLocalDir: input.workspaceLocalDir,
|
||||
workspaceRemoteDir: input.workspaceRemoteDir,
|
||||
assets: input.assets,
|
||||
onProgress: input.onProgress,
|
||||
});
|
||||
return {
|
||||
target,
|
||||
@@ -980,6 +989,7 @@ export async function prepareAdapterExecutionTargetRuntime(input: {
|
||||
assets: input.assets,
|
||||
installCommand: input.installCommand,
|
||||
detectCommand: input.detectCommand,
|
||||
onProgress: input.onProgress,
|
||||
});
|
||||
return {
|
||||
target,
|
||||
|
||||
@@ -61,6 +61,15 @@ export {
|
||||
redactCommandText,
|
||||
} from "./command-redaction.js";
|
||||
export { buildSandboxNpmInstallCommand } from "./sandbox-install-command.js";
|
||||
export { createRuntimeProgressReporter } from "./runtime-progress.js";
|
||||
export type {
|
||||
RuntimeProgressSink,
|
||||
RuntimeProgressPhase,
|
||||
RuntimeProgressDirection,
|
||||
RuntimeProgressTarget,
|
||||
RuntimeProgressReporter,
|
||||
RuntimeProgressReporterOptions,
|
||||
} from "./runtime-progress.js";
|
||||
export { inferOpenAiCompatibleBiller } from "./billing.js";
|
||||
// Keep the root adapter-utils entry browser-safe because the UI imports it.
|
||||
// The sandbox callback bridge stays available via its dedicated subpath export.
|
||||
|
||||
@@ -6,6 +6,7 @@ import {
|
||||
syncDirectoryToSsh,
|
||||
} from "./ssh.js";
|
||||
import { captureDirectorySnapshot } from "./workspace-restore-merge.js";
|
||||
import type { RuntimeProgressSink } from "./runtime-progress.js";
|
||||
|
||||
export interface RemoteManagedRuntimeAsset {
|
||||
key: string;
|
||||
@@ -20,7 +21,7 @@ export interface PreparedRemoteManagedRuntime {
|
||||
workspaceRemoteDir: string;
|
||||
runtimeRootDir: string;
|
||||
assetDirs: Record<string, string>;
|
||||
restoreWorkspace(): Promise<void>;
|
||||
restoreWorkspace(onProgress?: RuntimeProgressSink): Promise<void>;
|
||||
}
|
||||
|
||||
function asObject(value: unknown): Record<string, unknown> {
|
||||
@@ -69,6 +70,9 @@ export async function prepareRemoteManagedRuntime(input: {
|
||||
workspaceLocalDir: string;
|
||||
workspaceRemoteDir?: string;
|
||||
assets?: RemoteManagedRuntimeAsset[];
|
||||
// Upload progress sink. Threaded for the byte-counting transport rewrite; the
|
||||
// child task wires it into the workspace/asset transfers.
|
||||
onProgress?: RuntimeProgressSink;
|
||||
}): Promise<PreparedRemoteManagedRuntime> {
|
||||
const baseWorkspaceRemoteDir = input.workspaceRemoteDir ?? input.spec.remoteCwd;
|
||||
const workspaceRemoteDir = path.posix.join(
|
||||
@@ -84,6 +88,7 @@ export async function prepareRemoteManagedRuntime(input: {
|
||||
spec: input.spec,
|
||||
localDir: input.workspaceLocalDir,
|
||||
remoteDir: workspaceRemoteDir,
|
||||
onProgress: input.onProgress,
|
||||
});
|
||||
const restoreExclude = preparedWorkspace.gitBacked ? [".git", ".paperclip-runtime"] : [".paperclip-runtime"];
|
||||
const baselineSnapshot = await captureDirectorySnapshot(input.workspaceLocalDir, {
|
||||
@@ -101,6 +106,8 @@ export async function prepareRemoteManagedRuntime(input: {
|
||||
remoteDir,
|
||||
followSymlinks: asset.followSymlinks,
|
||||
exclude: asset.exclude,
|
||||
onProgress: input.onProgress,
|
||||
progressLabel: asset.key,
|
||||
});
|
||||
}
|
||||
} catch (error) {
|
||||
@@ -110,6 +117,7 @@ export async function prepareRemoteManagedRuntime(input: {
|
||||
remoteDir: workspaceRemoteDir,
|
||||
baselineSnapshot,
|
||||
restoreGitHistory: preparedWorkspace.gitBacked,
|
||||
onProgress: input.onProgress,
|
||||
});
|
||||
throw error;
|
||||
}
|
||||
@@ -120,13 +128,14 @@ export async function prepareRemoteManagedRuntime(input: {
|
||||
workspaceRemoteDir,
|
||||
runtimeRootDir,
|
||||
assetDirs,
|
||||
restoreWorkspace: async () => {
|
||||
restoreWorkspace: async (onProgress?: RuntimeProgressSink) => {
|
||||
await restoreWorkspaceFromSshExecution({
|
||||
spec: input.spec,
|
||||
localDir: input.workspaceLocalDir,
|
||||
remoteDir: workspaceRemoteDir,
|
||||
baselineSnapshot,
|
||||
restoreGitHistory: preparedWorkspace.gitBacked,
|
||||
onProgress,
|
||||
});
|
||||
},
|
||||
};
|
||||
|
||||
@@ -0,0 +1,281 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { createRuntimeProgressReporter } from "./runtime-progress.js";
|
||||
|
||||
const MB = 1024 * 1024;
|
||||
|
||||
function makeClock(start = 0) {
|
||||
let value = start;
|
||||
return {
|
||||
now: () => value,
|
||||
advance: (ms: number) => {
|
||||
value += ms;
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
describe("createRuntimeProgressReporter", () => {
|
||||
it("formats the message with phase, label, direction, target, percent and MB", async () => {
|
||||
const lines: string[] = [];
|
||||
const reporter = createRuntimeProgressReporter({
|
||||
sink: (line) => {
|
||||
lines.push(line);
|
||||
},
|
||||
phase: "Syncing",
|
||||
label: "workspace",
|
||||
direction: "to",
|
||||
target: "sandbox",
|
||||
});
|
||||
|
||||
await reporter.report(12.6 * MB, 31.4 * MB);
|
||||
|
||||
expect(lines).toEqual(["[paperclip] Syncing workspace to sandbox: 40% (12.6/31.4 MB)\n"]);
|
||||
});
|
||||
|
||||
it("omits the label when none is provided (e.g. git history)", async () => {
|
||||
const lines: string[] = [];
|
||||
const reporter = createRuntimeProgressReporter({
|
||||
sink: (line) => {
|
||||
lines.push(line);
|
||||
},
|
||||
phase: "Importing git history",
|
||||
direction: "to",
|
||||
target: "ssh",
|
||||
});
|
||||
|
||||
await reporter.report(4 * MB, 4 * MB);
|
||||
|
||||
expect(lines).toEqual(["[paperclip] Importing git history to ssh: 100% (4.0/4.0 MB)\n"]);
|
||||
});
|
||||
|
||||
it("suppresses intermediate emits that neither cross a step nor exceed the interval", async () => {
|
||||
const clock = makeClock();
|
||||
const lines: string[] = [];
|
||||
const reporter = createRuntimeProgressReporter({
|
||||
sink: (line) => {
|
||||
lines.push(line);
|
||||
},
|
||||
phase: "Syncing",
|
||||
label: "workspace",
|
||||
direction: "to",
|
||||
target: "sandbox",
|
||||
now: clock.now,
|
||||
});
|
||||
|
||||
// First report always emits (step 0 crossed).
|
||||
await reporter.report(1 * MB, 100 * MB); // 1%
|
||||
// Still within the same 10% step and under 2s -> suppressed.
|
||||
await reporter.report(2 * MB, 100 * MB); // 2%
|
||||
await reporter.report(5 * MB, 100 * MB); // 5%
|
||||
|
||||
expect(lines).toHaveLength(1);
|
||||
expect(lines[0]).toBe("[paperclip] Syncing workspace to sandbox: 1% (1.0/100.0 MB)\n");
|
||||
});
|
||||
|
||||
it("emits when the percentage crosses a 10% step", async () => {
|
||||
const clock = makeClock();
|
||||
const lines: string[] = [];
|
||||
const reporter = createRuntimeProgressReporter({
|
||||
sink: (line) => {
|
||||
lines.push(line);
|
||||
},
|
||||
phase: "Syncing",
|
||||
label: "workspace",
|
||||
direction: "to",
|
||||
target: "sandbox",
|
||||
now: clock.now,
|
||||
});
|
||||
|
||||
await reporter.report(1 * MB, 100 * MB); // 1% -> emit (step 0)
|
||||
await reporter.report(15 * MB, 100 * MB); // 15% -> crosses into step 1 -> emit
|
||||
|
||||
expect(lines).toHaveLength(2);
|
||||
expect(lines[1]).toBe("[paperclip] Syncing workspace to sandbox: 15% (15.0/100.0 MB)\n");
|
||||
});
|
||||
|
||||
it("emits on the time threshold even without a step crossing", async () => {
|
||||
const clock = makeClock();
|
||||
const lines: string[] = [];
|
||||
const reporter = createRuntimeProgressReporter({
|
||||
sink: (line) => {
|
||||
lines.push(line);
|
||||
},
|
||||
phase: "Syncing",
|
||||
label: "workspace",
|
||||
direction: "to",
|
||||
target: "sandbox",
|
||||
now: clock.now,
|
||||
});
|
||||
|
||||
await reporter.report(1 * MB, 100 * MB); // emit
|
||||
await reporter.report(2 * MB, 100 * MB); // suppressed (same step, no time elapsed)
|
||||
clock.advance(2000);
|
||||
await reporter.report(3 * MB, 100 * MB); // 3% same step, but 2s elapsed -> emit
|
||||
|
||||
expect(lines).toHaveLength(2);
|
||||
expect(lines[1]).toBe("[paperclip] Syncing workspace to sandbox: 3% (3.0/100.0 MB)\n");
|
||||
});
|
||||
|
||||
it("always emits the terminal 100% line via report reaching the total", async () => {
|
||||
const clock = makeClock();
|
||||
const lines: string[] = [];
|
||||
const reporter = createRuntimeProgressReporter({
|
||||
sink: (line) => {
|
||||
lines.push(line);
|
||||
},
|
||||
phase: "Restoring",
|
||||
label: "workspace",
|
||||
direction: "from",
|
||||
target: "sandbox",
|
||||
now: clock.now,
|
||||
});
|
||||
|
||||
await reporter.report(1 * MB, 100 * MB); // emit
|
||||
await reporter.report(100 * MB, 100 * MB); // terminal -> always emit
|
||||
|
||||
expect(lines[lines.length - 1]).toBe(
|
||||
"[paperclip] Restoring workspace from sandbox: 100% (100.0/100.0 MB)\n",
|
||||
);
|
||||
});
|
||||
|
||||
it("complete() emits the terminal 100% line even when intermediate emits were throttled", async () => {
|
||||
const clock = makeClock();
|
||||
const lines: string[] = [];
|
||||
const reporter = createRuntimeProgressReporter({
|
||||
sink: (line) => {
|
||||
lines.push(line);
|
||||
},
|
||||
phase: "Syncing",
|
||||
label: "workspace",
|
||||
direction: "to",
|
||||
target: "sandbox",
|
||||
now: clock.now,
|
||||
});
|
||||
|
||||
await reporter.report(1 * MB, 100 * MB); // emit
|
||||
await reporter.report(5 * MB, 100 * MB); // suppressed
|
||||
await reporter.complete();
|
||||
|
||||
expect(lines[lines.length - 1]).toBe(
|
||||
"[paperclip] Syncing workspace to sandbox: 100% (100.0/100.0 MB)\n",
|
||||
);
|
||||
});
|
||||
|
||||
it("complete() is idempotent and does not double-emit after a terminal report", async () => {
|
||||
const lines: string[] = [];
|
||||
const reporter = createRuntimeProgressReporter({
|
||||
sink: (line) => {
|
||||
lines.push(line);
|
||||
},
|
||||
phase: "Syncing",
|
||||
label: "workspace",
|
||||
direction: "to",
|
||||
target: "sandbox",
|
||||
});
|
||||
|
||||
await reporter.report(100 * MB, 100 * MB); // terminal
|
||||
await reporter.complete();
|
||||
await reporter.complete();
|
||||
|
||||
expect(lines).toHaveLength(1);
|
||||
});
|
||||
|
||||
it("reports MB-only (no percent) when the total is unknown, plus a completion line", async () => {
|
||||
const clock = makeClock();
|
||||
const lines: string[] = [];
|
||||
const reporter = createRuntimeProgressReporter({
|
||||
sink: (line) => {
|
||||
lines.push(line);
|
||||
},
|
||||
phase: "Restoring",
|
||||
label: "workspace",
|
||||
direction: "from",
|
||||
target: "ssh",
|
||||
now: clock.now,
|
||||
});
|
||||
|
||||
await reporter.report(2 * MB, null); // first emit
|
||||
await reporter.report(4 * MB, null); // suppressed (no time elapsed)
|
||||
clock.advance(2000);
|
||||
await reporter.report(8 * MB, null); // time elapsed -> emit
|
||||
await reporter.complete();
|
||||
|
||||
expect(lines).toEqual([
|
||||
"[paperclip] Restoring workspace from ssh: 2.0 MB\n",
|
||||
"[paperclip] Restoring workspace from ssh: 8.0 MB\n",
|
||||
"[paperclip] Restoring workspace from ssh: 8.0 MB\n",
|
||||
]);
|
||||
});
|
||||
|
||||
it("fail() emits a terminal failure marker with the last percent instead of a dangling line", async () => {
|
||||
const clock = makeClock();
|
||||
const lines: string[] = [];
|
||||
const reporter = createRuntimeProgressReporter({
|
||||
sink: (line) => {
|
||||
lines.push(line);
|
||||
},
|
||||
phase: "Syncing",
|
||||
label: "workspace",
|
||||
direction: "to",
|
||||
target: "ssh",
|
||||
now: clock.now,
|
||||
});
|
||||
|
||||
await reporter.report(40 * MB, 100 * MB); // emit at 40%
|
||||
await reporter.fail();
|
||||
|
||||
expect(lines).toEqual([
|
||||
"[paperclip] Syncing workspace to ssh: 40% (40.0/100.0 MB)\n",
|
||||
"[paperclip] Syncing workspace to ssh: failed at 40% (40.0/100.0 MB)\n",
|
||||
]);
|
||||
});
|
||||
|
||||
it("fail() falls back to an MB marker when the total is unknown", async () => {
|
||||
const lines: string[] = [];
|
||||
const reporter = createRuntimeProgressReporter({
|
||||
sink: (line) => {
|
||||
lines.push(line);
|
||||
},
|
||||
phase: "Restoring",
|
||||
direction: "from",
|
||||
target: "ssh",
|
||||
});
|
||||
|
||||
await reporter.report(3 * MB, null);
|
||||
await reporter.fail();
|
||||
|
||||
expect(lines.at(-1)).toBe("[paperclip] Restoring from ssh: failed after 3.0 MB\n");
|
||||
});
|
||||
|
||||
it("fail() is suppressed after a terminal completion and complete() after a failure", async () => {
|
||||
const lines: string[] = [];
|
||||
const reporter = createRuntimeProgressReporter({
|
||||
sink: (line) => {
|
||||
lines.push(line);
|
||||
},
|
||||
phase: "Syncing",
|
||||
label: "workspace",
|
||||
direction: "to",
|
||||
target: "ssh",
|
||||
});
|
||||
|
||||
await reporter.report(100 * MB, 100 * MB); // terminal complete
|
||||
await reporter.fail(); // suppressed — already completed
|
||||
expect(lines).toHaveLength(1);
|
||||
|
||||
const failLines: string[] = [];
|
||||
const failed = createRuntimeProgressReporter({
|
||||
sink: (line) => {
|
||||
failLines.push(line);
|
||||
},
|
||||
phase: "Syncing",
|
||||
label: "workspace",
|
||||
direction: "to",
|
||||
target: "ssh",
|
||||
});
|
||||
await failed.report(20 * MB, 100 * MB);
|
||||
await failed.fail();
|
||||
await failed.complete(); // suppressed — already failed
|
||||
expect(failLines).toHaveLength(2);
|
||||
expect(failLines.at(-1)).toContain("failed at 20%");
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,152 @@
|
||||
// Shared, throttled progress reporting for execution-target sync/restore.
|
||||
//
|
||||
// Transports (sandbox / SSH) own the byte counting and call `report()` as bytes
|
||||
// move; orchestrators own the per-phase label and direction. The reporter
|
||||
// throttles emits so a long transfer doesn't flood the log: a line is emitted
|
||||
// only when the percentage crosses a step boundary (default every 10%) or once
|
||||
// at least `minIntervalMs` has elapsed since the last emit. The terminal
|
||||
// completion line is always emitted via `complete()` (or when `report()` reaches
|
||||
// the known total).
|
||||
|
||||
/** A sink for fully-formatted progress lines (newline included). */
|
||||
export type RuntimeProgressSink = (line: string) => void | Promise<void>;
|
||||
|
||||
export type RuntimeProgressPhase =
|
||||
| "Syncing"
|
||||
| "Restoring"
|
||||
| "Importing git history"
|
||||
| "Exporting git history";
|
||||
|
||||
export type RuntimeProgressDirection = "to" | "from";
|
||||
|
||||
export type RuntimeProgressTarget = "sandbox" | "ssh";
|
||||
|
||||
export interface RuntimeProgressReporterOptions {
|
||||
sink: RuntimeProgressSink;
|
||||
phase: RuntimeProgressPhase;
|
||||
/** Optional per-phase label, e.g. "workspace" or an asset key. */
|
||||
label?: string;
|
||||
direction: RuntimeProgressDirection;
|
||||
target: RuntimeProgressTarget;
|
||||
/** Emit when the percentage crosses this step. Default 10. */
|
||||
stepPercent?: number;
|
||||
/** Emit when at least this many ms have elapsed since the last emit. Default 2000. */
|
||||
minIntervalMs?: number;
|
||||
/** Injectable clock for deterministic tests. Default `Date.now`. */
|
||||
now?: () => number;
|
||||
}
|
||||
|
||||
export interface RuntimeProgressReporter {
|
||||
/**
|
||||
* Report progress. Throttled: only emits on a step crossing or after
|
||||
* `minIntervalMs`. When `totalBytes` is known and `doneBytes` reaches it, the
|
||||
* terminal 100% line is emitted and the reporter is marked complete.
|
||||
*/
|
||||
report(doneBytes: number, totalBytes: number | null): Promise<void>;
|
||||
/**
|
||||
* Emit the terminal completion line if it hasn't been emitted yet. Idempotent.
|
||||
*/
|
||||
complete(doneBytes?: number, totalBytes?: number | null): Promise<void>;
|
||||
/**
|
||||
* Emit a terminal failure line if no terminal line has been emitted yet, so a
|
||||
* failed transfer leaves an explicit marker instead of a dangling percentage.
|
||||
* Idempotent and mutually exclusive with `complete()`.
|
||||
*/
|
||||
fail(doneBytes?: number, totalBytes?: number | null): Promise<void>;
|
||||
}
|
||||
|
||||
const BYTES_PER_MB = 1024 * 1024;
|
||||
|
||||
function formatMb(bytes: number): string {
|
||||
return (Math.max(0, bytes) / BYTES_PER_MB).toFixed(1);
|
||||
}
|
||||
|
||||
function clampPercent(value: number): number {
|
||||
if (!Number.isFinite(value)) return 0;
|
||||
return Math.min(100, Math.max(0, Math.round(value)));
|
||||
}
|
||||
|
||||
export function createRuntimeProgressReporter(
|
||||
options: RuntimeProgressReporterOptions,
|
||||
): RuntimeProgressReporter {
|
||||
const stepPercent = options.stepPercent && options.stepPercent > 0 ? options.stepPercent : 10;
|
||||
const minIntervalMs =
|
||||
options.minIntervalMs && options.minIntervalMs > 0 ? options.minIntervalMs : 2000;
|
||||
const now = options.now ?? Date.now;
|
||||
const prefix = `[paperclip] ${options.phase}${options.label ? ` ${options.label}` : ""} ${options.direction} ${options.target}`;
|
||||
|
||||
let lastEmitAt: number | null = null;
|
||||
let lastStep = -1;
|
||||
let lastDoneBytes = 0;
|
||||
let lastTotalBytes: number | null = null;
|
||||
let completed = false;
|
||||
|
||||
function buildLine(doneBytes: number, totalBytes: number | null): string {
|
||||
if (totalBytes != null && totalBytes > 0) {
|
||||
const pct = clampPercent((doneBytes / totalBytes) * 100);
|
||||
return `${prefix}: ${pct}% (${formatMb(doneBytes)}/${formatMb(totalBytes)} MB)\n`;
|
||||
}
|
||||
return `${prefix}: ${formatMb(doneBytes)} MB\n`;
|
||||
}
|
||||
|
||||
function buildFailLine(doneBytes: number, totalBytes: number | null): string {
|
||||
if (totalBytes != null && totalBytes > 0) {
|
||||
const pct = clampPercent((doneBytes / totalBytes) * 100);
|
||||
return `${prefix}: failed at ${pct}% (${formatMb(doneBytes)}/${formatMb(totalBytes)} MB)\n`;
|
||||
}
|
||||
return `${prefix}: failed after ${formatMb(doneBytes)} MB\n`;
|
||||
}
|
||||
|
||||
async function emit(doneBytes: number, totalBytes: number | null): Promise<void> {
|
||||
lastEmitAt = now();
|
||||
if (totalBytes != null && totalBytes > 0) {
|
||||
lastStep = Math.floor(((doneBytes / totalBytes) * 100) / stepPercent);
|
||||
}
|
||||
await options.sink(buildLine(doneBytes, totalBytes));
|
||||
}
|
||||
|
||||
return {
|
||||
async report(doneBytes, totalBytes) {
|
||||
lastDoneBytes = doneBytes;
|
||||
lastTotalBytes = totalBytes;
|
||||
if (completed) return;
|
||||
|
||||
const elapsedOk = lastEmitAt == null || now() - lastEmitAt >= minIntervalMs;
|
||||
|
||||
if (totalBytes != null && totalBytes > 0) {
|
||||
const terminal = doneBytes >= totalBytes;
|
||||
const step = Math.floor(((doneBytes / totalBytes) * 100) / stepPercent);
|
||||
const stepOk = step > lastStep;
|
||||
if (terminal || stepOk || elapsedOk) {
|
||||
await emit(doneBytes, totalBytes);
|
||||
}
|
||||
if (terminal) completed = true;
|
||||
return;
|
||||
}
|
||||
|
||||
// Unknown total: no step boundaries, throttle purely on elapsed time.
|
||||
if (elapsedOk) {
|
||||
await emit(doneBytes, totalBytes);
|
||||
}
|
||||
},
|
||||
async complete(doneBytes, totalBytes) {
|
||||
if (completed) return;
|
||||
completed = true;
|
||||
const total = totalBytes !== undefined ? totalBytes : lastTotalBytes;
|
||||
const done =
|
||||
doneBytes !== undefined
|
||||
? doneBytes
|
||||
: total != null && total > 0
|
||||
? total
|
||||
: lastDoneBytes;
|
||||
await options.sink(buildLine(done, total));
|
||||
},
|
||||
async fail(doneBytes, totalBytes) {
|
||||
if (completed) return;
|
||||
completed = true;
|
||||
const total = totalBytes !== undefined ? totalBytes : lastTotalBytes;
|
||||
const done = doneBytes !== undefined ? doneBytes : lastDoneBytes;
|
||||
await options.sink(buildFailLine(done, total));
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -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,6 +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 { createRuntimeProgressReporter, type RuntimeProgressSink } from "./runtime-progress.js";
|
||||
|
||||
const execFile = promisify(execFileCallback);
|
||||
|
||||
@@ -23,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>;
|
||||
@@ -38,7 +52,7 @@ export interface PreparedSandboxManagedRuntime {
|
||||
workspaceRemoteDir: string;
|
||||
runtimeRootDir: string;
|
||||
assetDirs: Record<string, string>;
|
||||
restoreWorkspace(): Promise<void>;
|
||||
restoreWorkspace(onProgress?: RuntimeProgressSink): Promise<void>;
|
||||
}
|
||||
|
||||
function asObject(value: unknown): Record<string, unknown> {
|
||||
@@ -261,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;
|
||||
@@ -270,6 +314,9 @@ export async function prepareSandboxManagedRuntime(input: {
|
||||
workspaceExclude?: string[];
|
||||
preserveAbsentOnRestore?: string[];
|
||||
assets?: SandboxManagedRuntimeAsset[];
|
||||
// Upload progress sink. Threaded for the byte-counting transport rewrite; the
|
||||
// child task wires it into writeFile/readFile.
|
||||
onProgress?: RuntimeProgressSink;
|
||||
}): Promise<PreparedSandboxManagedRuntime> {
|
||||
const workspaceRemoteDir = input.workspaceRemoteDir ?? input.spec.remoteCwd;
|
||||
const runtimeRootDir = path.posix.join(workspaceRemoteDir, ".paperclip-runtime", input.adapterKey);
|
||||
@@ -287,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(
|
||||
@@ -311,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)} && ` +
|
||||
@@ -334,7 +389,8 @@ export async function prepareSandboxManagedRuntime(input: {
|
||||
workspaceRemoteDir,
|
||||
runtimeRootDir,
|
||||
assetDirs,
|
||||
restoreWorkspace: async () => {
|
||||
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(
|
||||
@@ -345,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");
|
||||
|
||||
@@ -11,6 +11,7 @@ import {
|
||||
readSshEnvLabFixtureStatus,
|
||||
restoreWorkspaceFromSshExecution,
|
||||
runSshCommand,
|
||||
syncDirectoryFromSsh,
|
||||
syncDirectoryToSsh,
|
||||
startSshEnvLabFixture,
|
||||
stopSshEnvLabFixture,
|
||||
@@ -54,6 +55,31 @@ async function startSshEnvLabFixtureOrSkip(statePath: string, label: string) {
|
||||
}
|
||||
}
|
||||
|
||||
interface ParsedProgressLine {
|
||||
raw: string;
|
||||
percent: number | null;
|
||||
doneMb: number | null;
|
||||
totalMb: number | null;
|
||||
}
|
||||
|
||||
function parseProgressLine(line: string): ParsedProgressLine {
|
||||
const trimmed = line.trimEnd();
|
||||
const percentMatch = trimmed.match(/:\s*(\d+)%\s*\(([\d.]+)\/([\d.]+) MB\)$/);
|
||||
if (percentMatch) {
|
||||
return {
|
||||
raw: trimmed,
|
||||
percent: Number.parseInt(percentMatch[1]!, 10),
|
||||
doneMb: Number.parseFloat(percentMatch[2]!),
|
||||
totalMb: Number.parseFloat(percentMatch[3]!),
|
||||
};
|
||||
}
|
||||
const mbMatch = trimmed.match(/:\s*([\d.]+) MB$/);
|
||||
if (mbMatch) {
|
||||
return { raw: trimmed, percent: null, doneMb: Number.parseFloat(mbMatch[1]!), totalMb: null };
|
||||
}
|
||||
return { raw: trimmed, percent: null, doneMb: null, totalMb: null };
|
||||
}
|
||||
|
||||
describe("ssh env-lab fixture", () => {
|
||||
const cleanupDirs: string[] = [];
|
||||
|
||||
@@ -199,6 +225,142 @@ describe("ssh env-lab fixture", () => {
|
||||
expect(result.stdout).not.toContain("appledouble-present");
|
||||
}, SSH_FIXTURE_TEST_TIMEOUT_MS);
|
||||
|
||||
it("reports throttled upload progress with a clamped percent and terminal 100% line", async () => {
|
||||
const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-ssh-fixture-"));
|
||||
cleanupDirs.push(rootDir);
|
||||
const statePath = path.join(rootDir, "state.json");
|
||||
const localDir = path.join(rootDir, "local-overlay");
|
||||
|
||||
await mkdir(localDir, { recursive: true });
|
||||
// Multiple files large enough that tar emits several pipe chunks, so the
|
||||
// byte counter crosses several step boundaries before the stream closes.
|
||||
for (let index = 0; index < 4; index += 1) {
|
||||
await writeFile(path.join(localDir, `blob-${index}.bin`), Buffer.alloc(256 * 1024, index + 1));
|
||||
}
|
||||
|
||||
const started = await startSshEnvLabFixtureOrSkip(statePath, "SSH upload progress test");
|
||||
if (!started) return;
|
||||
const config = await buildSshEnvLabFixtureConfig(started);
|
||||
const remoteDir = path.posix.join(started.workspaceDir, "overlay-progress");
|
||||
|
||||
const lines: ParsedProgressLine[] = [];
|
||||
await syncDirectoryToSsh({
|
||||
spec: { ...config, remoteCwd: started.workspaceDir },
|
||||
localDir,
|
||||
remoteDir,
|
||||
onProgress: (line) => {
|
||||
lines.push(parseProgressLine(line));
|
||||
},
|
||||
progressLabel: "workspace",
|
||||
});
|
||||
|
||||
expect(lines.length).toBeGreaterThan(0);
|
||||
for (const line of lines) {
|
||||
expect(line.raw).toContain("Syncing workspace to ssh");
|
||||
}
|
||||
// Monotonically increasing byte counts.
|
||||
const doneSeries = lines.map((line) => line.doneMb ?? 0);
|
||||
for (let index = 1; index < doneSeries.length; index += 1) {
|
||||
expect(doneSeries[index]!).toBeGreaterThanOrEqual(doneSeries[index - 1]!);
|
||||
}
|
||||
// Percent clamped to <= 99% on every line emitted before the stream closed.
|
||||
for (const line of lines.slice(0, -1)) {
|
||||
if (line.percent != null) expect(line.percent).toBeLessThanOrEqual(99);
|
||||
}
|
||||
// Terminal completion line is 100% with matching done/total.
|
||||
const last = lines.at(-1)!;
|
||||
expect(last.percent).toBe(100);
|
||||
expect(last.doneMb).toBe(last.totalMb);
|
||||
}, SSH_FIXTURE_TEST_TIMEOUT_MS);
|
||||
|
||||
it("reports restore progress with a terminal completion line", async () => {
|
||||
const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-ssh-fixture-"));
|
||||
cleanupDirs.push(rootDir);
|
||||
const statePath = path.join(rootDir, "state.json");
|
||||
const localDir = path.join(rootDir, "local-overlay");
|
||||
const restoreDir = path.join(rootDir, "restore-target");
|
||||
|
||||
await mkdir(localDir, { recursive: true });
|
||||
await mkdir(restoreDir, { recursive: true });
|
||||
for (let index = 0; index < 4; index += 1) {
|
||||
await writeFile(path.join(localDir, `blob-${index}.bin`), Buffer.alloc(256 * 1024, index + 1));
|
||||
}
|
||||
|
||||
const started = await startSshEnvLabFixtureOrSkip(statePath, "SSH restore progress test");
|
||||
if (!started) return;
|
||||
const config = await buildSshEnvLabFixtureConfig(started);
|
||||
const spec = { ...config, remoteCwd: started.workspaceDir } as const;
|
||||
const remoteDir = path.posix.join(started.workspaceDir, "restore-source");
|
||||
|
||||
await syncDirectoryToSsh({ spec, localDir, remoteDir });
|
||||
|
||||
const lines: ParsedProgressLine[] = [];
|
||||
await syncDirectoryFromSsh({
|
||||
spec,
|
||||
remoteDir,
|
||||
localDir: restoreDir,
|
||||
onProgress: (line) => {
|
||||
lines.push(parseProgressLine(line));
|
||||
},
|
||||
progressLabel: "workspace",
|
||||
});
|
||||
|
||||
expect(lines.length).toBeGreaterThan(0);
|
||||
for (const line of lines) {
|
||||
expect(line.raw).toContain("Restoring workspace from ssh");
|
||||
}
|
||||
// Terminal completion line: either an exact 100% (probe succeeded) or a
|
||||
// final MB-received line (probe unavailable). Either is a valid terminal.
|
||||
const last = lines.at(-1)!;
|
||||
expect(last.percent === 100 || (last.percent === null && last.doneMb !== null)).toBe(true);
|
||||
// The restored files round-tripped through the byte-counting transport.
|
||||
await expect(readFile(path.join(restoreDir, "blob-0.bin"))).resolves.toEqual(
|
||||
Buffer.alloc(256 * 1024, 1),
|
||||
);
|
||||
}, SSH_FIXTURE_TEST_TIMEOUT_MS);
|
||||
|
||||
it("reports exact git-history import percentage from the known bundle size", async () => {
|
||||
const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-ssh-fixture-"));
|
||||
cleanupDirs.push(rootDir);
|
||||
const statePath = path.join(rootDir, "state.json");
|
||||
const localRepo = path.join(rootDir, "local-workspace");
|
||||
|
||||
await mkdir(localRepo, { recursive: true });
|
||||
await git(localRepo, ["init"]);
|
||||
await git(localRepo, ["checkout", "-b", "main"]);
|
||||
await git(localRepo, ["config", "user.name", "Paperclip Test"]);
|
||||
await git(localRepo, ["config", "user.email", "test@paperclip.dev"]);
|
||||
await writeFile(path.join(localRepo, "tracked.bin"), Buffer.alloc(256 * 1024, 7));
|
||||
await git(localRepo, ["add", "tracked.bin"]);
|
||||
await git(localRepo, ["commit", "-m", "initial"]);
|
||||
|
||||
const started = await startSshEnvLabFixtureOrSkip(statePath, "SSH git import progress test");
|
||||
if (!started) return;
|
||||
const config = await buildSshEnvLabFixtureConfig(started);
|
||||
const spec = { ...config, remoteCwd: started.workspaceDir } as const;
|
||||
|
||||
const lines: ParsedProgressLine[] = [];
|
||||
await prepareWorkspaceForSshExecution({
|
||||
spec,
|
||||
localDir: localRepo,
|
||||
remoteDir: started.workspaceDir,
|
||||
onProgress: (line) => {
|
||||
lines.push(parseProgressLine(line));
|
||||
},
|
||||
});
|
||||
|
||||
const importLines = lines.filter((line) => line.raw.includes("Importing git history to ssh"));
|
||||
expect(importLines.length).toBeGreaterThan(0);
|
||||
// Known bundle size -> exact percentage with no "workspace" label.
|
||||
for (const line of importLines) {
|
||||
expect(line.raw).not.toContain("workspace");
|
||||
expect(line.percent).not.toBeNull();
|
||||
}
|
||||
const lastImport = importLines.at(-1)!;
|
||||
expect(lastImport.percent).toBe(100);
|
||||
expect(lastImport.doneMb).toBe(lastImport.totalMb);
|
||||
}, SSH_FIXTURE_TEST_TIMEOUT_MS);
|
||||
|
||||
it("can dereference local symlinks while syncing to the remote fixture", async () => {
|
||||
const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-ssh-fixture-"));
|
||||
cleanupDirs.push(rootDir);
|
||||
|
||||
@@ -4,10 +4,17 @@ import { constants as fsConstants, createReadStream, createWriteStream, promises
|
||||
import net from "node:net";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { Transform } from "node:stream";
|
||||
import type { CommandManagedRuntimeRunner } from "./command-managed-runtime.js";
|
||||
import type { RunProcessResult } from "./server-utils.js";
|
||||
import type { DirectorySnapshot } from "./workspace-restore-merge.js";
|
||||
import { mergeDirectoryWithBaseline } from "./workspace-restore-merge.js";
|
||||
import {
|
||||
createRuntimeProgressReporter,
|
||||
type RuntimeProgressDirection,
|
||||
type RuntimeProgressPhase,
|
||||
type RuntimeProgressSink,
|
||||
} from "./runtime-progress.js";
|
||||
|
||||
export interface SshConnectionConfig {
|
||||
host: string;
|
||||
@@ -409,6 +416,152 @@ function tarSpawnEnv(): NodeJS.ProcessEnv {
|
||||
};
|
||||
}
|
||||
|
||||
// Converts a tar `--exclude` pattern into a regexp for the local-size estimate.
|
||||
// We only need approximate fidelity here (the estimate feeds a clamped percent),
|
||||
// so we support the literal names and `*`/`?` globs used in practice.
|
||||
function tarPatternToRegExp(pattern: string): RegExp {
|
||||
const escaped = pattern
|
||||
.replace(/[.+^${}()|[\]\\]/g, "\\$&")
|
||||
.replace(/\*/g, "[^/]*")
|
||||
.replace(/\?/g, "[^/]");
|
||||
return new RegExp(`^${escaped}$`);
|
||||
}
|
||||
|
||||
// Walks `localDir` summing regular-file sizes, mirroring tar's `--exclude`
|
||||
// handling (plus the implicit `._*`) and `followSymlinks` so the to-ssh upload
|
||||
// can report an estimated total before tar finishes producing the stream.
|
||||
async function estimateLocalDirSize(input: {
|
||||
localDir: string;
|
||||
exclude?: string[];
|
||||
followSymlinks?: boolean;
|
||||
}): Promise<number> {
|
||||
const regexes = ["._*", ...(input.exclude ?? [])].map(tarPatternToRegExp);
|
||||
const isExcluded = (relPath: string, base: string) =>
|
||||
regexes.some((regex) => regex.test(relPath) || regex.test(base));
|
||||
|
||||
let total = 0;
|
||||
const walk = async (dir: string, relative: string): Promise<void> => {
|
||||
const entries = await fs.readdir(dir, { withFileTypes: true }).catch(() => []);
|
||||
for (const entry of entries) {
|
||||
const entryRelative = relative ? `${relative}/${entry.name}` : entry.name;
|
||||
if (isExcluded(entryRelative, entry.name)) continue;
|
||||
const full = path.join(dir, entry.name);
|
||||
const stats = await (input.followSymlinks ? fs.stat(full) : fs.lstat(full)).catch(() => null);
|
||||
if (!stats) continue;
|
||||
if (stats.isDirectory()) {
|
||||
await walk(full, entryRelative);
|
||||
} else if (stats.isFile()) {
|
||||
total += stats.size;
|
||||
}
|
||||
}
|
||||
};
|
||||
await walk(input.localDir, "");
|
||||
return total;
|
||||
}
|
||||
|
||||
// Best-effort remote size probe for the from-ssh restore. `du -sk` is POSIX and
|
||||
// available on the BSD/Linux remotes we target; it over-counts (block-rounded,
|
||||
// includes excluded dirs) which keeps the reported percent safely below 100
|
||||
// until the stream actually closes. Returns null when unavailable so the caller
|
||||
// falls back to MB-received mode.
|
||||
async function probeRemoteDirSize(input: {
|
||||
spec: SshConnectionConfig;
|
||||
remoteDir: string;
|
||||
}): Promise<number | null> {
|
||||
try {
|
||||
const result = await runSshScript(
|
||||
input.spec,
|
||||
`du -sk ${shellQuote(input.remoteDir)} 2>/dev/null | cut -f1`,
|
||||
{ timeoutMs: 15_000, maxBuffer: 16 * 1024 },
|
||||
);
|
||||
const kilobytes = Number.parseInt(result.stdout.trim(), 10);
|
||||
return Number.isFinite(kilobytes) && kilobytes > 0 ? kilobytes * 1024 : null;
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
interface TransferProgress {
|
||||
// Backpressure-respecting counter to splice into a transport pipe.
|
||||
counter: Transform;
|
||||
// Last cumulative byte count observed by the counter.
|
||||
transferred: () => number;
|
||||
// Emit the terminal completion line. Idempotent.
|
||||
finish: () => Promise<void>;
|
||||
// Emit a terminal failure marker instead of a completion line. Idempotent.
|
||||
fail: () => Promise<void>;
|
||||
}
|
||||
|
||||
// Wraps a throttled progress reporter behind a counting Transform so transports
|
||||
// can `source.pipe(progress.counter).pipe(dest)`. When `totalBytes` is a known
|
||||
// exact size (e.g. a git bundle) the reporter emits an exact percentage. When it
|
||||
// is an estimate (tar upload / remote probe) we clamp the reported bytes to 99%
|
||||
// of the estimate so an inaccurate total never shows a premature 100%; `finish`
|
||||
// then emits the terminal 100% (or, in MB-only mode, the final MB) line.
|
||||
//
|
||||
// `totalBytes` may be a promise so an expensive size estimate (a local dir walk
|
||||
// or a remote `du` probe) runs concurrently with the transfer instead of
|
||||
// blocking the pipe from opening. Until it resolves the counter reports bytes in
|
||||
// MB-only mode, then adopts the percentage once the total is known; `finish`
|
||||
// awaits the estimate so the terminal 100% line is still guaranteed.
|
||||
function createTransferProgress(input: {
|
||||
onProgress: RuntimeProgressSink;
|
||||
phase: RuntimeProgressPhase;
|
||||
direction: RuntimeProgressDirection;
|
||||
label?: string;
|
||||
totalBytes: number | null | Promise<number | null>;
|
||||
estimated: boolean;
|
||||
}): TransferProgress {
|
||||
const reporter = createRuntimeProgressReporter({
|
||||
sink: input.onProgress,
|
||||
phase: input.phase,
|
||||
direction: input.direction,
|
||||
label: input.label,
|
||||
target: "ssh",
|
||||
});
|
||||
|
||||
let total: number | null = null;
|
||||
let cap: number | null = null;
|
||||
const applyTotal = (value: number | null) => {
|
||||
total = value != null && value > 0 ? value : null;
|
||||
cap = total != null && input.estimated ? Math.floor(total * 0.99) : null;
|
||||
};
|
||||
const totalReady: Promise<void> =
|
||||
input.totalBytes != null && typeof (input.totalBytes as Promise<number | null>).then === "function"
|
||||
? (input.totalBytes as Promise<number | null>).then(applyTotal, () => applyTotal(null))
|
||||
: (applyTotal(input.totalBytes as number | null), Promise.resolve());
|
||||
|
||||
let transferred = 0;
|
||||
let chain: Promise<void> = Promise.resolve();
|
||||
const enqueue = (work: () => Promise<void>) => {
|
||||
chain = chain.then(work).catch(() => undefined);
|
||||
};
|
||||
|
||||
const counter = new Transform({
|
||||
transform(chunk: Buffer, _encoding, callback) {
|
||||
transferred += chunk.length;
|
||||
const reported = cap != null ? Math.min(transferred, cap) : transferred;
|
||||
const totalSnapshot = total;
|
||||
enqueue(() => reporter.report(reported, totalSnapshot));
|
||||
callback(null, chunk);
|
||||
},
|
||||
});
|
||||
|
||||
return {
|
||||
counter,
|
||||
transferred: () => transferred,
|
||||
finish: async () => {
|
||||
await chain.catch(() => undefined);
|
||||
await totalReady.catch(() => undefined);
|
||||
await reporter.complete(total != null ? total : transferred, total).catch(() => undefined);
|
||||
},
|
||||
fail: async () => {
|
||||
await chain.catch(() => undefined);
|
||||
await reporter.fail(transferred, total).catch(() => undefined);
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
async function runSshScript(
|
||||
config: SshConnectionConfig,
|
||||
script: string,
|
||||
@@ -493,6 +646,7 @@ async function streamLocalFileToSsh(input: {
|
||||
spec: SshConnectionConfig;
|
||||
localFile: string;
|
||||
remoteScript: string;
|
||||
progress?: TransferProgress;
|
||||
}): Promise<void> {
|
||||
const auth = await createSshAuthArgs(input.spec);
|
||||
const sshArgs = [
|
||||
@@ -525,7 +679,12 @@ async function streamLocalFileToSsh(input: {
|
||||
});
|
||||
source.on("error", fail);
|
||||
ssh.on("error", fail);
|
||||
source.pipe(ssh.stdin ?? null);
|
||||
if (input.progress) {
|
||||
input.progress.counter.on("error", fail);
|
||||
source.pipe(input.progress.counter).pipe(ssh.stdin ?? null);
|
||||
} else {
|
||||
source.pipe(ssh.stdin ?? null);
|
||||
}
|
||||
ssh.on("close", (code) => {
|
||||
if (settled) return;
|
||||
settled = true;
|
||||
@@ -542,6 +701,7 @@ async function streamSshToLocalFile(input: {
|
||||
spec: SshConnectionConfig;
|
||||
remoteScript: string;
|
||||
localFile: string;
|
||||
progress?: TransferProgress;
|
||||
}): Promise<void> {
|
||||
const auth = await createSshAuthArgs(input.spec);
|
||||
const sshArgs = [
|
||||
@@ -569,7 +729,12 @@ async function streamSshToLocalFile(input: {
|
||||
reject(error);
|
||||
};
|
||||
|
||||
ssh.stdout?.pipe(sink);
|
||||
if (input.progress) {
|
||||
input.progress.counter.on("error", fail);
|
||||
ssh.stdout?.pipe(input.progress.counter).pipe(sink);
|
||||
} else {
|
||||
ssh.stdout?.pipe(sink);
|
||||
}
|
||||
ssh.stderr?.on("data", (chunk) => {
|
||||
sshStderr += String(chunk);
|
||||
});
|
||||
@@ -594,6 +759,7 @@ async function importGitWorkspaceToSsh(input: {
|
||||
localDir: string;
|
||||
remoteDir: string;
|
||||
snapshot: LocalGitWorkspaceSnapshot;
|
||||
onProgress?: RuntimeProgressSink;
|
||||
}): Promise<void> {
|
||||
const bundleDir = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-ssh-bundle-"));
|
||||
const bundlePath = path.join(bundleDir, "workspace.bundle");
|
||||
@@ -628,11 +794,31 @@ async function importGitWorkspaceToSsh(input: {
|
||||
`git -C ${shellQuote(input.remoteDir)} update-ref -d ${shellQuote(tempRef)} >/dev/null 2>&1 || true`,
|
||||
].join("\n");
|
||||
|
||||
await streamLocalFileToSsh({
|
||||
spec: input.spec,
|
||||
localFile: bundlePath,
|
||||
remoteScript: remoteSetupScript,
|
||||
});
|
||||
// The git bundle is a real local file of known size, so report an exact
|
||||
// percentage. No `workspace` label: the "Importing git history" phase is
|
||||
// already self-describing in the log line.
|
||||
const progress = input.onProgress
|
||||
? createTransferProgress({
|
||||
onProgress: input.onProgress,
|
||||
phase: "Importing git history",
|
||||
direction: "to",
|
||||
totalBytes: (await fs.stat(bundlePath)).size,
|
||||
estimated: false,
|
||||
})
|
||||
: null;
|
||||
|
||||
try {
|
||||
await streamLocalFileToSsh({
|
||||
spec: input.spec,
|
||||
localFile: bundlePath,
|
||||
remoteScript: remoteSetupScript,
|
||||
progress: progress ?? undefined,
|
||||
});
|
||||
await progress?.finish();
|
||||
} catch (error) {
|
||||
await progress?.fail();
|
||||
throw error;
|
||||
}
|
||||
} finally {
|
||||
await runLocalGit(input.localDir, ["update-ref", "-d", tempRef], {
|
||||
timeout: 10_000,
|
||||
@@ -648,6 +834,7 @@ async function exportGitWorkspaceFromSsh(input: {
|
||||
localDir: string;
|
||||
importedRef?: string;
|
||||
resetLocalWorkspace?: boolean;
|
||||
onProgress?: RuntimeProgressSink;
|
||||
}): Promise<string> {
|
||||
const bundleDir = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-ssh-bundle-"));
|
||||
const bundlePath = path.join(bundleDir, "workspace.bundle");
|
||||
@@ -665,11 +852,30 @@ async function exportGitWorkspaceFromSsh(input: {
|
||||
'cat "$tmp_bundle"',
|
||||
].join("\n");
|
||||
|
||||
await streamSshToLocalFile({
|
||||
spec: input.spec,
|
||||
remoteScript: exportScript,
|
||||
localFile: bundlePath,
|
||||
});
|
||||
// The remote bundle size isn't known before streaming, so report bytes
|
||||
// received (MB mode) with a terminal completion line.
|
||||
const progress = input.onProgress
|
||||
? createTransferProgress({
|
||||
onProgress: input.onProgress,
|
||||
phase: "Exporting git history",
|
||||
direction: "from",
|
||||
totalBytes: null,
|
||||
estimated: false,
|
||||
})
|
||||
: null;
|
||||
|
||||
try {
|
||||
await streamSshToLocalFile({
|
||||
spec: input.spec,
|
||||
remoteScript: exportScript,
|
||||
localFile: bundlePath,
|
||||
progress: progress ?? undefined,
|
||||
});
|
||||
await progress?.finish();
|
||||
} catch (error) {
|
||||
await progress?.fail();
|
||||
throw error;
|
||||
}
|
||||
|
||||
await runLocalGit(input.localDir, ["fetch", "--force", bundlePath, `refs/paperclip/ssh-sync/export:${importedRef}`], {
|
||||
timeout: 60_000,
|
||||
@@ -1049,6 +1255,8 @@ export async function syncDirectoryToSsh(input: {
|
||||
remoteDir: string;
|
||||
exclude?: string[];
|
||||
followSymlinks?: boolean;
|
||||
onProgress?: RuntimeProgressSink;
|
||||
progressLabel?: string;
|
||||
}): Promise<void> {
|
||||
const auth = await createSshAuthArgs(input.spec);
|
||||
const sshArgs = [
|
||||
@@ -1059,7 +1267,27 @@ export async function syncDirectoryToSsh(input: {
|
||||
`sh -c ${shellQuote(`mkdir -p ${shellQuote(input.remoteDir)} && tar -xf - -C ${shellQuote(input.remoteDir)}`)}`,
|
||||
];
|
||||
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
// tar's archive size isn't known until tar finishes, so estimate it from the
|
||||
// local file sizes and clamp the reported percent to 99% until the pipe closes.
|
||||
// The estimate walk runs concurrently with the transfer so it never delays the
|
||||
// pipe from opening on large workspaces.
|
||||
const progress = input.onProgress
|
||||
? createTransferProgress({
|
||||
onProgress: input.onProgress,
|
||||
phase: "Syncing",
|
||||
direction: "to",
|
||||
label: input.progressLabel,
|
||||
totalBytes: estimateLocalDirSize({
|
||||
localDir: input.localDir,
|
||||
exclude: input.exclude,
|
||||
followSymlinks: input.followSymlinks,
|
||||
}),
|
||||
estimated: true,
|
||||
})
|
||||
: null;
|
||||
|
||||
try {
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
const tarArgs = [
|
||||
...(input.followSymlinks ? ["-h"] : []),
|
||||
"-C",
|
||||
@@ -1111,7 +1339,12 @@ export async function syncDirectoryToSsh(input: {
|
||||
reject(error);
|
||||
};
|
||||
|
||||
tar.stdout?.pipe(ssh.stdin ?? null);
|
||||
if (progress) {
|
||||
progress.counter.on("error", fail);
|
||||
tar.stdout?.pipe(progress.counter).pipe(ssh.stdin ?? null);
|
||||
} else {
|
||||
tar.stdout?.pipe(ssh.stdin ?? null);
|
||||
}
|
||||
tar.stderr?.on("data", (chunk) => {
|
||||
tarStderr += String(chunk);
|
||||
});
|
||||
@@ -1131,7 +1364,12 @@ export async function syncDirectoryToSsh(input: {
|
||||
sshExitCode = code;
|
||||
maybeFinish();
|
||||
});
|
||||
}).finally(auth.cleanup);
|
||||
}).finally(auth.cleanup);
|
||||
await progress?.finish();
|
||||
} catch (error) {
|
||||
await progress?.fail();
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
export async function syncDirectoryFromSsh(input: {
|
||||
@@ -1140,6 +1378,8 @@ export async function syncDirectoryFromSsh(input: {
|
||||
localDir: string;
|
||||
exclude?: string[];
|
||||
preserveLocalEntries?: string[];
|
||||
onProgress?: RuntimeProgressSink;
|
||||
progressLabel?: string;
|
||||
}): Promise<void> {
|
||||
const auth = await createSshAuthArgs(input.spec);
|
||||
const stagingDir = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-ssh-sync-back-"));
|
||||
@@ -1155,6 +1395,21 @@ export async function syncDirectoryFromSsh(input: {
|
||||
`sh -c ${shellQuote(remoteTarScript)}`,
|
||||
];
|
||||
|
||||
// The remote tar size isn't known locally, so probe the remote directory for
|
||||
// an estimate (clamped to 99%). The probe runs concurrently with the transfer
|
||||
// so its round-trip never delays the restore; when it is unavailable we report
|
||||
// bytes received in MB mode with a terminal completion line.
|
||||
const progress = input.onProgress
|
||||
? createTransferProgress({
|
||||
onProgress: input.onProgress,
|
||||
phase: "Restoring",
|
||||
direction: "from",
|
||||
label: input.progressLabel,
|
||||
totalBytes: probeRemoteDirSize({ spec: input.spec, remoteDir: input.remoteDir }),
|
||||
estimated: true,
|
||||
})
|
||||
: null;
|
||||
|
||||
try {
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
const ssh = spawn("ssh", sshArgs, {
|
||||
@@ -1195,7 +1450,12 @@ export async function syncDirectoryFromSsh(input: {
|
||||
reject(error);
|
||||
};
|
||||
|
||||
ssh.stdout?.pipe(tar.stdin ?? null);
|
||||
if (progress) {
|
||||
progress.counter.on("error", fail);
|
||||
ssh.stdout?.pipe(progress.counter).pipe(tar.stdin ?? null);
|
||||
} else {
|
||||
ssh.stdout?.pipe(tar.stdin ?? null);
|
||||
}
|
||||
ssh.stderr?.on("data", (chunk) => {
|
||||
sshStderr += String(chunk);
|
||||
});
|
||||
@@ -1216,9 +1476,13 @@ export async function syncDirectoryFromSsh(input: {
|
||||
maybeFinish();
|
||||
});
|
||||
});
|
||||
await progress?.finish();
|
||||
|
||||
await clearLocalDirectory(input.localDir, input.preserveLocalEntries);
|
||||
await copyDirectoryContents(stagingDir, input.localDir);
|
||||
} catch (error) {
|
||||
await progress?.fail();
|
||||
throw error;
|
||||
} finally {
|
||||
await fs.rm(stagingDir, { recursive: true, force: true }).catch(() => undefined);
|
||||
await auth.cleanup();
|
||||
@@ -1229,6 +1493,7 @@ export async function prepareWorkspaceForSshExecution(input: {
|
||||
spec: SshRemoteExecutionSpec;
|
||||
localDir: string;
|
||||
remoteDir?: string;
|
||||
onProgress?: RuntimeProgressSink;
|
||||
}): Promise<{ gitBacked: boolean }> {
|
||||
const remoteDir = input.remoteDir ?? input.spec.remoteCwd;
|
||||
const gitSnapshot = await readLocalGitWorkspaceSnapshot(input.localDir);
|
||||
@@ -1239,12 +1504,15 @@ export async function prepareWorkspaceForSshExecution(input: {
|
||||
localDir: input.localDir,
|
||||
remoteDir,
|
||||
snapshot: gitSnapshot,
|
||||
onProgress: input.onProgress,
|
||||
});
|
||||
await syncDirectoryToSsh({
|
||||
spec: input.spec,
|
||||
localDir: input.localDir,
|
||||
remoteDir,
|
||||
exclude: [".git", ".paperclip-runtime"],
|
||||
onProgress: input.onProgress,
|
||||
progressLabel: "workspace",
|
||||
});
|
||||
await removeDeletedPathsOnSsh({
|
||||
spec: input.spec,
|
||||
@@ -1264,6 +1532,8 @@ export async function prepareWorkspaceForSshExecution(input: {
|
||||
localDir: input.localDir,
|
||||
remoteDir,
|
||||
exclude: [".paperclip-runtime"],
|
||||
onProgress: input.onProgress,
|
||||
progressLabel: "workspace",
|
||||
});
|
||||
return { gitBacked: false };
|
||||
}
|
||||
@@ -1274,6 +1544,7 @@ export async function restoreWorkspaceFromSshExecution(input: {
|
||||
remoteDir?: string;
|
||||
baselineSnapshot?: DirectorySnapshot;
|
||||
restoreGitHistory?: boolean;
|
||||
onProgress?: RuntimeProgressSink;
|
||||
}): Promise<void> {
|
||||
const remoteDir = input.remoteDir ?? input.spec.remoteCwd;
|
||||
if (input.baselineSnapshot) {
|
||||
@@ -1289,6 +1560,7 @@ export async function restoreWorkspaceFromSshExecution(input: {
|
||||
localDir: input.localDir,
|
||||
importedRef: importedRef ?? undefined,
|
||||
resetLocalWorkspace: false,
|
||||
onProgress: input.onProgress,
|
||||
})
|
||||
: null;
|
||||
await syncDirectoryFromSsh({
|
||||
@@ -1296,6 +1568,8 @@ export async function restoreWorkspaceFromSshExecution(input: {
|
||||
remoteDir,
|
||||
localDir: stagingDir,
|
||||
exclude: input.baselineSnapshot.exclude,
|
||||
onProgress: input.onProgress,
|
||||
progressLabel: "workspace",
|
||||
});
|
||||
await mergeDirectoryWithBaseline({
|
||||
baseline: input.baselineSnapshot,
|
||||
@@ -1330,6 +1604,7 @@ export async function restoreWorkspaceFromSshExecution(input: {
|
||||
spec: input.spec,
|
||||
remoteDir,
|
||||
localDir: input.localDir,
|
||||
onProgress: input.onProgress,
|
||||
});
|
||||
await syncDirectoryFromSsh({
|
||||
spec: input.spec,
|
||||
@@ -1337,6 +1612,8 @@ export async function restoreWorkspaceFromSshExecution(input: {
|
||||
localDir: input.localDir,
|
||||
exclude: [".git", ".paperclip-runtime"],
|
||||
preserveLocalEntries: [".git"],
|
||||
onProgress: input.onProgress,
|
||||
progressLabel: "workspace",
|
||||
});
|
||||
return;
|
||||
}
|
||||
@@ -1346,6 +1623,8 @@ export async function restoreWorkspaceFromSshExecution(input: {
|
||||
remoteDir,
|
||||
localDir: input.localDir,
|
||||
exclude: [".paperclip-runtime"],
|
||||
onProgress: input.onProgress,
|
||||
progressLabel: "workspace",
|
||||
});
|
||||
}
|
||||
|
||||
|
||||
@@ -485,6 +485,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
workspaceLocalDir: cwd,
|
||||
installCommand: SANDBOX_INSTALL_COMMAND,
|
||||
detectCommand: command,
|
||||
onProgress: (line) => onLog("stdout", line),
|
||||
assets: [
|
||||
{
|
||||
key: "skills",
|
||||
@@ -523,7 +524,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
executionCwd: effectiveExecutionCwd,
|
||||
});
|
||||
const restoreRemoteWorkspace = preparedExecutionTargetRuntime
|
||||
? () => preparedExecutionTargetRuntime.restoreWorkspace()
|
||||
? () => preparedExecutionTargetRuntime.restoreWorkspace((line) => onLog("stdout", line))
|
||||
: null;
|
||||
const effectivePromptBundleAddDir = executionTargetIsRemote
|
||||
? preparedExecutionTargetRuntime?.assetDirs.skills ??
|
||||
|
||||
@@ -397,6 +397,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
workspaceLocalDir: cwd,
|
||||
installCommand: SANDBOX_INSTALL_COMMAND,
|
||||
detectCommand: command,
|
||||
onProgress: (line) => onLog("stdout", line),
|
||||
assets: [
|
||||
{
|
||||
key: "home",
|
||||
@@ -414,7 +415,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
const executionTargetIsSandbox =
|
||||
runtimeExecutionTarget?.kind === "remote" && runtimeExecutionTarget.transport === "sandbox";
|
||||
const restoreRemoteWorkspace = preparedExecutionTargetRuntime
|
||||
? () => preparedExecutionTargetRuntime.restoreWorkspace()
|
||||
? () => preparedExecutionTargetRuntime.restoreWorkspace((line) => onLog("stdout", line))
|
||||
: null;
|
||||
let paperclipBridge: Awaited<ReturnType<typeof startAdapterExecutionTargetPaperclipBridge>> = null;
|
||||
const remoteCodexHome = executionTargetIsRemote
|
||||
|
||||
@@ -364,13 +364,15 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
workspaceLocalDir: cwd,
|
||||
installCommand: SANDBOX_INSTALL_COMMAND,
|
||||
detectCommand: command,
|
||||
onProgress: (line) => onLog("stdout", line),
|
||||
assets: [{
|
||||
key: "skills",
|
||||
localDir: localSkillsDir,
|
||||
followSymlinks: true,
|
||||
}],
|
||||
});
|
||||
restoreRemoteWorkspace = () => preparedExecutionTargetRuntime.restoreWorkspace();
|
||||
restoreRemoteWorkspace = () =>
|
||||
preparedExecutionTargetRuntime.restoreWorkspace((line) => onLog("stdout", line));
|
||||
effectiveExecutionCwd = preparedExecutionTargetRuntime.workspaceRemoteDir ?? effectiveExecutionCwd;
|
||||
refreshPaperclipWorkspaceEnvForExecution({
|
||||
env,
|
||||
|
||||
@@ -340,13 +340,15 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
workspaceLocalDir: cwd,
|
||||
installCommand: SANDBOX_INSTALL_COMMAND,
|
||||
detectCommand: command,
|
||||
onProgress: (line) => onLog("stdout", line),
|
||||
assets: [{
|
||||
key: "skills",
|
||||
localDir: localSkillsDir,
|
||||
followSymlinks: true,
|
||||
}],
|
||||
});
|
||||
restoreRemoteWorkspace = () => preparedExecutionTargetRuntime.restoreWorkspace();
|
||||
restoreRemoteWorkspace = () =>
|
||||
preparedExecutionTargetRuntime.restoreWorkspace((line) => onLog("stdout", line));
|
||||
effectiveExecutionCwd = preparedExecutionTargetRuntime.workspaceRemoteDir ?? effectiveExecutionCwd;
|
||||
refreshPaperclipWorkspaceEnvForExecution({
|
||||
env,
|
||||
|
||||
@@ -325,8 +325,10 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
timeoutSec,
|
||||
installCommand: ctx.runtimeCommandSpec?.installCommand ?? null,
|
||||
detectCommand: ctx.runtimeCommandSpec?.detectCommand ?? command,
|
||||
onProgress: (line) => onLog("stdout", line),
|
||||
});
|
||||
restoreRemoteWorkspace = () => preparedExecutionTargetRuntime.restoreWorkspace();
|
||||
restoreRemoteWorkspace = () =>
|
||||
preparedExecutionTargetRuntime.restoreWorkspace((line) => onLog("stdout", line));
|
||||
effectiveExecutionCwd = preparedExecutionTargetRuntime.workspaceRemoteDir ?? effectiveExecutionCwd;
|
||||
refreshPaperclipWorkspaceEnvForExecution({
|
||||
env,
|
||||
|
||||
@@ -377,6 +377,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
workspaceLocalDir: cwd,
|
||||
installCommand: SANDBOX_INSTALL_COMMAND,
|
||||
detectCommand: command,
|
||||
onProgress: (line) => onLog("stdout", line),
|
||||
assets: [
|
||||
{
|
||||
key: "skills",
|
||||
@@ -391,7 +392,8 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
: []),
|
||||
],
|
||||
});
|
||||
restoreRemoteWorkspace = () => preparedExecutionTargetRuntime.restoreWorkspace();
|
||||
restoreRemoteWorkspace = () =>
|
||||
preparedExecutionTargetRuntime.restoreWorkspace((line) => onLog("stdout", line));
|
||||
effectiveExecutionCwd = preparedExecutionTargetRuntime.workspaceRemoteDir ?? effectiveExecutionCwd;
|
||||
refreshPaperclipWorkspaceEnvForExecution({
|
||||
env: preparedRuntimeConfig.env,
|
||||
|
||||
@@ -417,6 +417,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
workspaceLocalDir: cwd,
|
||||
installCommand: SANDBOX_INSTALL_COMMAND,
|
||||
detectCommand: command,
|
||||
onProgress: (line) => onLog("stdout", line),
|
||||
assets: [
|
||||
{
|
||||
key: "skills",
|
||||
@@ -431,7 +432,8 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
: []),
|
||||
],
|
||||
});
|
||||
restoreRemoteWorkspace = () => preparedRemoteRuntime.restoreWorkspace();
|
||||
restoreRemoteWorkspace = () =>
|
||||
preparedRemoteRuntime.restoreWorkspace((line) => onLog("stdout", line));
|
||||
effectiveExecutionCwd = preparedRemoteRuntime.workspaceRemoteDir ?? effectiveExecutionCwd;
|
||||
refreshPaperclipWorkspaceEnvForExecution({
|
||||
env,
|
||||
|
||||
@@ -75,6 +75,7 @@ export async function resolveEnvironmentExecutionTarget(input: {
|
||||
timeoutMs,
|
||||
runner: input.environmentRuntime && input.lease
|
||||
? {
|
||||
supportsSingleStreamStdinProgress: false,
|
||||
execute: async (commandInput) => {
|
||||
const startedAt = new Date().toISOString();
|
||||
const result = await input.environmentRuntime!.execute({
|
||||
|
||||
Reference in New Issue
Block a user