effect-hatchet

Effect-native bindings for Hatchet, with an in-memory implementation for tests.

A single Hatchet tag with two constructors: one that talks to a real Hatchet engine, and one that runs your task graph entirely in-process. Define a task once with Effect + Schema; swap the layer to run it under Vitest with no engine.

Install

bun add effect-hatchet @hatchet-dev/typescript-sdk effect

@hatchet-dev/typescript-sdk is a peer dependency. It's only loaded at runtime when you use Hatchet.layer — tests using Hatchet.layerInMemory won't touch it.

Quick start

import { Effect, Schema as S } from "effect"
import { Hatchet, Task } from "effect-hatchet"

const greet = Task.make({
  name: "greet",
  input: S.Struct({ name: S.String }),
  output: S.Struct({ message: S.String }),
  fn: (input) => Effect.succeed({ message: `hello ${input.name}` }),
})

const program = Effect.gen(function* () {
  const hatchet = yield* Hatchet
  yield* hatchet.register(greet)
  yield* hatchet.startWorker()

  const result = yield* greet.run({ name: "world" })
  console.log(result.message) // "hello world"
})

program.pipe(
  Effect.provide(Hatchet.layer()),
  Effect.scoped,
  Effect.runPromise,
)

Hatchet.layer reads connection config from the standard Hatchet env vars (HATCHET_CLIENT_TOKEN, HATCHET_CLIENT_HOST_PORT, HATCHET_CLIENT_API_URL, HATCHET_CLIENT_TLS_STRATEGY) via Effect Config.

Test without a Hatchet engine

Swap Hatchet.layer for Hatchet.layerInMemory and the same program runs in-process — no worker, no gRPC, no engine.

import { Effect, Schema as S } from "effect"
import { expect, it } from "vitest"
import { Hatchet } from "effect-hatchet"

it("greets", () =>
  Effect.gen(function* () {
    const hatchet = yield* Hatchet
    yield* hatchet.register(greet)

    const result = yield* greet.run({ name: "world" })
    expect(result.message).toBe("hello world")
  }).pipe(
    Effect.provide(Hatchet.layerInMemory()),
    Effect.scoped,
    Effect.runPromise,
  ))

startWorker() is a no-op under layerInMemory, so the same bootstrap works in both layers. task.schedule honors real wall-clock delays via Effect.sleep — pair with TestClock for time-dependent tests.

Guides

The order below mirrors the lifecycle: define a task, register it, start the worker, run it, schedule it, handle errors.

Defining tasks

A Task is a unit of work. You define it with Task.make, then register it with the engine to make it runnable.

Only name and fn are required; everything else is optional.

Task.make({
  name: "send-email",                        // required
  input: S.Struct({                          // optional
    to: S.String,
    subject: S.String,
  }),
  output: S.Struct({ messageId: S.String }), // optional
  fn: (input, ctx) =>                        // required
    Effect.gen(function* () {
      const mailer = yield* Mailer
      const id = yield* mailer.send(input.to, input.subject)
      return { messageId: id }
    }),

  // Passed through to Hatchet (all optional):
  rateLimits: [{ key: "send-email", units: 1 }],
  concurrency: { expression: "input.to", maxRuns: 1 },
  on: { event: "user:created" },
  durable: true,
  executionTimeout: "6h",                    // defaults to "3h"
  scheduleTimeout: "10m",
  retries: 2,
})

Timeouts use TaskDuration, derived from the SDK's Duration (Go-duration strings), for both regular and durable tasks. Omitted scheduleTimeout and retries leave the SDK defaults unchanged. In memory, retries re-run fn immediately on failure; timeouts are not simulated.

Schema optionality

input and output are independent — supply either, both, or neither:

• Omit input → fn's input parameter is untyped (unknown); task.run accepts anything.
• Omit output → whatever fn returns is passed through verbatim, no encoding.
• Provide either → that side gets compile-time types and runtime validation. Input is decoded before fn runs; output is encoded before being returned to the caller.

fn shape

Effect.Effect<OUTPUT, ERROR, REQUIREMENTS>. Any R your task depends on (services, configs, loggers) is captured at register time and provided when the task runs — including under layerInMemory. So yield* Mailer inside fn works as long as the layer providing Mailer is in scope when you call register.

ctx

• runId: string — Hatchet's workflow run ID (a UUID under layerInMemory).
• signal: AbortSignal — aborts when the run is cancelled.
• putStream(data: string): Effect<void, TaskStreamError> — emits a live stream chunk.

Cancellation interrupts the task's Effect fiber in both layers, so Effect.ensuring and Effect.acquireRelease finalizers run. Use ctx.signal for external operations that accept an abort signal.

Errors

Any failure in fn — typed errors, schema decode errors on input, or unexpected throws — is logged and surfaced to the caller as TaskExecutionFailure with the original error in cause.

Registering tasks

