Cloudflare.Workflows reference

layerDurableObjectEngine

Source: packages/cloudflare/src/Workflows/DurableObjectEngine.ts Kind: Layer · Provides: WorkflowEngine

Runs Effect workflows on Durable Objects. Each workflow gets its own Durable Object class on the host Worker, and each execution is one instance of it, named by its execution id, with its journal in that instance’s SQLite.

A run replays the workflow from the top: finished activities return their stored results, and the first unfinished one runs and is stored. A run that waits on a DurableClock or DurableDeferred that isn’t ready suspends, and the clock firing or the deferred being completed runs it again. Every run happens in one durable callback per execution, so runs never overlap, and a run that fails or is cut short is retried.

layerDurableObjectEngine: Running workflows on Durable Objects

import * as Cloudflare from "@alchemy.run/cloudflare";
import * as Workflows from "alchemy/Workflows";
import * as Effect from "effect/Effect";
import * as Layer from "effect/Layer";
import * as HttpServerResponse from "effect/http/HttpServerResponse";
import * as Schema from "effect/Schema";
import { Activity, Workflow } from "effect/workflow";

const SendWelcome = Workflow.make("SendWelcome", {
  payload: { userId: Schema.String },
  idempotencyKey: ({ userId }) => userId,
});

const SendWelcomeLive = Workflows.layer(
  SendWelcome,
  Effect.fn(function* ({ userId }) {
    yield* Activity.make({ name: "send", execute: Effect.log(`welcome, ${userId}`) });
  }),
);

export default class AppWorker extends Cloudflare.Worker<AppWorker>()(
  "AppWorker",
  { main: import.meta.url },
  Effect.gen(function* () {
    const welcome = yield* Workflows.RunWorkflow(SendWelcome);
    return {
      fetch: Effect.gen(function* () {
        const run = yield* welcome.start({ userId: "u-1" });
        return HttpServerResponse.text(run.executionId);
      }),
    };
  }).pipe(
    Effect.provide(
      SendWelcomeLive.pipe(
        Layer.provideMerge(Cloudflare.Workflows.layerDurableObjectEngine()),
      ),
    ),
  ),
) {}

layerDurableObjectEngine: Running executions in a jurisdiction

import * as Cloudflare from "@alchemy.run/cloudflare";
import * as Layer from "effect/Layer";
import { SendWelcomeLive } from "./SendWelcome.ts";

const WorkflowsLive = SendWelcomeLive.pipe(
  Layer.provideMerge(Cloudflare.Workflows.layerDurableObjectEngine({ jurisdiction: "eu" })),
);

NativeEngine

Source: packages/cloudflare/src/Workflows/NativeEngine.ts Kind: Layer · Provides: WorkflowEngine

Run Effect workflows on Cloudflare Workflows. One Cloudflare Workflow class, NativeEngine, runs every workflow the Worker implements; each execution is one instance, named after the workflow and its execution id.

Steps run with step.do, so a finished step is replayed rather than run again. DurableClock.sleep is step.sleep, and DurableDeferred.await waits for an event that a run’s signal sends.

NativeEngine: Running workflows on Cloudflare Workflows

import * as Cloudflare from "@alchemy.run/cloudflare";
import * as Workflows from "alchemy/Workflows";
import * as Effect from "effect/Effect";
import * as Layer from "effect/Layer";
import * as HttpServerResponse from "effect/http/HttpServerResponse";

export default class AppWorker extends Cloudflare.Worker<AppWorker>()(
  "AppWorker",
  { main: import.meta.url },
  Effect.gen(function* () {
    const welcome = yield* Workflows.RunWorkflow(SendWelcome);
    return {
      fetch: Effect.gen(function* () {
        const run = yield* welcome.start({ userId: "u-1" });
        return HttpServerResponse.text(run.executionId);
      }),
    };
  }).pipe(
    Effect.provide(SendWelcomeLive.pipe(Layer.provideMerge(Cloudflare.Workflows.NativeEngine))),
  ),
) {}

Workflow

Source: packages/cloudflare/src/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: Restricting data location

jurisdiction keeps every instance’s execution and stored state inside the EU, the US, or FedRAMP-Moderate data centers. Cloudflare fixes it at creation, so changing it replaces the Workflow and deletes its instances.

EU-only Workflow

export default class GdprWorkflow extends Cloudflare.Workflow<GdprWorkflow>()(
  "GdprWorkflow",
  { jurisdiction: "eu" },
  Effect.gen(function* () {
    return Effect.fn(function* (input: { userId: string }) {
      return { processed: input.userId };
    });
  }),
) {}

EU-only Workflow on an async Worker

export const Worker = Cloudflare.Worker("Worker", {
  main: "./src/worker.ts",
  env: {
    GDPR_WORKFLOW: Cloudflare.Workflow("GdprWorkflow", {
      className: "GdprWorkflow",
      jurisdiction: "eu",
    }),
  },
});

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: Looking up an existing Workflow

Workflow.lookup binds a Workflow this stack doesn’t manage (hosted by another stack’s Worker, or created outside Alchemy), found by its account-global workflowName. Its class and host script are read from Cloudflare, and Params types the instance payload as on the form above. It is read during plan/deploy and never created, updated or deleted.

const consumer = yield* Cloudflare.Worker("Consumer", {
  main: "./src/consumer.ts",
  env: {
    MY_WORKFLOW: Cloudflare.Workflow.lookup<{ value: string }>({
      workflowName: "my-existing-workflow",
    }),
  },
});
type Env = Cloudflare.InferEnv<typeof consumer>; // MY_WORKFLOW: Workflow<{ value: string }>

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.