# Sagas & Compensation

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

> Retry policies, circuit breakers, compensation rollbacks, dead letter queues, and idempotency for resilient workflows.

## Overview

The saga module provides fault tolerance primitives for workflow nodes. When a multi-step workflow fails partway through, you need strategies to retry, isolate faults, roll back completed steps, and handle permanently failed operations.

```typescript
import {
  executeWithRetry,
  withRetry,
  CircuitBreaker,
  CompensationManager,
  InMemoryDLQ,
} from '@cogitator-ai/workflows';
```

## In the Executor

The executor applies retries, circuit breaking, dead-lettering and idempotency to every node when you pass them as execute options — no wrapping needed:

```typescript
const result = await executor.execute(workflow, input, {
  defaultRetry: { maxRetries: 3, backoff: 'exponential', initialDelay: 1000 },
  defaultCircuitBreaker: { threshold: 5, resetTimeout: 30_000 },
  deadLetterQueue: dlq,
  idempotencyStore: store,
});
```

A node's own `config.retries` / `retryDelay` take precedence over `defaultRetry`. Compensation is configured per node with `config.compensation` (see [Compensation in the Executor](#compensation-in-the-executor)). The primitives below also work on their own, outside workflows.

## Retry with Backoff

`executeWithRetry` wraps any async function with configurable retry logic, supporting constant, linear, and exponential backoff with jitter.

```typescript
import { executeWithRetry } from '@cogitator-ai/workflows';

const result = await executeWithRetry(
  async (attempt) => {
    const response = await fetch('https://api.example.com/data');
    if (!response.ok) throw new Error(`HTTP ${response.status}`);
    return response.json();
  },
  {
    maxRetries: 3,
    backoff: 'exponential',
    initialDelay: 1000,
    maxDelay: 30000,
    multiplier: 2,
    jitter: 0.1,
    onRetry: (info) => {
      console.log(`Attempt ${info.attempt} failed, retrying in ${info.delay}ms`);
    },
  }
);

if (result.success) {
  console.log('Data:', result.result);
} else {
  console.error(`Failed after ${result.attempts} attempts:`, result.error);
}
```

### RetryResult

```typescript
interface RetryResult<T> {
  success: boolean;
  result?: T;
  error?: Error;
  attempts: number;
  totalDuration: number;
  delays: number[];
}
```

### Backoff Strategies

| Strategy      | Behavior                | Example delays (1s base) |
| ------------- | ----------------------- | ------------------------ |
| `constant`    | Same delay every time   | 1s, 1s, 1s               |
| `linear`      | Delay grows linearly    | 1s, 2s, 3s               |
| `exponential` | Delay doubles each time | 1s, 2s, 4s               |

Jitter adds +/- randomness to prevent thundering herd. A jitter of `0.1` means +/-10% of the base delay.

### Retryable Error Detection

By default only transient errors are retried: network error codes (`ECONNRESET`, `ETIMEDOUT`, ...), a `statusCode` of 429/500/502/503/504, or a message mentioning a timeout, network or connection failure, a rate limit or one of those status codes. Any other error fails at once. Override with `isRetryable`:

```typescript
const result = await executeWithRetry(fn, {
  maxRetries: 5,
  isRetryable: (error) => {
    return error.message.includes('ECONNRESET') || error.message.includes('429');
  },
});
```

### withRetry Wrapper

Create a retryable version of any async function (it throws the last error when every attempt fails). `Retryable(options)` does the same as a method decorator, and `estimateRetryDuration(config)` returns the worst-case total delay of a policy.

```typescript
import { withRetry } from '@cogitator-ai/workflows';

const fetchWithRetry = withRetry(
  async (url: string) => {
    const res = await fetch(url);
    return res.json();
  },
  { maxRetries: 3, backoff: 'exponential', initialDelay: 500 }
);

const data = await fetchWithRetry('https://api.example.com/data');
```

## Circuit Breaker

The circuit breaker pattern prevents cascading failures by stopping requests to a failing service after a threshold is reached.

