ocaml.md

Effect Handlers in TypeScript

OCaml-style algebraic effects built on TypeScript generators. The fiber holds the iterator and a small state machine. The continuation is a one-shot, safety-checked handle. The handler is an async scheduler that drains synchronously-runnable work and parks while external wakers are pending.

Three pieces, each with one job:

• Fiber<T> — iteration state machine
• Continuation<R> — one-shot safety, typed by effect result
• Handler<E, T> — type-locked scheduler that runs any program over E

Effects are identified by reference identity (the constructor is the tag), and handlers are typed against the exact set of effects they handle. Forgetting a clause, supplying a wrong payload type, or resuming with a wrong value are all compile-time errors.

The correspondence

OCaml 5	TS generators
perform E	yield* perform(e)
try ... with effect E k -> ...	handler.run(comp) with clauses
continue k v	k.resume(v)
discontinue k e	k.throw(e)
One-shot continuations	Generators resume once

Implementation

// ---------- effects ----------

interface Effect<P, R> {
  ctor: EffectCtor<P, R>;
  payload: P;
}

interface EffectCtor<P, R> {
  (payload: P): Effect<P, R>;
  __p?: P;   // phantom, for inference
  __r?: R;   // phantom
}

// Each call creates a fresh function whose identity is the effect's tag.
function defEffect<P, R>(): EffectCtor<P, R> {
  const ctor = ((payload: P) => ({ ctor, payload })) as EffectCtor<P, R>;
  return ctor;
}

// A computation is a generator that may yield effects and returns a T.
type Eff<T> = Generator<Effect<any, any>, T, unknown>;

function* perform<P, R>(eff: Effect<P, R>): Generator<Effect<P, R>, R, unknown> {
  return (yield eff) as R;
}

// ---------- fiber ----------

type FiberState<T> =
  | { tag: "runnable" }
  | { tag: "suspended" }
  | { tag: "done"; value: T }
  | { tag: "aborted"; value: T };

type Input =
  | { kind: "value"; v: unknown }
  | { kind: "throw"; e: unknown };

class Fiber<T> {
  state: FiberState<T> = { tag: "runnable" };
  private input: Input = { kind: "value", v: undefined };

  constructor(private gen: Eff<T>) {}

  resumeWith(v: unknown) {
    this.input = { kind: "value", v };
    this.state = { tag: "runnable" };
  }

  throwWith(e: unknown) {
    this.input = { kind: "throw", e };
    this.state = { tag: "runnable" };
  }

  abortWith(value: T) {
    this.state = { tag: "aborted", value };
  }

  step(): Effect<any, any> | null {
    if (this.state.tag !== "runnable") {
      throw new Error(`step() on ${this.state.tag} fiber`);
    }
    const r = this.input.kind === "value"
      ? this.gen.next(this.input.v)
      : this.gen.throw(this.input.e);
    if (r.done) {
      this.state = { tag: "done", value: r.value };
      return null;
    }
    this.state = { tag: "suspended" };
    return r.value;
  }
}

// ---------- continuation ----------

class Continuation<R> {
  private used = false;

  constructor(
    private readonly fiber: Fiber<unknown>,
    private readonly wake: () => void,
    private readonly name: string,
  ) {}

  resume(value: R) {
    this.consume();
    this.fiber.resumeWith(value);
    this.wake();
  }

  throw(error: unknown) {
    this.consume();
    this.fiber.throwWith(error);
    this.wake();
  }

  private consume() {
    if (this.used) {
      throw new Error(`continuation for '${this.name}' already used`);
    }
    const s = this.fiber.state.tag;
    if (s !== "suspended") {
      throw new Error(`continuation for '${this.name}' invoked on ${s} fiber`);
    }
    this.used = true;
  }
}

// ---------- handler ----------

type Clause<P, R, T> = (
  payload: P,
  k: Continuation<R>,
  abort: (v: T) => void,
) => void;

// An effect bundle is any object whose values are effect constructors.
type Effects = Record<string, EffectCtor<any, any>>;

// Clauses for a bundle E producing T: one clause per key, with payload
// and resume types pinned to that key's constructor.
type ClausesFor<E extends Effects, T> = {
  [K in keyof E]: E[K] extends EffectCtor<infer P, infer R>
    ? Clause<P, R, T>
    : never;
};

interface Handler<E extends Effects, T> {
  run(comp: () => Eff<T>): Promise<T>;
}

