# Workflow Patterns

URL: https://cogitator.app/docs/workflows/patterns

> 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

```typescript
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:

```typescript
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 `MapItemResult`s), and `executeMap` / `executeReduce` run the phases directly.

### Map Options

```typescript
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:

```typescript
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:

```typescript
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

```typescript
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:

```typescript
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:

```typescript
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:

```typescript
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:

```typescript
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:

```typescript
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`:

```typescript
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:

```typescript
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:

```typescript
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`:

```typescript
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:

```typescript
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:

```typescript
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.