---
title: Pipelines
description: "Build durable multi-step workflows with DAG-based orchestration."
---

# Pipelines

Pipelines enable complex, multi-step agent workflows that survive server restarts, support human approval gates, and provide full observability into execution progress.

## Why Pipelines?

Running an AI agent to "clone a repo and run tests" is simple. But real-world workflows often require:

- **Sequential steps** — build before test before deploy
- **Parallel execution** — run linting and tests simultaneously
- **Human approval** — require sign-off before production deploy
- **Retry logic** — automatically retry flaky network calls
- **Crash recovery** — resume interrupted workflows after server restart
- **Shared state** — pass artifacts between stages

Pipelines provide all of this in a single, declarative API.

## Core Concepts

### Pipeline
A named, versioned workflow definition. Immutable once created — changes create a new version.

### Stage
A single unit of work in a pipeline. Stages can be:
- **command** — run a shell command in a sandbox
- **await_approval** — pause for human approval
- **gate** — check a condition (trust score, cost threshold)
- **eval** — run a Cage Eval scenario

### Run
A single execution of a pipeline with specific inputs. Runs are durable — they survive server restarts and can be paused indefinitely for approval.

### State
Key/value store scoped to a run. Stages can write artifacts (file paths, computed values) that subsequent stages read.

## Quick Start

### 1. Create a Pipeline

```bash
curl -X POST https://api.caged.dev/v1/pipelines \
  -H "Authorization: Bearer caged_sk_..." \
  -H "Content-Type: application/json" \
  -d '{
    "name": "ci-pipeline",
    "stages": [
      {"name": "build", "type": "command", "command": "npm run build"},
      {"name": "test", "type": "command", "command": "npm test", "depends_on": ["build"]},
      {"name": "lint", "type": "command", "command": "npm run lint", "depends_on": ["build"]},
      {"name": "deploy", "type": "command", "command": "npm run deploy", "depends_on": ["test", "lint"]}
    ],
    "defaults": {"template": "node-20"}
  }'
```

This creates a diamond DAG:
```
        build
       /     \
     test    lint
       \     /
       deploy
```

`test` and `lint` run in parallel after `build`. `deploy` waits for both.

### 2. Start a Run

```bash
curl -X POST https://api.caged.dev/v1/pipelines/pipe-abc123/runs \
  -H "Authorization: Bearer caged_sk_..." \
  -d '{"repo": "https://github.com/myorg/myapp", "branch": "main"}'
```

### 3. Monitor Progress

```bash
curl https://api.caged.dev/v1/pipelines/pipe-abc123/runs/run-xyz789/stages \
  -H "Authorization: Bearer caged_sk_..."
```

## Adding Approval Gates

Require human sign-off before sensitive operations:

```json
{
  "name": "approve-prod",
  "type": "await_approval",
  "depends_on": ["test"],
  "config": {
    "message": "Tests passed. Deploy to production?",
    "channels": ["slack", "dashboard"],
    "sla_timeout": "1h"
  }
}
```

When this stage is reached:
1. Run status changes to `paused`
2. Notifications sent to configured channels
3. Approvers see the request in dashboard / Slack
4. On approval, run resumes automatically
5. On rejection, run is canceled

## Retry Policies

Handle transient failures automatically:

```json
{
  "name": "upload",
  "type": "command",
  "command": "aws s3 cp dist.zip s3://bucket/",
  "retry": {
    "max_attempts": 3,
    "backoff": "5s",
    "max_backoff": "30s"
  }
}
```

Failed stages retry with exponential backoff until `max_attempts` is exhausted.

## Conditional Execution

Run stages only when conditions are met:

```json
{
  "name": "notify-failure",
  "type": "command",
  "command": "slack-notify 'Build failed!'",
  "depends_on": ["build"],
  "condition": {"on_failure": true}
}
```

Condition options:
- `on_success` — run only if all dependencies succeeded
- `on_failure` — run only if any dependency failed
- `if` — custom expression (coming soon)

## On Failure Behavior

