Spaces:

Mike0021
/

chat-ui-agent-demo

Paused

chat-ui-agent-demo / src /lib /server /agent /SandboxClient.ts
deploy: production agent journey fixes (part 2)
77374bd verified 2 months ago
31.1 kB
import { createHmac, randomBytes } from "node:crypto";

const SANDBOX_SERVER_PORT = 49_983;
const SANDBOX_MAX_LIFETIME_SECONDS = 24 * 60 * 60;
const SANDBOX_SERVER_BUCKET = "huggingface/sbx-server";
const SANDBOX_SERVER_MOUNT = "/.hf-sbx-server";
export const SANDBOX_EXEC_LIMITS = Object.freeze({
	/** Supports the 10 MiB HTML export after base64 expansion, with headroom. */
	stdoutBytes: 16 * 1024 * 1024,
	stderrBytes: 8 * 1024 * 1024,
	combinedOutputBytes: 20 * 1024 * 1024,
	/** sbx-server emits output incrementally; one JSON event must stay independently bounded. */
	ndjsonLineBytes: 1 * 1024 * 1024,
	/** Includes JSON framing and escaping, which can exceed decoded command output. */
	ndjsonStreamBytes: 32 * 1024 * 1024,
});
// A newly exposed hf.jobs route can briefly answer 404 from one edge even while
// the Job and sbx-server are healthy on another. A bounded retry prevents that
// routing miss from being mistaken for an application-level missing endpoint.
const SANDBOX_ROUTE_RETRY_DELAYS_MS = [100, 250, 500, 1_000, 2_000, 4_000] as const;

const SANDBOX_BOOTSTRAP = `set -e
d=/tmp/.sbx-server
if command -v wget >/dev/null 2>&1; then wget -q -O "$d" "$SBX_SERVER_URL"
elif command -v curl >/dev/null 2>&1; then curl -fsSL -o "$d" "$SBX_SERVER_URL"
else cp "$SBX_SERVER_MOUNT/sbx-server" "$d"; fi
chmod +x "$d"
unset SBX_SERVER_URL SBX_SERVER_MOUNT
exec "$d"`;

const TERMINAL_JOB_STAGES = new Set(["COMPLETED", "ERROR", "DELETED", "CANCELED"]);
const MAX_MATCHING_SANDBOX_JOBS = 16;

export type SandboxFlavor =
	| "cpu-basic"
	| "cpu-upgrade"
	| "cpu-performance"
	| "cpu-xl"
	| "t4-small"
	| "t4-medium"
	| "l4x1"
	| "l4x4"
	| "l40sx1"
	| "l40sx4"
	| "l40sx8"
	| "a10g-small"
	| "a10g-large"
	| "a10g-largex2"
	| "a10g-largex4"
	| "a100-large"
	| "a100x4"
	| "a100x8"
	| "h200"
	| "h200x2"
	| "h200x4"
	| "h200x8"
	| "rtx-pro-6000"
	| "rtx-pro-6000x2"
	| "rtx-pro-6000x4"
	| "rtx-pro-6000x8";

export interface SandboxVolume {
	type: "bucket" | "model" | "dataset" | "space";
	source: string;
	mountPath: string;
	readOnly?: boolean;
	revision?: string;
	path?: string;
}

export interface SandboxConnection {
	id: string;
	namespace: string;
	baseUrl: string;
	nonce: string;
	image: string;
	flavor: SandboxFlavor;
	createdAt: Date;
}

interface JobResponse {
	id: string;
	createdAt?: string;
	startedAt?: string;
	finishedAt?: string;
	flavor?: string;
	status?: {
		stage?: string;
		message?: string | null;
		exposeUrls?: string[] | null;
	};
	durations?: {
		schedulingSecs?: number | null;
		runningSecs?: number | null;
		totalSecs?: number | null;
	};
	owner?: { name?: string };
}

export interface SandboxJobInfo {
	id: string;
	stage: string;
	message?: string;
	flavor: string;
	createdAt?: Date;
	startedAt?: Date;
	finishedAt?: Date;
	schedulingSeconds?: number;
	runningSeconds?: number;
	totalSeconds?: number;
}

