// @ozaco/std

Effect

Structured-concurrency runtime for async work, built on generators and scopes.

import from@ozaco/std/effect

The effect system is a structured-concurrency runtime built on generators. An Operation<T> is a lazy, iterable description of work — synchronous or asynchronous — that you compose by yield*-ing one operation into another, much like await chains promises but without splitting the world into sync and async functions.

Operations always run inside a Scope. When a scope ends — normally, by failure, or by cancellation — every task spawned inside it is halted and every resource / ensure teardown runs, so background work and open handles can never leak. That is what makes the concurrency here "structured": a child task never outlives its parent.

run and main are the entry points. They execute an operation and hand back a Task<T>, which is a Promise<Result<T, unknown>> you can await, yield*, or halt(). Failures travel as Result.Failure values rather than as exceptions escaping across async boundaries.

On top of these primitives the module layers concurrency combinators (all, race, spawn), timing (sleep, retry, debounce, throttle), dependency injection through Context, and push-based data flow through signals, channels, queues, and streams.

// on this page

Running effects Constructors Combinators Concurrency Collections Scheduling & timing Resources & scope Context Signals, channels & queues Streams & subscriptions Debug Guards Types

Running effects

Entry points that execute an operation on the runtime.

run

Run an operation on the global scope, returning an awaitable, haltable `Task`.

signature

function run<T>(operation: () => Operation<T>): Task<T>

// returns A `Task<T>` — a `Promise<Result<T, unknown>>` that can also be `yield*`-ed or halted.

example

import { run, sleep } from '@ozaco/std/effect'

const task = run(function* () {
  yield* sleep(1000)
  return 'done'
})

const result = await task //=> Result<'done', unknown>

main

Run a program as the process entry point, wiring SIGINT/SIGTERM to clean halt.

signature

function main(body: (args: string[]) => Operation<void>): Promise<void>

Detects the host (Deno, Node, Bun, or browser), passes CLI args to body, and installs signal handlers so an interrupt unwinds the whole scope (running every teardown) before the process exits with the appropriate status code.

example

import { main } from '@ozaco/std/effect'

await main(function* (args) {
  console.log('argv:', args)
  // Ctrl-C halts the root task and runs all cleanups first
})

exit

Terminate the enclosing `main` program with a status code and optional message.

signature

function exit(status: number, message?: string): Operation<void>

Constructors

Turn functions, promises, and callbacks into operations.

call

Invoke a function and adapt its result — value, promise, or operation — into an operation.

signature

function call<T>(fn: () => Promise<T>): Operation<T>
function call<T>(fn: () => Operation<T>): Operation<T>
function call<T>(fn: () => T): Operation<T>
function call<T, TArgs extends unknown[]>(
  callable: (...args: TArgs) => T,
  ...args: TArgs
): Operation<T>

The universal bridge from ordinary code into the effect world. Promises are awaited, operations are delegated to, and plain values are returned as-is.

example

import { call, run } from '@ozaco/std/effect'

run(function* () {
  const res = yield* call(() => fetch('/api'))
  const json = yield* call(() => res.json())
  return json
})

action

Build an operation from an executor that resolves/rejects and returns a teardown.

signature

function action<T>(
  executor: (
    resolve: (value: T) => void,
    reject: (error: unknown) => void,
  ) => () => void,
  desc?: string,
): Operation<T>

The lowest-level constructor. The returned teardown callback runs when the operation is halted, so in-flight callback-based work (timers, sockets, listeners) is cancelled cleanly.

example

import { action } from '@ozaco/std/effect'

const delay = (ms: number) =>
  action<void>(resolve => {
    const id = setTimeout(resolve, ms)
    return () => clearTimeout(id) // cancelled on halt
  })

lift

Wrap a plain function so each call becomes an operation.

signature

function lift<TArgs extends unknown[], TReturn>(
  fn: (...args: TArgs) => TReturn,
): (...args: TArgs) => Operation<TReturn>

operation

Define a reusable effect from a generator; the result is both `yield*`-able and awaitable.

signature

