API Reference: Orchestration & Combinators

The Orchestration & Combinators API in Siglo provides functional combinators and workflow orchestrators to compose modular, resilient, and type-safe async workflows. Siglo combines sequential pipelines, parallel fan-out/fan-in, conditional branching, dynamic collections mapping, durable event suspension, and workflow AST inspection.

Core Orchestration
Parallel Combinators
Branching and Routing
- when(predicate, thenStep, elseStep?)
- branch(selectorOrBranches, branchesMap?, defaultBranch?)
Dynamic Mapping
- map(itemsOrAccessor, processItem, options?)
Durable Waits and Event Resumption
Graph Inspection and Visualization

Core Orchestration

`flow(...steps)`

Composes sequential execution units (Step, nested Flow, or combinator steps) into an executable workflow with compile-time accumulated context inference and typed step history ($steps).

Signatures

// 1 to 5+ steps with compile-time accumulated context inference
export function flow<TIn, T1Out, N1 extends string>(
    s1: Step<any, T1Out, N1>,
): Flow<TIn, FlowContextResult<MergeStepContext<TIn, T1Out>, Record<N1, T1Out>>>;

export function flow<TIn, T1Out, N1 extends string, T2Out, N2 extends string>(
    s1: Step<any, T1Out, N1>,
    s2: Step<any, T2Out, N2>,
): Flow<
    TIn,
    FlowContextResult<MergeStepContext<MergeStepContext<TIn, T1Out>, T2Out>, Record<N1, T1Out> & Record<N2, T2Out>>
>;

// Generic overload for dynamic or arbitrary step arrays
export function flow<TInput = any, TOutput = any>(...steps: FlowArg[]): Flow<TInput, TOutput>;

Parameters

...steps: (Step | Flow | Array<Step | Flow>)[] — Variable list or nested arrays of steps to execute sequentially.

Returns

A Flow<TInput, TOutput> object exposing execution method (.run()), durable controls (.resume(), .compensate()), guardrails (.networkPolicy()), live monitoring (.monitor(), .hook()), and export utilities (.toMermaid(), .toJSON()).

Context Accumulation & `$steps`

As each step executes, its output is shallow-merged into the accumulated context. Earlier step outputs remain accessible to subsequent steps. In addition, the return context includes a typed $steps property containing each step’s output keyed by step name.

Example

import { Step, flow } from '@siglo/core';

const fetchUser = Step.create('fetchUser', async (param: { userId: string }) => {
    return { name: 'Alice', role: 'admin' };
});

const checkPermissions = Step.create('checkPermissions', async (param: { role: string }) => {
    return { canApprove: param.role === 'admin' };
});

const sendAlert = Step.create('sendAlert', async (param: { name: string; canApprove: boolean }) => {
    return { alerted: true, timestamp: Date.now() };
});

export const authFlow = flow(fetchUser, checkPermissions, sendAlert);

const result = await authFlow.run({ userId: 'u_100' });
console.log(result.name); // 'Alice'
console.log(result.canApprove); // true
console.log(result.alerted); // true
console.log(result.$steps.fetchUser); // { name: 'Alice', role: 'admin' }

`isFlow(item)`

Type guard verifying whether a value implements the Flow interface.

Signature

export function isFlow(item: any): item is Flow;

Example

import { flow, isFlow, Step } from '@siglo/core';

const myStep = Step.create('noop', async () => ({}));
const myFlow = flow(myStep);

console.log(isFlow(myFlow)); // true
console.log(isFlow(myStep)); // false

`toPOJO(context)`

Strips internal tracked proxy wrappers, getters, and metadata from an execution context, returning a clean, serializable JavaScript object.

Signature

export function toPOJO<T = any>(context: T): T;

Example

import { toPOJO } from '@siglo/core';

const cleanData = toPOJO(rawExecutionContext);
console.log(JSON.stringify(cleanData));

Parallel Combinators

`parallel(steps, options?)`

Executes multiple steps, flows, or functions concurrently. By default, it operates in fail-fast mode: if any branch fails, all other in-flight branches are aborted via an internal AbortController.

Signature

export function parallel(...args: any[]): Step<any, any>;

