Agents

Durable AI Agents

Run an agent whose conversation, in-flight tool calls, and plan survive crashes, deploys, and restarts.

See the full example on GitHub. View on GitHub

A durable agent saves every step of its work as it goes, so a crash, deploy, or restart in the middle of an answer loses at most about the last second of streamed text. Here a travel assistant keeps its conversation, its plan, and its tool calls in its own SQLite database, and a chat client picks the answer back up after a reload.

What we’ll build:

A chatAgent per conversation, keyed by the conversation id, that keeps its history and plan in its own SQLite database.
Tools that call your API and don’t book a flight twice after a crash.
A chat client that sends each message straight to the agent with submit, streams the agent’s events, and picks the answer back up after a reload.

Install

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

Create the agent

Each conversation gets its own chatAgent, keyed by the conversation id:

import { arr, bool, obj, str } from "@earendil-works/pi-schema";
import { createRegistry, defineDoc, defineExtension, defineTool, section } from "@earendil-works/pi-durable";
import { pi } from "@rivet-dev/pi";
import { bookFlight, searchFlights } from "./tools";

// The agent's plan, saved with the conversation in the agent's SQLite database.
export const Plan = defineDoc<{ steps: { text: string; done: boolean }[] }>({
	kind: "travel.plan",
	version: 1,
	scope: "conversation",
	history: "latest",
	fork: "current",
	initial: () => ({ steps: [] }),
	// Store the whole document every 50 changes, so loading it replays at most 50.
	checkpointWhen: (_value, _ops, { deltasSinceBase }) => deltasSinceBase >= 50,
});

const updatePlan = defineTool({
	name: "update_plan",
	description: "Save your plan for a request that takes more than one step. Call it again whenever a step is done.",
	parameters: obj({
		steps: arr(obj({ text: str(), done: bool() })),
	}),
	// Saving the same plan twice is harmless.
	replay: "safe",
	execute: async ({ steps }, api, context) => {
		await api.commit(async (tx) => {
			(await tx.doc(Plan, api.conversationId)).steps = steps;
		}, context);
		return { output: [{ type: "text", text: "Plan saved." }] };
	},
});

// Shows the plan to the model before every model call, even after older messages are compacted.
const planSection = section("plan", async (input, context) => {
	const plan = await input.read.snapshot(Plan, input.conversationId, context);
	if (!plan?.steps.length) return undefined;
	return plan.steps.map((step) => `- [${step.done ? "x" : " "}] ${step.text}`).join("\n");
});

const extensions = createRegistry();
extensions.install(
	defineExtension({
		name: "travel",
		tools: [updatePlan, searchFlights, bookFlight],
		sections: [
			section("role", () => "You are a travel assistant. Ask the user to confirm before you book anything."),
			planSection,
		],
	}),
);

// One agent per conversation, keyed by the conversation id.
export const chatAgent = pi({
	model: "anthropic/claude-opus-5-5",
	registry: extensions,
	// Lets clients read the plan with harness.snapshot.
	documents: [Plan],
});

There’s no step that creates an agent. The client calls chatAgent.getOrCreate([conversationId]), so the first message creates the agent and every later message reaches the same one.
The agent is a Rivet Actor that runs Pi. Pi saves each model call and tool call to the Actor’s SQLite database before the next one starts, so there is no separate workflow engine, queue, or database to run.
An idle agent sleeps and uses no memory. The next message wakes it with its whole conversation.

Keep a plan

The plan is a document: typed state that Pi saves in the same database as the conversation.

import { arr, bool, obj, str } from "@earendil-works/pi-schema";
import { createRegistry, defineDoc, defineExtension, defineTool, section } from "@earendil-works/pi-durable";
import { pi } from "@rivet-dev/pi";
import { bookFlight, searchFlights } from "./tools";

// The agent's plan, saved with the conversation in the agent's SQLite database.
export const Plan = defineDoc<{ steps: { text: string; done: boolean }[] }>({
	kind: "travel.plan",
	version: 1,
	scope: "conversation",
	history: "latest",
	fork: "current",
	initial: () => ({ steps: [] }),
	// Store the whole document every 50 changes, so loading it replays at most 50.
	checkpointWhen: (_value, _ops, { deltasSinceBase }) => deltasSinceBase >= 50,
});

const updatePlan = defineTool({
	name: "update_plan",
	description: "Save your plan for a request that takes more than one step. Call it again whenever a step is done.",
	parameters: obj({
		steps: arr(obj({ text: str(), done: bool() })),
	}),
	// Saving the same plan twice is harmless.
	replay: "safe",
	execute: async ({ steps }, api, context) => {
		await api.commit(async (tx) => {
			(await tx.doc(Plan, api.conversationId)).steps = steps;
		}, context);
		return { output: [{ type: "text", text: "Plan saved." }] };
	},
});

// Shows the plan to the model before every model call, even after older messages are compacted.
const planSection = section("plan", async (input, context) => {
	const plan = await input.read.snapshot(Plan, input.conversationId, context);
	if (!plan?.steps.length) return undefined;
	return plan.steps.map((step) => `- [${step.done ? "x" : " "}] ${step.text}`).join("\n");
});

