@agentsmarket/pipeline-runtime

Executes PIPELINE.md files — multi-stage AI workflows.

Note (Phase 2): The production executor lives in @agentsmarket/server (Cloudflare Worker), which runs pipelines server-side at POST /v1/pipelines/:id/invoke with provenance signing. The bin/webhook.mjs HTTP server in this package is a local dev tool + self-hosting reference. Use it when you want to run pipelines outside the marketplace (e.g. for testing, custom deployments, or before publishing your pipeline to agentsmarket.world).

Quick start

import { MiniMaxProvider, runPipeline, runPipelineV2 } from '@agentsmarket/pipeline-runtime';

const provider = new MiniMaxProvider(); // reads MINIMAX_API_KEY env

const pipeline = {
  name: 'code-refactor',
  version: '1.0.0',
  stages: [
    {
      id: 'analyze',
      provider: 'minimax',        // R6: explicit provider declaration
      model: 'MiniMax-M3',
      system: 'You are a code analyzer.',
      prompt: 'Analyze this code: {{code}}. List 3 improvement opportunities.',
    },
    {
      id: 'refactor',
      provider: 'minimax',
      model: 'MiniMax-M3',
      depends_on: ['analyze'],
      system: 'You are a senior engineer.',
      prompt: 'Refactor the code based on: {{analyze}}. Original: {{code}}. Return only the refactored code.',
    },
  ],
};

const result = await runPipeline(pipeline, { code: '...' }, { provider });
console.log(result.refactor);

Provider ≠ Model (R6, v0.3)

As of R6 every LLM stage must declare both provider: and model:. Auto-dispatch by model name was removed. See docs/R6-PROVIDER-MODEL.md for the full design; quick reference:

stages:
  - id: analyze
    provider: minimax        # required — one of openai|anthropic|minimax|openrouter|replicate
    model: MiniMax-M3        # required — must be in PROVIDER_REGISTRY[provider].supportedModels
    prompt: ...

Use validateR6Spec(spec) (CLI / server) to fail-fast on invalid pairs:

import { validateR6Spec } from '@agentsmarket/pipeline-runtime';

const { errors } = validateR6Spec({
  stages: [{ id: 's0', provider: 'replicate', model: 'gpt-4o', prompt: 'x' }],
});
// errors[0]: "Stage 's0': Provider 'replicate' does not support model 'gpt-4o'.
//            Supported models for replicate: [...]"

For automated migration of legacy v0.2 pipelines:

import { migrateProviderField } from '@agentsmarket/pipeline-runtime';

const migrated = migrateProviderField(legacySpec); // adds provider: minimax to LLM stages

Provider abstraction

Pipelines reference models by name. The runtime picks a provider that knows that model:

• minimax — MiniMax (default). Models: MiniMax-M3, MiniMax-M2.7, MiniMax-M1. Env: MINIMAX_API_KEY.
• openai — OpenAI. Models: gpt-4o, gpt-4o-mini, o1-preview, o1-mini, gpt-3.5-turbo. Env: OPENAI_API_KEY.
• anthropic — Anthropic. Models: claude-3-5-sonnet-20241022, claude-3-5-haiku-20241022, claude-3-opus-20240229. Env: ANTHROPIC_API_KEY.
• openrouter — Multi-model gateway (supportedModels: ['*']). Env: OPENROUTER_API_KEY.
• replicate — Replicate (image gen). Models: black-forest-labs/flux-pro, black-forest-labs/flux-schnell, stability-ai/sdxl. Env: REPLICATE_API_TOKEN.

Single source of truth: PROVIDER_REGISTRY in src/provider-registry.ts.

Status

v0.5 — file output + shell stages shipped (P5 + P4). Server surfaces call_price_usdc + preferred_model + capabilities on every pipeline.

Output spec helpers (v0.5 — P1.1)

Pipelines declare a top-level output: block describing what they produce. Two concerns: format (text/json/zip/files) and delivery (inline or signed URL). validateOutputSpec(raw) returns the normalized spec or structured errors:

