Skip to main content
Sign In
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
Chat clientterminal or browserchatAgenthistory and planLLM providerAnthropicYour APIflights and bookingswatchEvents1submit2model calls3tool calls4pi.events stream5Saves each step to SQLite.Sleeps when idle and wakes from SQLite.One per conversationAgentCallReply

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 nothing. 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 rivetkit

Create the agent

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

import { Type } from "@earendil-works/pi-ai";
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: [] }),
});

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: Type.Object({
		steps: Type.Array(Type.Object({ text: Type.String(), done: Type.Boolean() })),
	}),
	// 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 { content: [{ 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 { Type } from "@earendil-works/pi-ai";
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: [] }),
});

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: Type.Object({
		steps: Type.Array(Type.Object({ text: Type.String(), done: Type.Boolean() })),
	}),
	// 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 { content: [{ 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 { Type } from "@earendil-works/pi-ai";
import { defineTool } from "@earendil-works/pi-durable";

export const searchFlights = defineTool({
	name: "search_flights",
	description: "Find flights between two airports on a date.",
	parameters: Type.Object({
		from: Type.String({ description: "Airport code, such as SFO." }),
		to: Type.String({ description: "Airport code, such as HND." }),
		date: Type.String({ 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 { content: [{ 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: Type.Object({ flightId: Type.String(), passenger: Type.String() }),
	// 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 { content: [{ 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

Register the agent and start the worker:

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 each event, and sends each message with submit:

import { createInterface } from "node:readline";
import { type AssistantMessage, contentText } from "@earendil-works/pi-ai";
import type { SnapshotEvent } from "@earendil-works/pi-durable";
import type { PiEventsFrame } 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();

// The text of the current answer printed so far.
let printed = "";

conn.on("pi.events", (frame) => {
	// RivetKit types event payloads as unknown, so name the frame's type.
	for (const event of (frame as PiEventsFrame).events) {
		// The agent restarted, such as after a crash, and sent a fresh snapshot.
		if (event.type === "snapshot") {
			console.log("\n[resumed]");
			showPartialAnswer(event);
		}
		if (event.type === "message_update") {
			for (const change of event.changes) {
				if (change.type === "text_delta") {
					process.stdout.write(change.delta);
					printed += change.delta;
				}
			}
		}
		if (event.type === "message_end") {
			// A short answer can arrive whole here, with no text_delta before it.
			const message = event.entry.model?.[0];
			if (message?.role === "assistant") process.stdout.write(textOf(message).slice(printed.length));
			printed = "";
		}
		if (event.type === "tool_execution_start") console.log(`\n> ${event.toolName}`);
		if (event.type === "run_end") console.log("\n");
	}
});

// Prints the conversation so far and the answer in progress, then streams the rest.
const root = await conn.harness.root();
const { snapshot } = await conn.conversation.watchEvents(root.id);
for (const entry of snapshot.entries) {
	const message = entry.model?.[0];
	if (message?.role === "user") console.log(`you: ${contentText(message.content, "")}`);
	if (message?.role === "assistant" && textOf(message)) console.log(`agent: ${textOf(message)}`);
}
showPartialAnswer(snapshot);

function showPartialAnswer(snapshot: SnapshotEvent) {
	const partial = snapshot.generation?.message;
	printed = partial ? textOf(partial) : "";
	process.stdout.write(printed);
}

// 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 });
}

function textOf(message: AssistantMessage) {
	return contentText(message.content, "");
}
  • Events go only to a connection that called conversation.watchEvents. Each pi.events frame carries a batch of events, such as text_delta changes while the model writes and tool_execution_start when a tool starts. See Session Lifecycle.
  • A short answer can arrive whole in message_end, with no text_delta before it, so the client prints the rest of the text there.
  • When the agent restarts while the client is connected, it sends a snapshot event on the same stream. 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 as events, 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, watchEvents returns a snapshot of the conversation before any events arrive:

// Prints the conversation so far and the answer in progress, then streams the rest.
const root = await conn.harness.root();
const { snapshot } = await conn.conversation.watchEvents(root.id);
for (const entry of snapshot.entries) {
	const message = entry.model?.[0];
	if (message?.role === "user") console.log(`you: ${contentText(message.content, "")}`);
	if (message?.role === "assistant" && textOf(message)) console.log(`agent: ${textOf(message)}`);
}
showPartialAnswer(snapshot);

function showPartialAnswer(snapshot: SnapshotEvent) {
	const partial = snapshot.generation?.message;
	printed = partial ? textOf(partial) : "";
	process.stdout.write(printed);
}
  • snapshot.entries holds every message so far, so a reload shows the whole conversation without a separate history call.
  • snapshot.generation.message is the answer the model is writing right now. The client prints it, sets printed to it, and the next text_delta continues 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:

  1. While the agent is writing an answer, kill the worker with pkill -9 -f server.ts.
  2. 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, and [resumed] appears in the client.
  3. Stop the client and run npx tsx client.ts trip-1 again. It prints the whole 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.
  • Session Lifecycle: every event in the stream.
  • Client SDK: connections, snapshots, and browser tokens.
Edit this page