MapReduce, sub-workflows, parallel execution, conditional branching, loops, fan-out/fan-in, scatter-gather, race, and fallback patterns.

Overview

Beyond basic DAG execution, Cogitator workflows support advanced composition patterns for data-parallel processing, nested workflows, and fault-tolerant execution strategies.

MapReduce

Process collections of items in parallel (map phase), then combine results (reduce phase). Supports concurrency limits, partial failure handling, per-item retries, and streaming reduce.

Basic MapReduce

import { readFile } from 'node:fs/promises';
import {
  WorkflowBuilder,
  collect,
  executeMapReduce,
  mapReduceNode,
  mapReduceWorkflowNode,
} from '@cogitator-ai/workflows';

type DocState = {
  documents: string[];
  summaries?: string[];
};

const summarizeDocs = mapReduceNode<DocState, string, string[]>('summarize-docs', {
  map: {
    items: (state) => state.documents,
    mapper: async (item) => {
      const content = await readFile(item as string, 'utf8');
      return `Summary of ${item}: ${content.slice(0, 100)}...`;
    },
    concurrency: 5,
    continueOnError: true,
  },
  reduce: collect<string>(),
});

const result = await executeMapReduce<DocState, string, string[]>(
  { documents: ['doc1.txt', 'doc2.txt', 'doc3.txt'] },
  summarizeDocs
);

console.log(`Processed: ${result.stats.successful}/${result.stats.total}`);
console.log('Summaries:', result.reduced);

Inside a workflow, wrap the configuration with mapReduceWorkflowNode; the node's output is the reduced value:

const workflow = new WorkflowBuilder<DocState>('docs')
  .initialState({ documents: [] })
  .addNode(
    'summarize-docs',
    mapReduceWorkflowNode(summarizeDocs, {
      stateMapper: (result) => ({ summaries: result.reduced }),
    })
  )
  .build();

mapNode with mapWorkflowNode runs only the map phase (the output is the array of MapItemResults), and executeMap / executeReduce run the phases directly.

Map Options

import { mapNode } from '@cogitator-ai/workflows';

type CrawlState = { urls: string[] };

const fetchPages = mapNode<CrawlState, string>('fetch-pages', {
  items: (state) => state.urls,
  mapper: async (item) => fetch(item as string).then((r) => r.text()),
  concurrency: 10,
  continueOnError: true,
  timeout: 30_000,
  retry: { maxAttempts: 3, delay: 1000, backoff: 'exponential' },
  filter: (item) => (item as string).startsWith('https'),
  transform: (item) => (item as string).trim(),
  onProgress: (progress) => {
    console.log(`${progress.completed}/${progress.total} (${progress.failed} failed)`);
  },
});

concurrency defaults to unlimited; timeout and retry apply per item (maxAttempts counts the first try).
Without continueOnError, the first failed item fails the whole map with its error (the error carries partialResults). With it, failed items appear in the results with success: false and error.
filter runs before mapping and drops items; transform changes an item right before its mapper call.
Every item produces a MapItemResult (index, item, result, success, error, duration), in input order.

Reduce Options

Option	Description
`initial`	Starting accumulator, or a function of the state
`reducer`	`(accumulator, item: MapItemResult, state) => accumulator`
`successOnly`	Reduce only successful items (default `true`)
`finalize`	Last transformation of the accumulator
`streaming`	Reduce items as they complete (in input order) instead of after the whole map

Built-in Reducers

The package includes common reducer factories:

import {
  collect,
  sum,
  count,
  first,
  last,
  groupBy,
  partition,
  flatMap,
  stats,
} from '@cogitator-ai/workflows';

const asArray = collect<string>();

const total = sum();

const howMany = count();

const firstResult = first<string>();

const lastResult = last<string>();

const grouped = groupBy<{ category: string }, string>((result) => result.category);

const split = partition<number>((result) => result > 0.5); // { pass, fail }

const flattened = flatMap<string>(); // each item returns string[]

const statistics = stats(); // { count, sum, avg, min, max }

Convenience Helpers

parallelMap, sequentialMap and batchedMap run a map with unlimited, single or batchSize concurrency and return the MapItemResult array:

import { parallelMap, sequentialMap, batchedMap } from '@cogitator-ai/workflows';

const pages = await parallelMap(
  state,
  (s) => s.urls,
  async (url) => fetch(url as string).then((r) => r.text())
);

const ordered = await sequentialMap(
  state,
  (s) => s.urls,
  async (url) => processItem(url)
);

const batched = await batchedMap(
  state,
  (s) => s.urls,
  async (url) => processItem(url),
  5,
  { continueOnError: true }
);

