Agent Swarms

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

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

export const agent = pi({
	model: "anthropic/claude-opus-5-5",
	registry: createRegistry(),
	actions: {
		discard: (c) => {
			c.destroy();
		},
	},
});

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

registry.start();

import { readFile } from "node:fs/promises";
import { ActorError, createClient } from "rivetkit/client";
import type { registry } from "./server";

const client = createClient<typeof registry>();

// How many forks may be alive at once. Each fork is a running agent.
const MAX_FORKS = 10;

const files = process.argv.slice(2);
const agent = client.agent.getOrCreate(["code-review"]);
await agent.prompt(
	"You review TypeScript for our team. We care about error handling and clear names.",
	{ requestId: crypto.randomUUID() },
);

// A turn in progress would be copied into every fork, so fork an idle agent.
await agent.harness.waitForIdle();
const agentId = await agent.resolve();

// One key per fork makes a retried fork safe: it can never make a second copy.
const runId = crypto.randomUUID();

const reports = await mapWithLimit(files, MAX_FORKS, async (path) => {
	const fork = await forkWithRetry(agentId, ["review", runId, path]);
	try {
		const source = await readFile(path, "utf8");
		const result = await fork.prompt(
			`Review ${path} in one line.\n\n${source}`,
			{
				requestId: `review:${runId}:${path}`,
			},
		);
		const line =
			result.status === "done"
				? result.text?.split("\n")[0]
				: result.reason;
		return `${path}: ${line ?? "no review"}`;
	} finally {
		await fork.discard();
	}
});

// The summary is a normal turn, so it lands in the agent's conversation.
const summary = await agent.prompt(
	`One-line reviews of every file:\n${reports.join("\n")}\n\nSummarize the biggest risks.`,
	{ requestId: crypto.randomUUID() },
);
console.log(summary.status === "done" ? summary.text : summary.reason);

/** Forks the agent, waiting and retrying when storage asks the caller to slow down. */
async function forkWithRetry(sourceId: string, key: string[]) {
	for (let attempt = 1; ; attempt++) {
		try {
			return await client.agent.fork(sourceId, { key });
		} catch (error) {
			if (!(error instanceof ActorError)) throw error;

			// The fork exists: an earlier attempt made it, or its copy is still running.
			if (
				error.group === "actor" &&
				(error.code === "duplicate_key" ||
					error.code === "fork_pending")
			) {
				return client.agent.get(key);
			}

			// Refused before any work. Wait as long as asked, with jitter.
			const overloaded =
				error.group === "depot" && error.code === "overloaded";
			if (!overloaded || attempt >= 5) throw error;
			const metadata = error.metadata as
				| { retry_after_ms?: number }
				| undefined;
			const waitMs =
				(metadata?.retry_after_ms ?? 200) * (1 + Math.random());
			await new Promise((resolve) => setTimeout(resolve, waitMs));
		}
	}
}

/** Runs `fn` on every item with at most `limit` running at once, keeping order. */
async function mapWithLimit<T, R>(
	items: T[],
	limit: number,
	fn: (item: T) => Promise<R>,
): Promise<R[]> {
	const results: R[] = new Array(items.length);
	let next = 0;
	const workers = Array.from(
		{ length: Math.min(limit, items.length) },
		async () => {
			while (next < items.length) {
				const index = next++;
				results[index] = await fn(items[index]);
			}
		},
	);
	await Promise.all(workers);
	return results;
}

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

npx tsx swarm.ts src/*.ts