Workflows

|Agent setup|Docs for agents

You create workflows with createWorkflow (reusable, with dependency inference) or use run() for one-off operations. Execution is always via workflow.run(fn), the workflow object is not callable.

Type Structure

createWorkflow returns a function that produces:

         ┌─── Your success return type
         │        ┌─── Union of all dependency error types + UnexpectedError
         ▼        ▼
Result<Value, 'NOT_FOUND' | 'EMAIL_FAILED' | UnexpectedError>
           ▲
           └─── Automatically inferred from dependencies!

createWorkflow

Creates a reusable workflow with automatic error type inference from dependencies. Two call shapes:

• Named: createWorkflow('checkout', { fetchUser, sendEmail }, options?), stable label in traces, events, and diagrams even if the variable is renamed.
• Deps-first: createWorkflow({ fetchUser, sendEmail }, options?), same inference; awaitly-analyze labels the workflow from the variable name (e.g. const checkout = createWorkflow({ ... })).

With inference

import { ok, err, type AsyncResult, createWorkflow } from 'awaitly';

const fetchUser = async (id: string): AsyncResult<User, 'NOT_FOUND'> => { /* ... */ };
const sendEmail = async (to: string): AsyncResult<void, 'EMAIL_FAILED'> => { /* ... */ };

// Named — explicit label for traces and diagrams
const workflow = createWorkflow('workflow', { fetchUser, sendEmail });

// Deps-first — label comes from the variable name in static analysis
const checkout = createWorkflow({ fetchUser, sendEmail });

const result = await workflow.run(async ({ steps }) => {
  const user = await steps.fetchUser('1');
  await steps.sendEmail(user.email);
  return user;
});
/*
result.error is automatically typed as:
'NOT_FOUND' | 'EMAIL_FAILED' | UnexpectedError
*/

How it works

// awaitly extracts error types from your dependencies:

fetchUser: AsyncResult<User, 'NOT_FOUND'>
                              ▲
                              │ extracted
                              ▼
sendEmail: AsyncResult<void, 'EMAIL_FAILED'>
                             ▲
                             │ extracted
                             ▼
workflow error = 'NOT_FOUND' | 'EMAIL_FAILED' | UnexpectedError

Why inference matters

With automatic inference, you never forget to handle an error type. Add a new dependency, and TypeScript immediately warns you about the new error cases.

Return raw values, not Result

Workflow executors return raw values. awaitly wraps them automatically. See Troubleshooting: Double-wrapping.

With options

const workflow = createWorkflow('workflow', { fetchUser, sendEmail },
  {
    cache: new Map(),                    // Enable step caching
    onEvent: (event) => console.log(event), // Event stream
    streamStore: createMemoryStreamStore(), // Enable streaming
  }
);

The value returned by createWorkflow is a workflow object with a single .run() method. Use workflow.run(fn) or workflow.run(name, fn) to execute; pass per-run options as workflow.run(fn, config).

run()

run(deps, fn) is the same deps-first idea without a name attached. Error types are inferred exactly as they are for createWorkflow. You do not supply type parameters:

import { run } from 'awaitly';

const result = await run({ fetchUser, sendEmail }, async (s) => {
  const user = await s.fetchUser('1');
  await s.sendEmail(user.email);
  return user;
});
// result.error is: 'NOT_FOUND' | 'EMAIL_FAILED' | UnexpectedError

No catchUnexpected on the deps-first form

run(deps, fn, options) accepts onError, onEvent, workflowId, workflowName, and context, but not catchUnexpected. Deps-first runs always carry UnexpectedError. To close the union, use createWorkflow(name, deps, { catchUnexpected }) or run.strict, where you supply E yourself:

const wf = createWorkflow('loadUser', { fetchUser, sendEmail }, {
  catchUnexpected: (thrown) => ({ type: 'UNEXPECTED', message: String(thrown) }),
});
// result.error: 'NOT_FOUND' | 'EMAIL_FAILED' | { type: 'UNEXPECTED', message: string }