function operation<Args extends unknown[], T>(
  fn: (...args: Args) => ManualOperation<T>,
  ...causes: string[]
): (...args: Args) => Future<T>

Returns a factory whose output is a Future<T>: yield* it to run inside another operation, or await it to run standalone and settle into a Result. Extra causes are appended to any failure.

example

import { operation, call } from '@ozaco/std/effect'

const loadUser = operation(function* (id: string) {
  const res = yield* call(() => fetch('/users/' + id))
  return yield* call(() => res.json())
}, 'loadUser')

await loadUser('42') //=> Result<User, unknown>

until

Await a promise as an operation, capturing rejections as failures.

signature

function until<T>(promise: Promise<T>, cause?: string): Operation<T>

suspend

Suspend the current operation forever (until its scope is torn down).

signature

function suspend(): Operation<void>

clone

Wrap an operation so it can be iterated again with a fresh iterator.

signature

function clone<T>(op: Operation<T>): Operation<T>

Combinators

Handle errors and transform outcomes.

attempt

Run an operation and capture success or failure as a `Result` instead of throwing.

signature

function attempt<T>(
  op: Operation<T> | (() => Operation<T>),
): Operation<Result<T, unknown>>

example

import { attempt, call, run } from '@ozaco/std/effect'
import { isSuccess } from '@ozaco/std/result'

run(function* () {
  const result = yield* attempt(() => call(() => risky()))
  return isSuccess(result) ? result.value : 'fallback'
})

recover

Run an operation, delegating to a handler operation on failure.

signature

function recover<T, R>(
  op: Operation<T> | (() => Operation<T>),
  handler: (failure: Result.Failure<unknown>) => Operation<R>,
): Operation<T | R>

mapError

Transform the failure of an operation, leaving success untouched.

signature

function mapError<T>(
  op: Operation<T>,
  mapper: (failure: Result.Failure<unknown>) => Result.Failure<unknown>,
): Operation<T>

validate

Validate a value against a Standard Schema, failing with formatted issues.

signature

function validate<Schema extends StandardSchemaV1>(
  schema: Schema,
  value: unknown,
): Future<StandardSchemaV1.InferOutput<Schema>>

Works with any Standard Schema library (zod, valibot, arktype, …). Async schemas are awaited, so both sync and async validators are supported. On failure the message is the formatted list of issues under std:validation.

Concurrency

Structured concurrency primitives — children are bound to their parent scope.

spawn

Start an operation as a child task of the current scope, returning its `Task`.

signature

function spawn<T>(op: () => Operation<T>): Operation<Task<T>>

The spawned task runs concurrently and is automatically halted when the enclosing scope ends. yield* the returned task to await its value.

example

import { spawn, sleep, run } from '@ozaco/std/effect'

run(function* () {
  const task = yield* spawn(function* () {
    yield* sleep(1000)
    return 42
  })
  return yield* task // await the child
})

all

Run operations concurrently and collect their values; a failure halts the rest.

signature

function all<T extends readonly Operation<unknown>[] | []>(
  ops: T,
): Operation<Helpers.All<T>>

// returns A tuple of the value yielded by each operation, preserving order and types.

example

import { all, call, run } from '@ozaco/std/effect'

run(function* () {
  const [a, b] = yield* all([
    call(() => fetch('/a')),
    call(() => fetch('/b')),
  ]) // both run at once; if either fails the other is halted
  return [a, b]
})

allSettled

Run operations concurrently and collect a `Result` for each, never short-circuiting.

signature

function allSettled<T extends readonly Operation<unknown>[] | []>(
  ops: T,
): Operation<Helpers.AllSettled<T>>

// returns A tuple of `Result<value, unknown>`, one per input operation.

race

Run operations concurrently, take the first to settle, and halt the losers.

signature

function race<T extends Operation<unknown>>(
  operations: readonly T[],
): Operation<Helpers.Yielded<T>>

The first operation to succeed or fail decides the outcome; all remaining operations are halted before race returns, so no work leaks. Throws if given an empty array.

example

import { race, sleep, call, run } from '@ozaco/std/effect'