export interface ParallelOptions {
    name?: string;
    mode?: 'fail-fast' | 'allSettled' | 'race';
    mergeStrategy?:
        'shallow' | 'deep' | 'error' | 'namespace' | ((target: any, source: any, branchKey: string | number) => any);
}

Options

Option	Type	Default	Description
`name`	`string`	`'parallel'`	Identifying name for the combinator node in AST and telemetry.
`mode`	`'fail-fast' \| 'allSettled' \| 'race'`	`'fail-fast'`	Execution mode governing failure and settle handling.
`mergeStrategy`	`'shallow' \| 'deep' \| 'error' \| 'namespace' \| Function`	`'shallow'`	Strategy for merging branch outputs into accumulated context.

Merge Strategies

'shallow' (default): Object.assign(target, branchOutput).
'deep': Recursively merges nested objects.
'error': Throws FieldCollisionException if two branches output differing values for the same key.
'namespace': Scopes each branch’s output under its branch name (target[branchName] = branchOutput).
(target, source, branchKey) => merged: Custom callback to resolve branch collisions.

Context Isolation

Each branch receives an isolated clone of the input context. Mutating context in one branch does not bleed into sibling branches.

Example

import { Step, flow, parallel } from '@siglo/core';

const fetchOrders = Step.create('orders', async () => ({ orders: [1, 2, 3] }));
const fetchInvoices = Step.create('invoices', async () => ({ invoices: ['inv_1'] }));

const prepareDashboard = flow(
    parallel([fetchOrders, fetchInvoices], {
        mergeStrategy: 'namespace',
        name: 'fetchDashboardData',
    }),
);

const result = await prepareDashboard.run();
console.log(result.orders); // { orders: [1, 2, 3] }
console.log(result.invoices); // { invoices: ['inv_1'] }
console.log(result.$branches.length); // 2

`all(steps, options?)`

Convenience alias for parallel(steps, { ...options, mode: 'fail-fast' }).

Signature

export function all(branches: any[], options?: ParallelOptions): Step<any, any>;

Example

import { all, Step, flow } from '@siglo/core';

const step1 = Step.create('s1', async () => ({ a: 1 }));
const step2 = Step.create('s2', async () => ({ b: 2 }));

const workflow = flow(all([step1, step2]));

`allSettled(steps, options?)`

Executes all branches concurrently without failing fast. Collects all results and errors, ensuring that every branch runs to completion or failure.

Signature

export function allSettled(branches: any[], options?: ParallelOptions): Step<any, any>;

Return Context Additions

$branches: BranchExecutionSummary[] containing execution duration, status, output, and error for each branch.
$settled: PromiseSettledResult<any>[] standard settled result objects.
Each branch output or error object keyed by branch name (e.g. result.auditStep = { error: ... }).

Example

import { Step, flow, allSettled } from '@siglo/core';

const notifySlack = Step.create('slack', async () => ({ slackSent: true }));
const notifyPagerDuty = Step.create('pagerDuty', async () => {
    throw new Error('PagerDuty rate limited');
});

const alertSystem = flow(allSettled([notifySlack, notifyPagerDuty]));

const result = await alertSystem.run();
console.log(result.slack.slackSent); // true
console.log(result.pagerDuty.error.message); // 'PagerDuty rate limited'
console.log(result.$branches);

`race(...steps)`

Executes multiple branches concurrently. Settles as soon as the first branch resolves or rejects. Losing branches are immediately aborted via cooperative AbortSignal cancellation.

Signature

export function race(...args: any[]): Step<any, any>;

Example

import { Step, flow, race } from '@siglo/core';

const primaryDns = Step.create('primaryDns', async () => {
    return { ip: '1.1.1.1', source: 'primary' };
});

const fallbackDns = Step.create('fallbackDns', async () => {
    await new Promise(r => setTimeout(r, 500));
    return { ip: '1.0.0.1', source: 'fallback' };
});

const resolveDns = flow(race([primaryDns, fallbackDns]));
const result = await resolveDns.run();
console.log(result.source); // 'primary'

Branching and Routing

`when(predicate, thenStep, elseStep?)`