Sub-Workflows

Execute entire workflows as nodes within a parent workflow. Supports state mapping between parent and child, error strategies, depth limits, and timeout.

Basic Sub-Workflow

import { subworkflowNode, subworkflowWorkflowNode } from '@cogitator-ai/workflows';

type ParentState = {
  query: string;
  detailedAnalysis?: string;
};

type ChildState = {
  input: string;
  output?: string;
};

const subConfig = subworkflowNode<ParentState, ChildState>('deep-analysis', {
  workflow: analysisWorkflow,
  inputMapper: (parentState) => ({ input: parentState.query }),
  outputMapper: (childResult, parentState) => ({
    ...parentState,
    detailedAnalysis: childResult.state.output,
  }),
  onError: 'retry',
  retryConfig: { maxAttempts: 3, delay: 2000, backoff: 'exponential' },
  maxDepth: 5,
  timeout: 60000,
});

builder.addNode('deep-analysis', subworkflowWorkflowNode(subConfig));

The state outputMapper returns replaces the parent state, so spread parentState into it. A child run that fails (result.error) counts as a failed attempt, so retryConfig / onError apply; a timed-out child run is cancelled, and aborting the parent run cancels the child. Nested depth is tracked automatically through the node context — a child deeper than maxDepth (default 10) throws MaxDepthExceededError.

executeSubworkflow(parentState, config, context) runs a config outside a workflow; it returns a SubworkflowResult (success, parentState, childResult, error, skipped, duration, depth). onStart, onComplete and onChildError hooks observe the child, and subworkflowWorkflowNode(config, { checkpointStore }) shares a checkpoint store with child runs (shareCheckpoints: false opts out).

Error Strategies

Strategy	Behavior
`'propagate'`	Re-throw the child error to the parent (default)
`'catch'`	`executeSubworkflow` returns `{ success: false, error }` without throwing
`'retry'`	Retry according to `retryConfig` (default 3 attempts, 1 s apart, linear), then re-throw
`'ignore'`	Return success with the parent state unchanged

Inside a workflow, subworkflowWorkflowNode with 'catch' lets the parent run continue: the parent state stays as it was and the node's output is { error: { name, message } }, so the node after it receives the error as ctx.input. With 'ignore' the node completes normally, also with the parent state unchanged.

Simple Sub-Workflow

When parent and child share the same state shape:

import { simpleSubworkflow } from '@cogitator-ai/workflows';

const sub = simpleSubworkflow('validation', validationWorkflow, {
  timeout: 30000,
});

Nested Sub-Workflow

Run the child on one key of the parent state; its final state is written back under that key:

import { nestedSubworkflow } from '@cogitator-ai/workflows';

type OrderState = { items: string[]; total?: number };

type AppState = {
  userId: string;
  orders: OrderState;
};

const sub = nestedSubworkflow<AppState, 'orders'>('process-orders', orderWorkflow, 'orders');

Conditional Sub-Workflow

The child runs only when condition holds; otherwise the node is skipped (output { skipped: true }) and the parent state is unchanged:

import { conditionalSubworkflow } from '@cogitator-ai/workflows';

type ReportState = { rawData: string; needsDeepAnalysis: boolean; extra?: unknown };
type ExtraState = { data: string; result?: string };

const sub = conditionalSubworkflow<ReportState, ExtraState>('optional-step', {
  workflow: optionalWorkflow,
  condition: (state) => state.needsDeepAnalysis,
  inputMapper: (state) => ({ data: state.rawData }),
  outputMapper: (result, state) => ({ ...state, extra: result.state.result }),
});

Parallel Sub-Workflows

Execute multiple sub-workflows concurrently with aggregation. The second type parameter is the children's state, so inputs and results stay typed. Add the configs to a builder with parallelSubworkflowsNode (its output is the run's stats: total, successful, failed, skipped), or run them with executeParallelSubworkflows(state, config, context).

Fan-Out / Fan-In

Distribute the same workflow across multiple inputs:

import { fanOutFanIn, parallelSubworkflowsNode } from '@cogitator-ai/workflows';

type BatchState = {
  items: string[];
  results: Record<string, string>;
};

type ItemState = { input: string; output?: string };

const config = fanOutFanIn<BatchState, ItemState>('process-batch', {
  workflow: itemWorkflow,
  getInputs: (state) =>
    state.items.map((item, i) => ({
      id: `item-${i}`,
      input: { input: item },
    })),
  aggregator: (results, state) => {
    const collected: Record<string, string> = {};
    for (const [id, result] of results) {
      collected[id] = result.state.output ?? '';
    }
    return { ...state, results: collected };
  },
  concurrency: 4,
  continueOnError: true,
});