Defining a task only creates a value; the engine doesn't know about it yet. hatchet.register(task) wires it in.

Required before any .run / .runNoWait / .schedule call. Calling those on an unregistered task is a defect (Effect.die), not a typed failure.

yield* hatchet.register(taskA)
yield* hatchet.register(taskB)
// or
yield* Effect.forEach([taskA, taskB], (t) => hatchet.register(t))

register captures the surrounding Effect runtime, so the task's R is satisfied by whichever layers are in scope at registration time, not at run time. If a task yields Mailer, make sure Mailer.Default (or equivalent) is provided before you call register.

Both layers require this — including layerInMemory, where it's how the in-memory engine learns about the task at all.

Starting the worker

The Hatchet engine queues work; a worker pulls work off the queue and runs it. hatchet.startWorker() boots that worker for the current process.

• Hatchet.layer: required. Without it, registered tasks exist client-side but no worker pulls them — task.run will block waiting for output that never arrives. The call forks the worker loop and returns; the surrounding scope keeps it alive.
• Hatchet.layerInMemory: no-op. In-memory runs happen directly in the Effect runtime, so there's nothing to start. The call is safe — that's the point, so the same bootstrap code works under both layers.

Call once, after all registrations:

yield* hatchet.register(taskA)
yield* hatchet.register(taskB)
yield* hatchet.startWorker()

Tasks registered after startWorker() aren't picked up by that worker.

Running tasks

Three call styles, all returning Effects that fail with TaskExecutionFailure:

// Wait for output
const result = yield* greet.run({ name: "world" })

// Fire-and-forget; get a handle you can await later
const handle = yield* greet.runNoWait({ name: "world" })
const later = yield* handle.output

// Enqueue for the future
const scheduled = yield* greet.schedule(
  new Date(Date.now() + 60_000),
  { name: "world" },
)

• run — blocks the Effect until the task produces output.
• runNoWait — returns a handle with handle.runId for logging or tracing the created Hatchet run; yield* handle.output later if you want to await the result. In-memory runs use synthetic IDs.
• schedule — enqueues for a future time, returns { id }. Cancel with hatchet.schedule.delete(id).

Input passes through the task's input schema (if any) before reaching fn. Output passes through the task's output schema before reaching the caller. Schema failures surface as TaskExecutionFailure.

Under Hatchet.layer, these dispatch through the Hatchet engine. Under Hatchet.layerInMemory, both run and runNoWait register a run and fork its task fiber into the layer's scope; run awaits its output. schedule honors real wall-clock delays via Effect.sleep (use TestClock for time-dependent tests).

Run status, cancellation, and streaming

const handle = yield* greet.runNoWait({ name: "world" })
const status = yield* hatchet.runs.getStatus(handle.runId)
const chunks = hatchet.runs.subscribeToStream(handle.runId) // Stream<string, RunsError>
yield* hatchet.runs.cancel(handle.runId)

• getStatus returns RunStatus: QUEUED | RUNNING | COMPLETED | FAILED | CANCELLED. Missing in-memory IDs (and detectable SDK HTTP 404s) fail with RunNotFound; other SDK failures use RunsError.
• cancel interrupts active runs; cancelling a finished in-memory run is a no-op.
• Cancelled runs fail task.run and handle.output with TaskExecutionFailure whose cause is RunCancelled({ runId }) in both layers.
• The live layer polls status after the SDK result resolves, with exponential delays from 100ms capped at 1s and a 30s deadline. Failed runs, status lookup errors, and runs that never become terminal fail with TaskExecutionFailure rather than returning output.
• subscribeToStream is live-only (no replay): start consuming before the task calls ctx.putStream. It ends when the SDK iterator ends, or when the in-memory run terminates. Unknown or already finished in-memory runs return an empty stream; the live layer preserves the SDK iterator's behavior/errors.

Crons and schedules

A cron is a recurring trigger driven by a cron expression. A schedule is a one-shot enqueue at a specific time (what task.schedule creates).

const cron = yield* hatchet.cron.create({
  workflowName: greet.name,
  name: "daily-greet",
  expression: "0 9 * * *",
  input: { name: "world" },
  additionalMetadata: { tier: "free" }, // optional
})

const all = yield* hatchet.cron.list({ workflowName: greet.name })

const firstPage = yield* hatchet.schedule.list({
  statuses: ["SCHEDULED"], // optional
  offset: 0,               // optional
  limit: 50,               // optional
})

yield* hatchet.cron.delete(cron.id)
yield* hatchet.schedule.delete(scheduled.id)