Conditional branching combinator that evaluates a predicate function against the current context. If true, executes thenStep; otherwise executes elseStep (if provided).

Signatures

export type BranchPredicate = (
    ctx: any,
    meta?: {
        attempt?: number;
        stackId?: string;
        namespace?: string;
    },
) => boolean | Promise<boolean>;

export function when(predicate: BranchPredicate, thenStep: any, elseStep?: any): Step<any, any>;

Metadata

The output context is enriched with:

$branchSelected: 'then' | 'else' | undefined
$skippedBranches: Array of branch route names that were bypassed.
$skipped: true if predicate evaluated to false and no elseStep was configured.

Example

import { Step, flow, when } from '@siglo/core';

const applyDiscount = Step.create('applyDiscount', async (ctx: { total: number }) => ({
    discount: ctx.total * 0.1,
}));

const noDiscount = Step.create('noDiscount', async () => ({
    discount: 0,
}));

const checkoutFlow = flow(when(ctx => ctx.isVipMember === true, applyDiscount, noDiscount));

const output = await checkoutFlow.run({ total: 200, isVipMember: true });
console.log(output.discount); // 20
console.log(output.$branchSelected); // 'then'

`branch(selectorOrBranches, branchesMap?, defaultBranch?)`

Dynamic multi-path router that selects a branch from a map using a selector function or context key.

Signatures

export type BranchSelector = (
    ctx: any,
    meta?: {
        attempt?: number;
        stackId?: string;
        namespace?: string;
    },
) => string | Promise<string>;

export function branch(selector: BranchSelector, branches: Record<string, any>, defaultBranch?: any): Step<any, any>;

export function branch(branchesMap: Record<string, any>, defaultBranch?: any): Step<any, any>;

Example

import { Step, flow, branch } from '@siglo/core';

const enterpriseFlow = Step.create('enterprise', async () => ({ slaTier: '24/7 Phone' }));
const standardFlow = Step.create('standard', async () => ({ slaTier: 'Email Only' }));
const defaultFlow = Step.create('community', async () => ({ slaTier: 'Forums' }));

const supportRouter = flow(
    branch(
        ctx => ctx.planType,
        {
            enterprise: enterpriseFlow,
            standard: standardFlow,
        },
        defaultFlow, // fallback
    ),
);

const result = await supportRouter.run({ planType: 'enterprise' });
console.log(result.slaTier); // '24/7 Phone'
console.log(result.$branchSelected); // 'enterprise'

Dynamic Mapping

`map(itemsOrAccessor, processItem, options?)`

Maps an array or async iterable through an item processing step or subflow with bounded worker concurrency, error aggregation, and order preservation.

Signatures

export type MapItemAccessor =
    | any[]
    | AsyncIterable<any>
    | Iterable<any>
    | ((ctx: any) => any[] | AsyncIterable<any> | Iterable<any> | Promise<any[] | AsyncIterable<any> | Iterable<any>>);

export interface MapOptions {
    name?: string;
    concurrency?: number; // Default: 10 (minimum: 1)
    stopOnError?: boolean; // Default: true (fails fast and aborts workers)
    preserveOrder?: boolean; // Default: true (results array matches input order)
}

export function map(itemsOrAccessor: MapItemAccessor, processItem: any, options?: MapOptions): Step<any, any>;

Process Item Context

Inside processItem, the step parameter receives { item, index, total, ...meta }.

Output

The map step outputs:

items: Array of processed results.
count: Total count of items.
errors: Array of caught errors (when stopOnError: false).

Example

import { Step, flow, map } from '@siglo/core';

const processOrder = Step.create('processOrder', async (ctx: { item: { id: string; amount: number } }) => {
    return { processedId: ctx.item.id, charged: ctx.item.amount };
});

const batchFlow = flow(
    map(ctx => ctx.pendingOrders, processOrder, {
        concurrency: 3,
        preserveOrder: true,
        stopOnError: true,
    }),
);

const result = await batchFlow.run({
    pendingOrders: [
        { id: 'ord_1', amount: 50 },
        { id: 'ord_2', amount: 120 },
        { id: 'ord_3', amount: 75 },
    ],
});

console.log(result.count); // 3
console.log(result.items); // [{ processedId: 'ord_1', charged: 50 }, ...]