run(function* () {
  return yield* race([
    call(() => fetch('/slow')),
    (function* () {
      yield* sleep(5000)
      return 'timeout'
    })(),
  ])
})

Collections

Operation-aware iteration over plain arrays.

map

Map each element through an operation, sequentially.

signature

function map<T, U>(
  arr: readonly T[],
  mapFn: (value: T, index: number) => Operation<U>,
): Operation<U[]>

mapPar

Map each element through an operation, all concurrently (via `all`).

signature

function mapPar<T, U>(
  arr: readonly T[],
  mapFn: (value: T, index: number) => Operation<U>,
): Operation<U[]>

filter

Keep elements whose predicate operation resolves truthy, sequentially.

signature

function filter<T>(
  arr: readonly T[],
  predicate: (value: T, index: number) => Operation<boolean>,
): Operation<T[]>

filterPar

Filter with a predicate operation evaluated concurrently for every element.

signature

function filterPar<T>(
  arr: readonly T[],
  predicate: (value: T, index: number) => Operation<boolean>,
): Operation<T[]>

some

Resolve true as soon as a predicate operation is truthy for any element.

signature

function some<T>(
  arr: readonly T[],
  predicate: (value: T, index: number) => Operation<boolean>,
): Operation<boolean>

reduce

Fold an array left-to-right with an operation-returning reducer.

signature

function reduce<T, U>(
  arr: readonly T[],
  reduceFn: (acc: U, value: T, index: number) => Operation<U>,
  initial: U,
): Operation<U>

toSorted

Return a sorted copy using an operation-returning comparator (stable insertion sort).

signature

function toSorted<T>(
  arr: readonly T[],
  compareFn: (a: T, b: T) => Operation<number>,
): Operation<T[]>

Scheduling & timing

Delays, retries, and convergence.

sleep

Pause the current operation for a number of milliseconds.

signature

function sleep(duration: number): Operation<void>

example

import { sleep, run } from '@ozaco/std/effect'

run(function* () {
  yield* sleep(500)
})

retry

Retry an operation on failure with configurable attempts, delay, and backoff.

signature

function retry<T>(op: () => Operation<T>, options?: RetryOptions): Operation<T>

Defaults to 3 attempts with no delay. Supply delay, backoff, and maxDelay for exponential backoff, or when to retry only certain failures. Throws the last failure once attempts are exhausted.

example

import { retry, call, run } from '@ozaco/std/effect'

run(function* () {
  return yield* retry(() => call(() => fetch('/flaky')), {
    attempts: 5,
    delay: 100,
    backoff: 2,
  })
})

when

Poll an assertion until it holds (non-`false`) or a timeout elapses.

signature

function when<T>(
  assertion: () => Operation<T>,
  options?: ConvergeOptions,
): Operation<ConvergeStats<T>>

Runs assertion every interval ms (default 10) until it returns a non-false value or timeout ms (default 2000) pass, then resolves with timing stats and the value — or fails with the last error.

always

Assert a condition stays true (never `false`) for the whole timeout window.

signature

function always<T>(
  assertion: () => Operation<T>,
  options?: ConvergeOptions,
): Operation<ConvergeStats<T>>

The inverse of when: fails the moment the assertion returns false; otherwise resolves with stats once the window (default 200 ms) elapses.

Resources & scope

Lifecycle management, cleanup, and scope control.

resource

Define an operation with a setup/teardown lifecycle tied to the calling scope.

signature

function resource<T>(
  op: (provide: (value: T) => Operation<void>) => Operation<void>,
): Operation<T>

Run acquisition, then call provide(value) to hand the value to the caller. provide suspends the resource until the calling scope tears down, at which point control returns for cleanup (typically in a finally). This guarantees every acquired resource is released.

example

import { resource, call } from '@ozaco/std/effect'

const useFile = (path: string) =>
  resource(function* (provide) {
    const handle = yield* call(() => open(path))
    try {
      yield* provide(handle) // suspends until the scope ends
    } finally {
      yield* call(() => handle.close())
    }
  })

