fix(runtime-core): reject oversized frames before they reach the sidecar

docs/content/docs/resource-limits.mdx

@@ -71,6 +71,14 @@ Library VMs share a sidecar process by default. Limits that apply to the whole
 process, such as `runtime.executor.maxActiveVms`, are set when creating a
 dedicated sidecar with `AgentOs.createSidecar()`, as in the example above.
 
+Each library request to the sidecar must fit in one protocol frame. The sidecar
+reports its frame limit (16 MiB) when the client connects, and the limit is not
+configurable. The client rejects a larger request with `SidecarFrameTooLarge`
+before it sends it, so the sidecar and the other VMs on it keep running. Binary
+file content is base64-encoded, so one `writeFile` or `writeFiles` call carries
+at most about 12 MiB of binary data. To move larger files, split them into
+several files or mount a host directory.
+
 ### Sidecar memory
 
 On Linux with glibc, the sidecar returns memory freed by a disposed VM's V8

packages/core/src/index.ts

@@ -2,6 +2,7 @@
 
 export { SidecarEventBufferOverflow } from "@rivet-dev/agentos-runtime-core/event-buffer";
 export {
+	SidecarFrameTooLarge,
 	SidecarProcessError,
 	SidecarProcessExited,
 	SidecarRejectedError,

packages/core/tests/sidecar-exit.test.ts

@@ -3,7 +3,7 @@ import { join, resolve } from "node:path";
 import { fileURLToPath } from "node:url";
 import { SidecarProcessExited } from "@rivet-dev/agentos-runtime-core/sidecar-client";
 import { expect, test } from "vitest";
-import { AgentOs } from "../src/index.js";
+import { AgentOs, SidecarFrameTooLarge } from "../src/index.js";
 import {
 	NativeSidecarProcessClient,
 	serializeRootFilesystemForSidecar,
@@ -143,3 +143,43 @@ test("a sidecar outlives more VM create and dispose cycles than its event buffer
 	}
 	await client.dispose();
 }, 60_000);
+
+test("an oversized request rejects without stopping the sidecar its VMs share", async () => {
+	const uncaught: unknown[] = [];
+	const recordUncaught = (error: unknown) => uncaught.push(error);
+	process.on("uncaughtException", recordUncaught);
+	const sidecar = await AgentOs.createSidecar();
+	const placement = { kind: "explicit", handle: sidecar } as const;
+	const writer = await AgentOs.create({
+		defaultSoftware: false,
+		sidecar: placement,
+	});
+	const bystander = await AgentOs.create({
+		defaultSoftware: false,
+		sidecar: placement,
+	});
+	try {
+		// Base64 content encoding puts this request over the sidecar frame limit.
+		const error = await writer.filesystem
+			.writeFile("/tmp/big.bin", new Uint8Array(13 * 1024 * 1024))
+			.catch((caught: unknown) => caught);
+		expect(error).toMatchObject({ name: "SidecarFrameTooLarge" });
+		expect(error).toBeInstanceOf(SidecarFrameTooLarge);
+		const { frameBytes, maxFrameBytes } = error as SidecarFrameTooLarge;
+		expect(frameBytes).toBeGreaterThan(maxFrameBytes);
+		expect((error as Error).message).toContain(`limit is ${maxFrameBytes}`);
+
+		// Another VM on the same sidecar keeps working, and so does the writer.
+		expect(await bystander.javascript.evaluate("1 + 1")).toMatchObject({
+			outcome: "succeeded",
+			value: 2,
+		});
+		await writer.filesystem.writeFile("/tmp/small.txt", "ok");
+		expect(uncaught).toEqual([]);
+	} finally {
+		process.off("uncaughtException", recordUncaught);
+		await writer.dispose().catch(() => {});
+		await bystander.dispose().catch(() => {});
+		await sidecar.dispose().catch(() => {});
+	}
+}, 60_000);

