From 4db5d95ecc7778ac380af9f62d58cb18d304feeb Mon Sep 17 00:00:00 2001 From: Dotta Date: Sat, 10 Oct 2026 09:12:06 -0500 Subject: [PATCH] Report bounded provider pack upload progress in run logs --- .../native-session-executor.test.ts | 4 ++++ .../native-runtime/native-session-executor.ts | 15 +++++++++++++++ 2 files changed, 19 insertions(+) diff --git a/server/src/services/native-runtime/native-session-executor.test.ts b/server/src/services/native-runtime/native-session-executor.test.ts index 84e36a6a62f..a92dbb2b5b2 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -2491,12 +2491,14 @@ describe("remote provider checkpoint restores", () => { return { exitCode: 0, timedOut: false, stdout: "", stderr: "" }; }); const runner = { execute }; + const onProgress = vi.fn(async (_completed: number, _total: number) => undefined); const staging = stageRemoteRunnerDirectory({ target: { kind: "remote", transport: "computer", remoteCwd: root, runner } as never, runner: runner as never, sourcePath, targetPath, mode: 0o700, + onProgress, }); if (corrupt) { await expect(staging).rejects.toThrow(); @@ -2505,6 +2507,8 @@ describe("remote provider checkpoint restores", () => { return; } await staging; + expect(onProgress.mock.calls[0]![0]).toBe(0); + expect(onProgress.mock.calls.at(-1)![0]).toBe(onProgress.mock.calls.at(-1)![1]); expect(await readFile(join(targetPath, "provider-pack.json"), "utf8")).toBe(manifest); expect((await lstat(join(targetPath, "bin", "provider"))).mode & 0o100).toBe(0o100); expect((await lstat(targetPath)).mode & 0o777).toBe(0o700); diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index b4b0dd42522..494e9767c5f 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -10317,6 +10317,7 @@ export async function stageRemoteRunnerDirectory(input: { targetPath: string; mode: number; excludeEntries?: readonly string[]; + onProgress?: (completedBytes: number, totalBytes: number) => Promise; }): Promise { const excludeArgs = archiveExcludeArgs(input.excludeEntries ?? []); if (input.runner.syncIn) { @@ -10378,11 +10379,16 @@ export async function stageRemoteRunnerDirectory(input: { }); await run(`umask 077; mkdir -p ${quote(posix.dirname(remoteArchive))} && : > ${quote(remoteArchive)}`); const digest = createHash("sha256"); + const totalBytes = lstatSync(archivePath).size; + let completedBytes = 0; + let progressAt = Date.now(); + await input.onProgress?.(0, totalBytes); let chunkIndex = 0; const stream = createReadStream(archivePath, { highWaterMark: 4 * 1024 * 1024 }); for await (const bytes of stream) { const chunk = bytes as Buffer; digest.update(chunk); + completedBytes += chunk.length; // Disjoint fixed offsets permit bounded parallel upload and idempotent // writes. The final digest rejects missing, duplicated, or partial bytes. pending.push(run(`base64 -d | dd of=${quote(remoteArchive)} bs=4194304 seek=${chunkIndex++} conv=notrunc 2>/dev/null`, @@ -10391,10 +10397,15 @@ export async function stageRemoteRunnerDirectory(input: { await Promise.all(pending); pending.length = 0; if (uploadError) throw uploadError; + if (Date.now() - progressAt >= 15_000) { + await input.onProgress?.(completedBytes, totalBytes); + progressAt = Date.now(); + } } } await Promise.all(pending); if (uploadError) throw uploadError; + await input.onProgress?.(totalBytes, totalBytes); const expectedDigest = digest.digest("hex"); await run(`test "$(if command -v sha256sum >/dev/null 2>&1; then sha256sum ${quote(remoteArchive)}; else shasum -a 256 ${quote(remoteArchive)}; fi | cut -d ' ' -f 1)" = ${quote(expectedDigest)} && ` + `umask 077 && mkdir -p ${quote(input.targetPath)} && ` + @@ -12266,12 +12277,16 @@ async function createRunnerdBackendWithinSessionClaim( "runner_remote_provider_artifact_incompatible: stale provider pack could not be replaced", ); } + await input.onLog?.("stderr", "[paperclip-runner] Preparing the provider pack for first use.\n"); await stageRemoteRunnerDirectory({ target: remoteTarget, runner: remoteCommandRunner, sourcePath: configuredProviderPackRoot, targetPath: stagedRemoteProviderPackRoot, mode: 0o700, + onProgress: async (completed, total) => { + await input.onLog?.("stderr", `[paperclip-runner] Uploading provider pack: ${Math.round(completed / 1024 / 1024)} / ${Math.round(total / 1024 / 1024)} MiB.\n`); + }, }); await measureNativeRunnerSpan(input.trace, "provider_pack.verify", () => verifyRemoteProviderPack(stagedRemoteProviderPackRoot),