builder.addNode('process-batch', parallelSubworkflowsNode(config));

Scatter-Gather

Send work to different specialized workflows and gather all results:

import { scatterGather } from '@cogitator-ai/workflows';

type AnalysisState = {
  text: string;
  sentiment?: string;
  entities?: string;
  topics?: string;
};

const config = scatterGather<AnalysisState, ItemState>('multi-analysis', {
  workflows: new Map([
    ['sentiment', sentimentWorkflow],
    ['entities', entityWorkflow],
    ['topics', topicWorkflow],
  ]),
  inputMapper: (state) => ({ input: state.text }),
  outputMapper: (results, state) => ({
    ...state,
    sentiment: results.get('sentiment')?.state.output,
    entities: results.get('entities')?.state.output,
    topics: results.get('topics')?.state.output,
  }),
  timeout: 30000,
});

scatterGather continues past failed children by default (continueOnError: true); the aggregators only see children that completed.

Custom Parallel Sets

parallelSubworkflows takes the definitions directly — a list or a function of the state — each with its own SubworkflowConfig:

import { parallelSubworkflows, subworkflowNode } from '@cogitator-ai/workflows';

const config = parallelSubworkflows<AnalysisState, ItemState>('checks', {
  subworkflows: (state) =>
    ['spelling', 'tone'].map((id) => ({
      id,
      config: subworkflowNode<AnalysisState, ItemState>(`check-${id}`, {
        workflow: id === 'spelling' ? spellingWorkflow : toneWorkflow,
        inputMapper: () => ({ input: state.text }),
        outputMapper: (_child, parentState) => parentState,
      }),
    })),
  concurrency: 2,
  onError: 'catch',
  continueOnError: true,
  aggregator: (results, state) => ({
    ...state,
    topics: [...results.keys()].join(', '),
  }),
  onProgress: (p) => console.log(`${p.completed}/${p.total}`),
});

Without continueOnError, the first failing child stops the set and fails the node. With it, the aggregator runs when at least one child succeeded.

Race and Fallback

raceSubworkflows runs several configs at once and returns the first successful result (cancelling the others), or null if none succeeds. fallbackSubworkflows tries them in order until one succeeds. Both take a SubworkflowContext, which a custom node builds from its runtime context:

import {
  customNode,
  fallbackSubworkflows,
  raceSubworkflows,
  type ExtendedNodeContext,
  type SubworkflowContext,
} from '@cogitator-ai/workflows';

const answer = customNode<ParentState>('answer', async (ctx) => {
  const ext = ctx as ExtendedNodeContext<ParentState>;
  const context: SubworkflowContext = {
    cogitator: ext.cogitator,
    parentWorkflowId: ctx.workflowId,
    parentRunId: ctx.workflowId,
    parentNodeId: ctx.nodeId,
    depth: ext.depth ?? 0,
    signal: ext.signal,
  };

  const fastest = await raceSubworkflows(ctx.state, [fastModelConfig, slowModelConfig], context);
  if (fastest) return { state: fastest.parentState };

  const fallback = await fallbackSubworkflows(
    ctx.state,
    [primaryConfig, lastResortConfig],
    context
  );
  return { state: fallback.parentState };
});

Conditional Branching

Use WorkflowBuilder.addConditional for state-based routing:

const workflow = new WorkflowBuilder<ReviewState>('review')
  .initialState({ code: '', severity: 'low' })
  .addNode('analyze', analyzeNode)
  .addConditional(
    'route',
    (state) => {
      if (state.severity === 'critical') return 'emergency-fix';
      if (state.severity === 'high') return 'senior-review';
      return 'auto-approve';
    },
    { after: ['analyze'] }
  )
  .addNode('emergency-fix', emergencyNode, { after: ['route'] })
  .addNode('senior-review', seniorNode, { after: ['route'] })
  .addNode('auto-approve', approveNode, { after: ['route'] })
  .build();

Multi-target routing returns several branches; every possible target lists the conditional in its after:

builder
  .addConditional(
    'fork',
    (state) => {
      const targets: string[] = ['log'];
      if (state.needsNotification) targets.push('notify');
      if (state.needsArchive) targets.push('archive');
      return targets;
    },
    { after: ['analyze'] }
  )
  .addNode('log', logNode, { after: ['fork'] })
  .addNode('notify', notifyNode, { after: ['fork'] })
  .addNode('archive', archiveNode, { after: ['fork'] });

Loops