export interface CreateSandboxOptions {
	hfToken: string;
	namespace: string;
	image?: string;
	flavor?: SandboxFlavor;
	idleTimeoutSeconds?: number | null;
	environment?: Record<string, string>;
	secrets?: Record<string, string>;
	volumes?: SandboxVolume[];
	/** Additional ownership labels. Reserved sandbox labels always win. */
	labels?: Record<string, string>;
	hubUrl?: string;
	startTimeoutMs?: number;
	pollIntervalMs?: number;
	fetch?: typeof fetch;
	onStatus?: (status: string) => void | Promise<void>;
	/** Runs as soon as Jobs returns an id, before readiness polling starts. */
	onCreated?: (connection: SandboxConnection) => void | Promise<void>;
}

export interface CancelMatchingSandboxJobsOptions {
	hfToken: string;
	namespace: string;
	labels: Record<string, string>;
	hubUrl?: string;
	fetch?: typeof fetch;
}

export interface SandboxExecResult {
	exitCode: number | null;
	stdout: string;
	stderr: string;
	signal?: number | null;
	timedOut: boolean;
	durationMs: number;
}

export interface SandboxProcess {
	/** Stable sbx-server registry handle (for example `p-7`). */
	id: string;
	/** Ephemeral operating-system PID, used only for process inspection. */
	pid: number;
	cmd: string | string[];
	tag?: string;
	startedAtMs?: number;
	running?: boolean;
	exitCode?: number | null;
}

export class SandboxApiError extends Error {
	constructor(
		message: string,
		readonly status?: number
	) {
		super(message);
		this.name = "SandboxApiError";
	}
}

/**
 * The HF token stays in this server-side object. Only the HMAC-derived sandbox
 * token is sent to sbx-server or placed in the Job's encrypted secrets.
 */
export class SandboxClient {
	readonly connection: SandboxConnection;

	private readonly hfToken: string;
	private readonly sandboxToken: string;
	private readonly hubUrl: string;
	private readonly requestFetch: typeof fetch;

	private constructor(
		connection: SandboxConnection,
		hfToken: string,
		options: { hubUrl?: string; fetch?: typeof fetch } = {}
	) {
		this.connection = {
			...connection,
			baseUrl: requireSandboxOrigin(connection.id, connection.baseUrl),
		};
		this.hfToken = hfToken;
		this.sandboxToken = deriveSandboxToken(hfToken, connection.nonce);
		this.hubUrl = trimTrailingSlash(options.hubUrl ?? "https://huggingface.co");
		this.requestFetch = options.fetch ?? fetch;
	}

	static connect(
		connection: SandboxConnection,
		hfToken: string,
		options: { hubUrl?: string; fetch?: typeof fetch } = {}
	): SandboxClient {
		return new SandboxClient(connection, hfToken, options);
	}

	static async create(options: CreateSandboxOptions): Promise<SandboxClient> {
		const requestFetch = options.fetch ?? fetch;
		const hubUrl = trimTrailingSlash(options.hubUrl ?? "https://huggingface.co");
		const image = options.image ?? "node:22-bookworm-slim";
		const flavor = options.flavor ?? "cpu-basic";
		const nonce = randomBytes(16).toString("hex");
		const sandboxToken = deriveSandboxToken(options.hfToken, nonce);

		const environment: Record<string, string> = {
			...(options.environment ?? {}),
			SBX_PORT: String(SANDBOX_SERVER_PORT),
			SBX_SERVER_URL: `${hubUrl}/buckets/${SANDBOX_SERVER_BUCKET}/resolve/sbx-server`,
			SBX_SERVER_MOUNT: SANDBOX_SERVER_MOUNT,
		};
		if (options.idleTimeoutSeconds !== null) {
			environment.SBX_IDLE_TIMEOUT = String(options.idleTimeoutSeconds ?? 10 * 60);
		}

		const jobSpec: Record<string, unknown> = {
			command: ["/bin/sh", "-c", SANDBOX_BOOTSTRAP],
			arguments: [],
			environment,
			secrets: { ...(options.secrets ?? {}), SBX_TOKEN: sandboxToken },
			flavor,
			timeoutSeconds: SANDBOX_MAX_LIFETIME_SECONDS,
			labels: {
				...(options.labels ?? {}),
				"hf-sandbox": "1",
				"hf-sandbox-mode": "dedicated",
				"hf-sandbox-nonce": nonce,
				"chat-ui-agent": "1",
			},
			volumes: [
				...(options.volumes ?? []),
				{
					type: "bucket",
					source: SANDBOX_SERVER_BUCKET,
					mountPath: SANDBOX_SERVER_MOUNT,
					readOnly: true,
				},
			],
			expose: { ports: [SANDBOX_SERVER_PORT] },
		};

		if (image.startsWith("hf.co/spaces/")) {
			jobSpec.spaceId = image.slice("hf.co/spaces/".length);
		} else if (image.startsWith("spaces/")) {
			jobSpec.spaceId = image.slice("spaces/".length);
		} else {
			jobSpec.dockerImage = image;
		}

		await options.onStatus?.("Creating sandbox");
		const response = await requestFetch(
			`${hubUrl}/api/jobs/${encodeURIComponent(options.namespace)}`,
			{
				method: "POST",
				headers: {
					Authorization: `Bearer ${options.hfToken}`,
					"Content-Type": "application/json",
				},
				body: JSON.stringify(jobSpec),
				redirect: "manual",
			}
		);
		const job = await readJsonResponse<JobResponse>(response, "create sandbox Job");
		if (!job.id) throw new SandboxApiError("Hugging Face Jobs returned no sandbox id");

		const exposedUrl = job.status?.exposeUrls
			?.map((url) => parseSandboxOrigin(job.id, url))
			.find((url): url is string => url !== undefined);
		const baseUrl =
			exposedUrl ??
			requireSandboxOrigin(job.id, `https://${job.id}--${SANDBOX_SERVER_PORT}.hf.jobs`);
		const client = new SandboxClient(
			{
				id: job.id,
				namespace: job.owner?.name ?? options.namespace,
				baseUrl,
				nonce,
				image,
				flavor,
				createdAt: new Date(),
			},
			options.hfToken,
			{ hubUrl, fetch: requestFetch }
		);

		try {
			await options.onCreated?.(client.connection);
			await client.waitUntilReady({
				timeoutMs: options.startTimeoutMs,
				pollIntervalMs: options.pollIntervalMs,
				onStatus: options.onStatus,
			});
			return client;
		} catch (err) {
			await client.kill().catch(() => undefined);
			throw err;
		}
	}