const extensions = createRegistry();
extensions.install(
	defineExtension({
		name: "travel",
		tools: [updatePlan, searchFlights, bookFlight],
		sections: [
			section("role", () => "You are a travel assistant. Ask the user to confirm before you book anything."),
			planSection,
		],
	}),
);

// One agent per conversation, keyed by the conversation id.
export const chatAgent = pi({
	model: "anthropic/claude-opus-5-5",
	registry: extensions,
	// Lets clients read the plan with harness.snapshot.
	documents: [Plan],
});

The model calls update_plan when a request takes more than one step, and again as each step is done. The tool writes the plan in one commit with api.commit.
The plan section adds the saved plan to the system prompt before every model call. When the context fills up and older messages are summarized, the plan stays in full.
documents: [Plan] lets a client read the plan with harness.snapshot("travel.plan", conversationId), such as to show a checklist next to the chat.

Add tools that call your API

import { obj, str } from "@earendil-works/pi-schema";
import { defineTool } from "@earendil-works/pi-durable";

export const searchFlights = defineTool({
	name: "search_flights",
	description: "Find flights between two airports on a date.",
	parameters: obj({
		from: str({ description: "Airport code, such as SFO." }),
		to: str({ description: "Airport code, such as HND." }),
		date: str({ description: "The date, as YYYY-MM-DD." }),
	}),
	// A search changes nothing, so running it again after a crash is harmless.
	replay: "safe",
	execute: async ({ from, to, date }) => {
		const flights = await findFlights(from, to, date);
		const text = flights.map((f) => `${f.id}: departs ${f.departs}, $${f.price}`).join("\n");
		return { output: [{ type: "text", text }], details: flights };
	},
});

export const bookFlight = defineTool({
	name: "book_flight",
	description: "Book a seat on a flight. Only call this after the user confirms the flight.",
	parameters: obj({ flightId: str(), passenger: str() }),
	// After a crash this runs again with the same idempotency key, so the API books only once.
	replay: "safe",
	execute: async ({ flightId, passenger }, api, context) => {
		// The memo is saved with this tool call before the booking, so a rerun gets the same key.
		const idempotencyKey = await api.memo("idempotency-key", crypto.randomUUID(), context);
		const booking = await createBooking(flightId, passenger, idempotencyKey);
		const text = `Booked ${passenger} on ${flightId}. Confirmation ${booking.confirmation}.`;
		return { output: [{ type: "text", text }], details: booking };
	},
});

// Mocks of your travel API. Replace them with calls to your own API.
async function findFlights(from: string, to: string, date: string) {
	return [
		{ id: `${from}-${to}-0810`, departs: `${date} 08:10`, price: 1240 },
		{ id: `${from}-${to}-1355`, departs: `${date} 13:55`, price: 980 },
	];
}

// Your API should return the first booking again when it sees the same idempotency key.
async function createBooking(flightId: string, passenger: string, idempotencyKey: string) {
	return { flightId, passenger, confirmation: idempotencyKey.slice(0, 6).toUpperCase() };
}

If the worker crashes while a tool runs, the agent decides on restart whether to run it again:

search_flights only reads, so it sets replay: "safe" and simply runs again.
book_flight changes something, so a second run must not book a second seat. api.memo saves an idempotency key with the tool call before the booking, and a rerun gets the same key. Your API returns the first booking when it sees that key again.
A tool without replay: "safe" isn’t run again. The model gets a result saying the call stopped partway and decides what to do. Use that for an API that has no idempotency key.
The mocks at the bottom of the file stand in for your API. Replace them with your own calls.

For work with several steps that must each finish once, such as charging a card and then sending a receipt, launch a task from the tool. A task saves a checkpoint after each step and resumes from the last one.

Start the agent

import { setup } from "rivetkit";
import { chatAgent } from "./chat-agent";

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

registry.start();

There are no HTTP routes. Clients call the agent’s actions directly through RivetKit.

Stream the answer

The client connects to the agent for its conversation, prints what changes in its view, and sends each message with submit:

import { createInterface } from "node:readline";
import { applyImmutable, type Op } from "@earendil-works/chord/delta";
import { contentText } from "@earendil-works/pi-ai";
import type { ConversationView, LiveState } from "@earendil-works/pi-durable";
import type { PiViewFrame } from "@rivet-dev/pi";
import { createClient } from "rivetkit/client";
import type { registry } from "./server";

// Run with the same conversation id again to reopen it, like a reload.
const conversationId = process.argv[2] ?? crypto.randomUUID();
console.log(`Conversation ${conversationId}`);

const client = createClient<typeof registry>();
const conn = client.chatAgent.getOrCreate([conversationId]).connect();

let view: ConversationView | undefined;
// Operations that arrive before the watch call returns.
const early: Op[] = [];
// What is already printed.
let printed = "";
const shownTools = new Set<string>();

