Introducing Pi Durable for Rivet

npm add @rivet-dev/pi @earendil-works/pi-durable rivetkit

import { createClient } from "rivetkit/client";
import type { registry } from "./server";

const client = createClient<typeof registry>();
const agent = client.agent.getOrCreate(["acme/api", "issue-1234"]);

const result = await agent.prompt("Find the failing test and fix it.", {
	requestId: "delivery-8f2c",
});

// A model error resolves with status "unanswered" and Pi's reason
console.log(result.status === "done" ? result.text : result.reason);

npm add @earendil-works/pi-ai

// Checkpoints in Pi Durable are explicit: each phase names the state it saves,
// so you, or an agent editing this file, can see exactly where a restart resumes.
import { Type } from "@earendil-works/pi-ai";
import { createRegistry, defineExtension, defineTask, defineTool } from "@earendil-works/pi-durable";
import { pi } from "@rivet-dev/pi";
import { setup } from "rivetkit";

type RefundState = { phase: "refund" } | { phase: "notify"; refundId: string };

// Two phases. Each one saves its progress before the next one starts
const Refund = defineTask<{ orderId: string }, RefundState, string>({
	name: "orders.refund",
	version: 1,
	initial: () => ({ phase: "refund" }),
	phases: {
		refund: async (task, runtime, context) => {
			// Keyed on the order, so the payment API refunds each order at most once
			const res = await fetch(`https://api.example.com/orders/${task.input.orderId}/refund`, {
				method: "POST",
				headers: { "idempotency-key": `refund-${task.input.orderId}` },
			});
			const { refundId } = (await res.json()) as { refundId: string };
			await runtime.commit(() => ({ status: "running", checkpoint: { phase: "notify", refundId } }), context);
		},
		notify: async (task, runtime, context) => {
			const { refundId } = task.state.checkpoint;
			await fetch(`https://api.example.com/orders/${task.input.orderId}/notify`, { method: "POST" });
			await runtime.commit(() => ({ status: "terminal", outcome: { status: "completed", result: refundId } }), context);
		},
	},
	abort: async (_task, runtime, context) => {
		await runtime.commit(() => ({ status: "terminal", outcome: { status: "aborted" } }), context);
	},
});

// The tool owns the refund task, so cancelling the call cancels the refund
const refundOrder = defineTool({
	name: "refund_order",
	description: "Refund an order and notify the customer",
	parameters: Type.Object({ orderId: Type.String() }),
	execute: async ({ orderId }, api, context) => {
		const id = await api.createTask(Refund, { orderId }, { ownership: { kind: "task", taskId: api.taskId } }, context);
		const { outcome } = (await api.waitForTask(id, context)).state;
		const text = outcome.status === "completed" ? `Refunded: ${outcome.result}` : outcome.status;
		return { content: [{ type: "text", text }] };
	},
});

const tools = createRegistry();
tools.install(defineExtension({ name: "orders", tools: [refundOrder], tasks: [Refund] }));

// One support agent per customer. Refund tasks are stored in its SQLite next to the conversation
const support = pi({
	model: "anthropic/claude-opus-5-5",
	registry: tools,
});

export const registry = setup({ use: { support } });

registry.start();

import { createClient } from "rivetkit/client";
import type { registry } from "./server";

const client = createClient<typeof registry>();
const support = client.support.getOrCreate(["customer-123"]);

// The ticket id doubles as the requestId, so a retried request reuses the first run
const result = await support.prompt("Order 4521 arrived broken. Please refund it.", {
	requestId: "ticket-881",
});
console.log(result.status === "done" ? result.text : result.reason);

import { createRegistry } from "@earendil-works/pi-durable";
import { pi } from "@rivet-dev/pi";
import { setup } from "rivetkit";

const agent = pi({
	model: "anthropic/claude-opus-5-5",
	registry: createRegistry(),
	onCreate: async (c) => {
		// Every morning at 9:00, prompt the agent
		await c.cron.set({
			name: "morning-digest",
			expression: "0 9 * * *",
			action: "prompt",
			args: ["Summarize yesterday's support tickets."],
		});
	},
});

export const registry = setup({ use: { agent } });

registry.start();

import { Type } from "@earendil-works/pi-ai";
import { createRegistry, defineDoc, defineExtension, defineTool } from "@earendil-works/pi-durable";
import { pi } from "@rivet-dev/pi";
import { setup } from "rivetkit";

// A typed document stored in the Actor's SQLite, next to the conversation
const Notes = defineDoc<{ items: string[] }>({
	kind: "app.notes",
	version: 1,
	scope: "conversation",
	history: "latest",
	fork: "current",
	initial: () => ({ items: [] }),
});

const saveNote = defineTool({
	name: "save_note",
	description: "Save a note for later",
	parameters: Type.Object({ text: Type.String() }),
	execute: async ({ text }, api, context) => {
		// Committed in the same transaction as the tool result
		await api.commit(async (tx) => {
			(await tx.doc(Notes, api.conversationId)).items.push(text);
		}, context);
		return { content: [{ type: "text", text: "Saved." }] };
	},
});

const tools = createRegistry();
tools.install(defineExtension({ name: "notes", tools: [saveNote] }));

// Clients read a document by kind with harness.snapshot("app.notes", conversationId)
const agent = pi({
	model: "anthropic/claude-opus-5-5",
	registry: tools,
	documents: [Notes],
});

export const registry = setup({ use: { agent } });

registry.start();