Naming the union

You rarely need to write the error type down, but when you do, an HTTP mapper, a shared handler, derive it with ErrorsOf<typeof deps> rather than retyping it. See What TypeScript gives you back.

When to use each

Both forms take deps first and both infer the error union. The difference is what becomes possible once the workflow has a name:

Scenario	Use
Composing a few operations, once	run
Named node in generated diagrams	createWorkflow
Step caching, resume, persistence	createWorkflow
Retries and timeouts via step helpers	createWorkflow
Swapping deps in tests	createWorkflow
Deps only known at call time	run

// Named: reusable, analyzable, testable
const checkout = createWorkflow('checkout', {
  validateOrder,
  processPayment,
  sendConfirmation,
});

await checkout.run(async ({ steps }) => { /* ... */ });
await checkout.run(async ({ steps }) => { /* a different flow, same deps */ });

Workflow options

cache

Enable step caching to avoid re-executing completed steps:

const workflow = createWorkflow('workflow', deps, {
  cache: new Map(),
});

const result = await workflow.run(async ({ step, deps }) => {
  // This step runs once, even if workflow is called multiple times
  const user = await step('fetchUser', () => deps.fetchUser('1'), { key: 'user:1' });
  return user;
});
/*
First call: fetchUser executes
Second call: returns cached result immediately
*/

Note

The key option is required for caching. Steps without keys always execute.

resumeState

Resume a workflow from saved state:

const workflow = createWorkflow('workflow', deps, {
  resumeState: savedState, // From a previous run
});

const result = await workflow.run(async ({ step, deps }) => {
  // Cached steps return their saved values
  const user = await step('fetchUser', () => deps.fetchUser('1'), { key: 'user:1' });
  return user;
});

See Persistence for details.

streamStore

Enable streaming within workflows:

import { createMemoryStreamStore } from 'awaitly/durable';

const workflow = createWorkflow('workflow', deps, {
  streamStore: createMemoryStreamStore(),
});

const result = await workflow.run(async ({ step, deps }) => {
  const writer = step.getWritable<string>({ namespace: 'tokens' });

  await writer.write('Hello');
  await writer.write(' World');
  await writer.close();
});

See Streaming for details.

graph

Declare the step and decision ids your workflow may execute. Any runtime id not in the graph fails the run immediately, so static diagrams and execution stay aligned. Accepts a plain string array or a WorkflowDiagramDSL from awaitly-analyze. Placeholder segments like item-{i} match any value in that slot (item-0, item-1, …).

const workflow = createWorkflow('checkout', deps, {
  graph: ['fetchUser', 'chargeCard', 'premium-check'],
});

// Per-run override
await workflow.run(async ({ step, deps }) => { /* ... */ }, {
  graph: ['fetchUser', 'item-{i}'],
});

// run() accepts graph too
await run(async ({ step }) => { /* ... */ }, { graph: ['fetchUser'] });

Pair with awaitly-analyze --assert-diagrammable in CI. See Static Analysis →.

onEvent

Subscribe to workflow events for logging, visualization, or debugging:

const workflow = createWorkflow('workflow', deps, {
  onEvent: (event) => {
    switch (event.type) {
      case 'step_start':
        console.log('Starting:', event.name);
        break;
      case 'step_complete':
        console.log('Completed:', event.name, event.durationMs, 'ms');
        break;
      case 'step_error':
        console.error('Failed:', event.name, event.error);
        break;
    }
  },
});
/*
Output:
Starting: Fetch user
Completed: Fetch user 45 ms
Starting: Send email
Failed: Send email EMAIL_FAILED
*/

catchUnexpected (custom unexpected type)

By default, thrown exceptions become an UnexpectedError. To replace it with your own type, pass catchUnexpected:

Default (UnexpectedError)

// Default - error union includes UnexpectedError
const workflow = createWorkflow('workflow', deps);

