effect-encore

name	effect-encore
description	Declarative actors and durable workflows for @effect/cluster. Use when building actor/entity definitions with effect-encore, wiring handlers, writing execute/send/peek/watch patterns, testing actors, defining workflows with the step DSL, or migrating from raw @effect/cluster Entity/Rpc/RpcGroup code.

effect-encore

Declarative actors and durable workflows for @effect/cluster. Unified call site for entities and workflows with a step DSL for durable orchestration.

Navigation

What are you working on?
├─ Defining an entity              → §Entity
├─ Defining a workflow             → §Workflow
├─ Identity / type guards          → §Type guards
├─ Wiring handlers / layers        → §Handle
├─ Calling / sending               → §Client
├─ Tracking execution (peek/watch) → §Peek
├─ Waiting for completion          → §WaitFor
├─ Testing                         → §Test
├─ Lifecycle (interrupt/resume/flush/redeliver) → §Lifecycle
├─ Delayed delivery (deliverAt)    → §DeliverAt
├─ Observability                   → §Observability
├─ v3 compatibility                → §v3
└─ Migrating from raw cluster      → §Migration

Core Concepts

• Unified call site. Entities and workflows share ref.execute(op) / ref.send(op). Callers don't care which is which.
• Delivery mode is the caller's choice. Every operation supports execute and send. The definition doesn't decide — the caller does.
• Value dispatch. Construct an operation value, pass it to ref.execute(op) or ref.send(op).
• One layer, two roles. Actor.toLayer(actor, handlers) = consumer + producer. Actor.toLayer(actor) = producer only.
• Step DSL. Workflow handlers receive (payload, step) — no raw Activity/DurableDeferred/DurableClock imports needed.
• Compiles to @effect/cluster. Not a new runtime. Actor.fromEntity produces an Entity, Actor.fromWorkflow wraps Workflow.make.

Entity

Multi-operation entity actor

const Counter = Actor.fromEntity("Counter", {
  Increment: {
    payload: { amount: Schema.Number },
    success: Schema.Number,
    id: (p) => String(p.amount),
  },
  GetCount: {
    success: Schema.Number,
    id: () => "singleton",
  },
});

// Constructors are on the object directly
Counter.Increment({ amount: 5 }); // → OperationValue
Counter.GetCount(); // zero-input, still callable

OperationDef fields

Field	Type	Required	Description
payload	Schema.Top | Schema.Struct.Fields	no	Inline fields or pre-built Schema.Class
success	Schema.Top	no	Success response schema (default: Void)
error	Schema.Top	no	Error schema (default: Never)
persisted	boolean	no	Persist to MessageStorage
id	(payload) => string | { entityId, primaryKey? }	yes	Routing and dedupe / exec ID key
deliverAt	(payload) => DateTime	no	Delayed delivery extractor

EntityActor properties

Property	Type	Description
Counter.name	"Counter"	Actor name (literal type)
Counter.type	"Counter"	Cluster entity type (same as name for entities)
Counter._tag	"EntityActor"	Discriminant for type guards
Counter.Increment(...)	Constructor	Returns OperationValue
Counter._meta.entity	Entity	Underlying cluster Entity
Counter._meta.definitions	Record	Raw operation definitions
Counter.Context	Context tag	DI tag for client factory
Counter.ref(id)	Method	yield* Counter.ref("id") → ActorRef
Counter.peek(execId)	Method	One-shot status check
Counter.watch(execId)	Method	Polling stream of status changes
Counter.waitFor(execId)	Method	Poll until terminal (or custom filter)
Counter.interrupt(id)	Method	Passivate entity (dies — not public API)
Counter.flush(id)	Method	Delete all messages + replies (entity-only)
Counter.redeliver(id)	Method	Clear read leases for reprocessing (entity-only)
Counter.of(handlers)	Identity fn	Type-safe handler construction (no redeclaring)
Counter.$is(tag)	Type guard	Counter.$is("Increment")(value)

Type guards

Actor.isEntity(actor); // narrows to EntityActor
Actor.isWorkflow(actor); // narrows to WorkflowActor

// .name — the actor's declared name
Counter.name; // "Counter"
ProcessOrder.name; // "ProcessOrder"

