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 71a256d3c8..43727f505e 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -368,6 +368,7 @@ import { sha256DirectoryTree, stageRemoteRunnerDirectory, stageRemoteRunnerFile, + nativeComputerToolForExecution, steerNativeSession, syncRemoteRunnerDirectoryOut, verifyNativeHarnessBackup, @@ -2466,6 +2467,17 @@ describe("remote provider checkpoint snapshots", () => { }); }); +describe("computer tool execution authority", () => { + it.each(["standard", "planning", "ask"] as const)("only installs mutation-capable desktop tools for standard work (%s)", (workMode) => { + const tool = { command: "cua-driver", args: ["mcp"] }; + const target = { kind: "remote", transport: "computer", computerTool: tool } as never; + const input = { task: { workMode }, executionMode: "default" } as NativeExecutionInput; + expect(nativeComputerToolForExecution(input, target)).toBe(workMode === "standard" ? tool : undefined); + expect(nativeComputerToolForExecution({ ...input, executionMode: "plan" } as NativeExecutionInput, target)).toBeUndefined(); + expect(nativeComputerToolForExecution(input, { kind: "remote", transport: "sandbox" } as never)).toBeUndefined(); + }); +}); + describe("remote provider checkpoint restores", () => { it("budgets artifact upload time and retains redacted staging diagnostics", async () => { const root = await mkdtemp(join(tmpdir(), "paperclip-runner-file-")); @@ -2514,7 +2526,7 @@ describe("remote provider checkpoint restores", () => { expect(await readFile(targetPath, "utf8")).toBe("previous verified binary"); } else { await stage; - expect(await readFile(targetPath)).toEqual(bytes); + expect((await readFile(targetPath)).equals(bytes)).toBe(true); expect((await lstat(targetPath)).mode & 0o777).toBe(0o700); } expect(maxInFlight).toBe(3); @@ -2522,6 +2534,31 @@ describe("remote provider checkpoint restores", () => { } finally { await rm(root, { recursive: true, force: true }); } }); + it("shares eight upload slots across concurrent artifact transfers", async () => { + const root = await mkdtemp(join(tmpdir(), "paperclip-runner-upload-slots-")); + const sourcePath = join(root, "source"); + await writeFile(sourcePath, randomBytes(20 * 1024 * 1024)); + let active = 0; + let peak = 0; + const execute = vi.fn(async (request: { command: string; args: string[]; stdin?: string }) => { + if (request.stdin) { active++; peak = Math.max(peak, active); } + try { + if (request.stdin) await new Promise((resolve) => setTimeout(resolve, 25)); + execFileSync(request.command, request.args, { input: request.stdin }); + return { exitCode: 0, timedOut: false, stdout: "", stderr: "" }; + } finally { if (request.stdin) active--; } + }); + try { + await Promise.all(["one", "two"].map(target => stageRemoteRunnerFile({ + target: { kind: "remote", transport: "computer" } as never, + runner: { execute } as never, sourcePath, targetPath: join(root, target), mode: 0o700, + }))); + expect(peak).toBe(8); + expect(active).toBe(0); + expect((await readFile(join(root, "one"))).equals(await readFile(join(root, "two")))).toBe(true); + } finally { await rm(root, { recursive: true, force: true }); } + }); + it.each([{ sizeMiB: 0, corrupt: false }, { sizeMiB: 65, corrupt: false }, { sizeMiB: 0, corrupt: true }])("stages a $sizeMiB MiB provider pack through a command-only computer runner (corrupt=$corrupt)", async ({ sizeMiB, corrupt }) => { const root = await mkdtemp(join(tmpdir(), "paperclip-computer-pack-")); const sourcePath = join(root, "source"); @@ -7559,7 +7596,7 @@ describe("native warm session supervision", () => { await closeWarmNativeSessionsForRun({ runId: owners[1]!.runId, reason: "fixture cleanup" }); }); - it.each(["model", "credential", "tools"])("replaces a computer warm runner after %s configuration admits a fresh owner", async (change) => { + it.each(["model", "credential", "tools", "read-only"])("replaces a computer warm runner after %s configuration admits a fresh owner", async (change) => { const name = `computer-config-change-${change}`; const current = { ...execution, binding: { ...execution.binding, runId: `${name}-one`, agentId: name, executionWorkspaceId: name }, session: { ...execution.session, normalizedSessionId: name, lifecyclePolicy: { mode: "warm" as const, idleTimeoutMs: 60_000 } } } as NativeExecutionInputV1; @@ -7580,7 +7617,8 @@ describe("native warm session supervision", () => { }); await executePaperclipNativeSession({ db: leaseDb(current), execution: current, runnerInstanceId: name, runnerExecutionTarget: target as never, managedAiCredentialIdentity: "credential-before" }); - const next = { ...current, binding: { ...current.binding, runId: `${name}-two` }, provider: change === "model" ? { ...current.provider, model: "changed-model" } : current.provider } as NativeExecutionInputV1; + const next = { ...current, binding: { ...current.binding, runId: `${name}-two` }, provider: change === "model" ? { ...current.provider, model: "changed-model" } : current.provider, + task: { ...current.task, workMode: change === "read-only" ? "ask" : "standard" } } as NativeExecutionInputV1; await executePaperclipNativeSession({ db: leaseDb(next), execution: next, runnerInstanceId: name, managedAiCredentialIdentity: change === "credential" ? "credential-after" : "credential-before", refreshTools: change === "tools", diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 0a301c8851..e5bef3d137 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -5430,6 +5430,7 @@ function nativeSessionConfigDigest( provider: execution.provider, driverKind: execution.session.driverKind, lifecyclePolicy: execution.session.lifecyclePolicy, + workMode: execution.task.workMode, executionMode: "executionMode" in execution ? execution.executionMode : "default", runtimeContextDigest: @@ -10249,6 +10250,16 @@ export function assertRemoteRunnerBuildMetadata( } } +export function nativeComputerToolForExecution( + execution: NativeExecutionInput, + target: AdapterExecutionTarget | null | undefined, +): AdapterComputerExecutionTarget["computerTool"] | undefined { + if (target?.kind !== "remote" || target.transport !== "computer" || + execution.task.workMode !== "standard" || + ("executionMode" in execution && execution.executionMode === "plan")) return undefined; + return target.computerTool; +} + export async function stageRemoteRunnerFile(input: { target: Extract; runner: CommandManagedRuntimeRunner; @@ -10298,6 +10309,25 @@ export async function stageRemoteRunnerFile(input: { } } +const REMOTE_ARTIFACT_UPLOAD_CONCURRENCY = 8; +let activeRemoteArtifactUploads = 0; +const remoteArtifactUploadWaiters: Array<() => void> = []; + +async function withRemoteArtifactUploadSlot(operation: () => Promise): Promise { + await new Promise((resolve) => { + if (activeRemoteArtifactUploads < REMOTE_ARTIFACT_UPLOAD_CONCURRENCY) { + activeRemoteArtifactUploads++; + resolve(); + } else remoteArtifactUploadWaiters.push(resolve); + }); + try { await operation(); } + finally { + const next = remoteArtifactUploadWaiters.shift(); + if (next) next(); + else activeRemoteArtifactUploads--; + } +} + async function uploadRemoteRunnerFileChunks(input: { sourcePath: string; remotePath: string; @@ -10323,9 +10353,11 @@ async function uploadRemoteRunnerFileChunks(input: { 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(remotePath)} bs=4194304 seek=${chunkIndex++} conv=notrunc 2>/dev/null`, - chunk.toString("base64")).catch((error) => { uploadError ??= error; })); - if (pending.length === 4) { + const offset = chunkIndex++; + pending.push(withRemoteArtifactUploadSlot(() => run( + `base64 -d | dd of=${quote(remotePath)} bs=4194304 seek=${offset} conv=notrunc 2>/dev/null`, + chunk.toString("base64"))).catch((error) => { uploadError ??= error; })); + if (pending.length === REMOTE_ARTIFACT_UPLOAD_CONCURRENCY) { await Promise.all(pending); pending.length = 0; if (uploadError) throw uploadError; @@ -13451,7 +13483,7 @@ async function createRunnerdBackendWithinSessionClaim( runnerFilesystemRoot: remoteRunnerFilesystemRoot ?? undefined, resumeWorkingDirectory: runnerExecution.workspace.cwd, externallySandboxed: adapterExecutionTargetIsCommandBacked(remoteTarget), - computerTool: remoteTarget?.transport === "computer" ? remoteTarget.computerTool : undefined, + computerTool: nativeComputerToolForExecution(input.execution, remoteTarget), persistentAgentHome: remoteTarget?.transport === "computer" ? remoteTarget.fileAuthority.agentHome : undefined, opencodeRuntimeDirectory: remoteRunnerFilesystemRoot ? posix.join(remoteRunnerFilesystemRoot, "opencode")