chore(pi): require RivetKit 2.3.26

examples/docs/package.json

@@ -17,13 +17,13 @@
 		"@rivet-dev/pi": "workspace:*",
 		"@rivet-dev/sandbox-adapter": "workspace:*",
 		"@rivet-dev/workflows": "1.0.0",
-		"@rivetkit/react": "2.3.23",
+		"@rivetkit/react": "2.3.26",
 		"@slack/web-api": "^8.2.0",
 		"discord-interactions": "^4.4.0",
 		"e2b": "^2.51.0",
 		"hono": "^4.13.12",
 		"react": "^19.3.0",
-		"rivetkit": "2.3.23"
+		"rivetkit": "2.3.26"
 	},
 	"devDependencies": {
 		"@types/node": "^24.0.0",

packages/pi/package.json

@@ -53,15 +53,15 @@
 	"peerDependencies": {
 		"@earendil-works/chord": "^1.0.0",
 		"@earendil-works/pi-durable": "^1.0.0",
-		"rivetkit": ">=2.3.23 <2.4.0"
+		"rivetkit": ">=2.3.26 <2.4.0"
 	},
 	"devDependencies": {
 		"@earendil-works/chord": "^1.1.0",
 		"@earendil-works/pi-ai": "^1.1.0",
 		"@earendil-works/pi-durable": "^1.1.0",
 		"@opentelemetry/sdk-trace-node": "2.11.0",
 		"@types/node": "^24.0.0",
-		"rivetkit": "2.3.23",
+		"rivetkit": "2.3.26",
 		"typescript": "^5.9.2",
 		"vitest": "^3.2.4"
 	},

packages/pi/tests/actor.test.ts

@@ -22,7 +22,7 @@ import type { SandboxProvider } from "@rivet-dev/sandbox-adapter";
 import { setup } from "rivetkit";
 import { setupTest } from "rivetkit/test";
 import { afterAll, beforeAll, describe, expect, test, vi } from "vitest";
-import { pi } from "../src/index.js";
+import { type PiEventsFrame, pi } from "../src/index.js";
 import { localSandboxProvider } from "./helpers/local-sandbox.js";
 import {
 	createMockModel,
@@ -500,6 +500,41 @@ describe("pi actor", () => {
 		expect(result).toMatchObject({ status: "done", text: "Hi there!" });
 	});
 
+	test("a watching connection gets a fresh snapshot when a failed commit reopens the harness, then the next run's batches", async (c) => {
+		const { client } = await setupTest(c, registry);
+		const conn = client.agent
+			.getOrCreate(["poison-watch", randomUUID()])
+			.connect();
+		const frames: PiEventsFrame[] = [];
+		// SAFETY: RivetKit types event payloads as unknown on the client, and pi() sends only PiEventsFrame on pi.events.
+		conn.on("pi.events", (frame) => void frames.push(frame as PiEventsFrame));
+		const root = await conn.harness.root();
+		await conn.conversation.watchEvents(root.id);
+		await conn.failCommitsOfPoison();
+
+		await expect(conn.writePoison()).rejects.toThrow();
+		expect(await conn.prompt("say hello")).toMatchObject({
+			status: "done",
+			text: "Hi there!",
+		});
+
+		// The reopened harness reattaches the watch: a seq 0 snapshot, then the run's batches in order.
+		await vi.waitFor(() =>
+			expect(
+				frames.some((frame) =>
+					frame.events.some((event) => event.type === "run_end"),
+				),
+			).toBe(true),
+		);
+		const reattached = frames.findIndex((frame) => frame.seq === 0);
+		expect(frames[reattached]?.events).toEqual([
+			expect.objectContaining({ type: "snapshot" }),
+		]);
+		expect(frames.slice(reattached).map((frame) => frame.seq)).toEqual(
+			frames.slice(reattached).map((_, index) => index),
+		);
+	});
+
 	test("two conversations prompted at the same time each get their own answer, and neither transcript holds the other's messages", async (c) => {
 		const { client } = await setupTest(c, registry);
 		const handle = client.agent.getOrCreate(["two", randomUUID()]);

packages/pi/tests/watch.test.ts

@@ -265,33 +265,38 @@ describe("pi watches", () => {
 		);
 	});
 
-	test("a connection that survives sleep gets its watch back with a fresh snapshot", async (c) => {
+	test("a connection that watches each time it opens gets the next run's events after sleep closes it", async (c) => {
 		const { client } = await setupTest(c, registry);
-		const key = ["reattach", randomUUID()];
+		const key = ["rewatch", randomUUID()];
 		const conn = client.agent.getOrCreate(key).connect();
 		const { frames: eventFrames, push } = received<PiEventsFrame>();
 		conn.on("pi.events", push);
 		const root = await conn.harness.root();
-		await conn.conversation.watchEvents(root.id);
+		// Sleep closes the connection, so the client watches again each time it opens.
+		const watches: Promise<unknown>[] = [];
+		conn.onOpen(() => {
+			watches.push(conn.conversation.watchEvents(root.id));
+		});
+		// The connection is already open, so onOpen runs in a microtask.
+		await vi.waitFor(() => expect(watches).toHaveLength(1));
+		await watches[0];
 
 		await conn.nap();
-		// Sleep completes after the nap action returns, and onSleep records it in this process.
-		await vi.waitFor(() =>
-			expect(sleeps.get(JSON.stringify(key)) ?? 0).toBe(1),
-		);
+		// The client reconnects on its own after sleep closes the connection, and that wakes the actor.
+		await vi.waitFor(() => expect(watches).toHaveLength(2));
+		await watches[1];
+		expect(sleeps.get(JSON.stringify(key))).toBe(1);
+		const afterWake = eventFrames.length;
 		expect(await conn.prompt("say hello")).toMatchObject({
 			status: "done",
 			text: "Hi there!",
 		});
 
-		// The reattached watch sends its snapshot when the actor wakes, before the prompt's batches.
-		await vi.waitFor(() => expect(hasEvent(eventFrames, "run_end")).toBe(true));
-		const reattached = eventFrames.findIndex((frame) => frame.seq === 0);
-		expect(eventFrames[reattached]?.events).toEqual([
-			expect.objectContaining({ type: "snapshot" }),
-		]);
-		expect(eventFrames.slice(reattached).map((frame) => frame.seq)).toEqual(
-			eventFrames.slice(reattached).map((_, index) => index),
+		// The new watch numbers its batches from 1, after the snapshot that watchEvents returned.
+		await vi.waitFor(() =>
+			expect(hasEvent(eventFrames.slice(afterWake), "run_end")).toBe(true),
 		);
+		const batches = eventFrames.slice(afterWake).map((frame) => frame.seq);
+		expect(batches).toEqual(batches.map((_, index) => index + 1));
 	});
 });