packages/runtime-core/src/frame-stream.ts

@@ -4,12 +4,18 @@ import {
 	encodeLengthPrefixedPayload,
 	tryDecodeLengthPrefixedPayload,
 } from "./framing.js";
+import { SidecarFrameTooLarge } from "./sidecar-errors.js";
 
 export interface StdioFrameTransportOptions<TReadFrame, TWriteFrame> {
 	stdin: Writable;
 	stdout: Readable;
 	encodeFrame: (frame: TWriteFrame) => Uint8Array;
 	decodeFrame: (payload: Uint8Array) => TReadFrame;
+	/**
+	 * Largest payload the peer accepts, or `undefined` while unknown. A larger
+	 * frame is rejected before it is written.
+	 */
+	maxFrameBytes?: () => number | undefined;
 }
 
 export interface FrameTransport<TReadFrame, TWriteFrame = TReadFrame> {
@@ -208,17 +214,21 @@ export class StdioFrameTransport<TReadFrame, TWriteFrame = TReadFrame>
 	private readonly stdout: Readable;
 	private readonly encodeFrame: (frame: TWriteFrame) => Uint8Array;
 	private readonly decodeFrame: (payload: Uint8Array) => TReadFrame;
+	private readonly maxFrameBytes: (() => number | undefined) | undefined;
 	private readonly sharedStdio: boolean;
 	private readonly frameListeners = new Set<(frame: TReadFrame) => void>();
 	private readonly errorListeners = new Set<(error: Error) => void>();
 	private readonly endListeners = new Set<() => void>();
 	private stdoutBuffer: ByteArray = new Uint8Array(0);
+	private pendingWrites = 0;
+	private disposed = false;
 
 	constructor(options: StdioFrameTransportOptions<TReadFrame, TWriteFrame>) {
 		this.stdin = options.stdin;
 		this.stdout = options.stdout;
 		this.encodeFrame = options.encodeFrame;
 		this.decodeFrame = options.decodeFrame;
+		this.maxFrameBytes = options.maxFrameBytes;
 		this.sharedStdio = this.stdin === (this.stdout as unknown as Writable);
 		if (!this.sharedStdio) {
 			this.stdin.on("error", this.handleError);
@@ -250,10 +260,24 @@ export class StdioFrameTransport<TReadFrame, TWriteFrame = TReadFrame>
 	}
 
 	async writeFrame(frame: TWriteFrame): Promise<void> {
+		if (this.disposed) {
+			throw new Error("stdio frame transport disposed");
+		}
 		const payload = this.encodeFrame(frame);
+		const maxFrameBytes = this.maxFrameBytes?.();
+		// The peer cannot resynchronize after an oversized length prefix, so
+		// writing one would end the stream for every user of it.
+		if (maxFrameBytes !== undefined && payload.length > maxFrameBytes) {
+			throw new SidecarFrameTooLarge({
+				frameBytes: payload.length,
+				maxFrameBytes,
+			});
+		}
 		const encoded = encodeLengthPrefixedPayload(payload);
+		this.pendingWrites += 1;
 		await new Promise<void>((resolve, reject) => {
 			this.stdin.write(encoded, (error) => {
+				this.pendingWrites -= 1;
 				if (error) {
 					reject(error);
 					return;
@@ -264,6 +288,10 @@ export class StdioFrameTransport<TReadFrame, TWriteFrame = TReadFrame>
 	}
 
 	dispose(): void {
+		if (this.disposed) {
+			return;
+		}
+		this.disposed = true;
 		if (!this.sharedStdio) {
 			this.stdin.off("error", this.handleError);
 		}
@@ -273,6 +301,14 @@ export class StdioFrameTransport<TReadFrame, TWriteFrame = TReadFrame>
 		this.frameListeners.clear();
 		this.errorListeners.clear();
 		this.endListeners.clear();
+		if (this.pendingWrites > 0) {
+			// A write still in flight rejects its own promise, but Node also emits
+			// the failure (for example EPIPE from a dead sidecar) as a stream
+			// `error` event. Without a listener that event crashes the host.
+			const stdin = this.stdin;
+			stdin.on("error", logLateStreamError);
+			stdin.once("close", () => stdin.off("error", logLateStreamError));
+		}
 	}
 
 	private readonly handleData = (chunk: ByteArray | string): void => {
@@ -319,6 +355,13 @@ export class StdioFrameTransport<TReadFrame, TWriteFrame = TReadFrame>
 	}
 }
 
+function logLateStreamError(error: Error): void {
+	console.error(
+		"[agentOS] sidecar stream failed after its frame transport was disposed:",
+		error,
+	);
+}
+
 function concatBytes(left: ByteArray, right: ByteArray): ByteArray {
 	if (left.length === 0) {
 		return right;

packages/runtime-core/src/protocol-client.ts

@@ -91,6 +91,11 @@ export class SidecarProtocolClient {
 		LiveSidecarRequestFrame
 	>;
 	private closedError: Error | null = null;
+	/**
+	 * Frame limit from the sidecar's authenticate response. Only the small
+	 * authenticate request is written before it is known.
+	 */
+	private maxFrameBytes: number | undefined;
 	private readonly closeListeners = new Set<(error: Error) => void>();
 	private readonly eventWaiters = new Set<{
 		matches: (event: LiveEventFrame) => boolean;
@@ -113,7 +118,11 @@ export class SidecarProtocolClient {
 			LiveEventFrame,
 			LiveSidecarRequestFrame
 		>({
-			frameTransport: resolveProtocolFrameTransport(options, this.payloadCodec),
+			frameTransport: resolveProtocolFrameTransport(
+				options,
+				this.payloadCodec,
+				() => this.maxFrameBytes,
+			),
 			encodeFrame: (frame) =>
 				encodeProtocolFramePayload(frame, this.payloadCodec),
 			decodeFrame: (payload) =>
@@ -223,6 +232,9 @@ export class SidecarProtocolClient {
 		if (response.payload.type === "rejected") {
 			throw new SidecarRejectedError(request.request_id, response.payload);
 		}
+		if (response.payload.type === "authenticated") {
+			this.maxFrameBytes = response.payload.max_frame_bytes;
+		}
 		return response;
 	}
 
@@ -424,6 +436,7 @@ export class SidecarProtocolClient {
 function resolveProtocolFrameTransport(
 	options: SidecarProtocolClientOptions,
 	payloadCodec: ProtocolFramePayloadCodec,
+	maxFrameBytes: () => number | undefined,
 ): FrameTransport<
 	LiveResponseFrame | LiveEventFrame | LiveSidecarRequestFrame,
 	LiveRequestFrame | LiveSidecarResponseFrame | LiveControlFrame
@@ -451,6 +464,7 @@ function resolveProtocolFrameTransport(
 			stdin: options.stdin,
 			stdout: options.stdout,
 			encodeFrame: (frame) => encodeProtocolFramePayload(frame, payloadCodec),
+			maxFrameBytes,
 			decodeFrame: (payload) =>
 				decodeProtocolFramePayload(payload, payloadCodec) as
 					| LiveResponseFrame
@@ -485,6 +499,7 @@ function resolveProtocolFrameTransport(
 			stdin: options.stdin,
 			stdout: options.stdout,
 			encodeFrame: (frame) => encodeProtocolFramePayload(frame, payloadCodec),
+			maxFrameBytes,
 			decodeFrame: (payload) => {
 				const frame = decodeProtocolFramePayload(payload, payloadCodec);
 				validateOrdinaryInboundFrame(frame);
@@ -498,6 +513,7 @@ function resolveProtocolFrameTransport(
 			stdin: options.control,
 			stdout: options.control,
 			encodeFrame: (frame) => encodeProtocolFramePayload(frame, payloadCodec),
+			maxFrameBytes,
 			decodeFrame: (payload) => {
 				const frame = decodeProtocolFramePayload(payload, payloadCodec);
 				validateControlInboundFrame(frame);

packages/runtime-core/src/sidecar-errors.ts

@@ -34,6 +34,24 @@ export class SidecarRejectedError extends Error {
 	}
 }
 
+/**
+ * A host frame exceeded the sidecar's negotiated `max_frame_bytes`. The client
+ * rejects it before writing, so the sidecar and its other VMs keep running.
+ */
+export class SidecarFrameTooLarge extends Error {
+	readonly frameBytes: number;
+	readonly maxFrameBytes: number;
+
+	constructor(options: { frameBytes: number; maxFrameBytes: number }) {
+		super(
+			`sidecar protocol frame is ${options.frameBytes} bytes, limit is ${options.maxFrameBytes} (sidecar max_frame_bytes); the request was not sent. Send less data per request, for example by writing large content in several smaller files or by mounting a host directory`,
+		);
+		this.name = "SidecarFrameTooLarge";
+		this.frameBytes = options.frameBytes;
+		this.maxFrameBytes = options.maxFrameBytes;
+	}
+}
+
 export class SidecarProcessExited extends Error {
 	readonly exitCode: number | null;
 	readonly signal: string | null;

packages/runtime-core/src/sidecar-process.ts

@@ -53,6 +53,7 @@ import type {
 
 export { SidecarEventBufferOverflow } from "./event-buffer.js";
 export {
+	SidecarFrameTooLarge,
 	SidecarProcessError,
 	SidecarProcessExited,
 	SidecarSilenceTimeout,

packages/runtime-core/tests/frame-stream.test.ts

@@ -1,5 +1,5 @@
-import { PassThrough } from "node:stream";
-import { describe, expect, test } from "vitest";
+import { PassThrough, Writable } from "node:stream";
+import { describe, expect, test, vi } from "vitest";
 import { StdioFrameTransport } from "../src/frame-stream.js";
 import { encodeLengthPrefixedPayload } from "../src/framing.js";
 
@@ -111,6 +111,41 @@ describe("stdio frame transport", () => {
 		expect(stdio.listenerCount("error")).toBe(initialErrorListeners);
 	});
 
+	test("a write that fails after dispose is logged instead of crashing the host", async () => {
+		let finishWrite: ((error?: Error | null) => void) | undefined;
+		const stdin = new Writable({
+			write(_chunk, _encoding, callback) {
+				finishWrite = callback;
+			},
+		});
+		const transport = new StdioFrameTransport<string>({
+			stdin,
+			stdout: new PassThrough(),
+			encodeFrame: (frame) => Buffer.from(frame, "utf8"),
+			decodeFrame: (payload) => Buffer.from(payload).toString("utf8"),
+		});
+		const uncaught: unknown[] = [];
+		const recordUncaught = (error: unknown) => uncaught.push(error);
+		process.on("uncaughtException", recordUncaught);
+		const logged = vi.spyOn(console, "error").mockImplementation(() => {});
+		try {
+			const write = transport.writeFrame("in flight");
+			transport.dispose();
+			const brokenPipe = Object.assign(new Error("write EPIPE"), {
+				code: "EPIPE",
+			});
+			finishWrite?.(brokenPipe);
+
+			await expect(write).rejects.toBe(brokenPipe);
+			await new Promise((resolve) => setImmediate(resolve));
+			expect(uncaught).toEqual([]);
+			expect(logged).toHaveBeenCalledWith(expect.any(String), brokenPipe);
+		} finally {
+			logged.mockRestore();
+			process.off("uncaughtException", recordUncaught);
+		}
+	});
+
 	test("reports decode errors", async () => {
 		const stdin = new PassThrough();
 		const stdout = new PassThrough();