Workflow

Source: src/Cloudflare/Workflows/Workflow.ts

A Cloudflare Workflow that orchestrates durable, multi-step tasks with automatic retries and at-least-once delivery.

A Workflow follows the same two-phase pattern as Workers and Durable Objects. The outer Effect.gen resolves shared dependencies. The inner Effect.fn is the workflow body — a function from a typed input payload to an Effect that runs steps using task, sleep, and sleepUntil. task takes the step name and Effect, plus an optional config object for retries, timeout, and a rollback handler.

Effect.gen(function* () {
  // Phase 1: resolve dependencies
  const notifier = yield* NotificationService;

  return Effect.fn(function* (input: { orderId: string }) {
    // Phase 2: workflow body (durable steps)
    const result = yield* Cloudflare.Workflows.task("process", doWork(input.orderId));
    yield* Cloudflare.Workflows.sleep("cooldown", "10 seconds");
    return result;
  });
})

Defining a Workflow

export default class MyWorkflow extends Cloudflare.Workflow<MyWorkflow>()(
  "MyWorkflow",
  Effect.gen(function* () {
    return Effect.fn(function* (input: { name: string }) {
      return { received: input.name };
    });
  }),
) {}

Step Primitives

Running a named task

const result = yield* Cloudflare.Workflows.task(
  "process-order",
  Effect.succeed({ orderId: "abc", total: 42 }),
);

Configuring retries and reading step context

const result = yield* Cloudflare.Workflows.task(
  "call-api",
  Effect.gen(function* () {
    const context = yield* Cloudflare.Workflows.WorkflowStepContext;
    return { attempt: context.attempt };
  }),
  { retries: { limit: 3, delay: "5 seconds", backoff: "linear" } },
);

Registering rollback

yield* Cloudflare.Workflows.task("reserve-inventory", reserveInventory, {
  rollback: ({ output }) =>
    output ? releaseInventory(output.reservationId) : Effect.void,
  rollbackConfig: { retries: { limit: 3, delay: "10 seconds" } },
});

Sleeping between steps

yield* Cloudflare.Workflows.sleep("cooldown", "30 seconds");

Waiting for an external event

const event = yield* Cloudflare.Workflows.waitForEvent<{ approved: boolean }>(
  "approval",
  { type: "approval", timeout: "1 day" },
);
// Same shape as the native step.waitForEvent result:
event.payload.approved;

Accessing env bindings inside a task

Bind a resource (e.g. Namespace, Bucket) in the workflow’s outer init phase to get a typed Effect-native client, then use it directly inside task. task threads the binding’s service requirement (WorkerEnvironment) through automatically so the inner Effect needs no extra plumbing.

Effect.gen(function* () {
  const kv = yield* Cloudflare.KV.ReadWriteNamespace(KV);

  return Effect.fn(function* (input: { roomId: string; message: string }) {
    const { roomId, message } = input;

    const stored = yield* Cloudflare.Workflows.task(
      "kv-roundtrip",
      Effect.gen(function* () {
        const key = `workflow:${roomId}`;
        yield* kv.put(key, message);
        return yield* kv.get(key);
      }).pipe(Effect.orDie),
    );

    return stored;
  });
});

Starting and Monitoring Instances

create mirrors Cloudflare’s native Workflow API: pass workflow input in params, pass id only when you need a deterministic instance ID, and omit id to let Cloudflare generate one.

Creating an instance from a Worker

const workflow = yield* MyWorkflow;
const instance = yield* workflow.create({ params: { orderId: "abc" } });

Creating an instance with id and retention

const instance = yield* workflow.create({
  id: "order-abc",
  params: { orderId: "abc" },
  retention: { successRetention: "1 day", errorRetention: "7 days" },
});

Creating a batch

const instances = yield* workflow.createBatch([
  { id: "order-a", params: { orderId: "a" } },
  { id: "order-b", params: { orderId: "b" } },
]);

Checking instance status

const workflow = yield* MyWorkflow;
const handle = yield* workflow.get(instanceId);
const status = yield* handle.status();