Durable Waits and Event Resumption

`waitForEvent(eventName, options)`

Suspends workflow execution in durable mode until an external event is delivered via resumeWorkflow(), or an optional timeout window elapses.

Signature

export interface WaitForEventOptions {
    correlationKey: string | ((ctx: any) => string);
    timeout?: string | number; // e.g. '24h', '30m', 60000
    timeoutMs?: number;
}

export function waitForEvent(eventName: string, options: WaitForEventOptions): Step<any, any>;

Semantics

When options.durable: true, reaching waitForEvent persists an execution checkpoint to storage and throws WorkflowSuspendedException.
The waiting instance is registered under eventName and the computed correlationKey.
If no event arrives before timeout, subsequent resumption throws ExpiredWaitException.

Example

import { Step, flow, waitForEvent } from '@siglo/core';

export const approvalWorkflow = flow(
    Step.create('createTicket', async (ctx: { ticketId: string }) => ({ status: 'pending' })),
    waitForEvent('managerApproval', {
        correlationKey: ctx => ctx.ticketId,
        timeout: '48h',
    }),
    Step.create('grantAccess', async (ctx: { approved: boolean; approver: string }) => ({
        granted: ctx.approved,
    })),
);

// Starts workflow and suspends at waitForEvent
await approvalWorkflow.run({ ticketId: 'ticket_42' }, { durable: true });

`wait(durationOrMs)`

Pauses workflow execution for a duration string (e.g. '5m', '2h', '1d') or millisecond count. In durable mode, suspends execution to storage. In in-memory mode, pauses using a timer.

Signature

export function wait(durationOrMs: number | string): Step<any, any>;

Example

import { Step, flow, wait } from '@siglo/core';

const delayedWorkflow = flow(
    Step.create('init', async () => ({ step: 1 })),
    wait('10m'), // In durable mode: suspends checkpoint. In memory: awaits timeout.
    Step.create('finalize', async () => ({ step: 2 })),
);

`resumeWorkflow(flow, key, payload?, options?)`

Restores and resumes a suspended durable workflow from storage using its correlation key or stack ID.

Signature

export interface ResumeOptions extends ExecutionOptions {
    migrations?: Record<string, (state: any) => any>;
    targetVersion?: string | number;
}

export function resumeWorkflow(
    flowInstance: Flow<any, any>,
    correlationKeyOrStackId: string,
    eventPayload: any = {},
    options?: ResumeOptions,
): Promise<any>;

Example

import { resumeWorkflow } from '@siglo/core';
import { approvalWorkflow } from './approval.workflow';

// Resumes the workflow where it left off, passing event payload
const finalResult = await resumeWorkflow(approvalWorkflow, 'ticket_42', { approved: true, approver: 'Alice' });

console.log(finalResult.granted); // true

Graph Inspection and Visualization

`toGraph(item)`

Compiles any step, flow, or combinator into a JSON-serializable WorkflowNode AST.

Signature

export function toGraph(item: any): WorkflowNode;

Node Types (`WorkflowNodeType`)

'step': Single step execution unit.
'sequence': Sequential chain (Flow).
'parallel': Concurrent fan-out.
'branch': Conditional or multi-path router.
'wait': Durable wait or event suspension.
'map': Dynamic array/iterable mapping.

`inspectGraph(item)`

Analyzes workflow AST topology and returns structural metrics including node count, depth, and directed edges.

Signature

export interface GraphInspectionResult {
    totalNodes: number;
    entryNodeId: string;
    exitNodeIds: string[];
    nodes: any[];
    edges: GraphInspectionEdge[];
}

export function inspectGraph(item: any): GraphInspectionResult;

Example

import { inspectGraph } from '@siglo/core';
import { authFlow } from './auth';

const metrics = inspectGraph(authFlow);
console.log(`Total Nodes: ${metrics.totalNodes}, Entry: ${metrics.entryNodeId}`);

`validateGraph(node)`

Validates graph structure, checking for cycles and unreachable nodes without throwing.

Signature

export interface GraphValidationResult {
    valid: boolean;
    errors: string[];
}