	/**
	 * Cancels the bounded set of live Jobs selected by exact ownership labels.
	 * The required agent label prevents callers from accidentally sweeping an
	 * account's unrelated Jobs.
	 */
	static async cancelMatchingJobs(options: CancelMatchingSandboxJobsOptions): Promise<string[]> {
		if (options.labels["chat-ui-agent"] !== "1") {
			throw new SandboxApiError("Refusing to cancel Jobs without the chat-ui agent label");
		}
		const labels = Object.entries(options.labels);
		if (
			labels.length < 2 ||
			labels.some(
				([key, value]) =>
					!key ||
					!value ||
					key.length > 128 ||
					value.length > 256 ||
					/[\r\n]/.test(`${key}${value}`)
			)
		) {
			throw new SandboxApiError("Invalid sandbox ownership labels");
		}

		const requestFetch = options.fetch ?? fetch;
		const hubUrl = trimTrailingSlash(options.hubUrl ?? "https://huggingface.co");
		const listUrl = new URL(`${hubUrl}/api/jobs/${encodeURIComponent(options.namespace)}`);
		listUrl.searchParams.append("stage", "RUNNING");
		listUrl.searchParams.append("stage", "SCHEDULING");
		for (const [key, value] of labels.sort(([left], [right]) => left.localeCompare(right))) {
			listUrl.searchParams.append("label", `${key}=${value}`);
		}
		const response = await requestFetch(listUrl, {
			headers: { Authorization: `Bearer ${options.hfToken}` },
			redirect: "manual",
		});
		const body = await readJsonResponse<unknown>(response, "list matching sandbox Jobs");
		if (!Array.isArray(body)) {
			throw new SandboxApiError("Hugging Face Jobs returned an invalid matching Job list");
		}
		const ids = body.map((job) => {
			const id =
				typeof job === "object" && job !== null && !Array.isArray(job)
					? (job as { id?: unknown }).id
					: undefined;
			if (typeof id !== "string" || !id || id.length > 256) {
				throw new SandboxApiError("Hugging Face Jobs returned an invalid matching Job id");
			}
			return id;
		});
		const uniqueIds = [...new Set(ids)];
		if (uniqueIds.length > MAX_MATCHING_SANDBOX_JOBS) {
			throw new SandboxApiError(
				`Refusing to cancel more than ${MAX_MATCHING_SANDBOX_JOBS} matching sandbox Jobs`
			);
		}

		for (const id of uniqueIds) {
			const cancelResponse = await requestFetch(
				`${hubUrl}/api/jobs/${encodeURIComponent(options.namespace)}/${encodeURIComponent(id)}/cancel`,
				{
					method: "POST",
					headers: { Authorization: `Bearer ${options.hfToken}` },
					redirect: "manual",
				}
			);
			if (!cancelResponse.ok && cancelResponse.status !== 404 && cancelResponse.status !== 409) {
				await throwResponseError(cancelResponse, "cancel matching sandbox Job");
			}
		}
		return uniqueIds;
	}