Sending events and restarting instances

const instance = yield* workflow.get(instanceId);
yield* instance.sendEvent({ type: "approval", payload: { approved: true } });
yield* instance.restart({ from: { name: "approval", type: "waitForEvent" } });

Triggering from a Worker

Wire the workflow into HTTP routes so callers can fire instances and poll for completion.

src/worker.ts

const notifier = yield* MyWorkflow;

return {
  fetch: Effect.gen(function* () {
    const request = yield* HttpServerRequest;

    if (request.url.startsWith("/workflow/start/")) {
      const id = request.url.split("/").pop()!;
      const instance = yield* notifier.create({ params: { orderId: id } });
      return HttpServerResponse.json({ instanceId: instance.id });
    }

    if (request.url.startsWith("/workflow/status/")) {
      const id = request.url.split("/").pop()!;
      const instance = yield* notifier.get(id);
      return HttpServerResponse.json(yield* instance.status());
    }

    return HttpServerResponse.text("Not Found", { status: 404 });
  }),
};

Binding in an Async Worker

When using an Async Worker (plain async fetch handler, no Effect runtime), declare Workflows in the env prop of the Worker resource. Pass a Workflow reference with a className matching the exported WorkflowEntrypoint subclass in your worker source file. If className is omitted, it defaults to the binding name. Use Cloudflare.InferEnv to get a fully typed env object that includes the workflow binding.

Declaring a Workflow binding in the stack

alchemy.run.ts

export type WorkerEnv = Cloudflare.InferEnv<typeof Worker>;

export const Worker = Cloudflare.Worker("Worker", {
  main: "./src/worker.ts",
  env: {
    MY_WORKFLOW: Cloudflare.Workflow<{ value: string }>("MyWorkflow", {
      className: "MyWorkflow",
    }),
  },
});

Using the Workflow from a plain async handler

src/worker.ts

import { WorkflowEntrypoint, type WorkflowEvent, type WorkflowStep } from "cloudflare:workers";
import type { WorkerEnv } from "../alchemy.run.ts";

export class MyWorkflow extends WorkflowEntrypoint<WorkerEnv, { value: string }> {
  async run(event: Readonly<WorkflowEvent<{ value: string }>>, step: WorkflowStep) {
    return await step.do("greet", async () => `Hello, ${event.payload.value}!`);
  }
}

export default {
  async fetch(request: Request, env: WorkerEnv) {
    const instance = await env.MY_WORKFLOW.create({ params: { value: "world" } });
    return Response.json({ instanceId: instance.id });
  },
};

Cross-Script Binding in an Async Worker

Async Workers can also bind to a Workflow hosted by another Worker script. The host Worker declares and exports the WorkflowEntrypoint class. The consumer Worker declares a Workflow with scriptName set to the host Worker’s script name. Cross-script references are bindings only — Alchemy does not drive putWorkflow for the foreign class, so deploy the host first.

const consumer = yield* Cloudflare.Worker("Consumer", {
  main: "./src/consumer.ts",
  env: {
    MY_WORKFLOW: Cloudflare.Workflow("MyWorkflow", {
      className: "MyWorkflow",
      scriptName: host.workerName,
    }),
  },
});

Testing Workflows

Workflows run asynchronously, so tests start an instance and poll until it reaches a terminal status. Keep polling bounded with Effect.repeat.

test(
  "workflow completes",
  Effect.gen(function* () {
    const { url } = yield* stack;

    const start = yield* HttpClient.post(`${url}/workflow/start/x`);
    const { instanceId } = (yield* start.json) as { instanceId: string };

    const status = yield* HttpClient.get(
      `${url}/workflow/status/${instanceId}`,
    ).pipe(
      Effect.flatMap((res) => res.json),
      Effect.map((json) => json as { status: string }),
      Effect.repeat({
        schedule: Schedule.spaced("2 seconds"),
        until: (status) =>
          status.status === "complete" || status.status === "errored",
        times: 30,
      }),
    );

    expect(status.status).toBe("complete");
  }),
  { timeout: 120_000 },
);