export function validateGraph(node: WorkflowNode | Flow | Step): GraphValidationResult;

`assertValidGraph(node)`

Asserts graph validity, throwing GraphValidationException if any cycles or structural defects are detected. Note that flow.run() does not automatically run graph assertions at runtime; validation is an explicit utility.

Signature

export function assertValidGraph(node: WorkflowNode | Flow | Step): void; // Throws GraphValidationException

`toMermaid(item, options?)`

Renders workflow structure as a Mermaid flowchart diagram with policy badges (e.g. [⏱️ 3000ms], [🔁 retry(3)], [⏪ saga]).

Signature

export interface MermaidExportOptions {
    direction?: 'TD' | 'LR';
    theme?: string;
}

export function toMermaid(item: any, options?: MermaidExportOptions): string;

Example

import { toMermaid } from '@siglo/core';
import { myWorkflow } from './pipeline';

const diagram = toMermaid(myWorkflow, { direction: 'TD' });
console.log(diagram);

`toDot(item, options?)`

Exports Graphviz DOT representation (options.graphName).

`toJSON(item)`

Serializes the entire workflow AST into a clean JSON string.

​API Reference: Orchestration & Combinators

​Table of Contents

​Core Orchestration

​flow(...steps)

​Signatures

​Parameters

​Returns

​Context Accumulation & $steps

​Example

​isFlow(item)

​Signature

​Example

​toPOJO(context)

​Signature

​Example

​Parallel Combinators

​parallel(steps, options?)

​Signature

​Options

​Merge Strategies

​Context Isolation

​Example

​all(steps, options?)

​Signature

​Example

​allSettled(steps, options?)

​Signature

​Return Context Additions

​Example

​race(...steps)

​Signature

​Example

​Branching and Routing

​when(predicate, thenStep, elseStep?)

​Signatures

​Metadata

​Example

​branch(selectorOrBranches, branchesMap?, defaultBranch?)

​Signatures

​Example

​Dynamic Mapping

​map(itemsOrAccessor, processItem, options?)

​Signatures

​Process Item Context

​Output

​Example

​Durable Waits and Event Resumption

​waitForEvent(eventName, options)

​Signature

​Semantics

​Example

​wait(durationOrMs)

​Signature

​Example

​resumeWorkflow(flow, key, payload?, options?)

​Signature

​Example

​Graph Inspection and Visualization

​toGraph(item)

​Signature

​Node Types (WorkflowNodeType)

​inspectGraph(item)

​Signature

​Example

​validateGraph(node)

​Signature

​assertValidGraph(node)

​Signature

​toMermaid(item, options?)

​Signature

​Example

​toDot(item, options?)

​toJSON(item)

​Related Topic Guides

API Reference: Orchestration & Combinators

Table of Contents

Core Orchestration

`flow(...steps)`

Signatures

Parameters

Returns

Context Accumulation & `$steps`

Example

`isFlow(item)`

Signature

Example

`toPOJO(context)`

Signature

Example

Parallel Combinators

`parallel(steps, options?)`

Signature

Options

Merge Strategies

Context Isolation

Example

`all(steps, options?)`

Signature

Example

`allSettled(steps, options?)`

Signature

Return Context Additions

Example

`race(...steps)`

Signature

Example

Branching and Routing

`when(predicate, thenStep, elseStep?)`

Signatures

Metadata

Example

`branch(selectorOrBranches, branchesMap?, defaultBranch?)`

Signatures

Example

Dynamic Mapping

`map(itemsOrAccessor, processItem, options?)`

Signatures

Process Item Context

Output

Example

Durable Waits and Event Resumption

`waitForEvent(eventName, options)`

Signature

Semantics

Example

`wait(durationOrMs)`

Signature

Example

`resumeWorkflow(flow, key, payload?, options?)`

Signature

Example

Graph Inspection and Visualization

`toGraph(item)`

Signature

Node Types (`WorkflowNodeType`)

`inspectGraph(item)`

Signature

Example

`validateGraph(node)`

Signature

`assertValidGraph(node)`

Signature

`toMermaid(item, options?)`

Signature

Example

`toDot(item, options?)`

`toJSON(item)`

Related Topic Guides