```typescript
import { CircuitBreaker, createCircuitBreaker } from '@cogitator-ai/workflows';

const breaker = createCircuitBreaker({
  threshold: 5,
  resetTimeout: 30000,
  successThreshold: 2,
  halfOpenMax: 3,
});
```

### States

```
CLOSED ──(failures >= threshold)──→ OPEN ──(timeout elapsed)──→ HALF_OPEN
  ↑                                                                │
  └──────────(successes >= successThreshold)────────────────────────┘
```

* **Closed** -- Normal operation. Failures are counted. Opens after `threshold` failures.
* **Open** -- All requests fail immediately with `CircuitBreakerOpenError`. After `resetTimeout` ms, transitions to half-open.
* **Half-open** -- Limited test requests allowed (`halfOpenMax`). If `successThreshold` consecutive successes, closes. Any failure reopens.

State is kept per id (the first argument of `execute`), so one breaker can guard several services. The defaults are `threshold: 5`, `resetTimeout: 30000`, `successThreshold: 2`, `halfOpenMax: 3`; `enabled: false` lets every call through.

### Usage

```typescript
import { CircuitBreakerOpenError } from '@cogitator-ai/workflows';

async function fetchPrices(): Promise<Prices> {
  try {
    return await breaker.execute('external-api', () => callExternalAPI());
  } catch (error) {
    if (error instanceof CircuitBreakerOpenError) {
      console.log(`Circuit ${error.state} for ${error.nodeId}, using fallback`);
      return fallbackPrices;
    }
    throw error;
  }
}
```

### Monitoring

```typescript
breaker.onEvent((event) => {
  switch (event.type) {
    case 'state_change':
      console.log(`${event.nodeId}: ${event.from} → ${event.to}`);
      break;
    case 'failure':
      console.error(`${event.nodeId} failed:`, event.error.message);
      break;
    case 'rejected':
      console.warn(`${event.nodeId} rejected: ${event.reason}`);
      break;
  }
});

const stats = breaker.getStats('external-api');
console.log(`State: ${stats.state}, Failures: ${stats.totalFailures}`);
```

## Compensation in the Executor

Give a node `config.compensation` and the executor rolls it back when a later node of the same run fails:

```typescript
import { WorkflowBuilder } from '@cogitator-ai/workflows';
import type { WorkflowState } from '@cogitator-ai/types';

interface OrderState extends WorkflowState {
  orderId: string;
  reservationId?: string;
  chargeId?: string;
}

const checkout = new WorkflowBuilder<OrderState>('checkout')
  .initialState({ orderId: '' })
  .addNode(
    'reserve',
    async (ctx) => {
      const reservation = await inventory.reserve(ctx.state.orderId);
      return { state: { reservationId: reservation.id }, output: reservation };
    },
    {
      config: {
        compensation: {
          compensate: async (_state, originalResult) => {
            await inventory.release((originalResult as { id: string }).id);
          },
        },
      },
    }
  )
  .addNode(
    'charge',
    async (ctx) => {
      const charge = await payments.charge(ctx.state.orderId);
      return { state: { chargeId: charge.id }, output: charge };
    },
    {
      after: ['reserve'],
      config: {
        compensation: {
          compensate: async (state) => payments.refund(state.chargeId!),
          compensateCondition: (_state, error) => error.message !== 'ALREADY_REFUNDED',
          compensateTimeout: 10_000,
        },
      },
    }
  )
  .addNode('ship', async (ctx) => ({ output: await shipping.ship(ctx.state.orderId) }), {
    after: ['charge'],
  })
  .build();

const result = await executor.execute(
  checkout,
  { orderId: 'ord_1' },
  {
    onCompensationStart: (nodeId) => console.log(`Rolling back ${nodeId}`),
    onCompensationComplete: (nodeId) => console.log(`Rolled back ${nodeId}`),
  }
);
```