conn.on("pi.view", (frame) => {
	// RivetKit types event payloads as unknown, so name the frame's type.
	const { value, ops = [] } = frame as PiViewFrame;
	// A value replaces everything, such as when the agent restarts after a crash.
	if (value) {
		console.log("\n[resumed]");
		view = value;
	} else if (view) view = applyImmutable(view, ops);
	else early.push(...ops);
	if (view) show(view);
});

function show(current: ConversationView) {
	const live = current.docs["pi.live"] as LiveState;
	for (const tool of live.tools ?? []) {
		if (shownTools.has(tool.callId)) continue;
		shownTools.add(tool.callId);
		console.log(`\n> ${tool.name}`);
	}
	// The answer in progress, or the newest entry once it is finished.
	const last = current.entries.at(-1)?.model?.[0];
	const answer = live.generation?.message ?? (last?.role === "assistant" ? last : undefined);
	const text = answer ? contentText(answer.content, "") : "";
	if (!text.startsWith(printed)) printed = "";
	process.stdout.write(text.slice(printed.length));
	printed = text;
}

// Prints the newest messages and the answer in progress, then streams the rest.
const root = await conn.harness.root();
const { value } = await conn.conversation.watch(root.id);
view = applyImmutable(value, early);
for (const entry of view.entries) {
	const message = entry.model?.[0];
	if (message?.role === "user") console.log(`you: ${contentText(message.content, "")}`);
	if (message?.role === "assistant") {
		printed = contentText(message.content, "");
		if (printed) console.log(`agent: ${printed}`);
	}
}
show(view);

// Each line you type goes to the agent. submit saves it and returns without waiting for the answer.
for await (const text of createInterface({ input: process.stdin })) {
	// One id per message. Resending the message with the same id doesn't run the agent twice.
	const requestId = crypto.randomUUID();
	await conn.submit(text, { requestId });
}

Frames go only to a connection that called conversation.watch. Each pi.view frame carries Chord operations, and applyImmutable applies them to the view. See Stream a conversation.
The pi.live document in the view holds the answer the model is writing and the tools that are running. Once the answer is finished, it is the newest entry, so the client prints whatever text it has not printed yet.
When the agent restarts while the client is connected, it sends a frame with a new value on the same connection. The client prints the answer in progress from it and keeps streaming.
submit saves the message and returns without waiting for the answer. The answer arrives in the view, so a slow answer never holds a call open.
Each message gets its own requestId. If the call is retried with the same id, such as after a network error, it reaches the same submission and the agent doesn’t answer twice.
A message sent while the agent is still answering waits as a follow-up and runs next.

This client reads your Rivet credentials from the environment, so it can reach any conversation. A browser should connect with a short-lived token that your backend issues for that user’s own agent. See Client SDK.

Pick up after a reload

When the client starts, conversation.watch returns the view of the conversation before any frames arrive:

// Prints the newest messages and the answer in progress, then streams the rest.
const root = await conn.harness.root();
const { value } = await conn.conversation.watch(root.id);
view = applyImmutable(value, early);
for (const entry of view.entries) {
	const message = entry.model?.[0];
	if (message?.role === "user") console.log(`you: ${contentText(message.content, "")}`);
	if (message?.role === "assistant") {
		printed = contentText(message.content, "");
		if (printed) console.log(`agent: ${printed}`);
	}
}
show(view);

view.entries holds the newest messages, so a reload shows them without a separate history call. Read older ones with conversation.entries.
The pi.live document’s generation.message is the answer the model is writing right now. The client prints it, and the next frames continue where it stopped.
The run never depended on the client. Closing the tab, or the worker restarting, doesn’t stop the answer.

Try it

Set your model key and start the worker:

export ANTHROPIC_API_KEY=sk-ant-...
npx tsx server.ts

In another terminal, start the client with a conversation id:

npx tsx client.ts trip-1

Type “Find me a flight from SFO to HND on December 10, then book the cheaper one for Ada Lovelace.” The agent saves a plan, searches, and asks you to confirm. Reply Yes, book it. Then test a crash:

While the agent is writing an answer, kill the worker with pkill -9 -f server.ts.
Start it again with npx tsx server.ts. The agent wakes from SQLite and carries on with the same run. The model call that was cut off is sent again after the retry backoff, and [resumed] appears in the client.
Stop the client and run npx tsx client.ts trip-1 again. It prints the conversation, including the booking, which was made once.

See Resuming a run for what happens to each kind of step.

Next steps

Multi-Agent Systems: split a job across specialist agents that each survive restarts.
Human in the Loop: wait for a person to approve the booking.
Schedules: check the flight’s status every morning without being asked.
Sandboxes: give the agent a computer to run code and browse.
React SDK: build the chat client in React.

Architecture: what survives sleep, crashes, and upgrades.
Pi: documents, tasks, and every action the agent has.
Custom Tools: how tools run and what replay does.
Stream a conversation: every event in the stream.
Client SDK: connections, views, and browser tokens.