// Two-step generic to lock T explicitly while inferring E from the bundle.
function handlerFor<T>() {
  return <E extends Effects>(
    effects: E,
    clauses: ClausesFor<E, T>,
  ): Handler<E, T> => {
    const dispatch = new Map<EffectCtor<any, any>, Clause<any, any, T>>();
    const nameOf = new Map<EffectCtor<any, any>, string>();
    for (const k in effects) {
      const ctor = effects[k] as EffectCtor<any, any>;
      dispatch.set(ctor, (clauses as any)[k]);
      nameOf.set(ctor, k);
    }

    return {
      async run(comp) {
        const fiber = new Fiber<T>(comp());
        let wake = () => {};
        const abort = (v: T) => { fiber.abortWith(v); wake(); };

        while (true) {
          while (fiber.state.tag === "runnable") {
            const eff = fiber.step();
            if (eff === null) break;
            const clause = dispatch.get(eff.ctor);
            const name = nameOf.get(eff.ctor) ?? "(unknown)";
            if (!clause) throw new Error(`unhandled effect: ${name}`);
            clause(
              eff.payload,
              new Continuation(fiber as Fiber<unknown>, wake, name),
              abort,
            );
          }

          switch (fiber.state.tag) {
            case "done":
            case "aborted":   return fiber.state.value;
            case "suspended": await new Promise<void>((r) => { wake = r; });
          }
        }
      },
    };
  };
}

The three-step shape

The design separates three concerns:

1. Define effects together. Each effect is a constructor function; group them in a plain object.
2. Build a handler for those effects. Type-checked against the bundle: missing clauses, wrong payload types, or wrong resume types are all compile errors. The handler is a standalone, reusable object.
3. Run any program against the handler. Programs are independent of handlers as long as they only perform effects in the bundle.

// 1. Define
const Get = defEffect<void, number>();
const Put = defEffect<number, void>();
const E = { Get, Put };

// 2. Build
let state = 0;
const handler = handlerFor<number>()(E, {
  Get: (_, k) => k.resume(state),
  Put: (v, k) => { state = v; k.resume(); },
});

// 3. Run
await handler.run(programA);
await handler.run(programB);

The Eff<T> alias

The full generator signature Generator<Effect<any, any>, T, unknown> shows up everywhere. The alias makes it bearable:

type Eff<T> = Generator<Effect<any, any>, T, unknown>;

The three type parameters:

• Effect<any, any> — what gets yielded. Any effect; the specific payload and result types are recovered through perform's typed return.
• T — what the computation eventually returns.
• unknown — the TNext slot, the type of values fed back via gen.next(value). TypeScript only has one slot for all yields, so this is unavoidably unknown and perform does the per-yield cast.

Read function* fetchUser(id: string): Eff<User> as "a computation that returns a User, possibly performing effects along the way" — the same way you'd read Promise<User>.

Examples

State

const Get = defEffect<void, number>();
const Put = defEffect<number, void>();
const State = { Get, Put };

function* counter(): Eff<number> {
  const x = yield* perform(Get(undefined));
  yield* perform(Put(x + 1));
  yield* perform(Put((yield* perform(Get(undefined))) * 10));
  return yield* perform(Get(undefined));
}

async function runCounter() {
  let state = 0;
  const handler = handlerFor<number>()(State, {
    Get: (_, k) => k.resume(state),         // k: Continuation<number>
    Put: (v, k) => { state = v; k.resume(); },  // v: number
  });
  const result = await handler.run(counter);
  console.log({ result, state });  // { result: 10, state: 10 }
}

Notice the inference: v is number without any cast, and k.resume() for Put takes no argument because R = void. The clauses match the effect signatures exactly.

Exceptions

abort terminates the computation with a handler-supplied value. k.throw throws into the generator so it can be caught with try / catch.

const Raise = defEffect<string, never>();
const Errors = { Raise };

function* safeDivide(a: number, b: number): Eff<number> {
  if (b === 0) yield* perform(Raise("division by zero"));
  return a / b;
}

type Result<T> = { ok: true; value: T } | { ok: false; error: string };

function* asResult<T>(comp: () => Eff<T>): Eff<Result<T>> {
  const value = yield* comp();
  return { ok: true, value };
}

async function runDivide() {
  const handler = handlerFor<Result<number>>()(Errors, {
    Raise: (msg, _k, abort) => abort({ ok: false, error: msg }),
  });

  console.log(await handler.run(() => asResult(() => safeDivide(10, 2))));
  // { ok: true, value: 5 }

  console.log(await handler.run(() => asResult(() => safeDivide(10, 0))));
  // { ok: false, error: "division by zero" }
}

Same Raise effect, different handler, recoverable exception:

function* mightFail(): Eff<string> {
  try {
    yield* perform(Raise("kaboom"));
    return "unreachable";
  } catch (e) {
    return `recovered from: ${(e as Error).message}`;
  }
}

async function runRecover() {
  const handler = handlerFor<string>()(Errors, {
    Raise: (msg, k) => k.throw(new Error(msg)),
  });
  console.log(await handler.run(mightFail));
  // "recovered from: kaboom"
}

The meaning of Raise lives in the handler. The computation doesn't know whether it aborts or recovers.

