Allow repository ZIP uploads up to 5 GiB (#194)
* Allow ZIP uploads up to 5 GiB and stream them onto Docker computers Repository archives could not be picked at all: application/zip was not an allowed attachment type. ZIPs are now allowed with their own 5 GiB cap, upload as a stream, and reach the bot's Docker computer through a new streamed supervisor write that commits only after the body ended cleanly. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Tgvrj34NMnaqvtgvruYQZZ * Use a disallowed type for the invalid-attachment check now that ZIP is allowed Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Tgvrj34NMnaqvtgvruYQZZ --------- Co-authored-by: Claude <noreply@anthropic.com>
This commit is contained in:
parent
81c4bfd7e3
commit
797906dde2
@ -175,7 +175,7 @@ describe("upload timeouts", () => {
|
|||||||
|
|
||||||
it("gives the server time for large uploads but keeps the header timeout", () => {
|
it("gives the server time for large uploads but keeps the header timeout", () => {
|
||||||
const server = createServer(API_SERVER_OPTIONS);
|
const server = createServer(API_SERVER_OPTIONS);
|
||||||
expect(server.requestTimeout).toBe(30 * 60 * 1000);
|
expect(server.requestTimeout).toBe(2 * 60 * 60 * 1000);
|
||||||
expect(server.headersTimeout).toBe(createServer().headersTimeout);
|
expect(server.headersTimeout).toBe(createServer().headersTimeout);
|
||||||
});
|
});
|
||||||
|
|
||||||
|
|||||||
@ -1,8 +1,8 @@
|
|||||||
import type { AdapterContext, ArtifactByteRange, ArtifactStore } from "@rakazo/adapter-kit";
|
import type { AdapterContext, ArtifactByteRange, ArtifactStore } from "@rakazo/adapter-kit";
|
||||||
import { ArtifactTooLargeError } from "@rakazo/adapter-kit";
|
import { ArtifactTooLargeError } from "@rakazo/adapter-kit";
|
||||||
import { artifactMaxBytes, storeArtifactStream } from "@rakazo/adapters";
|
import { storeArtifactStream } from "@rakazo/adapters";
|
||||||
import type { Actor } from "@rakazo/contracts";
|
import type { Actor } from "@rakazo/contracts";
|
||||||
import { attachmentKindOf, isAllowedAttachmentMimeType } from "@rakazo/contracts";
|
import { artifactMaxBytes, attachmentKindOf, isAllowedAttachmentMimeType } from "@rakazo/contracts";
|
||||||
import type { PrismaClient } from "@rakazo/db";
|
import type { PrismaClient } from "@rakazo/db";
|
||||||
import { createGroupRepos, createRepos, IsolationError, requireMembership } from "@rakazo/db";
|
import { createGroupRepos, createRepos, IsolationError, requireMembership } from "@rakazo/db";
|
||||||
import type { Context, Hono } from "hono";
|
import type { Context, Hono } from "hono";
|
||||||
@ -11,11 +11,11 @@ import { cancelBody } from "./http-body.js";
|
|||||||
export const ARTIFACT_HTTP_PATH = "/api/artifacts";
|
export const ARTIFACT_HTTP_PATH = "/api/artifacts";
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Node's default 5-minute requestTimeout cuts off a 500 MiB video upload on any uplink slower than
|
* Node's default 5-minute requestTimeout cuts off a large upload on a slow uplink; 2 hours fits a
|
||||||
* about 14 Mbit/s; 30 minutes fits it at about 2.5 Mbit/s. headersTimeout keeps its default, and
|
* 5 GiB repository ZIP at about 6 Mbit/s. headersTimeout keeps its default, and the per-upload
|
||||||
* the per-upload stall timeout stops a slow-drip client from holding a connection that long.
|
* stall timeout stops a slow-drip client from holding a connection that long.
|
||||||
*/
|
*/
|
||||||
export const API_SERVER_OPTIONS = { requestTimeout: 30 * 60 * 1000 };
|
export const API_SERVER_OPTIONS = { requestTimeout: 2 * 60 * 60 * 1000 };
|
||||||
/** An upload that sends no bytes for this long is aborted and its partial file deleted. */
|
/** An upload that sends no bytes for this long is aborted and its partial file deleted. */
|
||||||
export const ARTIFACT_UPLOAD_STALL_TIMEOUT_MS = 60_000;
|
export const ARTIFACT_UPLOAD_STALL_TIMEOUT_MS = 60_000;
|
||||||
|
|
||||||
|
|||||||
@ -80,3 +80,32 @@ test("an image in the main conversation opens and downloads from the artifact ro
|
|||||||
await download.click();
|
await download.click();
|
||||||
expect((await saved).suggestedFilename()).toBe("pixel.png");
|
expect((await saved).suggestedFilename()).toBe("pixel.png");
|
||||||
});
|
});
|
||||||
|
|
||||||
|
test("a repository ZIP above the inline cap streams up as a file", async ({ page }, testInfo) => {
|
||||||
|
const stamp = Date.now();
|
||||||
|
await signup(page, `zip-upload-${stamp}@rakazo.test`, "password12", "Zip Upload");
|
||||||
|
await completeOnboarding(page);
|
||||||
|
|
||||||
|
const input = page.locator('input[type="file"]');
|
||||||
|
await expect(input).toHaveAttribute("accept", /application\/zip/);
|
||||||
|
const upload = page.waitForResponse(
|
||||||
|
(response) =>
|
||||||
|
response.request().method() === "POST" && /\/api\/artifacts\?/.test(response.url()),
|
||||||
|
);
|
||||||
|
// Windows reports ZIPs under the legacy type; 12 MiB is above the 10 MiB inline cap.
|
||||||
|
await input.setInputFiles({
|
||||||
|
name: "repo.zip",
|
||||||
|
mimeType: "application/x-zip-compressed",
|
||||||
|
buffer: Buffer.alloc(12 * 1024 * 1024, 1),
|
||||||
|
});
|
||||||
|
await expect(page.getByRole("button", { name: "Remove repo.zip" })).toBeVisible();
|
||||||
|
await page.getByRole("combobox", { name: /Message/ }).fill("here is the repo");
|
||||||
|
await page.keyboard.press("Enter");
|
||||||
|
const response = await upload;
|
||||||
|
expect(response.status()).toBe(201);
|
||||||
|
expect(response.request().headers()["content-type"]).toBe("application/zip");
|
||||||
|
await expect(page.getByTestId("transcript").getByText("repo.zip")).toBeVisible({
|
||||||
|
timeout: 30_000,
|
||||||
|
});
|
||||||
|
await captureScreenshot(page, testInfo, "zip-upload");
|
||||||
|
});
|
||||||
|
|||||||
@ -1,8 +1,8 @@
|
|||||||
import { t } from "@lingui/core/macro";
|
import { t } from "@lingui/core/macro";
|
||||||
import {
|
import {
|
||||||
ARTIFACT_MAX_BYTES_BY_KIND,
|
|
||||||
ArtifactSchema,
|
ArtifactSchema,
|
||||||
ATTACHMENT_MAX_BYTES,
|
ATTACHMENT_MAX_BYTES,
|
||||||
|
artifactMaxBytes,
|
||||||
attachmentKindOf,
|
attachmentKindOf,
|
||||||
} from "@rakazo/contracts";
|
} from "@rakazo/contracts";
|
||||||
import type { ArtifactTarget } from "./artifact-open";
|
import type { ArtifactTarget } from "./artifact-open";
|
||||||
@ -12,7 +12,7 @@ type UploadedArtifact = ReturnType<typeof ArtifactSchema.parse>;
|
|||||||
|
|
||||||
/** Largest file of this type the composer accepts. */
|
/** Largest file of this type the composer accepts. */
|
||||||
export function attachmentMaxBytes(mimeType: string): number {
|
export function attachmentMaxBytes(mimeType: string): number {
|
||||||
return ARTIFACT_MAX_BYTES_BY_KIND[attachmentKindOf(mimeType)];
|
return artifactMaxBytes(mimeType);
|
||||||
}
|
}
|
||||||
|
|
||||||
/** Media and anything above the inline RPC cap upload as a raw stream instead of base64. */
|
/** Media and anything above the inline RPC cap upload as a raw stream instead of base64. */
|
||||||
|
|||||||
@ -41,7 +41,8 @@ import { previewMessageText } from "./message-preview";
|
|||||||
import type { PendingAttachment } from "./shell-types";
|
import type { PendingAttachment } from "./shell-types";
|
||||||
import { FALLBACK_BOT_COLOR } from "./shell-types";
|
import { FALLBACK_BOT_COLOR } from "./shell-types";
|
||||||
|
|
||||||
const ATTACHMENT_ACCEPT = ATTACHMENT_ALLOWED_MIME_TYPES.join(",");
|
// ".zip" too: Windows reports ZIPs as application/x-zip-compressed.
|
||||||
|
const ATTACHMENT_ACCEPT = [...ATTACHMENT_ALLOWED_MIME_TYPES, ".zip"].join(",");
|
||||||
|
|
||||||
const addButtonClass =
|
const addButtonClass =
|
||||||
"tactile-press size-10 shrink-0 rounded-full border border-border bg-foreground/5 text-foreground/80 transition-all hover:bg-foreground/10 hover:text-foreground flex items-center justify-center cursor-pointer";
|
"tactile-press size-10 shrink-0 rounded-full border border-border bg-foreground/5 text-foreground/80 transition-all hover:bg-foreground/10 hover:text-foreground flex items-center justify-center cursor-pointer";
|
||||||
|
|||||||
48
infra/sandboxes/supervisor/src/file-stream.docker.test.ts
Normal file
48
infra/sandboxes/supervisor/src/file-stream.docker.test.ts
Normal file
@ -0,0 +1,48 @@
|
|||||||
|
import { execFileSync } from "node:child_process";
|
||||||
|
import { randomUUID } from "node:crypto";
|
||||||
|
import { Readable } from "node:stream";
|
||||||
|
import Docker from "dockerode";
|
||||||
|
import { expect, it } from "vitest";
|
||||||
|
import { writeContainerFile } from "./index.js";
|
||||||
|
|
||||||
|
const MiB = 1024 * 1024;
|
||||||
|
|
||||||
|
function chunks(count: number, failAfter?: number) {
|
||||||
|
return Readable.from(
|
||||||
|
(async function* () {
|
||||||
|
for (let index = 0; index < count; index += 1) {
|
||||||
|
if (index === failAfter) throw new Error("upload aborted");
|
||||||
|
yield Buffer.alloc(MiB, index % 256);
|
||||||
|
}
|
||||||
|
})(),
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
// Opt-in: needs Docker and a local Python image, but no network or credentials.
|
||||||
|
it.skipIf(process.env.VERIFY_DOCKER_FILE_STREAM !== "1")(
|
||||||
|
"streams large files into a container and leaves nothing behind when the stream fails",
|
||||||
|
async () => {
|
||||||
|
const image = process.env.VERIFY_DOCKER_PYTHON_IMAGE ?? "python:3.12-alpine";
|
||||||
|
const name = `rakazo-file-stream-test-${randomUUID()}`;
|
||||||
|
const docker = (...args: string[]) => execFileSync("docker", args, { encoding: "utf8" });
|
||||||
|
docker("run", "-d", "--name", name, image, "sh", "-c", "mkdir -p /home/rakazo && sleep 600");
|
||||||
|
try {
|
||||||
|
const container = new Docker().getContainer(name);
|
||||||
|
await writeContainerFile(container, "/home/rakazo/attachments/repo.zip", chunks(64));
|
||||||
|
expect(docker("exec", name, "stat", "-c", "%s %a", "/home/rakazo/attachments/repo.zip")).toBe(
|
||||||
|
`${64 * MiB} 600\n`,
|
||||||
|
);
|
||||||
|
|
||||||
|
await expect(
|
||||||
|
writeContainerFile(container, "/home/rakazo/attachments/broken.zip", chunks(64, 8)),
|
||||||
|
).rejects.toThrow("upload aborted");
|
||||||
|
expect(docker("exec", name, "ls", "/home/rakazo/attachments")).toBe("repo.zip\n");
|
||||||
|
|
||||||
|
await writeContainerFile(container, "/home/rakazo/run.sh", Buffer.from("echo hi\n"), true);
|
||||||
|
expect(docker("exec", name, "stat", "-c", "%s %a", "/home/rakazo/run.sh")).toBe("8 700\n");
|
||||||
|
} finally {
|
||||||
|
docker("rm", "-f", name);
|
||||||
|
}
|
||||||
|
},
|
||||||
|
120_000,
|
||||||
|
);
|
||||||
@ -201,6 +201,12 @@ describe("sandbox supervisor HTTP boundary", () => {
|
|||||||
expect(supervisorRequestBodyLimit("POST", "/computers/id/files/")).toBe(
|
expect(supervisorRequestBodyLimit("POST", "/computers/id/files/")).toBe(
|
||||||
MAX_SUPERVISOR_FILE_REQUEST_BYTES,
|
MAX_SUPERVISOR_FILE_REQUEST_BYTES,
|
||||||
);
|
);
|
||||||
|
expect(supervisorRequestBodyLimit("POST", "/computers/id/files/stream")).toBe(
|
||||||
|
Number.POSITIVE_INFINITY,
|
||||||
|
);
|
||||||
|
expect(supervisorRequestBodyLimit("PUT", "/computers/id/files/stream")).toBe(
|
||||||
|
MAX_SUPERVISOR_REQUEST_BYTES,
|
||||||
|
);
|
||||||
expect(supervisorRequestBodyLimit("PUT", "/computers/id/files")).toBe(
|
expect(supervisorRequestBodyLimit("PUT", "/computers/id/files")).toBe(
|
||||||
MAX_SUPERVISOR_REQUEST_BYTES,
|
MAX_SUPERVISOR_REQUEST_BYTES,
|
||||||
);
|
);
|
||||||
|
|||||||
@ -3,6 +3,9 @@ import { existsSync } from "node:fs";
|
|||||||
import { mkdir } from "node:fs/promises";
|
import { mkdir } from "node:fs/promises";
|
||||||
import http from "node:http";
|
import http from "node:http";
|
||||||
import path from "node:path";
|
import path from "node:path";
|
||||||
|
import { Readable } from "node:stream";
|
||||||
|
import { pipeline } from "node:stream/promises";
|
||||||
|
import type { ReadableStream as NodeReadableStream } from "node:stream/web";
|
||||||
import { fileURLToPath } from "node:url";
|
import { fileURLToPath } from "node:url";
|
||||||
import { serve } from "@hono/node-server";
|
import { serve } from "@hono/node-server";
|
||||||
import {
|
import {
|
||||||
@ -103,14 +106,22 @@ const computerScreens = new Map<string, Map<string, ScreenAssignment>>();
|
|||||||
export const MAX_SUPERVISOR_REQUEST_BYTES = 1024 * 1024;
|
export const MAX_SUPERVISOR_REQUEST_BYTES = 1024 * 1024;
|
||||||
export const MAX_SUPERVISOR_FILE_REQUEST_BYTES = 16 * 1024 * 1024 + 64 * 1024;
|
export const MAX_SUPERVISOR_FILE_REQUEST_BYTES = 16 * 1024 * 1024 + 64 * 1024;
|
||||||
|
|
||||||
/** Keep normal control requests small while allowing the existing 16 MiB file payload. */
|
/**
|
||||||
|
* Keep normal control requests small while allowing the existing 16 MiB file payload. Streamed
|
||||||
|
* file writes go straight to disk, so their size is bounded by the artifact that is copied.
|
||||||
|
*/
|
||||||
export function supervisorRequestBodyLimit(method: string, pathname: string): number {
|
export function supervisorRequestBodyLimit(method: string, pathname: string): number {
|
||||||
return method === "POST" && /^\/computers\/[^/]+\/files\/?$/.test(pathname)
|
if (method !== "POST") return MAX_SUPERVISOR_REQUEST_BYTES;
|
||||||
|
if (/^\/computers\/[^/]+\/files\/stream\/?$/.test(pathname)) return Number.POSITIVE_INFINITY;
|
||||||
|
return /^\/computers\/[^/]+\/files\/?$/.test(pathname)
|
||||||
? MAX_SUPERVISOR_FILE_REQUEST_BYTES
|
? MAX_SUPERVISOR_FILE_REQUEST_BYTES
|
||||||
: MAX_SUPERVISOR_REQUEST_BYTES;
|
: MAX_SUPERVISOR_REQUEST_BYTES;
|
||||||
}
|
}
|
||||||
|
|
||||||
const limitSupervisorRequestBody: MiddlewareHandler = async (c, next) => {
|
const limitSupervisorRequestBody: MiddlewareHandler = async (c, next) => {
|
||||||
|
if (supervisorRequestBodyLimit(c.req.method, c.req.path) === Number.POSITIVE_INFINITY) {
|
||||||
|
return next();
|
||||||
|
}
|
||||||
return bodyLimit({
|
return bodyLimit({
|
||||||
maxSize: supervisorRequestBodyLimit(c.req.method, c.req.path),
|
maxSize: supervisorRequestBodyLimit(c.req.method, c.req.path),
|
||||||
onError: (context) => context.json({ error: "Request body is too large." }, 413),
|
onError: (context) => context.json({ error: "Request body is too large." }, 413),
|
||||||
@ -594,6 +605,31 @@ app.post("/computers/:id/files", async (c) => {
|
|||||||
}
|
}
|
||||||
});
|
});
|
||||||
|
|
||||||
|
app.post("/computers/:id/files/stream", async (c) => {
|
||||||
|
const body = c.req.raw.body;
|
||||||
|
if (!body) return c.json({ error: "file is empty" }, 400);
|
||||||
|
try {
|
||||||
|
const { container } = await managedContainer(
|
||||||
|
c.req.param("id"),
|
||||||
|
c.req.header("x-rakazo-bot-id"),
|
||||||
|
c.req.header("x-rakazo-space-id"),
|
||||||
|
);
|
||||||
|
const relative = normalizeWorkspaceRelative(c.req.query("path") ?? "");
|
||||||
|
if (!relative) throw new Error("file path is required");
|
||||||
|
await writeContainerFile(
|
||||||
|
container,
|
||||||
|
workspaceTarget(relative),
|
||||||
|
Readable.fromWeb(body as NodeReadableStream<Uint8Array>),
|
||||||
|
c.req.query("executable") === "true",
|
||||||
|
);
|
||||||
|
return c.json({ ok: true });
|
||||||
|
} catch (error) {
|
||||||
|
await body.cancel().catch(() => undefined);
|
||||||
|
const message = error instanceof Error ? error.message : String(error);
|
||||||
|
return c.json({ error: message }, 400);
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
app.get("/computers/:id/screen", async (c) => {
|
app.get("/computers/:id/screen", async (c) => {
|
||||||
const id = c.req.param("id");
|
const id = c.req.param("id");
|
||||||
try {
|
try {
|
||||||
@ -803,7 +839,9 @@ function startSupervisor() {
|
|||||||
computerResourceLimits();
|
computerResourceLimits();
|
||||||
const port = Number(process.env.SUPERVISOR_PORT ?? 7091);
|
const port = Number(process.env.SUPERVISOR_PORT ?? 7091);
|
||||||
const hostname = process.env.SUPERVISOR_HOST ?? "127.0.0.1";
|
const hostname = process.env.SUPERVISOR_HOST ?? "127.0.0.1";
|
||||||
const server = serve({ fetch: app.fetch, hostname, port }, () => {
|
// Streamed file writes copy whole repository archives; Node's 5-minute default would cut them.
|
||||||
|
const serverOptions = { requestTimeout: 2 * 60 * 60 * 1000 };
|
||||||
|
const server = serve({ fetch: app.fetch, hostname, port, serverOptions }, () => {
|
||||||
logger.info("supervisor listening", { "http.host": hostname, "http.port": port });
|
logger.info("supervisor listening", { "http.host": hostname, "http.port": port });
|
||||||
});
|
});
|
||||||
let stopping = false;
|
let stopping = false;
|
||||||
@ -1402,21 +1440,57 @@ async function observeContainer(container: Docker.Container, display = ":1") {
|
|||||||
return parseObservation(result.stdout);
|
return parseObservation(result.stdout);
|
||||||
}
|
}
|
||||||
|
|
||||||
async function writeContainerFile(
|
/**
|
||||||
|
* Writes beside the target and renames it into place, so a failed write never leaves a truncated
|
||||||
|
* file. A buffer is complete before the write starts and commits in the same exec; a stream only
|
||||||
|
* commits once it ended cleanly. Python copies stdin in chunks, so large streams stay off-heap.
|
||||||
|
*/
|
||||||
|
export async function writeContainerFile(
|
||||||
container: Docker.Container,
|
container: Docker.Container,
|
||||||
target: string,
|
target: string,
|
||||||
content: Buffer,
|
content: Buffer | Readable,
|
||||||
executable = false,
|
executable = false,
|
||||||
) {
|
) {
|
||||||
const script = [
|
const partial = `${target}.part-${randomUUID()}`;
|
||||||
"import os, sys",
|
const commit = [
|
||||||
"target = sys.argv[1]",
|
`os.chmod(sys.argv[1], ${executable ? "0o700" : "0o600"})`,
|
||||||
"os.makedirs(os.path.dirname(target), exist_ok=True)",
|
"os.replace(sys.argv[1], sys.argv[2])",
|
||||||
"with open(target, 'wb') as f: f.write(sys.stdin.buffer.read())",
|
];
|
||||||
`os.chmod(target, ${executable ? "0o700" : "0o600"})`,
|
const write = [
|
||||||
].join("\n");
|
"import os, shutil, sys",
|
||||||
|
"os.makedirs(os.path.dirname(sys.argv[1]), exist_ok=True)",
|
||||||
|
"with open(sys.argv[1], 'wb') as f: shutil.copyfileobj(sys.stdin.buffer, f, 1024 * 1024)",
|
||||||
|
];
|
||||||
|
if (Buffer.isBuffer(content)) {
|
||||||
|
await runContainerPython(
|
||||||
|
container,
|
||||||
|
[...write, ...commit].join("\n"),
|
||||||
|
[partial, target],
|
||||||
|
content,
|
||||||
|
);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
try {
|
||||||
|
await runContainerPython(container, write.join("\n"), [partial], content);
|
||||||
|
} catch (error) {
|
||||||
|
await runContainerPython(
|
||||||
|
container,
|
||||||
|
"import os, sys\ntry: os.remove(sys.argv[1])\nexcept FileNotFoundError: pass",
|
||||||
|
[partial],
|
||||||
|
).catch(() => undefined);
|
||||||
|
throw error;
|
||||||
|
}
|
||||||
|
await runContainerPython(container, ["import os, sys", ...commit].join("\n"), [partial, target]);
|
||||||
|
}
|
||||||
|
|
||||||
|
async function runContainerPython(
|
||||||
|
container: Docker.Container,
|
||||||
|
script: string,
|
||||||
|
args: string[],
|
||||||
|
stdin: Buffer | Readable = Buffer.alloc(0),
|
||||||
|
) {
|
||||||
const exec = await container.exec({
|
const exec = await container.exec({
|
||||||
Cmd: ["python3", "-c", script, target],
|
Cmd: ["python3", "-c", script, ...args],
|
||||||
AttachStdin: true,
|
AttachStdin: true,
|
||||||
AttachStdout: true,
|
AttachStdout: true,
|
||||||
AttachStderr: true,
|
AttachStderr: true,
|
||||||
@ -1426,14 +1500,37 @@ async function writeContainerFile(
|
|||||||
const stream = await exec.start({ hijack: true, stdin: true });
|
const stream = await exec.start({ hijack: true, stdin: true });
|
||||||
const chunks: Buffer[] = [];
|
const chunks: Buffer[] = [];
|
||||||
stream.on("data", (data: Buffer) => chunks.push(data));
|
stream.on("data", (data: Buffer) => chunks.push(data));
|
||||||
stream.end(content);
|
const ended = new Promise<void>((resolve, reject) => {
|
||||||
await new Promise<void>((resolve, reject) => {
|
|
||||||
stream.on("end", resolve);
|
stream.on("end", resolve);
|
||||||
|
stream.on("close", resolve);
|
||||||
stream.on("error", reject);
|
stream.on("error", reject);
|
||||||
});
|
});
|
||||||
const inspect = await exec.inspect();
|
if (Buffer.isBuffer(stdin)) {
|
||||||
|
stream.end(stdin);
|
||||||
|
} else {
|
||||||
|
try {
|
||||||
|
await pipeline(stdin, stream);
|
||||||
|
} catch (error) {
|
||||||
|
// pipeline already destroyed the exec stream with this error; `ended` must not reject unseen.
|
||||||
|
ended.catch(() => undefined);
|
||||||
|
stream.destroy();
|
||||||
|
await waitForExecExit(exec);
|
||||||
|
throw error;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
await ended;
|
||||||
|
const inspect = await waitForExecExit(exec);
|
||||||
if ((inspect.ExitCode ?? 0) !== 0) {
|
if ((inspect.ExitCode ?? 0) !== 0) {
|
||||||
const output = demuxDockerStream(Buffer.concat(chunks));
|
const output = demuxDockerStream(Buffer.concat(chunks));
|
||||||
throw new Error(output.stderr || output.stdout || "file write failed");
|
throw new Error(output.stderr || output.stdout || "file write failed");
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/** Docker reports the exit code a moment after the attached stream closes. */
|
||||||
|
async function waitForExecExit(exec: Docker.Exec) {
|
||||||
|
for (let attempt = 0; ; attempt += 1) {
|
||||||
|
const inspect = await exec.inspect();
|
||||||
|
if (!inspect.Running || attempt >= 50) return inspect;
|
||||||
|
await new Promise((resolve) => setTimeout(resolve, 100));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@ -2,7 +2,7 @@ import { constants } from "node:fs";
|
|||||||
import { open } from "node:fs/promises";
|
import { open } from "node:fs/promises";
|
||||||
import type { AdapterContext, ArtifactStore, ArtifactStreamPut } from "@rakazo/adapter-kit";
|
import type { AdapterContext, ArtifactStore, ArtifactStreamPut } from "@rakazo/adapter-kit";
|
||||||
import { ArtifactTooLargeError } from "@rakazo/adapter-kit";
|
import { ArtifactTooLargeError } from "@rakazo/adapter-kit";
|
||||||
import { ARTIFACT_MAX_BYTES_BY_KIND, attachmentKindOf } from "@rakazo/contracts";
|
import { artifactMaxBytes } from "@rakazo/contracts";
|
||||||
import { validateAttachmentMimeType } from "@rakazo/core";
|
import { validateAttachmentMimeType } from "@rakazo/core";
|
||||||
import type { PrismaClient } from "@rakazo/db";
|
import type { PrismaClient } from "@rakazo/db";
|
||||||
|
|
||||||
@ -30,11 +30,6 @@ type StoreArtifactDeps = {
|
|||||||
artifacts: ArtifactStore;
|
artifacts: ArtifactStore;
|
||||||
};
|
};
|
||||||
|
|
||||||
/** Largest artifact of this MIME type that may be streamed or generated. */
|
|
||||||
export function artifactMaxBytes(mimeType: string): number {
|
|
||||||
return ARTIFACT_MAX_BYTES_BY_KIND[attachmentKindOf(mimeType)];
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Streams bytes into the artifact store and records the artifact row; the stored blob is removed
|
* Streams bytes into the artifact store and records the artifact row; the stored blob is removed
|
||||||
* again when the row cannot be written. Rejects disallowed MIME types and anything above the
|
* again when the row cannot be written. Rejects disallowed MIME types and anything above the
|
||||||
|
|||||||
@ -37,6 +37,13 @@ export function computerFileTransferMaxBytes(computer: Pick<ComputerRef, "kind">
|
|||||||
: REMOTE_COMPUTER_FILE_TRANSFER_MAX_BYTES;
|
: REMOTE_COMPUTER_FILE_TRANSFER_MAX_BYTES;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/** Docker computers also take streamed writes, so files copied onto them have no transfer cap. */
|
||||||
|
export function computerFileWriteMaxBytes(computer: Pick<ComputerRef, "kind">): number {
|
||||||
|
return computer.kind === "docker"
|
||||||
|
? Number.POSITIVE_INFINITY
|
||||||
|
: computerFileTransferMaxBytes(computer);
|
||||||
|
}
|
||||||
|
|
||||||
/** Streams a file into the workspace; providers without streaming writes receive its bytes. */
|
/** Streams a file into the workspace; providers without streaming writes receive its bytes. */
|
||||||
export async function writeComputerFileStream(
|
export async function writeComputerFileStream(
|
||||||
sandbox: SandboxProvider,
|
sandbox: SandboxProvider,
|
||||||
|
|||||||
@ -1,3 +1,5 @@
|
|||||||
|
import { createServer } from "node:http";
|
||||||
|
import type { AddressInfo } from "node:net";
|
||||||
import type { ProcessEvent } from "@rakazo/adapter-kit";
|
import type { ProcessEvent } from "@rakazo/adapter-kit";
|
||||||
import { afterEach, describe, expect, it, vi } from "vitest";
|
import { afterEach, describe, expect, it, vi } from "vitest";
|
||||||
import {
|
import {
|
||||||
@ -348,4 +350,42 @@ describe("Docker page browser", () => {
|
|||||||
}),
|
}),
|
||||||
);
|
);
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it("streams file writes to the supervisor without buffering them", async () => {
|
||||||
|
const received: { url?: string; bytes: number; type?: string } = { bytes: 0 };
|
||||||
|
const server = createServer((request, response) => {
|
||||||
|
received.url = request.url;
|
||||||
|
received.type = request.headers["content-type"];
|
||||||
|
request.on("data", (chunk: Buffer) => {
|
||||||
|
received.bytes += chunk.byteLength;
|
||||||
|
});
|
||||||
|
request.on("end", () => {
|
||||||
|
response.setHeader("content-type", "application/json");
|
||||||
|
response.end('{"ok":true}');
|
||||||
|
});
|
||||||
|
});
|
||||||
|
await new Promise<void>((resolve) => server.listen(0, "127.0.0.1", resolve));
|
||||||
|
try {
|
||||||
|
const { port } = server.address() as AddressInfo;
|
||||||
|
const provider = new DockerSandboxProvider(`http://127.0.0.1:${port}`, "test-token");
|
||||||
|
const chunk = new Uint8Array(1024 * 1024);
|
||||||
|
await provider.writeFileStream(
|
||||||
|
{ id: "computer", botId: "bot", kind: "docker", providerRef: "computer" },
|
||||||
|
{
|
||||||
|
path: "attachments/repo.zip",
|
||||||
|
stream: (async function* () {
|
||||||
|
for (let index = 0; index < 8; index += 1) yield chunk;
|
||||||
|
})(),
|
||||||
|
},
|
||||||
|
context,
|
||||||
|
);
|
||||||
|
expect(received).toEqual({
|
||||||
|
url: "/computers/computer/files/stream?path=attachments%2Frepo.zip",
|
||||||
|
bytes: 8 * 1024 * 1024,
|
||||||
|
type: "application/octet-stream",
|
||||||
|
});
|
||||||
|
} finally {
|
||||||
|
await new Promise((resolve) => server.close(resolve));
|
||||||
|
}
|
||||||
|
});
|
||||||
});
|
});
|
||||||
|
|||||||
@ -1,4 +1,5 @@
|
|||||||
import path from "node:path";
|
import path from "node:path";
|
||||||
|
import { Readable } from "node:stream";
|
||||||
import type {
|
import type {
|
||||||
AdapterContext,
|
AdapterContext,
|
||||||
CommandRequest,
|
CommandRequest,
|
||||||
@ -11,6 +12,7 @@ import type {
|
|||||||
PageBrowserCommand,
|
PageBrowserCommand,
|
||||||
PageBrowserResult,
|
PageBrowserResult,
|
||||||
PortableFile,
|
PortableFile,
|
||||||
|
PortableFileStream,
|
||||||
ProcessEvent,
|
ProcessEvent,
|
||||||
SandboxProvider,
|
SandboxProvider,
|
||||||
ScreenRequest,
|
ScreenRequest,
|
||||||
@ -18,6 +20,7 @@ import type {
|
|||||||
} from "@rakazo/adapter-kit";
|
} from "@rakazo/adapter-kit";
|
||||||
import { boundedSandboxCommandTimeoutMs, resolveSupervisorToken } from "@rakazo/core";
|
import { boundedSandboxCommandTimeoutMs, resolveSupervisorToken } from "@rakazo/core";
|
||||||
import { outgoingCorrelationHeaders } from "@rakazo/logging";
|
import { outgoingCorrelationHeaders } from "@rakazo/logging";
|
||||||
|
import { byteChunks } from "./byte-streams.js";
|
||||||
import {
|
import {
|
||||||
boundedComputerActions,
|
boundedComputerActions,
|
||||||
clampRounded,
|
clampRounded,
|
||||||
@ -397,6 +400,26 @@ export class DockerSandboxProvider implements SandboxProvider {
|
|||||||
if (!res.ok) throw new Error(`sandbox file write failed: ${res.status}`);
|
if (!res.ok) throw new Error(`sandbox file write failed: ${res.status}`);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/** Streams the body to the supervisor, which writes it to disk without buffering it. */
|
||||||
|
async writeFileStream(computer: ComputerRef, file: PortableFileStream, context: AdapterContext) {
|
||||||
|
const query = new URLSearchParams({ path: normalizeWorkspacePath(file.path) });
|
||||||
|
if (file.executable) query.set("executable", "true");
|
||||||
|
const res = await fetch(this.url(`/computers/${computer.id}/files/stream?${query}`), {
|
||||||
|
method: "POST",
|
||||||
|
headers: {
|
||||||
|
...this.headers(context, computer.botId),
|
||||||
|
"content-type": "application/octet-stream",
|
||||||
|
},
|
||||||
|
body: Readable.toWeb(Readable.from(byteChunks(file.stream))) as unknown as BodyInit,
|
||||||
|
duplex: "half",
|
||||||
|
signal: context.signal,
|
||||||
|
} as RequestInit & { duplex: "half" });
|
||||||
|
if (!res.ok) {
|
||||||
|
const detail = await safeBody(res, context.signal);
|
||||||
|
throw new Error(`sandbox file write failed: ${res.status} ${detail}`.trim());
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
async *exportWorkspace(computer: ComputerRef, context: AdapterContext) {
|
async *exportWorkspace(computer: ComputerRef, context: AdapterContext) {
|
||||||
yield* this.walkWorkspace(computer, "", context);
|
yield* this.walkWorkspace(computer, "", context);
|
||||||
}
|
}
|
||||||
|
|||||||
@ -227,7 +227,7 @@ describe("current-turn thread files", () => {
|
|||||||
|
|
||||||
const remote = await materializeCurrentTurnFiles(deps, blocks, {
|
const remote = await materializeCurrentTurnFiles(deps, blocks, {
|
||||||
context,
|
context,
|
||||||
computer: { id: "computer-1", botId: "bot-1", kind: "docker", providerRef: "docker-1" },
|
computer: { id: "computer-1", botId: "bot-1", kind: "e2b", providerRef: "e2b-1" },
|
||||||
computerMode: "dedicated",
|
computerMode: "dedicated",
|
||||||
});
|
});
|
||||||
expect(remote.map((file) => [file.artifactId, file.path])).toEqual([
|
expect(remote.map((file) => [file.artifactId, file.path])).toEqual([
|
||||||
@ -254,6 +254,24 @@ describe("current-turn thread files", () => {
|
|||||||
]);
|
]);
|
||||||
expect(writeFile).toHaveBeenCalledTimes(3);
|
expect(writeFile).toHaveBeenCalledTimes(3);
|
||||||
expect(currentTurnFilesInstruction(local)).not.toContain("too large");
|
expect(currentTurnFilesInstruction(local)).not.toContain("too large");
|
||||||
|
|
||||||
|
// Docker computers take streamed writes, so large files reach them too.
|
||||||
|
const writeFileStream = vi.fn().mockResolvedValue(undefined);
|
||||||
|
const docker = await materializeCurrentTurnFiles(
|
||||||
|
{ ...deps, sandbox: { writeFile, writeFileStream } as unknown as SandboxProvider },
|
||||||
|
blocks,
|
||||||
|
{
|
||||||
|
context,
|
||||||
|
computer: { id: "computer-3", botId: "bot-1", kind: "docker", providerRef: "docker-1" },
|
||||||
|
computerMode: "dedicated",
|
||||||
|
},
|
||||||
|
);
|
||||||
|
expect(docker.map((file) => file.path)).toEqual([
|
||||||
|
"attachments/img-1.png",
|
||||||
|
"attachments/aud-1.mp3",
|
||||||
|
"attachments/vid-1.mp4",
|
||||||
|
]);
|
||||||
|
expect(writeFileStream).toHaveBeenCalledTimes(3);
|
||||||
});
|
});
|
||||||
|
|
||||||
it("skips loading when a turn has no attachments", async () => {
|
it("skips loading when a turn has no attachments", async () => {
|
||||||
|
|||||||
@ -7,18 +7,18 @@ import type {
|
|||||||
} from "@rakazo/adapter-kit";
|
} from "@rakazo/adapter-kit";
|
||||||
import { ArtifactTooLargeError } from "@rakazo/adapter-kit";
|
import { ArtifactTooLargeError } from "@rakazo/adapter-kit";
|
||||||
import type { ComputerMode, MessageBlock } from "@rakazo/contracts";
|
import type { ComputerMode, MessageBlock } from "@rakazo/contracts";
|
||||||
import { ATTACHMENT_MAX_BYTES } from "@rakazo/contracts";
|
import { ATTACHMENT_MAX_BYTES, artifactMaxBytes } from "@rakazo/contracts";
|
||||||
import {
|
import {
|
||||||
attachmentExtensionForMimeType,
|
attachmentExtensionForMimeType,
|
||||||
inferAttachmentMimeType,
|
inferAttachmentMimeType,
|
||||||
messageBlockForArtifact,
|
messageBlockForArtifact,
|
||||||
} from "@rakazo/core";
|
} from "@rakazo/core";
|
||||||
import type { PrismaClient } from "@rakazo/db";
|
import type { PrismaClient } from "@rakazo/db";
|
||||||
import { artifactMaxBytes, storeArtifactStream } from "./artifact-files.js";
|
import { storeArtifactStream } from "./artifact-files.js";
|
||||||
import { discardByteStream } from "./byte-streams.js";
|
import { discardByteStream } from "./byte-streams.js";
|
||||||
import {
|
import {
|
||||||
assertTeamWorkspacePathAllowed,
|
assertTeamWorkspacePathAllowed,
|
||||||
computerFileTransferMaxBytes,
|
computerFileWriteMaxBytes,
|
||||||
resolveBotWorkspacePath,
|
resolveBotWorkspacePath,
|
||||||
writeComputerFileStream,
|
writeComputerFileStream,
|
||||||
} from "./computer-support.js";
|
} from "./computer-support.js";
|
||||||
@ -141,7 +141,7 @@ export async function materializeCurrentTurnFiles(
|
|||||||
},
|
},
|
||||||
});
|
});
|
||||||
const byId = new Map(rows.map((row) => [row.id, row]));
|
const byId = new Map(rows.map((row) => [row.id, row]));
|
||||||
const transferMaxBytes = computerFileTransferMaxBytes(input.computer);
|
const transferMaxBytes = computerFileWriteMaxBytes(input.computer);
|
||||||
const materialized: MaterializedThreadFile[] = [];
|
const materialized: MaterializedThreadFile[] = [];
|
||||||
|
|
||||||
for (const block of attachmentBlocks) {
|
for (const block of attachmentBlocks) {
|
||||||
|
|||||||
@ -1,5 +1,12 @@
|
|||||||
import { describe, expect, it } from "vitest";
|
import { describe, expect, it } from "vitest";
|
||||||
import { attachmentKindOf, MessageBlock, validateThreadsSendInput } from "./index.js";
|
import {
|
||||||
|
ARCHIVE_MAX_BYTES,
|
||||||
|
ARTIFACT_MAX_BYTES_BY_KIND,
|
||||||
|
artifactMaxBytes,
|
||||||
|
attachmentKindOf,
|
||||||
|
MessageBlock,
|
||||||
|
validateThreadsSendInput,
|
||||||
|
} from "./index.js";
|
||||||
|
|
||||||
describe("attachment contracts", () => {
|
describe("attachment contracts", () => {
|
||||||
it("parses image and file message blocks", () => {
|
it("parses image and file message blocks", () => {
|
||||||
@ -22,6 +29,14 @@ describe("attachment contracts", () => {
|
|||||||
).toMatchObject({ kind: "file", size: 1234 });
|
).toMatchObject({ kind: "file", size: 1234 });
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it("gives ZIP archives a far larger cap than other files", () => {
|
||||||
|
expect(attachmentKindOf("application/zip")).toBe("file");
|
||||||
|
expect(artifactMaxBytes("application/zip")).toBe(ARCHIVE_MAX_BYTES);
|
||||||
|
expect(ARCHIVE_MAX_BYTES).toBe(5 * 1024 ** 3);
|
||||||
|
expect(artifactMaxBytes("application/pdf")).toBe(ARTIFACT_MAX_BYTES_BY_KIND.file);
|
||||||
|
expect(artifactMaxBytes("video/mp4")).toBe(ARTIFACT_MAX_BYTES_BY_KIND.video);
|
||||||
|
});
|
||||||
|
|
||||||
it("requires text or attachments for threads.send", () => {
|
it("requires text or attachments for threads.send", () => {
|
||||||
expect(validateThreadsSendInput({ text: "hello" })).toBe(true);
|
expect(validateThreadsSendInput({ text: "hello" })).toBe(true);
|
||||||
expect(validateThreadsSendInput({ artifactIds: ["art_1"] })).toBe(true);
|
expect(validateThreadsSendInput({ artifactIds: ["art_1"] })).toBe(true);
|
||||||
|
|||||||
@ -28,6 +28,7 @@ export const ATTACHMENT_FILE_MIME_TYPES = [
|
|||||||
"application/json",
|
"application/json",
|
||||||
"text/vtt",
|
"text/vtt",
|
||||||
"application/x-subrip",
|
"application/x-subrip",
|
||||||
|
"application/zip",
|
||||||
] as const;
|
] as const;
|
||||||
|
|
||||||
export const ATTACHMENT_ALLOWED_MIME_TYPES = [
|
export const ATTACHMENT_ALLOWED_MIME_TYPES = [
|
||||||
@ -50,6 +51,16 @@ export const ARTIFACT_MAX_BYTES_BY_KIND: Record<AttachmentKind, number> = {
|
|||||||
file: 25 * 1024 * 1024,
|
file: 25 * 1024 * 1024,
|
||||||
};
|
};
|
||||||
|
|
||||||
|
/** A ZIP carries a whole project or repository, so it gets a far larger cap than other files. */
|
||||||
|
export const ARCHIVE_MAX_BYTES = 5 * 1024 * 1024 * 1024;
|
||||||
|
|
||||||
|
/** Largest artifact of this MIME type that may be streamed or generated. */
|
||||||
|
export function artifactMaxBytes(mimeType: string): number {
|
||||||
|
return mimeType === "application/zip"
|
||||||
|
? ARCHIVE_MAX_BYTES
|
||||||
|
: ARTIFACT_MAX_BYTES_BY_KIND[attachmentKindOf(mimeType)];
|
||||||
|
}
|
||||||
|
|
||||||
export type AttachmentMimeType = (typeof ATTACHMENT_ALLOWED_MIME_TYPES)[number];
|
export type AttachmentMimeType = (typeof ATTACHMENT_ALLOWED_MIME_TYPES)[number];
|
||||||
|
|
||||||
export function isAttachmentImageMimeType(mimeType: string): boolean {
|
export function isAttachmentImageMimeType(mimeType: string): boolean {
|
||||||
|
|||||||
@ -13,7 +13,10 @@ import {
|
|||||||
|
|
||||||
describe("attachment helpers", () => {
|
describe("attachment helpers", () => {
|
||||||
it("rejects unsupported mime types and empty payloads", () => {
|
it("rejects unsupported mime types and empty payloads", () => {
|
||||||
expect(() => validateAttachmentMimeType("application/zip")).toThrow(AttachmentValidationError);
|
expect(() => validateAttachmentMimeType("application/x-msdownload")).toThrow(
|
||||||
|
AttachmentValidationError,
|
||||||
|
);
|
||||||
|
expect(() => validateAttachmentMimeType("application/zip")).not.toThrow();
|
||||||
expect(() => decodeAttachmentBase64("")).toThrow(AttachmentValidationError);
|
expect(() => decodeAttachmentBase64("")).toThrow(AttachmentValidationError);
|
||||||
expect(() => decodeAttachmentBase64("aGVsbG8=trailing-junk")).toThrow(
|
expect(() => decodeAttachmentBase64("aGVsbG8=trailing-junk")).toThrow(
|
||||||
AttachmentValidationError,
|
AttachmentValidationError,
|
||||||
@ -53,7 +56,12 @@ describe("attachment helpers", () => {
|
|||||||
expect(inferAttachmentMimeType("notes.md", "")).toBe("text/markdown");
|
expect(inferAttachmentMimeType("notes.md", "")).toBe("text/markdown");
|
||||||
expect(inferAttachmentMimeType("notes.markdown", "text/plain")).toBe("text/markdown");
|
expect(inferAttachmentMimeType("notes.markdown", "text/plain")).toBe("text/markdown");
|
||||||
expect(inferAttachmentMimeType("notes.md", "application/pdf")).toBe("application/pdf");
|
expect(inferAttachmentMimeType("notes.md", "application/pdf")).toBe("application/pdf");
|
||||||
expect(inferAttachmentMimeType("archive.zip", "")).toBeNull();
|
expect(inferAttachmentMimeType("repo.zip", "")).toBe("application/zip");
|
||||||
|
// Windows reports ZIPs under a legacy type.
|
||||||
|
expect(inferAttachmentMimeType("repo.ZIP", "application/x-zip-compressed")).toBe(
|
||||||
|
"application/zip",
|
||||||
|
);
|
||||||
|
expect(inferAttachmentMimeType("archive.tar", "")).toBeNull();
|
||||||
});
|
});
|
||||||
|
|
||||||
it("scopes current-turn images to user-triggered runs", () => {
|
it("scopes current-turn images to user-triggered runs", () => {
|
||||||
|
|||||||
@ -182,6 +182,7 @@ const EXTENSION_MIME_TYPES: Record<string, AttachmentMimeType> = {
|
|||||||
".oga": "audio/ogg",
|
".oga": "audio/ogg",
|
||||||
".vtt": "text/vtt",
|
".vtt": "text/vtt",
|
||||||
".srt": "application/x-subrip",
|
".srt": "application/x-subrip",
|
||||||
|
".zip": "application/zip",
|
||||||
};
|
};
|
||||||
|
|
||||||
const MIME_TYPE_EXTENSIONS: Record<AttachmentMimeType, string> = {
|
const MIME_TYPE_EXTENSIONS: Record<AttachmentMimeType, string> = {
|
||||||
@ -204,6 +205,7 @@ const MIME_TYPE_EXTENSIONS: Record<AttachmentMimeType, string> = {
|
|||||||
"audio/ogg": ".ogg",
|
"audio/ogg": ".ogg",
|
||||||
"text/vtt": ".vtt",
|
"text/vtt": ".vtt",
|
||||||
"application/x-subrip": ".srt",
|
"application/x-subrip": ".srt",
|
||||||
|
"application/zip": ".zip",
|
||||||
};
|
};
|
||||||
|
|
||||||
export function inferAttachmentMimeType(
|
export function inferAttachmentMimeType(
|
||||||
|
|||||||
@ -94,8 +94,8 @@ describeAttachments("chat attachments", () => {
|
|||||||
|
|
||||||
const badMime = await raw(app, cookie, "artifacts/create", {
|
const badMime = await raw(app, cookie, "artifacts/create", {
|
||||||
botId: bot.id,
|
botId: bot.id,
|
||||||
name: "evil.zip",
|
name: "evil.exe",
|
||||||
mimeType: "application/zip",
|
mimeType: "application/x-msdownload",
|
||||||
contentBase64: Buffer.from("zip").toString("base64"),
|
contentBase64: Buffer.from("zip").toString("base64"),
|
||||||
});
|
});
|
||||||
expect(badMime.status).toBeGreaterThanOrEqual(400);
|
expect(badMime.status).toBeGreaterThanOrEqual(400);
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user