	get id(): string {
		return this.connection.id;
	}

	get proxyHeaders(): Record<string, string> {
		return {
			Authorization: `Bearer ${this.hfToken}`,
			"X-Sandbox-Token": this.sandboxToken,
		};
	}

	async health(): Promise<boolean> {
		try {
			const response = await this.request("/health", {
				signal: AbortSignal.timeout(5_000),
			});
			const healthy = response.ok;
			await response.body?.cancel().catch(() => undefined);
			return healthy;
		} catch {
			return false;
		}
	}

	async waitUntilReady(
		options: {
			timeoutMs?: number;
			pollIntervalMs?: number;
			onStatus?: (status: string) => void | Promise<void>;
		} = {}
	): Promise<void> {
		const timeoutMs = options.timeoutMs ?? 120_000;
		const pollIntervalMs = options.pollIntervalMs ?? 250;
		const deadline = Date.now() + timeoutMs;
		let nextJobCheckAt = 0;

		await options.onStatus?.("Waiting for sandbox");
		while (Date.now() < deadline) {
			if (await this.health()) {
				await options.onStatus?.("Sandbox ready");
				return;
			}

			if (Date.now() >= nextJobCheckAt) {
				nextJobCheckAt = Date.now() + 2_000;
				const job = await this.getJob();
				const stage = job.status?.stage ?? "STARTING";
				await options.onStatus?.(`Sandbox ${stage.toLowerCase()}`);
				if (TERMINAL_JOB_STAGES.has(stage)) {
					throw new SandboxApiError(
						`Sandbox ${this.id} stopped during startup (${stage}${
							job.status?.message ? `: ${job.status.message}` : ""
						})`
					);
				}
			}
			await delay(pollIntervalMs);
		}
		throw new SandboxApiError(`Sandbox ${this.id} did not become ready within ${timeoutMs}ms`);
	}

	/** Authoritative server-side Jobs timing used by the UI cost meter. */
	async inspect(): Promise<SandboxJobInfo> {
		const job = await this.getJob();
		return {
			id: job.id,
			stage: job.status?.stage ?? "UNKNOWN",
			...(job.status?.message ? { message: job.status.message } : {}),
			flavor: job.flavor ?? this.connection.flavor,
			...(job.createdAt ? { createdAt: new Date(job.createdAt) } : {}),
			...(job.startedAt ? { startedAt: new Date(job.startedAt) } : {}),
			...(job.finishedAt ? { finishedAt: new Date(job.finishedAt) } : {}),
			...(typeof job.durations?.schedulingSecs === "number"
				? { schedulingSeconds: job.durations.schedulingSecs }
				: {}),
			...(typeof job.durations?.runningSecs === "number"
				? { runningSeconds: job.durations.runningSecs }
				: {}),
			...(typeof job.durations?.totalSecs === "number"
				? { totalSeconds: job.durations.totalSecs }
				: {}),
		};
	}

