From 07e98d2b2cf35d380cde5854c9f69b72d2b2617c Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Sat, 20 Jun 2026 13:03:42 -0700 Subject: [PATCH] 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 --- .../src/command-managed-runtime.test.ts | 231 +++++++++---- .../src/command-managed-runtime.ts | 134 +++++++- .../adapter-utils/src/execution-target.ts | 12 +- packages/adapter-utils/src/index.ts | 9 + .../src/remote-managed-runtime.ts | 13 +- .../src/runtime-progress.test.ts | 281 ++++++++++++++++ .../adapter-utils/src/runtime-progress.ts | 152 +++++++++ .../src/sandbox-managed-runtime.test.ts | 83 +++++ .../src/sandbox-managed-runtime.ts | 72 +++- .../adapter-utils/src/ssh-fixture.test.ts | 162 +++++++++ packages/adapter-utils/src/ssh.ts | 311 +++++++++++++++++- .../claude-local/src/server/execute.ts | 3 +- .../codex-local/src/server/execute.ts | 3 +- .../cursor-local/src/server/execute.ts | 4 +- .../gemini-local/src/server/execute.ts | 4 +- .../adapters/grok-local/src/server/execute.ts | 4 +- .../opencode-local/src/server/execute.ts | 4 +- .../adapters/pi-local/src/server/execute.ts | 4 +- .../services/environment-execution-target.ts | 1 + 19 files changed, 1377 insertions(+), 110 deletions(-) create mode 100644 packages/adapter-utils/src/runtime-progress.test.ts create mode 100644 packages/adapter-utils/src/runtime-progress.ts diff --git a/packages/adapter-utils/src/command-managed-runtime.test.ts b/packages/adapter-utils/src/command-managed-runtime.test.ts index df0c55d2..a910fb80 100644 --- a/packages/adapter-utils/src/command-managed-runtime.test.ts +++ b/packages/adapter-utils/src/command-managed-runtime.test.ts @@ -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((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; - stdin?: string; - timeoutMs?: number; - }> = []; - const runner = { - execute: async (input: { - command: string; - args?: string[]; - cwd?: string; - env?: Record; - stdin?: string; - timeoutMs?: number; - }): Promise => { - 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); + }); }); diff --git a/packages/adapter-utils/src/command-managed-runtime.ts b/packages/adapter-utils/src/command-managed-runtime.ts index dbcd5ab9..a74789cc 100644 --- a/packages/adapter-utils/src/command-managed-runtime.ts +++ b/packages/adapter-utils/src/command-managed-runtime.ts @@ -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; + } = {}, + ) => { 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 ` 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 { 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, }); } diff --git a/packages/adapter-utils/src/execution-target.ts b/packages/adapter-utils/src/execution-target.ts index 8b6d5bbb..86d3f1a8 100644 --- a/packages/adapter-utils/src/execution-target.ts +++ b/packages/adapter-utils/src/execution-target.ts @@ -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; - restoreWorkspace(): Promise; + restoreWorkspace(onProgress?: RuntimeProgressSink): Promise; } 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 { 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, diff --git a/packages/adapter-utils/src/index.ts b/packages/adapter-utils/src/index.ts index fa5b7632..303c028e 100644 --- a/packages/adapter-utils/src/index.ts +++ b/packages/adapter-utils/src/index.ts @@ -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. diff --git a/packages/adapter-utils/src/remote-managed-runtime.ts b/packages/adapter-utils/src/remote-managed-runtime.ts index 900824b4..6adb2561 100644 --- a/packages/adapter-utils/src/remote-managed-runtime.ts +++ b/packages/adapter-utils/src/remote-managed-runtime.ts @@ -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; - restoreWorkspace(): Promise; + restoreWorkspace(onProgress?: RuntimeProgressSink): Promise; } function asObject(value: unknown): Record { @@ -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 { 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, }); }, }; diff --git a/packages/adapter-utils/src/runtime-progress.test.ts b/packages/adapter-utils/src/runtime-progress.test.ts new file mode 100644 index 00000000..24048404 --- /dev/null +++ b/packages/adapter-utils/src/runtime-progress.test.ts @@ -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%"); + }); +}); diff --git a/packages/adapter-utils/src/runtime-progress.ts b/packages/adapter-utils/src/runtime-progress.ts new file mode 100644 index 00000000..8cdd7ce4 --- /dev/null +++ b/packages/adapter-utils/src/runtime-progress.ts @@ -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; + +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; + /** + * Emit the terminal completion line if it hasn't been emitted yet. Idempotent. + */ + complete(doneBytes?: number, totalBytes?: number | null): Promise; + /** + * 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; +} + +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 { + 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)); + }, + }; +} diff --git a/packages/adapter-utils/src/sandbox-managed-runtime.test.ts b/packages/adapter-utils/src/sandbox-managed-runtime.test.ts index bd3be700..b6c4a085 100644 --- a/packages/adapter-utils/src/sandbox-managed-runtime.test.ts +++ b/packages/adapter-utils/src/sandbox-managed-runtime.test.ts @@ -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) | 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); diff --git a/packages/adapter-utils/src/sandbox-managed-runtime.ts b/packages/adapter-utils/src/sandbox-managed-runtime.ts index 81bd90b9..e374121a 100644 --- a/packages/adapter-utils/src/sandbox-managed-runtime.ts +++ b/packages/adapter-utils/src/sandbox-managed-runtime.ts @@ -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; +} + export interface SandboxManagedRuntimeClient { makeDir(remotePath: string): Promise; - writeFile(remotePath: string, bytes: ArrayBuffer): Promise; - readFile(remotePath: string): Promise; + writeFile(remotePath: string, bytes: ArrayBuffer, options?: SandboxTransferProgressOptions): Promise; + readFile( + remotePath: string, + options?: SandboxTransferProgressOptions, + ): Promise; listFiles(remotePath: string): Promise; remove(remotePath: string): Promise; run(command: string, options: { timeoutMs: number }): Promise; @@ -38,7 +52,7 @@ export interface PreparedSandboxManagedRuntime { workspaceRemoteDir: string; runtimeRootDir: string; assetDirs: Record; - restoreWorkspace(): Promise; + restoreWorkspace(onProgress?: RuntimeProgressSink): Promise; } function asObject(value: unknown): Record { @@ -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 } { + 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 { 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"); diff --git a/packages/adapter-utils/src/ssh-fixture.test.ts b/packages/adapter-utils/src/ssh-fixture.test.ts index 3fb8f15f..ba2da732 100644 --- a/packages/adapter-utils/src/ssh-fixture.test.ts +++ b/packages/adapter-utils/src/ssh-fixture.test.ts @@ -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); diff --git a/packages/adapter-utils/src/ssh.ts b/packages/adapter-utils/src/ssh.ts index 3a5cc644..d3efffd7 100644 --- a/packages/adapter-utils/src/ssh.ts +++ b/packages/adapter-utils/src/ssh.ts @@ -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 { + 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 => { + 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 { + 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; + // Emit a terminal failure marker instead of a completion line. Idempotent. + fail: () => Promise; +} + +// 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; + 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 = + input.totalBytes != null && typeof (input.totalBytes as Promise).then === "function" + ? (input.totalBytes as Promise).then(applyTotal, () => applyTotal(null)) + : (applyTotal(input.totalBytes as number | null), Promise.resolve()); + + let transferred = 0; + let chain: Promise = Promise.resolve(); + const enqueue = (work: () => Promise) => { + 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 { 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 { 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 { 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 { 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 { 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((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((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 { 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((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 { 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", }); } diff --git a/packages/adapters/claude-local/src/server/execute.ts b/packages/adapters/claude-local/src/server/execute.ts index d238bbe6..57848b73 100644 --- a/packages/adapters/claude-local/src/server/execute.ts +++ b/packages/adapters/claude-local/src/server/execute.ts @@ -485,6 +485,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise onLog("stdout", line), assets: [ { key: "skills", @@ -523,7 +524,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise preparedExecutionTargetRuntime.restoreWorkspace() + ? () => preparedExecutionTargetRuntime.restoreWorkspace((line) => onLog("stdout", line)) : null; const effectivePromptBundleAddDir = executionTargetIsRemote ? preparedExecutionTargetRuntime?.assetDirs.skills ?? diff --git a/packages/adapters/codex-local/src/server/execute.ts b/packages/adapters/codex-local/src/server/execute.ts index 2c52aa2b..ce99fda3 100644 --- a/packages/adapters/codex-local/src/server/execute.ts +++ b/packages/adapters/codex-local/src/server/execute.ts @@ -397,6 +397,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise onLog("stdout", line), assets: [ { key: "home", @@ -414,7 +415,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise preparedExecutionTargetRuntime.restoreWorkspace() + ? () => preparedExecutionTargetRuntime.restoreWorkspace((line) => onLog("stdout", line)) : null; let paperclipBridge: Awaited> = null; const remoteCodexHome = executionTargetIsRemote diff --git a/packages/adapters/cursor-local/src/server/execute.ts b/packages/adapters/cursor-local/src/server/execute.ts index 800a499f..40c3e8ff 100644 --- a/packages/adapters/cursor-local/src/server/execute.ts +++ b/packages/adapters/cursor-local/src/server/execute.ts @@ -364,13 +364,15 @@ export async function execute(ctx: AdapterExecutionContext): Promise 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, diff --git a/packages/adapters/gemini-local/src/server/execute.ts b/packages/adapters/gemini-local/src/server/execute.ts index 0a3c0da6..4f9d5486 100644 --- a/packages/adapters/gemini-local/src/server/execute.ts +++ b/packages/adapters/gemini-local/src/server/execute.ts @@ -340,13 +340,15 @@ export async function execute(ctx: AdapterExecutionContext): Promise 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, diff --git a/packages/adapters/grok-local/src/server/execute.ts b/packages/adapters/grok-local/src/server/execute.ts index 0d25845a..56e70a70 100644 --- a/packages/adapters/grok-local/src/server/execute.ts +++ b/packages/adapters/grok-local/src/server/execute.ts @@ -325,8 +325,10 @@ export async function execute(ctx: AdapterExecutionContext): Promise onLog("stdout", line), }); - restoreRemoteWorkspace = () => preparedExecutionTargetRuntime.restoreWorkspace(); + restoreRemoteWorkspace = () => + preparedExecutionTargetRuntime.restoreWorkspace((line) => onLog("stdout", line)); effectiveExecutionCwd = preparedExecutionTargetRuntime.workspaceRemoteDir ?? effectiveExecutionCwd; refreshPaperclipWorkspaceEnvForExecution({ env, diff --git a/packages/adapters/opencode-local/src/server/execute.ts b/packages/adapters/opencode-local/src/server/execute.ts index b7ae92bc..247480b1 100644 --- a/packages/adapters/opencode-local/src/server/execute.ts +++ b/packages/adapters/opencode-local/src/server/execute.ts @@ -377,6 +377,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise onLog("stdout", line), assets: [ { key: "skills", @@ -391,7 +392,8 @@ export async function execute(ctx: AdapterExecutionContext): Promise preparedExecutionTargetRuntime.restoreWorkspace(); + restoreRemoteWorkspace = () => + preparedExecutionTargetRuntime.restoreWorkspace((line) => onLog("stdout", line)); effectiveExecutionCwd = preparedExecutionTargetRuntime.workspaceRemoteDir ?? effectiveExecutionCwd; refreshPaperclipWorkspaceEnvForExecution({ env: preparedRuntimeConfig.env, diff --git a/packages/adapters/pi-local/src/server/execute.ts b/packages/adapters/pi-local/src/server/execute.ts index 51579662..30d4d16a 100644 --- a/packages/adapters/pi-local/src/server/execute.ts +++ b/packages/adapters/pi-local/src/server/execute.ts @@ -417,6 +417,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise onLog("stdout", line), assets: [ { key: "skills", @@ -431,7 +432,8 @@ export async function execute(ctx: AdapterExecutionContext): Promise preparedRemoteRuntime.restoreWorkspace(); + restoreRemoteWorkspace = () => + preparedRemoteRuntime.restoreWorkspace((line) => onLog("stdout", line)); effectiveExecutionCwd = preparedRemoteRuntime.workspaceRemoteDir ?? effectiveExecutionCwd; refreshPaperclipWorkspaceEnvForExecution({ env, diff --git a/server/src/services/environment-execution-target.ts b/server/src/services/environment-execution-target.ts index 33ea7015..0e50bd7b 100644 --- a/server/src/services/environment-execution-target.ts +++ b/server/src/services/environment-execution-target.ts @@ -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({