• workflowName is the target task's name. The task must be registered for the cron to fire under Hatchet.layer.
• schedule.delete swallows missing-ID errors under both layers — safe to call defensively.
• schedule.list makes exactly one call and returns one page — it does not accumulate every schedule for you. The result is { schedules, pagination? }: schedules is that page's ScheduledRun[], and pagination (when the source reports it) is { currentPage?, nextPage?, numPages? }, all 1-indexed. nextPage is not an end-of-results sentinel — the server always populates it, clamping it back to currentPage on the last page instead of omitting it, so looping while (pagination.nextPage is defined) will hang forever. Walk every page by advancing offset to (pagination.nextPage - 1) * limit while pagination.nextPage > pagination.currentPage, and also stop once a page comes back with an empty schedules array — the zero-result case never satisfies the clamp, since numPages is 0 and nextPage just keeps counting up on every call. statuses filters by ScheduledRunStatus (PENDING, RUNNING, SUCCEEDED, FAILED, CANCELLED, QUEUED, SCHEDULED); results are ordered by triggerAt.
• Under layerInMemory, a schedule whose run hasn't fired yet reports workflowRunStatus: "SCHEDULED" — the same default that statuses: ["SCHEDULED"] filters on, so matching entries always carry that value rather than leaving the field absent. offset/limit and the returned pagination follow the same 1-indexed semantics as Hatchet.layer, so pagination logic written against one layer works against the other.
• Under layerInMemory, crons are stored but don't auto-fire on a schedule. Use hatchet.cron._testFire(cron.id) in tests to manually fire a registered cron's task; it's in-memory only and dies under Hatchet.layer.
• Testing gotcha: inside an it.layer(...) block (the @effect/vitest fixture that builds a layer once and shares it across every test in the block), writing Effect.provide(Hatchet.layerInMemory()) inside one of those tests does not get you a fresh instance — it's the same Layer reference it.layer already built, so Effect's layer memoization silently reuses the shared one. Symptom: schedule/cron counts come out too high, because schedules created by earlier tests are still sitting in what looks like a fresh instance. Fix: wrap it in Layer.fresh(Hatchet.layerInMemory()) (from effect's Layer module) when a test needs a genuinely isolated instance.

Events

An event is a named trigger: pushing one fires every task registered with on: { event: <key> } for that key.

const UserCreated = Event.make({
  key: "user:created",
  payload: S.Struct({ userId: S.String }),
})

const onUserCreated = Task.make({
  name: "on-user-created",
  input: UserCreated.payload,   // schema written once, referenced here
  on: { event: UserCreated },   // typed reference, not a bare string
  fn: (input) => Effect.succeed(`welcomed ${input.userId}`),
})

yield* hatchet.register(onUserCreated)
yield* hatchet.startWorker()

yield* hatchet.event.push(UserCreated, { userId: "user-1" })

• Event.make's key is the wire key events are pushed and matched on; payload is decoded before fn runs, same as Task.make's own input.
• on: { event } and event.push's first argument both accept the typed Event or a plain string key — mix and match freely.

Using a plain string key also works

const onUserCreated = Task.make({
  name: "on-user-created",
  input: S.Struct({ userId: S.String }),
  on: { event: "user:created" },
  fn: (input) => Effect.succeed(`welcomed ${input.userId}`),
})

yield* hatchet.register(onUserCreated)
yield* hatchet.startWorker()

yield* hatchet.event.push("user:created", { userId: "user-1" })

• event.push is fire-and-forget under both layers: it returns { id } immediately without waiting for the triggered run(s) to finish, matching real Hatchet's own semantics. Await onUserCreated.run(...) directly if you need the result.
• Pushing a key with no registered listeners is a no-op, not an error.
• Unlike crons, event-triggered tasks fire under layerInMemory too — no _testFire-style workaround needed.

Errors

Tagged errors you can Effect.catchTag on:

Error	Raised by
TaskExecutionFailure	task.run, task.runNoWait, task.schedule
TaskStreamError	ctx.putStream
RunNotFound	hatchet.runs.getStatus
RunsError	hatchet.runs.getStatus, cancel, subscribeToStream
CronCreateError	hatchet.cron.create
CronDeleteError	hatchet.cron.delete
CronListError	hatchet.cron.list
ScheduleListError	hatchet.schedule.list
ScheduleDeleteError	hatchet.schedule.delete
EventPushError	hatchet.event.push

TaskExecutionFailure.cause carries the original error from your fn (typed failure, schema decode error, or unexpected throw).

Worker affinity

To prefer running enqueued tasks on the same worker that submitted them:

Hatchet.layer({ runPrefersThisWorker: true })

The layer tags the worker with a per-process instance id and requires that label on dispatched runs. Useful when tasks need access to in-process state on the submitting worker.

Development

pnpm test runs the in-memory suite — no Docker required. pnpm test:real runs the same portable tests against a real hatchet-lite server, booted in Docker via testcontainers; it requires a running Docker daemon and is not part of the default pnpm test.

Status

Pre-1.0. Surface is shaped against production usage but expect changes as more Hatchet features (multi-step workflows, parent/child runs, richer ctx) get surfaced.