Orchestration

Workflow Automation

Orchestrate multi-step agent tasks with durable workflows.

Orchestrate multi-step agent tasks with durable workflows that survive crashes and restarts. Build them with RivetKit’s workflow() run handler, where each ctx.step() is recorded, retried, and resumed independently, and the output of one step can feed into the next.

Basic workflow

A workflow is the durable run handler of an actor. Wrap it in workflow() and drive a multi-step agent task as an ordered series of steps: clone the repo, let an agent fix the bug, then run the tests. Each actor instance is one workflow run initialized with creation input, so no application queue is required.

Session creation and prompting happen within the step that uses them, so a session never has to outlive the work it backs (sessions are ephemeral and would not survive a replay). Steps reach the agentOS VM, a separate actor, through ctx.client().

import { agentOS, setup } from "@rivet-dev/agentos";
import pi from "@agentos-software/pi";
import { actor } from "rivetkit";
import { type WorkflowStepContextOf, workflow } from "rivetkit/workflow";

const vm = agentOS({ software: [pi] });

interface BugFixInput {
	repo: string;
	issue: string;
}

// Each actor instance is one durable workflow run. Its creation input is stored
// in actor state, so no queue is needed to trigger or order agent prompts.
const bugFixer = actor({
	state: {
		repo: "",
		issue: "",
		status: "running" as "running" | "complete",
		exitCode: null as number | null,
	},
	onCreate: (c, input: BugFixInput) => {
		c.state.repo = input.repo;
		c.state.issue = input.issue;
	},
	run: workflow(async (ctx) => {
		await ctx.step("clone-repo", (step) => cloneRepo(step, step.state.repo));
		await ctx.step("fix-bug", (step) =>
			fixBugWithAgent(step, step.state.issue),
		);
		const exitCode = await ctx.step("run-tests", (step) => runTests(step));
		await ctx.step("record-result", async (step) => {
			step.state.exitCode = exitCode;
			step.state.status = "complete";
		});
	}),
	actions: {
		getState: (c) => c.state,
	},
});

async function cloneRepo(
	step: WorkflowStepContextOf<typeof bugFixer>,
	repo: string,
): Promise<void> {
	const agent = step.client<typeof registry>().vm.getOrCreate("bug-fixer");
	await agent.exec(`git clone ${repo} /home/agentos/repo`);
}

async function fixBugWithAgent(
	step: WorkflowStepContextOf<typeof bugFixer>,
	issue: string,
): Promise<void> {
	const agent = step.client<typeof registry>().vm.getOrCreate("bug-fixer");
	await agent.openSession({
		agent: "pi",
		env: { ANTHROPIC_API_KEY: process.env.ANTHROPIC_API_KEY! },
	});
	await agent.prompt({
		content: [{ type: "text", text: `Fix the bug described in issue: ${issue}` }],
	});
}

async function runTests(
	step: WorkflowStepContextOf<typeof bugFixer>,
): Promise<number> {
	const agent = step.client<typeof registry>().vm.getOrCreate("bug-fixer");
	const tests = await agent.exec("cd /home/agentos/repo && npm test");
	return tests.exitCode;
}

import { randomUUID } from "node:crypto";
import { createClient } from "@rivet-dev/agentos/client";
import type { registry } from "./server";

const client = createClient<typeof registry>({ endpoint: "http://localhost:6420" });

// Creating one actor starts one durable workflow with immutable input.
const handle = await client.bugFixer.create(randomUUID(), {
	input: {
		repo: "https://github.com/example/repo.git",
		issue: "Fix the login redirect bug",
	},
});

let state = await handle.getState();
while (state.status !== "complete") {
	await new Promise((resolve) => setTimeout(resolve, 1_000));
	state = await handle.getState();
}
console.log("Exit code:", state.exitCode);

Agent chaining

Output of one agent session feeds into the next. Each session is created and completed within its own step, and data passes between steps through the VM filesystem (a review file) and step return values.

interface CodeReviewInput {
	filePath: string;
}

const codeReviewer = actor({
	state: {
		filePath: "",
		status: "running" as "running" | "complete",
	},
	onCreate: (c, input: CodeReviewInput) => {
		c.state.filePath = input.filePath;
	},
	run: workflow(async (ctx) => {
		await ctx.step("review", (step) =>
			reviewCode(step, step.state.filePath),
		);
		const review = await ctx.step("read-review", (step) => readReview(step));
		await ctx.step("fix", (step) => applyReview(step, review));
		await ctx.step("record-review", async (step) => {
			step.state.status = "complete";
		});
	}),
	actions: {
		getState: (c) => c.state,
	},
});

async function reviewCode(
	step: WorkflowStepContextOf<typeof codeReviewer>,
	filePath: string,
): Promise<void> {
	const agent = step.client<typeof registry>().vm.getOrCreate("reviewer");
	await agent.openSession({
		agent: "pi",
		env: { ANTHROPIC_API_KEY: process.env.ANTHROPIC_API_KEY! },
	});
	await agent.prompt({
		content: [
			{
				type: "text",
				text: `Review ${filePath} and write findings to /home/agentos/review.md`,
			},
		],
	});
}

async function readReview(
	step: WorkflowStepContextOf<typeof codeReviewer>,
): Promise<string> {
	const agent = step.client<typeof registry>().vm.getOrCreate("reviewer");
	return new TextDecoder().decode(
		await agent.readFile("/home/agentos/review.md"),
	);
}

async function applyReview(
	step: WorkflowStepContextOf<typeof codeReviewer>,
	review: string,
): Promise<void> {
	const agent = step.client<typeof registry>().vm.getOrCreate("reviewer");
	await agent.prompt({
		content: [
			{ type: "text", text: `Apply the following review feedback:\n\n${review}` },
		],
	});
}

export const registry = setup({ use: { vm, bugFixer, codeReviewer } });
registry.start();

import { randomUUID } from "node:crypto";
import { createClient } from "@rivet-dev/agentos/client";
import type { registry } from "./server";

const client = createClient<typeof registry>({ endpoint: "http://localhost:6420" });
const handle = await client.codeReviewer.create(randomUUID(), {
	input: { filePath: "/home/agentos/src/auth.ts" },
});

let state = await handle.getState();
while (state.status !== "complete") {
	await new Promise((resolve) => setTimeout(resolve, 1_000));
	state = await handle.getState();
}

Recommendations

Build the actor’s run handler with workflow() so each ctx.step() is durable: recorded, retried, and resumed independently across crashes and restarts.
Keep step names stable across code changes. Renaming a step breaks replay for in-progress workflows.
Create and close sessions within the step that uses them. Sessions are ephemeral, so keep their lifetime scoped to one unit of work.
Pass data between steps via the filesystem or step return values, not session state.
Keep state changes and other actor-local side effects inside ctx.step() callbacks; use non-step workflow code only for orchestration.
Reach the agentOS VM, a separate actor, from inside a step with ctx.client().
See Workflows for the full workflow API reference including timers, joins, and races.