import { validateOutputSpec } from '@agentsmarket/pipeline-runtime';

const r = validateOutputSpec({
  format: 'files',
  files: ['*.md', 'report.json'],
  delivery: 'signed_url',
  expires_in: 3600,
});
// r.ok === true, r.value = { format: 'files', files: [...], delivery: 'signed_url', expires_in: 3600 }

Limits: expires_in 60 ≤ seconds ≤ 604800 (7 days). Globs use * and ** with the standard "no slash past **" semantics.

File output collection (v0.5 — P1.2 / P5)

When a stage declares output_format: structured_json and includes a produced_files array, extractProducedFiles(stageOutputs, spec) collects the base64-encoded files and validates them against output.files globs:

import { extractProducedFiles, globToRegex } from '@agentsmarket/pipeline-runtime';

const stageOutput = JSON.stringify({
  summary: 'done',
  produced_files: [
    { name: 'report.md', content_type: 'text/markdown', content_b64: 'IyBS...' },
  ],
});

const r = extractProducedFiles(
  { stage1: stageOutput },
  { format: 'files', files: ['*.md'], delivery: 'signed_url', expires_in: 3600 },
);
// r.files = [{ stage_id: 'stage1', name: 'report.md', size: 7, content_type: 'text/markdown', content_b64: '...' }]
// r.errors = [], r.warnings = []

Limits: 10 MB/file, 50 MB total per invoke, 100 files max. Names reject /, \, .. (flat namespace). Bad base64 / oversized / path-traversal return errors that abort the pipeline; unmatched globs return warnings.

Server uses this on POST /v1/pipelines/:id/invoke (P5) to upload files to R2 and return HMAC-signed URLs in output_files[].

Shell stage helpers (v0.5 — P4, CLI-only)

Stages can declare shell: "cmd args..." or shell: { cmd, args?, cwd?, timeout_ms? }. Executed only via pipeline call --local; server invokes return 400 SHELL_UNSUPPORTED_ON_SERVER.

import { NodeShellExecutor, ThrowingShellExecutor } from '@agentsmarket/pipeline-runtime';

// CLI passes a real executor
const cliDeps = { isLocal: true, shellExecutor: new NodeShellExecutor(), ... };

// Server passes a throwing stub (Workers can't spawn)
const serverDeps = { isLocal: false, shellExecutor: new ThrowingShellExecutor('server cannot spawn'), ... };

NodeShellExecutor uses child_process.spawn(cmd, args, { shell: false, timeout }): SIGTERM on timeout_ms expiry (maps to ShellStageTimeoutError), non-zero exit → ShellStageExecError, stdout/stderr truncated at 1 MB each.

Two forms, two different things:

The string form is a real shell because the stage is called shell: — a line that auto-splits into an argv would make && an ordinary argument and quotes literal. The translation happens once, in normalizeShellStage, so every executor (local Node, remote sandbox) runs the exact same argv. The object form stays shell-free: it is the form to use when the program must receive precisely these arguments.

stages:
  - id: lint-and-test
    shell: "npm run lint && npm test"
  - id: pytest
    shell:
      cmd: pytest
      args: ["-x", "tests/"]
      cwd: ./backend
      timeout_ms: 120000

Limits: timeout_ms default 30000, max 600000 (10 min). Output cap 1 MB per stream. The string form runs under POSIX /bin/sh (SHELL_BIN).

One collision to know about: ${VAR} is substituted by the runtime (v0.3 env-var substitution) before the YAML is parsed, so ${VAR} inside a shell line is resolved from the run's environment — not by the shell. That is what you want for shell: "curl -H 'Bearer ${API_KEY}' …", but a shell construct that wants ${...} for itself (a positional parameter, ${1}) never reaches the shell.

Status

v0.5 — file output (P5) + shell stages (P4) shipped. R6 (Provider ≠ Model) shipped in v0.3. Linear + DAG stage execution. Retry, parallel branches, checkpoint.