	async exec(
		cmd: string | string[],
		options: {
			cwd?: string;
			environment?: Record<string, string>;
			timeoutSeconds?: number;
			stdin?: string;
			onStdout?: (chunk: string) => void;
			onStderr?: (chunk: string) => void;
			signal?: AbortSignal;
		} = {}
	): Promise<SandboxExecResult> {
		const response = await this.request("/v1/exec", {
			method: "POST",
			headers: { "Content-Type": "application/json" },
			body: JSON.stringify({
				cmd,
				shell: typeof cmd === "string",
				...(options.cwd ? { cwd: options.cwd } : {}),
				...(options.environment ? { env: options.environment } : {}),
				...(options.timeoutSeconds ? { timeout: options.timeoutSeconds } : {}),
				...(options.stdin !== undefined ? { stdin: options.stdin } : {}),
			}),
			signal: options.signal,
		});

		const stdout: string[] = [];
		const stderr: string[] = [];
		let stdoutBytes = 0;
		let stderrBytes = 0;
		let sawEvent = false;
		let sawStart = false;
		let terminal:
			Pick<SandboxExecResult, "exitCode" | "signal" | "timedOut" | "durationMs"> | undefined;
		for await (const event of parseNdjson(response)) {
			if (event.event === "start") {
				if (sawStart) throw new SandboxApiError("Sandbox returned more than one start event");
				if (sawEvent) {
					throw new SandboxApiError("Sandbox returned a start event after another event");
				}
				if (
					!hasExactKeys(event, ["event", "pid"]) ||
					typeof event.pid !== "number" ||
					!Number.isSafeInteger(event.pid) ||
					event.pid <= 0
				) {
					throw new SandboxApiError("Sandbox returned an invalid start event");
				}
				sawStart = true;
				sawEvent = true;
			} else if (event.event === "ping") {
				if (terminal) throw new SandboxApiError("Sandbox returned ping after its exit event");
				if (!hasExactKeys(event, ["event"])) {
					throw new SandboxApiError("Sandbox returned an invalid ping event");
				}
				sawEvent = true;
			} else if (event.event === "stdout") {
				sawEvent = true;
				if (typeof event.data !== "string") {
					throw new SandboxApiError("Sandbox returned stdout without string data");
				}
				if (terminal) throw new SandboxApiError("Sandbox returned stdout after its exit event");
				const chunkBytes = Buffer.byteLength(event.data, "utf8");
				assertExecOutputLimit("stdout", stdoutBytes, chunkBytes, SANDBOX_EXEC_LIMITS.stdoutBytes);
				assertExecOutputLimit(
					"combined output",
					stdoutBytes + stderrBytes,
					chunkBytes,
					SANDBOX_EXEC_LIMITS.combinedOutputBytes
				);
				stdoutBytes += chunkBytes;
				stdout.push(event.data);
				options.onStdout?.(event.data);
			} else if (event.event === "stderr") {
				sawEvent = true;
				if (typeof event.data !== "string") {
					throw new SandboxApiError("Sandbox returned stderr without string data");
				}
				if (terminal) throw new SandboxApiError("Sandbox returned stderr after its exit event");
				const chunkBytes = Buffer.byteLength(event.data, "utf8");
				assertExecOutputLimit("stderr", stderrBytes, chunkBytes, SANDBOX_EXEC_LIMITS.stderrBytes);
				assertExecOutputLimit(
					"combined output",
					stdoutBytes + stderrBytes,
					chunkBytes,
					SANDBOX_EXEC_LIMITS.combinedOutputBytes
				);
				stderrBytes += chunkBytes;
				stderr.push(event.data);
				options.onStderr?.(event.data);
			} else if (event.event === "exit") {
				sawEvent = true;
				if (terminal) throw new SandboxApiError("Sandbox returned more than one exit event");
				assertOptionalFiniteNumber(event, "exit_code", true);
				assertOptionalFiniteNumber(event, "signal", true);
				assertOptionalFiniteNumber(event, "duration_ms", false);
				if (event.timed_out !== undefined && typeof event.timed_out !== "boolean") {
					throw new SandboxApiError("Sandbox returned an invalid timed_out value");
				}
				terminal = {
					exitCode: typeof event.exit_code === "number" ? event.exit_code : null,
					signal: typeof event.signal === "number" ? event.signal : null,
					timedOut: event.timed_out === true,
					durationMs: typeof event.duration_ms === "number" ? event.duration_ms : 0,
				};
			} else {
				throw new SandboxApiError("Sandbox returned an unsupported NDJSON event");
			}
		}
		if (!terminal) throw new SandboxApiError("Connection lost while executing sandbox command");
		return {
			...terminal,
			stdout: stdout.join(""),
			stderr: stderr.join(""),
		};
	}

	async spawn(
		cmd: string | string[],
		options: { cwd?: string; environment?: Record<string, string>; tag?: string } = {}
	): Promise<SandboxProcess> {
		const response = await this.request("/v1/processes", {
			method: "POST",
			headers: { "Content-Type": "application/json" },
			body: JSON.stringify({
				cmd,
				shell: typeof cmd === "string",
				...(options.cwd ? { cwd: options.cwd } : {}),
				...(options.environment ? { env: options.environment } : {}),
				...(options.tag ? { tag: options.tag } : {}),
			}),
		});
		return parseSandboxProcess(
			await readJsonResponse<unknown>(response, "spawn sandbox process"),
			"spawn sandbox process"
		);
	}

