zopu-code/.agents/skills/ai-agent-workspace/reference/agent-os/workflows.md

Workflow Automation

Source: src/content/docs/agent-os/workflows.mdx Canonical URL: https://rivet.dev/docs/agent-os/workflows Description: Orchestrate multi-step agent tasks with durable workflows.

---

• Durable workflows that survive crashes and restarts
• Multi-step orchestration with sessions, file operations, and process execution
• Error handling and retry via ctx.step() for each operation
• Agent chaining where output of one session feeds into the next

Basic workflow

Use the actor workflow() primitive to orchestrate a multi-step agent task. Each step is durable and will resume from where it left off after a restart.

Session creation and prompting must happen within the same step because sessions are ephemeral and won't survive a replay.

import { agentOs } from "rivetkit/agent-os";
import common from "@rivet-dev/agent-os-common";
import pi from "@rivet-dev/agent-os-pi";
import { actor, setup, workflow } from "rivetkit";

const automator = actor({
  workflows: {
    fixBug: workflow<{ repo: string; issue: string }>(),
  },
  run: async (c) => {
    for await (const message of c.workflow.iter("fixBug")) {
      const { repo, issue } = message.body;
      const agentHandle = c.actors.vm.getOrCreate([`fix-${issue}`]);

      // Step 1: Clone the repo
      await c.step("clone-repo", async () => {
        return agentHandle.exec(`git clone ${repo} /home/user/repo`);
      });

      // Step 2: Agent fixes the bug (session lives within this step)
      await c.step("fix-bug", async () => {
        const session = await agentHandle.createSession("pi", {
          env: { ANTHROPIC_API_KEY: process.env.ANTHROPIC_API_KEY! },
        });
        await agentHandle.sendPrompt(
          session.sessionId,
          `Fix the bug described in issue: ${issue}`,
        );
        await agentHandle.closeSession(session.sessionId);
      });

      // Step 3: Run tests
      const tests = await c.step("run-tests", async () => {
        return agentHandle.exec("cd /home/user/repo && npm test");
      });

      console.log("Tests exit code:", tests.exitCode);
      await message.complete();
    }
  },
});

const vm = agentOs({
  options: { software: [common, pi] },
});

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

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

const client = createClient<typeof registry>("http://localhost:6420");
const handle = client.automator.getOrCreate(["main"]);

// Trigger the workflow
await handle.send("fixBug", {
  repo: "https://github.com/example/repo.git",
  issue: "Fix the login redirect bug",
});

Agent chaining

Output of one agent session feeds into the next. Each session is created and completed within its own step.

import { agentOs } from "rivetkit/agent-os";
import common from "@rivet-dev/agent-os-common";
import pi from "@rivet-dev/agent-os-pi";
import { actor, setup, workflow } from "rivetkit";

const pipeline = actor({
  workflows: {
    codeReview: workflow<{ filePath: string }>(),
  },
  run: async (c) => {
    for await (const message of c.workflow.iter("codeReview")) {
      const agentHandle = c.actors.vm.getOrCreate([`review-${Date.now()}`]);

      // Step 1: Agent reviews code and writes findings to a file
      await c.step("review", async () => {
        const session = await agentHandle.createSession("pi", {
          env: { ANTHROPIC_API_KEY: process.env.ANTHROPIC_API_KEY! },
        });
        await agentHandle.sendPrompt(
          session.sessionId,
          `Review the code at ${message.body.filePath} and write your findings to /home/user/review.md`,
        );
        await agentHandle.closeSession(session.sessionId);
      });

      // Step 2: Read the review from the filesystem
      const review = await c.step("read-review", async () => {
        const content = await agentHandle.readFile("/home/user/review.md");
        return new TextDecoder().decode(content);
      });

      // Step 3: Second session applies fixes based on the review
      await c.step("fix", async () => {
        const session = await agentHandle.createSession("pi", {
          env: { ANTHROPIC_API_KEY: process.env.ANTHROPIC_API_KEY! },
        });
        await agentHandle.sendPrompt(
          session.sessionId,
          `Apply the following review feedback:\n\n${review}`,
        );
        await agentHandle.closeSession(session.sessionId);
      });

      await message.complete();
    }
  },
});

const vm = agentOs({
  options: { software: [common, pi] },
});

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

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

const client = createClient<typeof registry>("http://localhost:6420");
const handle = client.pipeline.getOrCreate(["main"]);

await handle.send("codeReview", { filePath: "/home/user/src/auth.ts" });

Recommendations

• Create and close sessions within the same step. Sessions are ephemeral and won't exist after a workflow replays.
• Pass data between steps via the filesystem or step return values, not session state.
• Keep step names stable across code changes. Renaming a step breaks replay for in-progress workflows.
• Use separate actors for the workflow orchestrator and the agentOS VM.
• See Workflows for the full workflow API reference including timers, joins, and races.

Source doc path: /docs/agent-os/workflows