ensure

signature

function ensure(fn: () => Operation<unknown> | void): Operation<void>

example

import { ensure, run } from '@ozaco/std/effect'

run(function* () {
  yield* ensure(() => console.log('runs on success, failure, or halt'))
  // ... work
})

scoped

Run an operation in a fresh child scope, halting its children when the block exits.

signature

function scoped<T>(operation: () => Operation<T>): Operation<T>

example

import { scoped, spawn, run } from '@ozaco/std/effect'

run(function* () {
  yield* scoped(function* () {
    yield* spawn(worker) // halted when this block returns
    yield* doWork()
  })
})

useScope

Obtain the `Scope` the current operation is running in.

signature

function useScope(): Operation<Scope>

createScope

Create a child scope (and its disposer) rooted under a parent scope.

signature

function createScope(
  parent?: Scope,
): Scope & AsyncDisposable & [Scope, () => Future<void>]

Returns a scope that is also destructurable as [scope, dispose] and usable with await using. Defaults to the global scope as parent.

const

global

The root scope that `run` uses by default.

signature

const global: Scope

withResolvers

Create an operation plus external `resolve`/`reject` to settle it (a deferred).

signature

function withResolvers<T>(desc?: string): {
  operation: Operation<T>
  resolve(value: T): void
  reject(error: unknown): void
}

withHost

Dispatch to a runtime-specific operation for Deno, Node, Bun, or the browser.

signature

function withHost<T>(op: {
  deno(): Operation<T>
  node(): Operation<T>
  bun(): Operation<T>
  browser(): Operation<T>
}): Operation<T>

useAbortSignal

An `AbortSignal` that aborts when the current task is halted or fails, not on success.

signature

function useAbortSignal(): Operation<AbortSignal>

Pass the signal to fetch or other cancellable APIs. It fires only on cancellation/failure teardown, so a value carried out on the success path (e.g. a Response body) stays readable.

example

import { useAbortSignal, call, run } from '@ozaco/std/effect'

run(function* () {
  const signal = yield* useAbortSignal()
  return yield* call(() => fetch('/data', { signal }))
})

Context

Scoped dependency injection — values flow down through scopes.

createContext

Create a named `Context` for reading/writing a value within a scope.

signature

function createContext<T>(name: string, defaultValue?: T): Context<T>

A Context exposes get, set, expect, delete, and with operations. Values are stored on the current scope and inherited by descendants.

example

import { createContext, run } from '@ozaco/std/effect'

const Db = createContext<Database>('db')

run(function* () {
  yield* Db.set(connect())
  const db = yield* Db.expect() // throws if unset
})

useContext

Read a context value, accepting either a `Context` or a `{ context }` holder.

signature

function useContext<T>(ctx: Context<T> | { context: Context<T> }): Operation<T>

markContextAsSnapshot

Flag a context so its value is captured when a child scope is snapshotted.

signature

function markContextAsSnapshot<T>(context: Context<T>): Context<T>

Signals, channels & queues

Push-based primitives for multicasting and buffering values.

createSignal

A multicast event source: `send`/`close` synchronously, subscribe as a `Stream`.

signature

function createSignal<T, TClose = never>(): Signal<T, TClose>

send and close are plain synchronous methods; iterating the signal (it is a Stream) yields a fresh, independently-buffered subscription per consumer.

example

import { createSignal, each, run } from '@ozaco/std/effect'

const clicks = createSignal<MouseEvent>()

run(function* () {
  for (const event of yield* each(clicks)) {
    handle(event)
    yield* each.next()
  }
})

createChannel

Like a signal, but `send`/`close` are operations rather than sync methods.

signature

function createChannel<T, TClose = void>(): Channel<T, TClose>

createQueue

A single unbounded FIFO queue you push to and pull from as a `Subscription`.

signature

function createQueue<T, TClose>(): Queue<T, TClose>

add/close enqueue items; next() is an operation that resolves immediately if buffered, or suspends until the next item arrives.

const

SignalQueueFactory

Context holding the queue constructor signals use, overridable per scope.

signature