	async listProcesses(options: { signal?: AbortSignal } = {}): Promise<SandboxProcess[]> {
		const response = await this.request("/v1/processes", { signal: options.signal });
		const body = await readJsonResponse<unknown>(response, "list sandbox processes");
		if (!Array.isArray(body)) {
			throw new SandboxApiError("Invalid process list while trying to list sandbox processes");
		}
		return body.map((process) => parseSandboxProcess(process, "list sandbox processes"));
	}

	async killProcess(processId: string): Promise<void> {
		if (!validSandboxProcessId(processId)) {
			throw new SandboxApiError("Invalid sandbox process id");
		}
		const response = await this.request(`/v1/processes/${encodeURIComponent(processId)}`, {
			method: "DELETE",
		});
		const body = await readJsonResponse<unknown>(response, "stop sandbox process");
		if (!isRecord(body) || body.id !== processId || body.ok !== true) {
			throw new SandboxApiError("Invalid response while trying to stop sandbox process");
		}
	}

	async writeFile(
		path: string,
		content: string | ArrayBuffer,
		mode?: string,
		options: { signal?: AbortSignal } = {}
	): Promise<void> {
		const params = new URLSearchParams({ path });
		if (mode) params.set("mode", mode);
		await this.request(`/v1/files/write?${params.toString()}`, {
			method: "PUT",
			body: content,
			signal: options.signal,
		});
	}

	async readFile(path: string): Promise<Uint8Array> {
		const params = new URLSearchParams({ path });
		const response = await this.request(`/v1/files/read?${params.toString()}`);
		return new Uint8Array(await response.arrayBuffer());
	}

	async readTextFile(path: string): Promise<string> {
		return new TextDecoder().decode(await this.readFile(path));
	}

	proxyUrlFor(port: number, path = "/", scheme: "https" | "wss" = "https"): string {
		const normalizedPath = path.startsWith("/") ? path : `/${path}`;
		const queryAt = normalizedPath.indexOf("?");
		const pathname = queryAt === -1 ? normalizedPath : normalizedPath.slice(0, queryAt);
		const search = queryAt === -1 ? "" : normalizedPath.slice(queryAt);
		const target = new URL(this.connection.baseUrl);
		target.protocol = `${scheme}:`;
		target.pathname = `/v1/proxy/${port}${pathname}`;
		target.search = search;
		return target.toString();
	}

	async proxy(port: number, path: string, init: RequestInit = {}): Promise<Response> {
		return this.fetchSandboxRoute(
			this.proxyUrlFor(port, path),
			{
				...init,
				headers: authenticatedHeaders(init.headers, this.proxyHeaders),
				redirect: "manual",
			},
			false
		);
	}

	async kill(): Promise<void> {
		const response = await this.requestFetch(
			`${this.hubUrl}/api/jobs/${encodeURIComponent(this.connection.namespace)}/${encodeURIComponent(
				this.id
			)}/cancel`,
			{
				method: "POST",
				headers: { Authorization: `Bearer ${this.hfToken}` },
				redirect: "manual",
			}
		);
		if (!response.ok && response.status !== 404 && response.status !== 409) {
			await throwResponseError(response, "cancel sandbox Job");
		}
	}

	private async getJob(): Promise<JobResponse> {
		const response = await this.requestFetch(
			`${this.hubUrl}/api/jobs/${encodeURIComponent(this.connection.namespace)}/${encodeURIComponent(
				this.id
			)}`,
			{
				headers: { Authorization: `Bearer ${this.hfToken}` },
				redirect: "manual",
			}
		);
		return readJsonResponse<JobResponse>(response, "inspect sandbox Job");
	}

	private async request(path: string, init: RequestInit = {}): Promise<Response> {
		const response = await this.fetchSandboxRoute(
			`${this.connection.baseUrl}${path}`,
			{
				...init,
				headers: authenticatedHeaders(init.headers, this.proxyHeaders),
				redirect: "manual",
			},
			true
		);
		if (!response.ok) await throwResponseError(response, "call sandbox");
		return response;
	}

	private async fetchSandboxRoute(
		url: string,
		init: RequestInit,
		retryIdempotentWrites: boolean
	): Promise<Response> {
		const method = (init.method ?? "GET").toUpperCase();
		const canRetryRouteMiss =
			method === "GET" ||
			method === "HEAD" ||
			(retryIdempotentWrites && (method === "PUT" || method === "DELETE"));
		for (let attempt = 0; ; attempt += 1) {
			const response = await this.requestFetch(url, init);
			const retryDelay = SANDBOX_ROUTE_RETRY_DELAYS_MS[attempt];
			if (!canRetryRouteMiss || response.status !== 404 || retryDelay === undefined) {
				return response;
			}
			await response.body?.cancel().catch(() => undefined);
			await delay(retryDelay, init.signal);
		}
	}
}