Control what happens when a stage fails:

```json
{
  "defaults": {
    "on_failure": "continue"
  }
}
```

- `stop` (default) — fail the entire run
- `continue` — mark stage failed but continue other branches

## Environment Variables

Pass environment to all stages:

```bash
curl -X POST https://api.caged.dev/v1/pipelines/pipe-abc123/runs \
  -H "Authorization: Bearer caged_sk_..." \
  -d '{
    "env": {
      "NODE_ENV": "production",
      "AWS_REGION": "us-east-1"
    }
  }'
```

Per-stage environment overrides pipeline-level:

```json
{
  "name": "deploy",
  "type": "command",
  "command": "deploy.sh",
  "env": {"DEPLOY_TARGET": "us-west-2"}
}
```

## Crash Recovery

Pipelines automatically recover from server restarts:

1. Server crashes mid-run
2. On restart, `RecoverRuns()` is called
3. Incomplete runs resume from last completed stage
4. Paused runs (awaiting approval) remain paused

No manual intervention required.

## Events and Observability

Pipeline lifecycle events are emitted to the event pipeline:

- `pipeline.run.created`
- `pipeline.run.started`
- `pipeline.run.completed`
- `pipeline.run.failed`
- `pipeline.stage.started`
- `pipeline.stage.completed`
- `pipeline.stage.failed`

These integrate with:
- Session replay — see full stage execution
- Cost tracking — aggregate LLM + compute costs
- Trust scoring — score agent behavior per stage

## Best Practices

### 1. Keep Stages Atomic
Each stage should do one thing. Don't put "build and test" in a single command.

### 2. Use Dependencies Explicitly
Even if stages naturally run in order, declare `depends_on` for clarity and parallelization.

### 3. Set Appropriate Timeouts
Don't let runaway commands block pipelines. Default timeout is 10 minutes.

```json
{"timeout": "30m"}
```

### 4. Add Approval Gates for Production
Never auto-deploy to production. Add an `await_approval` stage.

### 5. Use Gate Stages for Guardrails
Check trust scores or cost limits before expensive operations:

```json
{
  "name": "cost-check",
  "type": "gate",
  "config": {"cost_below": 10.00}
}
```

## SDK Usage

### TypeScript

```typescript
import { Caged } from '@caged-dev/sdk';

const caged = new Caged({ apiKey: 'caged_sk_...' });

// Create pipeline
const pipeline = await caged.pipelines.create({
  name: 'my-pipeline',
  stages: [
    { name: 'build', type: 'command', command: 'npm run build' },
    { name: 'test', type: 'command', command: 'npm test', dependsOn: ['build'] }
  ]
});

// Start run
const run = await caged.pipelines.startRun(pipeline.id, {
  repo: 'https://github.com/myorg/myapp',
  branch: 'main'
});

// Poll status
let status = run.status;
while (status === 'pending' || status === 'running') {
  await new Promise(r => setTimeout(r, 5000));
  const updated = await caged.pipelines.getRun(pipeline.id, run.id);
  status = updated.status;
}

console.log(`Run completed with status: ${status}`);
```

### Python

```python
from caged import Caged

caged = Caged(api_key="caged_sk_...")

# Create pipeline
pipeline = caged.pipelines.create(
    name="my-pipeline",
    stages=[
        {"name": "build", "type": "command", "command": "npm run build"},
        {"name": "test", "type": "command", "command": "npm test", "depends_on": ["build"]}
    ]
)

# Start run
run = caged.pipelines.start_run(pipeline.id, repo="https://github.com/myorg/myapp")

# Wait for completion
run.wait()
print(f"Run completed with status: {run.status}")
```

## Next Steps

- [Pipelines API Reference](/api-reference/pipelines) — full endpoint documentation
- [Approvals Guide](/guides/approvals) — HITL approval configuration
- [Eval Guide](/guides/eval) — integrate Cage Eval into pipelines
- [CI Pipeline Recipe](/recipes/ci-pipeline) — full CI/CD example