Logging

const Log = defEffect<string, void>();
const log = (msg: string) => perform(Log(msg));

const StateAndLog = { Get, Put, Log };

function* task(): Eff<number> {
  yield* log("starting");
  const x = yield* perform(Get(undefined));
  yield* log(`got ${x}`);
  yield* perform(Put(x + 1));
  yield* log("done");
  return x + 1;
}

async function runWithLog() {
  let state = 5;
  const logs: string[] = [];
  const handler = handlerFor<number>()(StateAndLog, {
    Get: (_, k) => k.resume(state),
    Put: (v, k) => { state = v; k.resume(); },
    Log: (msg, k) => { logs.push(msg); k.resume(); },
  });
  const result = await handler.run(task);
  console.log({ result, logs });
  // { result: 6, logs: ["starting", "got 5", "done"] }
}

Dependency injection (reader)

interface Env { apiBase: string; userId: string }
const Ask = defEffect<void, Env>();
const Reader = { Ask };

function* greet(): Eff<string> {
  const env = yield* perform(Ask(undefined));
  return `hello, ${env.userId} (via ${env.apiBase})`;
}

async function runGreet() {
  const env: Env = { apiBase: "https://api.example.com", userId: "alice" };
  const handler = handlerFor<string>()(Reader, {
    Ask: (_, k) => k.resume(env),
  });
  console.log(await handler.run(greet));
  // "hello, alice (via https://api.example.com)"
}

Async

Await is just another clause. The scheduler doesn't know promises exist — the clause arranges for the promise's .then to call k.resume, and the fiber parks until that happens.

const Await = defEffect<Promise<unknown>, unknown>();
const await_ = <T>(p: Promise<T>) =>
  perform(Await(p as Promise<unknown>)) as unknown as Eff<T>;

const Async = { Await, Log };

function* fetchUser(id: string): Eff<string> {
  yield* log(`fetching ${id}`);
  const res  = (yield* await_(fetch(`https://api.example.com/users/${id}`))) as Response;
  const json = (yield* await_(res.json())) as { name: string };
  yield* log(`got ${json.name}`);
  return json.name;
}

async function runFetch() {
  const logs: string[] = [];
  const handler = handlerFor<string>()(Async, {
    Await: (p, k) => p.then((v) => k.resume(v), (e) => k.throw(e)),
    Log:   (msg, k) => { logs.push(msg); k.resume(); },
  });
  const name = await handler.run(() => fetchUser("42"));
  console.log({ name, logs });
}

Real vs test handlers

The seam between business logic and infrastructure sits at the effect handler. The same program runs against real fetch in production and against fixtures in tests, with no parameterization at the function level — and the handlers have the same type, so any program from the bundle works against either.

interface HttpReq { method: "GET" | "POST"; url: string; body?: unknown }
interface HttpRes { status: number; body: unknown }

const Http = defEffect<HttpReq, HttpRes>();
const httpGet = (url: string) => perform(Http({ method: "GET", url }));

const Api = { Http, Log, Raise };

interface User { id: string; name: string }
interface Post { id: string; title: string }

function* fetchUserWithPosts(userId: string): Eff<{ user: User; posts: Post[] }> {
  yield* log(`fetching user ${userId}`);
  const userRes = yield* httpGet(`/users/${userId}`);
  if (userRes.status !== 200) {
    yield* perform(Raise(`user ${userId} not found`));
  }
  const user = userRes.body as User;

  yield* log(`fetching posts for ${user.name}`);
  const postsRes = yield* httpGet(`/users/${userId}/posts`);
  const posts = postsRes.body as Post[];

  yield* log(`done: ${posts.length} posts`);
  return { user, posts };
}

Production handler — real fetch:

type ApiResult = Result<{ user: User; posts: Post[] }>;

function makeRealHandler() {
  const logs: string[] = [];

  const handler = handlerFor<ApiResult>()(Api, {
    Http: (req, k) => {
      const init: RequestInit = { method: req.method };
      if (req.body !== undefined) {
        init.headers = { "content-type": "application/json" };
        init.body = JSON.stringify(req.body);
      }
      fetch(`https://api.example.com${req.url}`, init).then(
        async (response) => {
          const body = await response.json().catch(() => null);
          k.resume({ status: response.status, body });
        },
        (err) => k.throw(err),
      );
    },
    Log:   (msg, k) => { logs.push(msg); k.resume(); },
    Raise: (msg, _k, abort) => abort({ ok: false, error: msg }),
  });

  return { handler, logs };
}

Test handler — fixtures with call recording:

interface MockRoute {
  method: "GET" | "POST";
  url: string;
  status?: number;
  body?: unknown;
}

