@cascade-flow/workflow

Workflow authoring SDK for CascadeFlow - provides type-safe step definition helpers and utilities for building workflows.

Installation

bun add @cascade-flow/workflow

API Reference

Step Definition

defineStep(spec)

Default step definition helper for workflows without typed input.

import { defineStep } from "@cascade-flow/workflow";

export const step = defineStep({
  name: "my-step",
  dependencies: {},
  exportOutput: true,
  fn: async ({ ctx }) => {
    return { result: "value" };
  }
});

baseDefineStep<TInput>()

Factory function that creates a typed defineStep function for a workflow with typed input.

Usage pattern:

1. Create a workflow-specific defineStep.ts:

// workflows/my-workflow/define-step.ts
import { baseDefineStep } from "@cascade-flow/workflow";
import type { WorkflowInput } from "./input-schema";

export const defineStep = baseDefineStep<WorkflowInput>();

2. Use it in your steps:

// workflows/my-workflow/steps/my-step/step.ts
import { defineStep } from "../../define-step";

export const step = defineStep({
  fn: async ({ ctx }) => {
    // ctx.input is now typed as WorkflowInput!
    const message = ctx.input.message; // Fully typed
    return { processed: message };
  }
});

Skip and Optional Dependencies

Skip

Error class for intentionally skipping steps. Skipped steps:

• Are marked as "skipped" (not failed)
• Cascade to dependent steps (they also get skipped)
• Do not trigger retries

import { Skip } from "@cascade-flow/workflow";

export const step = defineStep({
  fn: async ({ ctx }) => {
    if (!ctx.input.shouldProcess) {
      throw new Skip("Processing not needed");
    }
    return { result: "processed" };
  }
});

With metadata:

throw new Skip("Feature flag disabled", {
  flag: "new-processor",
  environment: process.env.NODE_ENV
});

optional(step)

Marks a dependency as optional. Optional dependencies:

• Won't cause cascade skips when they are skipped
• Will be undefined in the dependencies object if skipped
• Will have their normal output value if they complete successfully

import { defineStep, optional } from "@cascade-flow/workflow";
import { step as validation } from "../validation/step.ts";
import { step as enrichment } from "../enrichment/step.ts";

export const step = defineStep({
  dependencies: {
    validation,                      // Required - will cascade skip
    enrichment: optional(enrichment) // Optional - won't cascade skip
  },
  fn: async ({ dependencies }) => {
    // dependencies.validation is guaranteed to exist
    // dependencies.enrichment is StepOutput | undefined
    if (dependencies.enrichment) {
      // Use enrichment data
    } else {
      // Fallback when enrichment was skipped
    }
    return { result: "processed" };
  }
});

isOptional(dep)

Checks if a dependency is marked as optional. (Internal utility)

Types

RunnerContext

Runtime context available to every step:

type RunnerContext = {
  runId: string;
  workflow: {
    slug: string; // Workflow directory name
    name: string; // Friendly name from workflow.json
  };
  input?: unknown; // Validated workflow input
  log: (...args: any[]) => void;
};

InferStepOutput<T>

Type utility for inferring a step's output type from its function signature.

JSONValue, JSONArray, JSONObject, JSONPrimitive

Types representing JSON-serializable values. All step outputs must be JSON-serializable.

StepOutput

Alias for JSONValue - the type constraint for step return values.

OptionalDependency<T>

Type representing an optional dependency wrapper.

Step Configuration

The defineStep spec accepts:

{
  name?: string;                // Display name (defaults to directory name)
  dependencies?: Record<...>;   // Other steps this step depends on
  exportOutput?: boolean;       // Include in final workflow output
  maxRetries?: number;          // Retry count on failure (default: 0)
  retryDelayMs?: number;        // Delay between retries (default: 0)
  timeoutMs?: number;           // Max execution time (default: 300000ms / 5 min)
  fn: (args) => Promise<Output> | Output;  // Step implementation
}

Organizing Steps in Groups

Steps can be organized in nested directory structures for better organization:

workflows/my-workflow/steps/
├── step-1/step.ts                           # Flat step
└── data-processing/                         # Group directory
    ├── extract/
    │   └── fetch-data/step.ts              # Nested step
    └── transform/
        └── normalize/step.ts                # Nested step

Key Points:

• Groups are purely organizational (no special behavior)
• Flat and nested steps can coexist in the same workflow
• Step IDs use paths: "step-1" (flat) or "data-processing/extract/fetch-data" (nested)
• Dependencies use relative imports:

// In steps/notifications/send/step.ts
import { step as fetchData } from "../../data-processing/extract/fetch-data/step.ts";

export const step = defineStep({
  dependencies: { fetchData },
  fn: async ({ dependencies }) => {
    // Use data from nested group
    return { sent: true, data: dependencies.fetchData };
  }
});

Examples

Simple Step

import { defineStep } from "@cascade-flow/workflow";

export const step = defineStep({
  exportOutput: true,
  fn: async () => {
    return { timestamp: Date.now() };
  }
});

Step with Dependencies

import { defineStep } from "@cascade-flow/workflow";
import { step as stepA } from "../step-a/step.ts";
import { step as stepB } from "../step-b/step.ts";

export const step = defineStep({
  dependencies: { stepA, stepB },
  fn: async ({ dependencies }) => {
    // Both dependencies are available and fully typed
    const combined = {
      a: dependencies.stepA.value,
      b: dependencies.stepB.result
    };
    return combined;
  }
});

Step with Retry Logic

import { defineStep } from "@cascade-flow/workflow";

export const step = defineStep({
  maxRetries: 3,
  retryDelayMs: 1000,
  fn: async () => {
    const response = await fetch("https://api.example.com/data");
    if (!response.ok) throw new Error("API request failed");
    return await response.json();
  }
});

Conditional Skip

import { defineStep, Skip } from "@cascade-flow/workflow";

export const step = defineStep({
  fn: async ({ ctx }) => {
    if (!ctx.input.featureEnabled) {
      throw new Skip("Feature disabled");
    }

    // Process when enabled
    return { status: "processed" };
  }
});

Checkpoints

Checkpoints save intermediate progress within a step, persisting results across retries:

import { defineStep } from "@cascade-flow/workflow";

export const step = defineStep({
  fn: async ({ ctx }) => {
    for (let i = 0; i < items.length; i++) {
      const result = await ctx.checkpoint?.("process-item", async () => {
        return processItem(items[i]);
      });
    }
    return { processed: items.length };
  }
});

For parallel checkpoints, use awaitCheckpoints() to ensure all complete before errors propagate:

import { defineStep, awaitCheckpoints } from "@cascade-flow/workflow";

export const step = defineStep({
  fn: async ({ ctx }) => {
    const [a, b] = await awaitCheckpoints([
      ctx.checkpoint?.("task-a", async () => fetchData()),
      ctx.checkpoint?.("task-b", async () => validate()),
    ]);
    return { a, b };
  }
});

See Also

• Main Documentation
• Runner Package - Execution engine
• Examples - Example workflows