export function deriveSandboxToken(hfToken: string, nonce: string): string {
	return createHmac("sha256", hfToken).update(`hf-sandbox:${nonce}`).digest("hex");
}

async function* parseNdjson(response: Response): AsyncGenerator<Record<string, unknown>> {
	if (!response.body) throw new SandboxApiError("Sandbox response had no body");
	const reader = response.body.getReader();
	const decoder = new TextDecoder("utf-8", { fatal: true });
	const lineState = { value: "", bytes: 0 };
	let streamBytes = 0;
	let completed = false;
	try {
		for (;;) {
			const { done, value } = await reader.read();
			if (done) {
				for (const event of decodeNdjsonText(decoder.decode(), lineState)) yield event;
				completed = true;
				break;
			}
			streamBytes += value.byteLength;
			if (streamBytes > SANDBOX_EXEC_LIMITS.ndjsonStreamBytes) {
				throw new SandboxApiError(
					`Sandbox command stream exceeded ${formatByteLimit(SANDBOX_EXEC_LIMITS.ndjsonStreamBytes)}`
				);
			}
			for (const event of decodeNdjsonText(decoder.decode(value, { stream: true }), lineState)) {
				yield event;
			}
		}
		if (lineState.value.trim()) {
			yield parseNdjsonEvent(lineState.value.replace(/\r$/, ""));
		}
	} catch (cause) {
		if (cause instanceof SandboxApiError) throw cause;
		throw new SandboxApiError("Sandbox returned malformed UTF-8 NDJSON");
	} finally {
		if (!completed) await reader.cancel().catch(() => undefined);
		reader.releaseLock();
	}
}

function* decodeNdjsonText(
	text: string,
	state: { value: string; bytes: number }
): Generator<Record<string, unknown>> {
	let offset = 0;
	for (;;) {
		const newline = text.indexOf("\n", offset);
		if (newline === -1) break;
		const segment = text.slice(offset, newline);
		appendNdjsonSegment(state, segment);
		const line = state.value.replace(/\r$/, "");
		state.value = "";
		state.bytes = 0;
		if (line) {
			yield parseNdjsonEvent(line);
		}
		offset = newline + 1;
	}
	appendNdjsonSegment(state, text.slice(offset));
}

function appendNdjsonSegment(state: { value: string; bytes: number }, segment: string): void {
	if (!segment) return;
	const segmentBytes = Buffer.byteLength(segment, "utf8");
	if (state.bytes + segmentBytes > SANDBOX_EXEC_LIMITS.ndjsonLineBytes) {
		throw new SandboxApiError(
			`Sandbox command event exceeded ${formatByteLimit(SANDBOX_EXEC_LIMITS.ndjsonLineBytes)}`
		);
	}
	state.value += segment;
	state.bytes += segmentBytes;
}

function parseNdjsonEvent(line: string): Record<string, unknown> {
	let value: unknown;
	try {
		value = JSON.parse(line);
	} catch {
		throw new SandboxApiError("Sandbox returned malformed NDJSON");
	}
	if (typeof value !== "object" || value === null || Array.isArray(value)) {
		throw new SandboxApiError("Sandbox returned a non-object NDJSON event");
	}
	return value as Record<string, unknown>;
}

function hasExactKeys(event: Record<string, unknown>, expected: readonly string[]): boolean {
	const keys = Object.keys(event);
	return keys.length === expected.length && expected.every((key) => keys.includes(key));
}

function assertExecOutputLimit(
	name: string,
	currentBytes: number,
	chunkBytes: number,
	limitBytes: number
): void {
	if (chunkBytes > limitBytes - currentBytes) {
		throw new SandboxApiError(`Sandbox command ${name} exceeded ${formatByteLimit(limitBytes)}`);
	}
}

function assertOptionalFiniteNumber(
	event: Record<string, unknown>,
	key: string,
	integer: boolean
): void {
	const value = event[key];
	if (value === undefined || value === null) return;
	if (
		typeof value !== "number" ||
		!Number.isFinite(value) ||
		value < 0 ||
		(integer && !Number.isSafeInteger(value))
	) {
		throw new SandboxApiError(`Sandbox returned an invalid ${key} value`);
	}
}

function formatByteLimit(bytes: number): string {
	return `${bytes / (1024 * 1024)} MiB`;
}

