fix(runtime-core): release a disposed VM's buffered sidecar events

packages/core/src/index.ts

@@ -1,5 +1,6 @@
 // @rivet-dev/agentos
 
+export { SidecarEventBufferOverflow } from "@rivet-dev/agentos-runtime-core/event-buffer";
 export {
 	SidecarProcessError,
 	SidecarProcessExited,

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

@@ -1,7 +1,18 @@
 import type { ChildProcess } from "node:child_process";
+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 {
+	NativeSidecarProcessClient,
+	serializeRootFilesystemForSidecar,
+} from "../src/sidecar/rpc-client.js";
+
+const REPO_ROOT = fileURLToPath(new URL("../../..", import.meta.url));
+const SIDECAR_BINARY = process.env.AGENTOS_SIDECAR_BIN
+	? resolve(process.env.AGENTOS_SIDECAR_BIN)
+	: join(REPO_ROOT, "target/debug/agentos-native-sidecar");
 
 // Kills the VM's real sidecar process. This reaches into internals only to find
 // the process; the behavior under test is the public evaluate() result.
@@ -92,3 +103,43 @@ test("a sidecar whose transport fails is stopped, not leaked", async () => {
 	await broken.dispose().catch(() => {});
 	await fresh.dispose();
 }, 60_000);
+
+test("a sidecar outlives more VM create and dispose cycles than its event buffer holds", async () => {
+	const eventBufferCapacity = 8;
+	const client = NativeSidecarProcessClient.spawn({
+		cwd: REPO_ROOT,
+		command: SIDECAR_BINARY,
+		eventBufferCapacity,
+	});
+	try {
+		const session = await client.authenticateAndOpenSession();
+		// Each cycle leaves lifecycle events that no caller waits for.
+		for (let cycle = 0; cycle < eventBufferCapacity * 2; cycle += 1) {
+			const vm = await client.createVm(session, {
+				runtime: "java_script",
+				config: {
+					env: {},
+					rootFilesystem: serializeRootFilesystemForSidecar(),
+				},
+			});
+			await client.waitForEvent(
+				{
+					type: "vm_lifecycle",
+					ownership: {
+						scope: "vm",
+						connection_id: session.connectionId,
+						session_id: session.sessionId,
+						vm_id: vm.vmId,
+					},
+					state: "ready",
+				},
+				10_000,
+			);
+			await client.disposeVm(session, vm);
+		}
+	} catch (error) {
+		await client.dispose().catch(() => {});
+		throw error;
+	}
+	await client.dispose();
+}, 60_000);

packages/runtime-core/src/event-buffer.ts

@@ -141,6 +141,17 @@ export class SidecarEventBuffer<TEvent extends LiveSidecarEventFrame> {
 	}
 
 	buffer(event: TEvent): SidecarEventBufferOverflow | null {
+		// `disposed` is the last event the sidecar emits for a VM. Nothing can
+		// claim that VM's unclaimed events afterwards, so release them instead of
+		// holding them for the sidecar's lifetime.
+		if (
+			event.payload.type === "vm_lifecycle" &&
+			event.payload.state === "disposed" &&
+			event.ownership.scope === "vm"
+		) {
+			this.releaseOwner(event.ownership);
+			return null;
+		}
 		if (this.bufferedEvents.size >= this.capacity) {
 			return new SidecarEventBufferOverflow({
 				capacity: this.capacity,
@@ -186,6 +197,22 @@ export class SidecarEventBuffer<TEvent extends LiveSidecarEventFrame> {
 		return null;
 	}
 
+	private releaseOwner(ownership: LiveOwnershipScope): void {
+		const queue = this.bufferedEventQueues.get(ANY_BUFFERED_EVENT_KEY);
+		if (!queue) {
+			return;
+		}
+		for (const eventId of [...queue]) {
+			const record = this.bufferedEvents.get(eventId);
+			if (
+				record &&
+				ownershipMatchesSelector(ownership, record.event.ownership)
+			) {
+				this.remove(eventId);
+			}
+		}
+	}
+
 	private takeFromKey(key: string): TEvent | null {
 		const queue = this.bufferedEventQueues.get(key);
 		if (!queue) {

secure-exec/src/index.ts

@@ -33,6 +33,7 @@ export {
 	hostDirMount,
 	KernelError,
 	nodeModulesMount,
+	SidecarEventBufferOverflow,
 	SidecarProcessError,
 	SidecarProcessExited,
 	SidecarRejectedError,