Cloudflare.Workflows reference

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

Workflow: Defining a Workflow

Minimal workflow

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

Preserving an existing Workflow name

An existing Workflow has no ownership marker, so the first deployment must opt in with --adopt. Keep fixed names unique per independently managed stack and stage; unlike the default, they are not stage-scoped by Alchemy.

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

Setting a step limit

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

Workflow: Scheduling instances

Each matching cron expression creates a new instance automatically — no Worker Cron Trigger or scheduled handler required.

export default class HourlyWorkflow extends Cloudflare.Workflow<HourlyWorkflow>()(
  "HourlyWorkflow",
  { schedules: ["0 * * * *"] },
  Effect.gen(function* () {
    return Effect.fn(function* () {
      const event = yield* Cloudflare.Workflows.WorkflowEvent;
      if (event.schedule) {
        return { cron: event.schedule.cron };
      }
      return {};
    });
  }),
) {}

Workflow: 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;
  });
});

Workflow: 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" } });

Workflow: 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 });
  }),
};

Workflow: 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",
      workflowName: "my-existing-workflow",
    }),
  },
});

Native cron schedules on an async Workflow binding

export const Worker = Cloudflare.Worker("Worker", {
  main: "./src/worker.ts",
  env: {
    HOURLY: Cloudflare.Workflow("HourlyWorkflow", {
      className: "HourlyWorkflow",
      schedules: ["0 * * * *"],
    }),
  },
});

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

Workflow: Consuming the Workflow’s Physical Name

An external/async Worker declaration exposes each Workflow binding on worker.env with the Workflow’s account-global workflowName as an Output of the current deploy. Pass the binding directly as a Queue subscription’s source; it deploys after the Workflow is registered, including on the first deployment. Explicit source descriptors remain supported for Workflows referenced by physical name.

const worker = yield* Cloudflare.Worker("Worker", {
  main: "./src/worker.ts",
  env: {
    FILE_URL_INGESTION: Cloudflare.Workflow("FileUrlIngestion", {
      className: "FileUrlIngestionWorkflow",
    }),
  },
});

yield* Cloudflare.Queues.Subscription("WorkflowEvents", {
  source: worker.env.FILE_URL_INGESTION,
  events: ["instance.completed", "instance.errored"],
  queueId: queue.queueId,
});

Workflow: Referencing a Deployed Workflow

Workflow.ref reads the same persisted resource as WorkflowResource.ref. Use its logical ID (including any namespace), not its physical name or Worker env key. Omitting stack and stage uses the current stack/stage. Deploy the host first; a reference does not register or own the Workflow and returns resource attributes, not a runtime WorkflowHandle.

yield* Cloudflare.Queues.Subscription("WorkflowEvents", {
  source: yield* Cloudflare.Workflow.ref("Ingestion", {
    stack: "workflow-host",
    stage: "production",
  }),
  events: ["instance.completed", "instance.errored"],
  queueId: queue.queueId,
});

Workflow: 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,
      // Repeat the host's explicit or preserved physical name when it
      // does not use Alchemy's current generated default.
      workflowName: "my-existing-workflow",
    }),
  },
});

Workflow: 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 },
);

Workflow: Observing and Deleting Instances

const instance = yield* workflow.get("report-123");
const events = yield* instance.subscribe({ filter: ["workflow_queued"] }).pipe(
  Stream.take(1),
  Stream.runCollect,
);
yield* instance.delete();
const result = yield* workflow.deleteBatch(["report-456", "report-789"]);
// result.deleted contains successful IDs; result.errors contains per-instance failures.