const SignalQueueFactory: Context<typeof createQueue>

Streams & subscriptions

Build, adapt, and consume streams of values.

stream

Turn an async iterable into a `Stream`.

signature

function stream<T, R>(iterable: AsyncIterable<T>): Stream<T, R>

into

Adapt a sync or async iterable into a `Stream`, optionally transforming each item.

signature

function into<T, R = T>(
  source: Iterable<T> | AsyncIterable<T>,
  transform?: (item: T) => Operation<R>,
): Stream<R, void>

The source is drained by a background task; async iterators are closed on teardown so their own cleanup runs.

collect

Drain a stream or subscription fully into an array.

signature

function collect<R>(
  source: Subscription<R, void> | Stream<R, void>,
  transform?: (item: R) => Operation<unknown>,
): Future<R[]>

interval

A `Stream` that emits `void` every N milliseconds until torn down.

signature

function interval(milliseconds: number): Stream<void, never>

on

A `Stream` of events from an `EventTarget`, auto-removing the listener on teardown.

signature

function on<T extends EventTarget, K extends string>(
  target: T,
  name: K,
): Stream<Event, never>

example

import { on, each, run } from '@ozaco/std/effect'

run(function* () {
  for (const event of yield* each(on(window, 'resize'))) {
    onResize(event)
    yield* each.next()
  }
})

once

Wait for a single event from an `EventTarget`.

signature

function once<T extends EventTarget, K extends string>(
  target: T,
  name: K,
): Operation<Event>

debounce

Emit only the latest value after `ms` of quiet, cancelling pending timers on halt.

signature

function debounce<T>(source: Stream<T, unknown>, ms: number): Stream<T, never>

throttle

Emit at most one value per `ms` window, dropping the rest.

signature

function throttle<T>(source: Stream<T, unknown>, ms: number): Stream<T, never>

each

Iterate a stream in a `for...of` loop, pairing with `each.next()` per step.

signature

function each<T>(stream: Stream<T, unknown>): Operation<Iterable<T>>
each.next(): Operation<void>

Yields the current value; you must yield* each.next() before continuing the loop, which advances the subscription and closes it when the stream is done.

streamForEach

Consume a stream, running a callback (optionally an operation) for each item.

signature

function streamForEach<T>(
  stream: Stream<T, unknown>,
  fn: (item: T) => unknown,
): Operation<void>

forEachSubscriptionEvent

Pull a subscription to completion, running a callback per emitted value.

signature

function forEachSubscriptionEvent<T>(
  source: Subscription<T, unknown>,
  fn: (value: T) => unknown,
): Operation<void>

Debug

Trace effect execution.

debug

Enable/disable effect tracing for the current scope.

signature

function debug(
  enabled?: boolean | ((desc: string) => void) | 'force-silence',
): Operation<void>

Pass true to log each effect, a function to receive descriptions, false to disable, or force-silence to suppress inherited debugging.

enableGlobalDebug

Turn on effect tracing globally, with the default logger or a custom sink.

signature

function enableGlobalDebug(enabled?: boolean | ((desc: string) => void)): void

getGlobalDebug

Read the current global debug sink, if any.

signature

function getGlobalDebug(): ((desc: string) => void) | undefined

Guards

Narrow unknown values to effect types.

isOperation

Narrow a value to an `Operation` (anything with a `Symbol.iterator`).

signature

function isOperation<T>(value: unknown): value is Operation<T>

isContext

Narrow a value to a `Context`.

signature

function isContext(value: unknown): value is Context<any>

isScope

Narrow a value to a `Scope`.

signature

function isScope(value: unknown): value is Scope

isSubscription

Narrow a value to a `Subscription` (has a `next` method).

signature

function isSubscription<T>(value: unknown): value is Subscription<T, void>

isStream

Narrow a value to a `Stream`.

signature

function isStream<T>(value: unknown): value is Stream<T, void>

isSnapshotContext

Test whether a context was flagged with `markContextAsSnapshot`.

signature

function isSnapshotContext(context: Context<unknown>): boolean

Types

