Guides
View as Markdown Open in ChatGPT Open in Claude

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

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

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

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:

{
  "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:

{
  "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:

{
  "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:

{
  "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:

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:

{
  "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.

{"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:

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

SDK Usage

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

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

Was this page helpful?