// .type — the cluster entity type (how it's registered with @effect/cluster)
Counter.type; // "Counter"
ProcessOrder.type; // "Workflow/ProcessOrder"

Reserved operation names

_tag, _meta, $is, Context, ref, name, type, of, peek, watch, interrupt, executionId, flush, redeliver — compile-time type guard + runtime check.

Pre-built Schema.Class payload

Escape hatch for custom symbol implementations. The id/deliverAt options in OperationDef are ignored when using a Schema.Class — symbols must be on the class.

class CustomPayload extends Schema.Class<CustomPayload>("CustomPayload")({
  id: Schema.String,
}) {
  [PrimaryKey.symbol]() {
    return this.id;
  }
}

const MyActor = Actor.fromEntity("MyActor", {
  Run: { payload: CustomPayload, success: Schema.Void, persisted: true, id: (p) => p.id },
});

Workflow

Workflow actor (single-op "Run")

const ProcessOrder = Actor.fromWorkflow("ProcessOrder", {
  payload: { orderId: Schema.String },
  success: OrderResult,
  error: OrderError,
  id: (p) => p.orderId,
});

// Single constructor — always "Run"
ProcessOrder.Run({ orderId: "ord-1" }); // → OperationValue

WorkflowDef fields

Field	Type	Required	Description
payload	Schema.Struct.Fields	yes	Workflow input fields
success	Schema.Top	no	Success schema (default: Void)
error	Schema.Top	no	Error schema (default: Never)
id	(payload) => string	yes	Deterministic execution ID
signals	SignalDefs	no	Declarative signal definitions (see below)
captureDefects	boolean	no	Capture defects as workflow failures
suspendOnFailure	boolean	no	Suspend workflow on failure instead of fail
suspendedRetrySchedule	Schedule	no	Retry schedule for suspended workflows

WorkflowActor properties

All entity properties plus:

Property	Type	Description
ProcessOrder.name	"ProcessOrder"	Workflow name
ProcessOrder.type	"Workflow/ProcessOrder"	Cluster entity type (prefixed)
ProcessOrder._tag	"WorkflowActor"	Discriminant for type guards
ProcessOrder.resume(execId)		Resume suspended workflow
ProcessOrder.executionId(payload)		Compute deterministic execution ID
ProcessOrder.Approval		Signal property (from signals on WorkflowDef)

Declarative signals

Signals are defined on WorkflowDef.signals and become typed properties on the actor — single source of truth:

const ProcessOrder = Actor.fromWorkflow("ProcessOrder", {
  payload: { orderId: Schema.String },
  success: OrderResult,
  id: (p) => p.orderId,
  signals: {
    Approval: { success: Schema.String, error: ApprovalError },
    Cancel: {}, // void signal — Schema.Void/Schema.Never defaults
  },
});

// Typed properties on the actor
ProcessOrder.Approval.token; // Effect<Token, never, WorkflowInstance>
ProcessOrder.Approval.await; // Effect<string, ApprovalError, ...>
ProcessOrder.Approval.succeed({ token, value }); // Effect<void, never, WorkflowEngine>
ProcessOrder.Approval.fail({ token, error }); // Effect<void, never, WorkflowEngine>
ProcessOrder.Approval.tokenFromExecutionId(id); // Token (pure)

// Inside the workflow handler — reference the actor's signals directly
Actor.toLayer(ProcessOrder, (payload, step) =>
  Effect.gen(function* () {
    const token = yield* ProcessOrder.Approval.token;
    yield* step.run("send-email", { do: sendEmail({ token }), success: Schema.Void });
    const decision = yield* ProcessOrder.Approval.await;
  }),
);

// External resolution
const token = ProcessOrder.Approval.tokenFromExecutionId(execId);
yield * ProcessOrder.Approval.succeed({ token, value: "approved" });

Signal names must not collide with reserved workflow properties (Run, ref, peek, etc.).

Step DSL

Workflow handlers receive (payload, step) — a WorkflowStepContext that wraps upstream Activity, DurableDeferred, and DurableClock. No direct upstream imports needed.

step.run — execute a durable activity

Always provide success and error schemas. Activities serialize results through JSON — explicit schemas ensure durable round-tripping and typed decode. The shorthand (step.run(id, effect)) uses Schema.Unknown internally, which accepts any JSON-safe value but loses type safety on decode. Use it for prototyping; prefer full options for production workflows.