const result = await workflow.run(async ({ step, deps }) => { /* ... */ });
// result.error: 'NOT_FOUND' | 'ORDER_FAILED' | UnexpectedError

Custom unexpected type

// Custom unexpected - your shape replaces UnexpectedError
const workflow = createWorkflow('workflow', deps, {
  catchUnexpected: (cause) => ({
    type: 'UNEXPECTED',
    cause
  })
});

const result = await workflow.run(async ({ step, deps }) => { /* ... */ });
// result.error: 'NOT_FOUND' | 'ORDER_FAILED' | { type: 'UNEXPECTED', cause: unknown }

run() with catchUnexpected

// Works with run() too
const result = await run<User, 'NOT_FOUND' | 'EMAIL_FAILED'>(
  async ({ step }) => {
    const user = await step('fetchUser', () => fetchUser('1'));
    await step('sendEmail', () => sendEmail(user.email));
    return user;
  },
  { catchUnexpected: (cause) => ({ type: 'UNEXPECTED', cause }) }
);
// result.error: 'NOT_FOUND' | 'EMAIL_FAILED' | { type: 'UNEXPECTED', cause: unknown }

With catchUnexpected:

• Your custom type replaces UnexpectedError in the error union.
• Use for exhaustive switch statements or when you want a single typed shape for all unexpected cases.

When to use catchUnexpected

Most users don’t need catchUnexpected. The default UnexpectedError works well. It’s a TaggedError you can check with isUnexpectedError(result.error), and the original thrown value is available via result.error.cause. Use catchUnexpected only when you need a custom shape.

signal (Cancellation)

Cancel workflows externally using an AbortSignal:

import { isWorkflowCancelled } from 'awaitly';

const controller = new AbortController();
const workflow = createWorkflow('workflow', deps, {
  signal: controller.signal,
});

// Start workflow
const resultPromise = workflow.run(async ({ step, deps }) => {
  const user = await step('fetchUser', () => deps.fetchUser('1'), { key: 'fetch-user' });
  await step('sendEmail', () => deps.sendEmail(user.email), { key: 'send-email' });
  return user;
});

// Cancel from outside (e.g., timeout, user action)
setTimeout(() => controller.abort('timeout'), 5000);

const result = await resultPromise;
if (!result.ok && isWorkflowCancelled(result.cause)) {
  console.log('Cancelled:', result.cause.reason);
}
/*
Output (if cancelled):
Cancelled: timeout
*/

When cancellation applies

• Abort signaled before workflow starts
• Abort occurs between steps
• A step throws AbortError (e.g., from fetch respecting the signal)
• Abort fires during the last step but the step completes successfully

Typed errors are preserved: If a step returns err("KNOWN_ERROR") even while abort is signaled, that typed error is returned, not masked as cancellation. With the default error mapper, check result.cause (not result.error) for WorkflowCancelledError when handling cancellation.

Passing signal to operations:

Steps using step.withTimeout can receive the workflow signal:

const result = await workflow.run(async ({ step, deps }) => {
  // Signal fires on EITHER timeout OR workflow cancellation
  const data = await step.withTimeout(
    'fetchUrl',
    (signal) => fetch(url, { signal }),
    { ms: 3000, signal: true }
  );
  return data;
});

Or access the signal directly via context:

const result = await workflow.run(async ({ step, deps, ctx }) => {
  // Manual signal access
  const response = await fetch(url, { signal: ctx.signal });

  // Check if cancelled
  if (ctx.signal?.aborted) {
    return err('CANCELLED');
  }
  return ok(response);
});

clock

Time source for retry delays, step.sleep, and step.withTimeout. Defaults to the system clock. Per-run clock overrides creation-time.

import { createTestClock } from 'awaitly/testing';

const clock = createTestClock();
const workflow = createWorkflow('workflow', deps, { clock });

const pending = workflow.run(async ({ step }) => {
  await step.sleep('wait', '1s');
  return 'ok';
});
clock.advance(1000);
const result = await pending;