function makeTestHandler(routes: MockRoute[]) {
  const calls: HttpReq[] = [];
  const logs: string[] = [];

  const handler = handlerFor<ApiResult>()(Api, {
    Http: (req, k) => {
      calls.push(req);
      const route = routes.find((r) => r.method === req.method && r.url === req.url);
      if (!route) return k.throw(new Error(`unmocked: ${req.method} ${req.url}`));
      k.resume({ status: route.status ?? 200, body: route.body ?? null });
    },
    Log:   (msg, k) => { logs.push(msg); k.resume(); },
    Raise: (msg, _k, abort) => abort({ ok: false, error: msg }),
  });

  return { handler, calls, logs };
}

async function testHappyPath() {
  const t = makeTestHandler([
    { method: "GET", url: "/users/u1",       body: { id: "u1", name: "Alice" } },
    { method: "GET", url: "/users/u1/posts", body: [{ id: "p1", title: "Hello" }] },
  ]);

  const result = await t.handler.run(() => asResult(() => fetchUserWithPosts("u1")));

  console.assert(result.ok === true);
  if (result.ok) {
    console.assert(result.value.user.name === "Alice");
    console.assert(result.value.posts.length === 1);
  }
  console.assert(t.calls.length === 2);
  console.assert(t.calls[0].url === "/users/u1");
  console.assert(t.logs.length === 3);
}

async function testNotFound() {
  const t = makeTestHandler([
    { method: "GET", url: "/users/missing", status: 404, body: null },
  ]);

  const result = await t.handler.run(() => asResult(() => fetchUserWithPosts("missing")));

  console.assert(result.ok === false);
  if (!result.ok) console.assert(result.error === "user missing not found");
  console.assert(t.calls.length === 1);
}

The test asserts both the result and the trace. The mock is a real implementation of Http, not a global patch. Tests resolve synchronously through the inner drain loop — latency is microseconds, not milliseconds.

What the type system catches

Forgetting a clause:

const handler = handlerFor<number>()(StateAndLog, {
  Get: (_, k) => k.resume(state),
  Put: (v, k) => { state = v; k.resume(); },
  // missing Log
});
// ^^^ Property 'Log' is missing in type '...' but required in type 'ClausesFor<...>'.

Wrong payload type:

const handler = handlerFor<number>()(StateAndLog, {
  Get: (_, k) => k.resume(state),
  Put: (v, k) => { state = v.toUpperCase(); k.resume(); },
  //                       ^^^^^^^^^^^ Property 'toUpperCase' does not exist on type 'number'.
  Log: (m, k) => { logs.push(m); k.resume(); },
});

Wrong resume value:

const handler = handlerFor<number>()(StateAndLog, {
  Get: (_, k) => k.resume("oops"),
  //                      ^^^^^^ Argument of type 'string' is not assignable to parameter of type 'number'.
  // ...
});

Extra clause not in bundle (with --strict):

const handler = handlerFor<number>()(StateAndLog, {
  Get: ..., Put: ..., Log: ...,
  Other: (_, k) => k.resume(undefined),
  // ^^^ Object literal may only specify known properties, and 'Other' does not exist...
});

Safety at runtime

Continuation<R> enforces, at runtime:

• One-shot use. Calling k.resume (or k.throw) twice throws with a message naming the effect.
• State validity. Resuming a continuation after the fiber has been aborted or completed throws.

These guards catch the most common handler bugs at the point of misuse with a labeled error, rather than silently corrupting state or hanging.

What's not caught:

• Forgetting to call k.resume or abort. The fiber stays suspended, the scheduler parks on wake, and nothing wakes it. The program hangs. Catching this requires either a timeout, or a pending-external-wakers counter so the scheduler can distinguish "waiting" from "stuck".
• Cross-fiber misuse. With one fiber this is moot. With multiple fibers (a natural extension), continuations would close over their fiber.

What the scheduler does

In about thirty lines: pick a runnable fiber, step it, dispatch its effect to the matching clause via reference-identity lookup, repeat until done or aborted; park on a promise when nothing is runnable.

The scheduler knows nothing about promises, channels, sleep, joins, mutexes. Each of those is a clause — handler-level policy. The scheduler is fixed; the language of effects is open. The same program runs under a production scheduler, a deterministic test scheduler, or a journal-replay scheduler, with no changes to the program text.

Extending it

The natural next steps, all additive:

• Multiple fibers. Replace the single fiber with a ready queue. Continuations close over which fiber they target. Fork becomes a clause that pushes a new fiber onto the queue.
• External waker tracking. Add a pending counter and a sched.park() / sched.unpark() pair so the scheduler can detect "suspended with no waker" as a handler bug instead of hanging.
• Cancellation. A fiber-local flag; step checks it and calls gen.return() to run finally blocks.
• Structured concurrency. A nursery effect that forks children into a group, joins all on exit, propagates errors.

None of these change the core. The fiber state machine, the safe continuation, and the typed handler bundle are the spine.