// Full options (recommended) — explicit schemas for durable serialization
const result =
  yield *
  step.run("process", {
    do: processPayment(payload),
    success: PaymentResult,
    error: PaymentError,
    undo: (value, cause) => refundPayment(value.paymentId),
    retry: { times: 3 }, // or a Schedule
  });

// Shorthand — infallible, uses Schema.Unknown (quick & dirty, prototyping)
const result = yield * step.run("validate", validateOrder(payload.orderId));

// Shorthand with undo (saga compensation) — same Schema.Unknown caveat
const charge =
  yield *
  step.run("charge", chargeCard(payload.cardId, payload.amount), (chargeResult, cause) =>
    refundCard(chargeResult.chargeId),
  );

step.sleep — durable sleep

yield * step.sleep("cooldown", "30 minutes");
yield * step.sleep("delay", "5 seconds", { inMemoryThreshold: "2 seconds" });

step.race — first activity to complete

import { Activity } from "effect/unstable/workflow"; // only needed for race step types

const winner =
  yield *
  step.race("fastest", [
    Activity.make({ name: "a", success: Schema.String, execute: taskA }),
    Activity.make({ name: "b", success: Schema.String, execute: taskB }),
  ]);

step.executionId — current execution ID

const id = step.executionId; // string

step.attempt — current attempt number

const attempt = yield * step.attempt; // Effect<number>

step.suspend — suspend the workflow

yield * step.suspend; // Effect<never>

step.idempotencyKey — derived key

const key = yield * step.idempotencyKey("sub-step"); // "executionId/sub-step"
const keyWithAttempt = yield * step.idempotencyKey("sub-step", { includeAttempt: true });

step.scope / step.provideScope / step.addFinalizer

const scope = yield * step.scope;
yield * step.provideScope(effectNeedingScope);
yield * step.addFinalizer((exit) => cleanup(exit));

Full workflow handler example

const ProcessOrderLive = Actor.toLayer(ProcessOrder, (payload, step) =>
  Effect.gen(function* () {
    const validated = yield* step.run("validate", {
      do: validateOrder(payload.orderId),
      success: ValidatedOrder,
    });

    const charge = yield* step.run("charge", {
      do: chargeCard(validated.cardId, validated.amount),
      success: ChargeResult,
      error: ChargeError,
      undo: (charge, cause) => refundCard(charge.chargeId),
      retry: { times: 3 },
    });

    yield* step.sleep("cooldown", "5 seconds");

    return { orderId: payload.orderId, chargeId: charge.chargeId, status: "ok" };
  }),
);

Handle

Entity handlers — Actor.toLayer

// Consumer + producer — registers handlers AND provides Context
const CounterLive = Actor.toLayer(Counter, {
  Increment: ({ operation }) => Effect.succeed(operation.amount + 1),
  GetCount: () => Effect.succeed(42),
});

// From Effect context (yield services) — use .of for type-safe handlers
const CounterLive = Actor.toLayer(
  Counter,
  Effect.gen(function* () {
    const db = yield* Database;
    return Counter.of({
      Increment: ({ operation }) => db.increment(operation.amount),
      GetCount: () => db.getCount(),
    });
  }),
);

.of is a typed identity function — it infers handler types from the actor's operation defs so you never need to manually annotate { operation: { amount: number } }. Especially useful when yielding services in Effect.gen where TS can't infer the return type from Actor.toLayer.

Workflow handler — Actor.toLayer

const ProcessOrderLive = Actor.toLayer(ProcessOrder, (payload, step) =>
  Effect.gen(function* () {
    const validated = yield* step.run("validate", {
      do: validateOrder(payload.orderId),
      success: ValidatedOrder,
    });
    return { orderId: payload.orderId, status: "ok" };
  }),
);

Producer-only (client layer)

For services that send messages to actors they don't handle:

const CounterClient = Actor.toLayer(Counter);

HandlerOptions (entity only)

Actor.toLayer(actor, handlers, {
  spanAttributes: { team: "platform" },
  maxIdleTime: 60_000,
  concurrency: 10,
  mailboxCapacity: 100,
});

Handler shape (entity)

Handlers receive { operation, request }:

• operation — typed operation value with _tag and payload fields spread
• request — cluster request metadata (headers, requestId, etc.)

