Composition

Composition: uses: accepts pipeline IDs (v0.3+)

A stage’s uses: field can reference a published pipeline instead of a SKILL. The referenced pipeline’s stages are fetched and inlined into the parent pipeline at invocation time, with id-prefixing so they cannot collide with parent stage ids.

Syntax

stages:
  - id: refactor
    uses: code_rewriter@0.2.1

  - id: research
    uses: pipeline:web-research-helper@1.2.0

  - id: enrich
    uses: pipeline:enrichment-helper

The pipeline: prefix (COMPOSITION_USES_PREFIX in code) is what disambiguates a pipeline ref from a skill ref. The parser (tryParsePipelineUses) returns null for any uses: string that does not start with pipeline:, leaving it for the skill-loader to handle.

Identifier grammar

Resolution semantics

When runPipelineV2 encounters a stage with uses: pipeline:NAME@VERSION, it:

1. Calls tryParsePipelineUses → { id, version? } or null
2. Calls resolvePipelineRef(ref, fetcher) → fetcher hits the marketplace API
3. If fetcher returns null → throws PipelineFetchError (carries id + version)
4. Calls expandComposition(child.stages, fetcher, depth + 1) — recursive
5. injectNestedStages prefixes each child stage id with <parent_id>.<child_id>
6. Non-dotted depends_on entries in the child are rewritten to use the prefix; already-dotted paths (referencing sibling refs) are preserved verbatim

Depth limit

MAX_COMPOSITION_DEPTH = 4 (hard-coded constant). At depth 4 the expander throws CompositionDepthError. This guards against malicious or accidental cycles (A → B → A → B → ...) and accidental exponential blow-up.

if (depth >= MAX_COMPOSITION_DEPTH) {
  throw new CompositionDepthError(depth);
}

2-level nesting example

stages:
  - id: classify
    provider: openai
    model: gpt-4o-mini
    prompt: "Classify: {{ inputs.text }}"

  - id: extract
    provider: openai
    model: gpt-4o-mini
    depends_on: [classify]
    prompt: "Extract entities from class={{ stages.classify.outputs.category }}"

stages:
  - id: preprocess
    provider: openai
    prompt: "Trim: {{ inputs.raw }}"

  - id: research            # will be expanded to: research.classify, research.extract
    uses: pipeline:helper@1
    depends_on: [preprocess]

  - id: synthesize
    provider: openai
    depends_on: [research.extract]
    prompt: "Synthesize from {{ stages.research.extract.outputs.entities }}"

After expansion the parent has 4 stages: preprocess, research.classify, research.extract, synthesize. Downstream stages reference expanded ids via the full dotted path (stages.research.extract.outputs.entities).

Recursion / cycle detection

The depth limit (MAX_COMPOSITION_DEPTH = 4) is the only cycle guard — there is no per-pipeline visited-set. A chain pipeline:A → pipeline:B → pipeline:A is blocked because depth 4 is reached before re-entering A. For practical use cases (a pipeline references its helper once or twice), depth 4 is more than enough.

When does the fetch happen?

At invoke-time, via the optional fetchPipeline dependency injected into runPipelineV2. The marketplace server implements this fetcher against its pipeline registry (POST /v1/pipelines/:id). The stored YAML in D1 is not pre-expanded at publish-time — original uses: pipeline:NAME@VERSION syntax is preserved verbatim in provenance for reproducibility.

This means:

• Author can re-publish a referenced pipeline; consumers always get the latest unless version is pinned
• Marketplace controls access (private / granted_to / public visibility)
• Network failure on referenced pipeline → PipelineFetchError → pipeline aborts and buyer is refunded (same failure model as skill fetch)

Behavioral oracle

• packages/pipeline-runtime/src/composition.ts — implementation: tryParsePipelineUses, resolvePipelineRef, injectNestedStages, expandComposition, CompositionDepthError, PipelineFetchError, MAX_COMPOSITION_DEPTH = 4, COMPOSITION_USES_PREFIX = 'pipeline:'
• packages/pipeline-runtime/tests/composition.test.ts — parser + ref-resolver + injector unit tests (prefix correctness, depends_on rewriting, immutability)
• packages/pipeline-runtime/tests/composition-expand.test.ts — expander: single-level inlining, recursive expansion, MAX_COMPOSITION_DEPTH enforcement, PipelineFetchError metadata, mixed composition + plain stages
• packages/pipeline-runtime/tests/integration-r10-x.test.ts — end-to-end through runPipelineV2 with fetchPipeline injected: full DAG build from composed spec, fail-fast on missing referenced pipeline