* `compensation` is a `CompensationConfig`: `compensate(state, originalResult)` (required to register anything; `originalResult` is the node's output), `compensateCondition?(state, error)` to skip a step, `compensateOrder?` (`'reverse'` by default, `'parallel'` or `'forward'`) and `compensateTimeout?` in ms.
* It runs when a node fails after its retries: every completed node that has a `compensate` is rolled back (the failed node itself is not), in reverse completion order by default, once the nodes still in flight have settled. Nodes passed over on a resume count as completed. An aborted run — `signal`, or a `WorkflowManager` pause or cancel — is not compensated.
* `onCompensationStart(nodeId)` and `onCompensationComplete(nodeId)` fire around each step that runs (steps skipped by `compensateCondition` report neither); `onCompensationComplete` fires whether the step succeeded or not. A failing step is logged and does not stop the others; `result.error` stays the node's original error.

## Compensation Manager

`CompensationManager` is the same rollback logic as a standalone object, for code outside the executor or for custom flows. When a multi-step workflow fails, compensation rolls back already-completed steps in reverse order.

```typescript
import { CompensationManager, createCompensationManager } from '@cogitator-ai/workflows';

type OrderState = {
  orderId: string;
  reservationId?: string;
  chargeId?: string;
};

const manager = createCompensationManager<OrderState>();

manager.registerCompensation('reserve-inventory', async (state) => {
  await inventoryService.release(state.reservationId!);
});

manager.registerCompensation(
  'charge-payment',
  async (state) => {
    await paymentService.refund(state.chargeId!);
  },
  {
    retries: 2,
    timeout: 10000,
  }
);
```

### Recording Completions

As nodes complete, mark them so the manager knows which steps to compensate. The value you pass is handed to the compensation function as its second argument (`originalResult`):

```typescript
manager.markCompleted('reserve-inventory', { reservationId: 'res_123' });
manager.markCompleted('charge-payment', { chargeId: 'ch_456' });
```

### Triggering Compensation

When a downstream step fails:

```typescript
const report = await manager.compensate(
  currentState,
  'ship-order',
  new Error('Shipping service unavailable')
);

console.log(`Compensated ${report.compensated.length} steps`);
console.log(`All successful: ${report.allSuccessful}`);

if (report.partialFailures.length > 0) {
  console.error('Failed to compensate:', report.partialFailures);
}
```

Only steps that were marked completed (and are not the failed step) are compensated. Compensation executes in reverse completion order by default; each step can set `order: 'parallel'` (consecutive parallel steps run together) or `'forward'` (run last, in completion order). A step that throws is retried `retries` times with a short backoff, and `timeout` fails a step that takes too long; failed steps are listed in `partialFailures` instead of stopping the rest. An optional fourth argument, `CompensationHooks` (`{ onStepStart?(nodeId), onStepComplete?(result) }`), observes each step that runs — the executor's `onCompensationStart` / `onCompensationComplete` are built on it. `getSummary()` and `reset()` help when you reuse a manager across runs, and `registerFromConfig(nodeId, { compensate, compensateCondition, compensateOrder, compensateTimeout })` registers from a `CompensationConfig`.

### CompensationBuilder

Fluent API for building compensation configurations:

```typescript
import { compensationBuilder } from '@cogitator-ai/workflows';

const manager = compensationBuilder<OrderState>()
  .addStep('reserve', async (state) => {
    await inventoryService.release(state.reservationId!);
  })
  .addConditionalStep(
    'charge',
    async (state) => {
      await paymentService.refund(state.chargeId!);
    },
    (state, error) => {
      return error.message !== 'ALREADY_REFUNDED';
    }
  )
  .build();
```

## Dead Letter Queue

Failed operations that exhaust all retries go to a DLQ for manual inspection and replay. With the `deadLetterQueue` execute option the executor adds an entry (state, input, error, attempts) for every node that finally fails.

```typescript
import { InMemoryDLQ, FileDLQ, createDLQEntry } from '@cogitator-ai/workflows';

const dlq = new InMemoryDLQ({ defaultTTL: 7 * 24 * 60 * 60 * 1000 });

const entry = createDLQEntry(
  'process-payment',
  'wf_abc123',
  'checkout-workflow',
  currentState,
  new Error('Payment gateway timeout'),
  { attempts: 3, maxAttempts: 3, tags: ['payment', 'critical'] }
);

const id = await dlq.add(entry);
```

### Querying

```typescript
const entries = await dlq.list({
  workflowName: 'checkout-workflow',
  minAttempts: 3,
  tags: ['critical'],
  limit: 10,
});

for (const entry of entries) {
  console.log(`${entry.nodeId}: ${entry.error.message} (${entry.attempts} attempts)`);
}
```

`get(id)`, `remove(id)`, `count(filters)` and `clear()` manage entries. `retry(id)` records another attempt on the entry (`attempts`, `lastAttempt`); `manager.retryDeadLetter(dlq, id)` runs the node again and calls it. Entries expire after `defaultTTL` (7 days by default), and the in-memory queue keeps at most `maxSize` entries (10 000).

### Retrying a dead letter

`manager.retryDeadLetter(dlq, id)` runs a dead-lettered node again. It replays the node's run from the failed node (the nodes before it keep their checkpointed results and do not run twice), or runs the workflow again from its input when the run failed before its first checkpoint. The attempt is recorded on the entry first, and the entry is removed when the retry succeeds, so an entry that fails again stays with one more attempt. The workflow must be registered with the manager and the manager needs a `checkpointStore`. Pass the execute options the run needs, such as `approvalStore` or `timerStore`, as the third argument:

```typescript
const manager = createWorkflowManager({ cogitator, checkpointStore, runStore });
await manager.execute(checkout, input, { deadLetterQueue: dlq });

for (const entry of await dlq.list({ workflowName: 'checkout-workflow' })) {
  const result = await manager.retryDeadLetter(dlq, entry.id, { approvalStore });
  if (result.error) console.log(`${entry.nodeId} failed again: ${result.error.message}`);
}
```

### Postgres DLQ

`PostgresDLQ` keeps the queue in Postgres, so failed nodes survive restarts and every process sees the same queue. It takes a `pg` Pool or Client, creates its table on first use, filters in SQL and deletes expired entries with `cleanupExpired()`:

```typescript
import pg from 'pg';
import { PostgresDLQ } from '@cogitator-ai/workflows';

const dlq = new PostgresDLQ({
  client: new pg.Pool({ connectionString: process.env.DATABASE_URL }),
  table: 'checkout_dead_letters', // default cogitator_workflow_dead_letters
  defaultTTL: 14 * 24 * 60 * 60 * 1000,
});
```

### File-based DLQ

For persistence across restarts:

```typescript
const dlq = new FileDLQ('/var/data/dlq', { defaultTTL: 30 * 24 * 60 * 60 * 1000 });
```

## Idempotency

Ensure operations are safe to retry by preventing duplicate execution:

```typescript
import { InMemoryIdempotencyStore, generateCustomKey, idempotent } from '@cogitator-ai/workflows';

const store = new InMemoryIdempotencyStore({ ttl: 24 * 60 * 60 * 1000 });

async function processPayment(orderId: string, amount: number) {
  const key = generateCustomKey('payment', orderId, amount);
  return idempotent(store, key, () => paymentGateway.charge(orderId, amount));
}

// safe to call multiple times
await processPayment('order_123', 99.99);
await processPayment('order_123', 99.99);
```

`idempotent(store, key, fn)` runs `fn` once per key: a later call returns the stored result, or rethrows the stored error if the first call failed, and concurrent calls with the same key share one execution. `generateIdempotencyKey(workflowId, nodeId, input?)` and `generateCustomKey(...parts)` build stable hashed keys (object keys are sorted first). `Idempotent(store, keyFn)` is the method-decorator form; `FileIdempotencyStore(directory)` keeps records on disk. Records expire after `ttl` (24 hours by default).