Client

actor(id) — get a ref (entity)

const ref = yield * Counter.ref("counter-1");

Requires Counter.Context in the environment (provided by Actor.toLayer or Actor.toTestLayer).

actor() — get a ref (workflow)

const ref = yield * ProcessOrder.ref();

Workflow actors are nullary — no entity ID needed.

execute — block for reply

const result = yield * ref.execute(Counter.Increment({ amount: 5 }));

Return type inferred from operation's success schema. Error type from error schema.

send — fire-and-forget with ExecId

const execId = yield * ref.send(Counter.Increment({ amount: 5 }));
// execId: ExecId<number, never> — branded string with phantom types

Returns ExecId<Success, Error> immediately. Send error channel does NOT include handler errors (discard semantics).

ExecId

type ExecId<Success = unknown, Error = unknown> = string & {
  readonly [ExecIdBrand]: { readonly success: Success; readonly error: Error };
};

At runtime, just a string. Format:

• Entity: "entityId\0operationTag\0primaryKey" (null-byte separated — opaque, don't parse)
• Workflow: upstream workflow execution id derived from id(payload)

Peek

peek — one-shot status check

// On the actor object — takes ExecId, returns typed PeekResult
const status = yield * Counter.peek(execId);
// status: PeekResult<number, never> — types flow from ExecId brand

Entity peek requires MessageStorage | Sharding in context. Workflow peek requires WorkflowEngine in context.

watch — polling stream

const stream = Counter.watch(execId, { interval: Duration.millis(200) });

Emits on status changes, completes on terminal result.

PeekResult

type PeekResult<A = unknown, E = unknown> =
  | { _tag: "Pending" }
  | { _tag: "Success"; value: A }
  | { _tag: "Failure"; error: E }
  | { _tag: "Interrupted" }
  | { _tag: "Defect"; cause: unknown }
  | { _tag: "Suspended" }; // workflow-only at runtime

Guards: isPending, isSuccess, isFailure, isSuspended, isTerminal.

WaitFor

waitFor — poll until terminal

const result = yield * Counter.waitFor(execId);
// result: PeekResult<number, never> — guaranteed terminal

With custom filter and schedule:

const result =
  yield *
  Counter.waitFor(execId, {
    filter: (r) => r._tag === "Success", // stop only on success
    schedule: Schedule.spaced("500 millis"), // custom poll interval
  });

Default: polls every 200ms, stops on any terminal result (isTerminal).

Entity waitFor requires MessageStorage | Sharding in context. Workflow waitFor requires WorkflowEngine in context.

Test

Entity test — Actor.toTestLayer

const CounterTest = Layer.provide(
  Actor.toTestLayer(Counter, {
    Increment: ({ operation }) => Effect.succeed(operation.amount + 1),
    GetCount: () => Effect.succeed(42),
  }),
  TestShardingConfig,
);

const test = it.scopedLive.layer(CounterTest);

test("increments", () =>
  Effect.gen(function* () {
    const ref = yield* Counter.ref("counter-1");
    const result = yield* ref.execute(Counter.Increment({ amount: 5 }));
    expect(result).toBe(6);
  }));

Workflow test — Actor.toTestLayer

WorkflowEngine.layerMemory provided automatically:

const GreeterTest = Actor.toTestLayer(Greeter, (payload, step) =>
  Effect.succeed(`hello ${payload.name}`),
);

it.scopedLive.layer(GreeterTest)("greets", () =>
  Effect.gen(function* () {
    const ref = yield* Greeter.ref();
    const result = yield* ref.execute(Greeter.Run({ name: "world" }));
    expect(result).toBe("hello world");
  }),
);

Dynamic / inline test layers

Provide the layer around the full usage — don't let ActorRef escape the provider scope:

it.scopedLive("dynamic test", () =>
  Effect.gen(function* () {
    const calls = yield* Ref.make<Array<string>>([]);

    const Tracker = Actor.fromEntity("Tracker", {
      Track: {
        payload: { item: Schema.String },
        success: Schema.String,
        id: (p) => p.item,
      },
    });

    const TrackerTest = Layer.provide(
      Actor.toTestLayer(Tracker, {
        Track: ({ operation }) =>
          Ref.update(calls, (arr) => [...arr, operation.item]).pipe(
            Effect.as(`tracked: ${operation.item}`),
          ),
      }),
      TestShardingConfig,
    );

    return yield* Effect.gen(function* () {
      const ref = yield* Tracker.ref("t-1");
      const result = yield* ref.execute(Tracker.Track({ item: "widget" }));
      expect(result).toBe("tracked: widget");
    }).pipe(Effect.provide(TrackerTest));
  }),
);

Scope gotcha

Effect.provide(effect, scopedLayer) creates a private scope. If you provide only around actor("id"), the ref escapes the scope → "All fibers interrupted without error". Provide around the entire block.

Lifecycle

// Entity: flush all messages + replies (hard delete)
yield * Order.flush("ord-1");
// Requires: MessageStorage | Sharding

// Entity: redeliver — clear read leases so unprocessed messages re-enter polling
yield * Order.redeliver("ord-1");
// Requires: MessageStorage | Sharding

// Entity: passivate (not yet supported — dies at runtime)
Order.interrupt("ord-1");

// Workflow: cancel + resume (via WorkflowEngine)
yield * ProcessOrder.interrupt("exec-id");
yield * ProcessOrder.resume("exec-id");

// Workflow-only
const execId = yield * ProcessOrder.executionId({ orderId: "ord-1" });

Entity vs Workflow lifecycle:

Operation	Entity	Workflow
flush	Deletes all messages + replies via MessageStorage.clearAddress	Not available — would corrupt execution state
redeliver	Clears read leases via MessageStorage.resetAddress	Not available
interrupt	Dies (passivation not public API)	Cancels via WorkflowEngine
resume	N/A	Resumes via WorkflowEngine

Shard group: flush/redeliver use entity.getShardGroup(entityId) to compute the correct shard ID (defaults to "default"). This is the same derivation the cluster uses for message routing.

DeliverAt

const Scheduled = Actor.fromEntity("Scheduled", {
  Process: {
    payload: { id: Schema.String, deliverAt: Schema.DateTimeUtc },
    id: (p) => p.id,
    deliverAt: (p) => p.deliverAt,
    persisted: true,
  },
});

Observability

Cluster creates spans EntityType(entityId).RpcTag automatically. No custom middleware needed. Pass extra attributes via HandlerOptions.spanAttributes.

v3

Import from effect-encore/v3. Same API, different import paths:

v4	v3
effect/unstable/cluster	@effect/cluster
effect/unstable/rpc	@effect/rpc
effect/unstable/workflow	@effect/workflow
Schema.Top	Schema.Schema.Any
Context.Service	Context.GenericTag

Migration

From raw @effect/cluster

Before (raw cluster)	After (effect-encore)
Custom Schema.Class with PrimaryKey.symbol + DeliverAt.symbol	id + deliverAt in OperationDef
Rpc.make + RpcGroup.make + Entity.fromRpcGroup	Actor.fromEntity(name, operations)
entity.toLayer(Effect.gen(...)) with entity.of({...})	Actor.toLayer(actor, handlers)
Context.Tag + makeClientLayer per entity	Actor.toLayer(actor) — provides actor.Context
client(entityId).Op(payload, { discard: true })	ref.send(Actor.Op(payload)) — returns ExecId<S, E>
Manual getMessageStatus(primaryKey) with empty address fields	actor.peek(execId) with correct compound key
Custom RpcMiddleware for spans	Not needed — cluster creates spans automatically
Entity.makeTestClient + manual RpcClient mapping	Actor.toTestLayer(actor, handlers) — returns typed Layer
Workflow.make + manual client wiring	Actor.fromWorkflow(name, def) — unified call site
Activity/DurableDeferred/DurableClock direct imports	step.run/step.sleep + declarative signals on def

From effect-encore v1 (Actor.make era)

Before	After
Actor.make("Name", defs)	Actor.fromEntity("Name", defs)
primaryKey optional	id mandatory on all entity operations
ref.call(op)	ref.execute(op)
ref.cast(op) → CastReceipt	ref.send(op) → ExecId<S, E>
peek(actor, receipt)	actor.peek(execId)
watch(actor, receipt)	actor.watch(execId)
import { Workflow } from "..."	Step DSL via (payload, step) handler
Workflow.workflow(name, opts)	Actor.fromWorkflow(name, def)