Policy wrappers (retry(), timeout()) do not pick up this clock automatically. Pass clock at wrap time, or use step.retry / step.withTimeout / step.sleep inside the run. See Retries & Timeouts.

Execution-time options

While createWorkflow accepts options at creation time, you may need different options for specific runs. Pass a config as the second argument to workflow.run(fn, config), or use a named run with workflow.run("run-id", fn, config) for logging and tracing:

const workflow = createWorkflow('workflow', { fetchUser }, {
  onEvent: defaultLogger,  // Creation-time default
});

// Normal run - uses creation-time options
await workflow.run(async ({ step, deps }) => { /* ... */ });

// Per-run options override creation-time
await workflow.run(async ({ step, deps }) => {
  const user = await step('fetchUser', () => deps.fetchUser('1'));
  return user;
}, {
  onEvent: vizualizer.handleEvent,  // Override for this run only
  signal: controller.signal,         // Add signal for this run
});

// Named run (workflowId for logging, tracing, resume)
await workflow.run('onboard-user-1', async ({ step, deps }) => {
  return await step('fetchUser', () => deps.fetchUser('1'));
});

Override semantics

Execution-time options override creation-time options:

const workflow = createWorkflow('workflow', deps, {
  onEvent: handlerA,        // Creation-time
  signal: signalA,
});

// handlerB overrides handlerA for this run
// signalA is NOT used (signalB takes over)
await workflow.run(fn, {
  onEvent: handlerB,        // Wins
  signal: signalB,         // Wins
});

// Back to creation-time defaults
await workflow.run(fn);  // Uses handlerA, signalA

Available execution options (RunConfig)

Option	Description
onEvent	Event handler for this run
onError	Error handler for this run
signal	AbortSignal for this run
clock	Clock for retry delays, step.sleep, and step.withTimeout
createContext	Context factory for this run
resumeState	Resume state for this run (can be async factory)
deps	Partial override of creation-time deps
snapshot	Restore from a previously saved snapshot
shouldRun	Concurrency control hook
onBeforeStart	Pre-start hook
onBeforeStep	Pre-step hook. Runs before the operation and before a cached or replayed value is read; throwing aborts the step
onAfterStep	Post-step hook for checkpointing. Does not fire for steps served from a snapshot
streamStore	Stream store for this run
graph	Declared workflow graph for this run (overrides creation-time)

Note

Options like cache and catchUnexpected are creation-time only. They define the workflow’s fundamental behavior and cannot be changed per-run.

Composing workflows

Workflows are functions. Compose them naturally:

const validateInput = createWorkflow('workflow', { validateEmail, validatePassword });
const processPayment = createWorkflow('workflow', { chargeCard, saveReceipt });

const checkout = createWorkflow('workflow', { validateEmail,
  validatePassword,
  chargeCard,
  saveReceipt,
});

const result = await checkout.run(async ({ step, deps }) => {
  // Use steps from all dependencies
  await step('validateEmail', () => deps.validateEmail(input.email));
  await step('validatePassword', () => deps.validatePassword(input.password));
  await step('chargeCard', () => deps.chargeCard(input.amount));
  await step('saveReceipt', () => deps.saveReceipt(input.orderId));
});

Workflow input (closure)

Pass input to a workflow by capturing it in a closure and calling workflow.run(fn):

const workflow = createWorkflow('workflow', deps);

const userId = '123';
const requestId = 'abc';

const result = await workflow.run(async ({ step, deps }) => {
  console.log('User ID:', userId);
  const user = await step('fetchUser', () => deps.fetchUser(userId));
  return user;
});

Callback context

The workflow callback receives a context object { step, deps, ctx }. Capture any run-specific input in the closure. Per-run execution context (e.g. for events) is set via createContext in options or workflow.run(fn, { createContext }), and is available as ctx.

Next

Learn about Tagged Errors →