The core interfaces of the effect system.

iface

Operation

A lazy, iterable unit of work driven by the runtime via `yield*`.

signature

interface Operation<T> {
  [Symbol.iterator](): Iterator<Effect | Result.Failure, T, unknown>
}

type

ManualOperation

A generator producing an operation — the shape of an operation body.

signature

type ManualOperation<T> = Generator<Effect | Result.Failure, T, unknown>

iface

Future

An operation that is also a promise settling to a `Result`.

signature

interface Future<T> extends Operation<T>, Promise<Result<T, unknown>> {}

iface

Task

A running operation you can await, `yield*`, halt, or dispose.

signature

interface Task<T> extends Future<T> {
  halt(): Future<void>
  [Symbol.asyncDispose](): Promise<void>
}

iface

Scope

The lifetime boundary that owns tasks and context values.

signature

interface Scope {
  run<T>(operation: () => Operation<T>): Task<T>
  safeRun<T>(operation: () => Operation<T>): Promise<Result<T, unknown>>
  spawn<T>(operation: () => Operation<T>): Operation<Task<T>>
  get<T>(context: Context<T>): T | undefined
  set<T>(context: Context<T>, value: T): T
  expect<T>(context: Context<T>): T
  delete<T>(context: Context<T>): boolean
  hasOwn<T>(context: Context<T>): boolean
}

iface

Context

A named, scoped slot for a value, with operations to read and write it.

signature

interface Context<T> {
  name: string
  defaultValue?: T
  get(): Operation<T | undefined>
  set(value: T): Operation<T>
  expect(): Operation<T>
  delete(): Operation<boolean>
  with<R>(value: T, operation: (value: T) => Operation<R>): Operation<R>
}

iface

Subscription

A pull-based reader whose `next()` yields iterator results as an operation.

signature

interface Subscription<T, TDone> {
  next(): Operation<IteratorResult<T, TDone>>
}

type

Stream

An operation that produces a fresh `Subscription` on each consumption.

signature

type Stream<T, TReturn> = Operation<Subscription<T, TReturn>>

iface

Signal

A `Stream` with synchronous `send`/`close`.

signature

interface Signal<T, TClose> extends Stream<T, TClose> {
  send(value: T): void
  close(value: TClose): void
}

iface

Channel

A `Stream` whose `send`/`close` are operations.

signature

interface Channel<T, TClose> extends Stream<T, TClose> {
  send(message: T): Operation<void>
  close(value: TClose): Operation<void>
}

iface

Queue

A `Subscription` you can push into via `add`/`close`.

signature

interface Queue<T, TClose> extends Subscription<T, TClose> {
  add(item: T): void
  close(value: TClose): void
}

iface

RetryOptions

Options controlling `retry`: attempts, delay, backoff, maxDelay, and `when`.

signature

interface RetryOptions {
  attempts?: number // default 3
  delay?: number // ms, default 0
  backoff?: number // multiplier, default 1
  maxDelay?: number // ms, default 30_000
  when?: (error: Result.Failure<unknown>) => boolean
}

iface

ConvergeOptions

Timing options for `when` / `always`.

signature

interface ConvergeOptions {
  timeout?: number
  interval?: number
}

iface

ConvergeStats

Timing report returned by `when` / `always`.

signature

interface ConvergeStats<T> {
  start: number
  end: number
  elapsed: number
  runs: number
  timeout: number
  interval: number
  value: T
}

Helpers

Supporting type utilities used throughout the effect API.

signature

namespace Helpers { /* Effect, Executor, Provide, All, AllSettled, Yielded, … */ }

Internal-leaning type helpers surfaced for advanced typing. Notable members:

•All<T> / AllSettled<T> — result tuples for all / allSettled.
•Yielded<T> — the value type an Operation yields.
•Executor<T> — the (resolve, reject) => teardown shape used by action.
•Provide<T> — the provide callback passed to resource.
•WithResolvers<T> / FutureWithResolvers<T> — deferred shapes.
•Exit, HostOperation<T>, Coroutine, ScopeInternal — runtime internals.