Use WorkflowBuilder.addLoop for iterative refinement:

type DraftState = {
  draft: string;
  score: number;
  iteration: number;
};

const workflow = new WorkflowBuilder<DraftState>('iterative-writer')
  .initialState({ draft: '', score: 0, iteration: 0 })
  .addNode(
    'write',
    agentNode(writer, {
      inputMapper: (state) =>
        state.draft
          ? `Improve this draft:\n${state.draft}`
          : 'Write an initial draft about AI agents',
      stateMapper: (r) => ({ draft: r.output }),
    })
  )
  .addNode(
    'evaluate',
    agentNode(critic, {
      inputMapper: (state) => `Score 1-10, reply with the number only:\n${state.draft}`,
      stateMapper: (r) => ({ score: parseInt(r.output, 10) || 0 }),
    }),
    { after: ['write'] }
  )
  .addNode(
    'count',
    functionNode('count', async (state) => state.iteration + 1, {
      stateMapper: (n) => ({ iteration: n as number }),
    }),
    { after: ['evaluate'] }
  )
  .addLoop('quality-gate', {
    condition: (state) => state.score < 8 && state.iteration < 5,
    back: 'write',
    exit: 'publish',
    after: ['count'],
  })
  .addNode(
    'publish',
    functionNode('publish', async (state) => {
      return { finalDraft: state.draft, iterations: state.iteration };
    })
  )
  .build();

The loop checks the condition after count completes. If true, execution jumps back to write. If false, it proceeds to publish. The maxIterations option on WorkflowExecutor (default 100 steps) provides a safety net against infinite loops.

Parallel Execution

Use WorkflowBuilder.addParallel to fan out, then converge with shared dependencies:

const workflow = new WorkflowBuilder<MultiSearchState>('parallel-search')
  .initialState({ query: '' })
  .addNode('prepare', prepareNode)
  .addParallel('search', ['web', 'academic', 'github'], { after: ['prepare'] })
  .addNode('web', webSearchNode)
  .addNode('academic', academicSearchNode)
  .addNode('github', githubSearchNode)
  .addNode(
    'merge',
    functionNode('merge', async (state) => {
      return {
        webResults: state.webResults,
        academicResults: state.academicResults,
        githubResults: state.githubResults,
      };
    }),
    { after: ['web', 'academic', 'github'] }
  )
  .build();

The merge node has after: ['web', 'academic', 'github'] so it waits for all parallel branches to complete before executing — even when branches have different lengths, the join runs exactly once, after every upstream node that can still reach it has finished. Its ctx.input contains the outputs of all its dependencies.

Workflow Patterns

Overview

MapReduce

Basic MapReduce

Map Options

Reduce Options

Built-in Reducers

Convenience Helpers

Sub-Workflows

Basic Sub-Workflow

Error Strategies

Simple Sub-Workflow

Nested Sub-Workflow

Conditional Sub-Workflow

Parallel Sub-Workflows

Fan-Out / Fan-In

Scatter-Gather

Custom Parallel Sets

Race and Fallback

Conditional Branching

Loops

Parallel Execution

On this page

Workflow Patterns

OverviewCopy Anchor Link

MapReduceCopy Anchor Link

Basic MapReduceCopy Anchor Link

Map OptionsCopy Anchor Link

Reduce OptionsCopy Anchor Link

Built-in ReducersCopy Anchor Link

Convenience HelpersCopy Anchor Link

Sub-WorkflowsCopy Anchor Link

Basic Sub-WorkflowCopy Anchor Link

Error StrategiesCopy Anchor Link

Simple Sub-WorkflowCopy Anchor Link

Nested Sub-WorkflowCopy Anchor Link

Conditional Sub-WorkflowCopy Anchor Link

Parallel Sub-WorkflowsCopy Anchor Link

Fan-Out / Fan-InCopy Anchor Link

Scatter-GatherCopy Anchor Link

Custom Parallel SetsCopy Anchor Link

Race and FallbackCopy Anchor Link

Conditional BranchingCopy Anchor Link

LoopsCopy Anchor Link

Parallel ExecutionCopy Anchor Link

On this page

Overview

MapReduce

Basic MapReduce

Map Options

Reduce Options

Built-in Reducers

Convenience Helpers

Sub-Workflows

Basic Sub-Workflow

Error Strategies

Simple Sub-Workflow

Nested Sub-Workflow

Conditional Sub-Workflow

Parallel Sub-Workflows

Fan-Out / Fan-In

Scatter-Gather

Custom Parallel Sets

Race and Fallback

Conditional Branching

Loops

Parallel Execution