pipeline

name	pipeline
description	This skill should be used when the user asks to "create a pipeline", "add a pipeline step", "configure pipeline LLM settings", "debug a pipeline", "design a pipeline architecture", or works with claudi pipeline definitions in `.pipelines/definitions/`. Covers the builder API, step types, StepContext, InputSchema, LLMConfig, and pipeline-level configuration.

Claudi Pipeline Authoring

A pipeline is a declarative execution plan that orchestrates Claude agents and TypeScript scripts into reproducible, resumable workflows. Pipeline files live in .pipelines/definitions/*.ts and export a Pipeline object built with the fluent builder API.

.pipelines/
  definitions/
    release.ts        # Named export: export const release = buildPipeline({...}).build()
    test-unit.ts      # Named export: export const testUnit = buildPipeline({...}).build()
    _shared.ts        # Underscore prefix = not a pipeline, shared helpers

Files prefixed with _ are ignored during discovery.

import { buildPipeline, NestedBuilder, z } from 'claudi/sdk';
import type { PipelineStep, StepContext } from 'claudi/sdk';

Design Principle: Scripts First, Agents Only When Necessary

Agent steps are expensive, non-deterministic, and slow. Script steps are free, deterministic, and fast. Default to scripts. Use agents only for tasks that genuinely require reasoning, creativity, or natural language understanding.

Before creating an agent step, ask: "Can this be done with TypeScript code?" If yes, make it a script.

Offload to scripts:

• File discovery, glob scanning, directory traversal
• Running shell commands (bun test, tsc, eslint, git)
• Parsing output (JSON, coverage reports, test results)
• Triage and classification with clear rules
• Data transformation between steps
• File I/O (reading templates, writing generated output)
• Validation and pre-checks
• Aggregation of results from parallel branches

Reserve agents for:

• Understanding code semantics (what does this function do?)
• Generating new code or tests
• Making judgment calls (is this a bug or intentional?)
• Creative tasks (writing documentation, designing APIs)
• Fixing issues that require reading and reasoning about context

Anti-pattern: Using an agent to read a file, extract data, and return JSON when a script with readFileSync + regex/AST parsing would be deterministic and instant.

Pattern: Sandwich agents between scripts — script to gather context, agent to reason, script to apply the result:

.script('gather', { execute: gatherContext, outputKey: 'context' })
.agent('reason', { prompt: (ctx) => `Given:\n${JSON.stringify(ctx.outputs.context)}`, ... })
.script('apply', { execute: (ctx) => applyChanges(ctx.outputs.reason), outputKey: 'result' })

Builder API

buildPipeline() returns a fluent PipelineBuilder that accumulates output types as steps are chained. Each .agent() or .script() call with outputKey + outputSchema extends the TOutputs type parameter.

const pipeline = buildPipeline({
  id: 'my-pipeline',
  version: '1.0.0',
  description: 'Does something useful',
  inputSchema: z.object({ target: z.string() }),
  maxBudgetUsd: 10,
})
  .agent('step-1', { ... })
  .script('step-2', { ... })
  .sequence('step-3', { ... })          // raw config or nested builder callback
  .parallel('step-4', { ... })          // raw config or nested builder callback
  .loop('refine', { ... })
  .map('fan-out', { ... })              // fan-out over dynamic list → parallel
  .gate('review', { ... })              // draft/critique/approve → loop
  .converge('implement', { ... })       // generate/fix convergence → loop
  .build();                             // => Pipeline<TInput, TOutputs>

Type Accumulation

When providing outputKey + outputSchema, the builder extends TOutputs:

.agent('analyze', {
  description: 'Analyze code',
  prompt: 'Analyze this code...',
  llmConfig: { model: 'sonnet', maxTurns: 10 },
  outputKey: 'analysis',
  outputSchema: z.object({ issues: z.array(z.string()) }),
})
// After this step, TOutputs = { analysis: { issues: string[] } }

When a sequence or parallel generates children dynamically, use .outputs<T>() to declare their types:

.parallel('dynamic-work', { steps: (ctx) => buildDynamicSteps(ctx) })
.outputs<{ [key: `task-${string}`]: TaskResult }>()

Nested Builder Callbacks

.sequence() and .parallel() accept a callback that receives a NestedBuilder. Output types accumulate through the chain:

.sequence('setup', (seq) =>
  seq
    .agent('scan', { ..., outputKey: 'scan', outputSchema: ScanSchema })
    .script('filter', { ..., outputKey: 'filtered' }),
  { description: 'Setup phase' }
)
// TOutputs now includes { scan: ScanOutput; filtered: unknown }

Pattern-Aware Primitives

High-level methods that compile to existing domain types (no runner changes):

Method	Compiles To	Use Case
.map(id, config)	parallel with dynamic steps function	Fan-out over a list of items
.gate(id, config)	loop with 3 agent children	Draft/critique/approve approval loop
.converge(id, config)	loop with iteration-aware agent + script	Generate/fix until convergence (e.g., test-fix)

// .map() — fan-out
.map('process', {
  description: 'Process each item',
  over: (ctx) => ctx.outputs.items!,       // list to iterate
  step: (item) => ({ type: 'agent', ... }), // step factory per item
  maxConcurrency: 3,
  outputKey: 'results',
  outputSchema: ResultSchema,
})

// .gate() — approval loop
.gate('review', {
  description: 'Review until approved',
  maxIterations: 3,
  draft:    { description: 'Write',    prompt: '...', outputKey: 'draft',    outputSchema: DraftSchema,    llmConfig: { ... } },
  critique: { description: 'Critique', prompt: '...', outputKey: 'critique', outputSchema: CritiqueSchema, llmConfig: { ... } },
  approve:  { description: 'Approve',  prompt: '...', outputKey: 'approval', outputSchema: ApprovalSchema, llmConfig: { ... } },
})

// .converge() — test-fix convergence
.converge('impl', {
  description: 'Implement until tests pass',
  maxIterations: 5,
  generate: { description: 'Generate', prompt: '...', outputKey: 'code', outputSchema: CodeSchema, llmConfig: { ... } },
  fix:      { description: 'Fix',      prompt: '...', outputKey: 'code', outputSchema: CodeSchema, llmConfig: { ... } },
  check: async (ctx) => runTests(),          // script function
  checkOutputKey: 'testResult',
  until: (result) => result.allPassed,       // exit condition on check output
  afterGenerate: async (ctx) => writeFiles(), // optional hook between generate and check
})

Step Types

Type	Purpose	Key Fields
agent	Claude interaction	prompt, llmConfig, outputKey?, outputSchema?
script	TypeScript execution	execute(ctx) => Promise<T>, outputKey?
sequence	Serial child steps	steps (static array or (ctx) => steps[]), isCheckpoint?
parallel	Concurrent child steps	steps (static or dynamic), maxConcurrency?, onFailure?
loop	Iterative refinement	steps, until, maxIterations, outputKey?

All steps share: type, id, description, outputKey?. Composite steps (sequence, parallel, loop) accept either a static array or a function (ctx) => PipelineStep[] for dynamic step generation.

Deep dive: Consult references/steps.md for complete reference on each step type, nesting patterns, and helper factory functions.

StepContext

Every step callback receives a StepContext<TInput, TOutputs>:

interface StepContext<TInput, TOutputs> {
  runId: string; // Unique run identifier
  sessionId?: string; // SDK session for conversation continuity
  inputs: TInput; // Validated pipeline inputs (read-only)
  variables: Record<string, unknown>; // Mutable scratchpad
  outputs: Partial<TOutputs>; // Typed outputs from completed steps
  config: PipelineConfig; // Resolved pipeline configuration
  stepPath: string[]; // Path from root to current step
  updateProgress?: (stepPath: string[], status: ProgressStatus) => void;
  completedTasks?: Array<{ phaseId: string; taskId: string }>; // Resume support
}

Data Flow

• ctx.inputs — Read-only validated input from CLI args. Typed by inputSchema.
• ctx.outputs — Keyed by each step's outputKey. Primary structured data channel.
• ctx.variables — Mutable scratchpad for lightweight ad-hoc data passing.

Prefer outputs with outputKey for structured, typed data. Use variables for quick coordination.

Pipeline-Level Configuration

buildPipeline({
  id: 'my-pipeline',
  version: '1.0.0',
  description: 'Pipeline description',
  inputSchema: InputSchema,
  maxBudgetUsd: 30,
  enableTodos: true,
  llmConfig: { model: 'sonnet', maxTurns: 20 }, // Default for all agent steps
  rateLimits: { requestsPerMinute: 10 },
});

Config merges: claudi.config.json → pipeline llmConfig → step llmConfig (later wins).

Deep dive: Consult references/llm-config.md for the full LLMConfig interface, model selection, tools, permissions, output schemas, sessions, MCP servers, and retry.

Input Schema

Define inputs with Zod. The schema drives CLI argument parsing, validation, and TypeScript inference:

const InputSchema = z.object({
  target: z.string().describe('Source file or directory to process'),
  maxConcurrency: z.number().min(1).max(10).default(3).describe('Max parallel tasks'),
  threshold: z.number().min(0).max(100).default(95).describe('Coverage threshold %'),
});

• Use .describe() on every field — it becomes the --help text
• Use .default() to make fields optional on the CLI
• Keep schemas flat (no nested objects) for clean CLI arg mapping

Minimal Example

import { buildPipeline, z } from 'claudi/sdk';

export const research = buildPipeline({
  id: 'research',
  version: '1.0.0',
  description: 'Research a topic and produce a summary',
  inputSchema: z.object({ topic: z.string().describe('Topic to research') }),
  maxBudgetUsd: 5,
})
  .agent('gather', {
    description: 'Gather information',
    prompt: (ctx) => `Research the following topic thoroughly: ${ctx.inputs.topic}`,
    llmConfig: { model: 'sonnet', maxTurns: 15, tools: ['Read', 'Grep', 'Glob'], permissionMode: 'bypassPermissions' },
    outputKey: 'research',
  })
  .agent('summarize', {
    description: 'Produce a structured summary',
    prompt: (ctx) => `Summarize the research:\n${JSON.stringify(ctx.outputs.research, null, 2)}`,
    llmConfig: { model: 'sonnet', maxTurns: 10, tools: ['Write'], permissionMode: 'bypassPermissions' },
    outputKey: 'summary',
  })
  .build();

Run: claudi run research --topic "hexagonal architecture patterns"

Reference Files

For detailed documentation beyond this overview, consult:

• references/steps.md — Complete reference for all 5 step types (agent, script, sequence, parallel, loop), dynamic config, nesting patterns, step ID rules, and helper factory functions
• references/llm-config.md — Full LLMConfig interface: model selection, tools, permissions, output schemas, system prompts, human input, budgets, retry, MCP servers, sessions, and streaming
• references/patterns.md — Reusable pipeline architecture patterns: fan-out/fan-in, staged workflow, conditional branching, batch processing with triage, pre-check/fix loops, shared infrastructure modules, gate/converge primitives, and gotchas
• references/workflow.md — Interactive pipeline creation workflow: understand goal, choose pattern, design step tree, scaffold code, validate with checklist