mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-17 10:06:42 +02:00
fix(runtime): bound shared uploads and restrict computer tools to writable tasks
This commit is contained in:
1 parent
9ef482b1d0
commit
e498b05159
2 files changed
+77
-7
No files matched your search
@@ -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",
|
||||
|
||||
@@ -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<AdapterExecutionTarget, { kind: "remote" }>;
|
||||
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<void>): Promise<void> {
|
||||
await new Promise<void>((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")
|
||||
|
||||
Reference in new issue
Block a user