function parseSandboxProcess(value: unknown, action: string): SandboxProcess {
	if (!isRecord(value)) {
		throw new SandboxApiError(`Invalid process while trying to ${action}`);
	}
	const { id, pid, cmd, tag, started_at_ms, running, exit_code } = value;
	if (
		!validSandboxProcessId(id) ||
		typeof pid !== "number" ||
		!Number.isSafeInteger(pid) ||
		pid <= 0 ||
		!validSandboxProcessCommand(cmd) ||
		(tag !== undefined && tag !== null && typeof tag !== "string") ||
		(started_at_ms !== undefined &&
			(typeof started_at_ms !== "number" ||
				!Number.isSafeInteger(started_at_ms) ||
				started_at_ms < 0)) ||
		(running !== undefined && typeof running !== "boolean") ||
		(exit_code !== undefined &&
			exit_code !== null &&
			(typeof exit_code !== "number" || !Number.isSafeInteger(exit_code)))
	) {
		throw new SandboxApiError(`Invalid process while trying to ${action}`);
	}
	return {
		id,
		pid,
		cmd,
		...(typeof tag === "string" ? { tag } : {}),
		...(typeof started_at_ms === "number" ? { startedAtMs: started_at_ms } : {}),
		...(typeof running === "boolean" ? { running } : {}),
		...(exit_code === null || typeof exit_code === "number" ? { exitCode: exit_code } : {}),
	};
}

function validSandboxProcessId(value: unknown): value is string {
	return typeof value === "string" && /^[A-Za-z0-9._~-]{1,256}$/.test(value);
}

function validSandboxProcessCommand(value: unknown): value is string | string[] {
	return (
		typeof value === "string" ||
		(Array.isArray(value) && value.length > 0 && value.every((part) => typeof part === "string"))
	);
}

function isRecord(value: unknown): value is Record<string, unknown> {
	return typeof value === "object" && value !== null && !Array.isArray(value);
}

async function readJsonResponse<T>(response: Response, action: string): Promise<T> {
	if (!response.ok) await throwResponseError(response, action);
	try {
		return (await response.json()) as T;
	} catch {
		throw new SandboxApiError(`Invalid JSON while trying to ${action}`, response.status);
	}
}

async function throwResponseError(response: Response, action: string): Promise<never> {
	let detail = response.statusText;
	try {
		const body = (await response.json()) as { error?: string; message?: string };
		detail = body.error ?? body.message ?? detail;
	} catch {
		// Preserve the status text for non-JSON proxy errors.
	}
	throw new SandboxApiError(
		`Failed to ${action} (${response.status}${detail ? `: ${detail}` : ""})`,
		response.status
	);
}

function trimTrailingSlash(value: string): string {
	return value.replace(/\/+$/, "");
}

function requireSandboxOrigin(jobId: string, candidate: string): string {
	const origin = parseSandboxOrigin(jobId, candidate);
	if (!origin) {
		throw new SandboxApiError(`Invalid sandbox URL for Job ${jobId}`);
	}
	return origin;
}

function parseSandboxOrigin(jobId: string, candidate: string): string | undefined {
	let url: URL;
	try {
		url = new URL(candidate);
	} catch {
		return undefined;
	}

	const expectedHostname = `${jobId}--${SANDBOX_SERVER_PORT}.hf.jobs`;
	if (
		url.protocol !== "https:" ||
		url.username !== "" ||
		url.password !== "" ||
		url.hostname !== expectedHostname ||
		url.port !== "" ||
		url.pathname !== "/" ||
		url.search !== "" ||
		url.hash !== "" ||
		(candidate !== url.origin && candidate !== `${url.origin}/`)
	) {
		return undefined;
	}

	return url.origin;
}

function authenticatedHeaders(
	input: HeadersInit | undefined,
	credentials: Record<string, string>
): Headers {
	const headers = new Headers(input);
	for (const [name, value] of Object.entries(credentials)) headers.set(name, value);
	return headers;
}

function delay(ms: number, signal?: AbortSignal | null): Promise<void> {
	if (signal?.aborted) return Promise.reject(signal.reason);
	return new Promise((resolve, reject) => {
		const timeout = setTimeout(done, ms);
		function done() {
			signal?.removeEventListener("abort", aborted);
			resolve();
		}
		function aborted() {
			clearTimeout(timeout);
			signal?.removeEventListener("abort", aborted);
			reject(signal?.reason);
		}
		signal?.addEventListener("abort", aborted, { once: true });
	});
}