Skip to content

pipeline-dsl.md

Latest commit

 

History

History
1030 lines (871 loc) · 52.8 KB

pipeline-dsl.md

File metadata and controls

1030 lines (871 loc) · 52.8 KB
Raw
type reference
topic runtime
audience
human
agent
search_hints
pipeline DSL
pipeline grammar
EBNF
formal grammar
agent generation grammar
transform step
tool step
agent step
call step
match step
fold step
for_each step
parallel step
expr
R1 expression
verify schema
run_pipeline
collect
on_settle
safety.spawn.max_pipeline_fan_out_depth
safety.spawn.max_pipeline_spawns
parse_json

Pipeline DSL reference

Normative grammar for a pipeline definition — the step kinds, the compositional primitives, the expression language they evaluate against, the schema/verify: schema mechanism, and the run_pipeline tool that launches a pipeline. See Pipelines for the why/architecture, and Pipeline registration for how a definition reaches a session.

Document shape

A pipeline definition is one or more ----separated YAML documents:

  • Exactly one pipeline: document — the pipeline itself.
  • Zero or more schema: documents — named schemas the pipeline's steps can reference via verify: schema (see Schemas).
schema: Review
fields:
  passed: {type: bool}
  notes: {type: string}
---
pipeline: review_and_report
description: Review a document and summarize the verdict.
steps:
  - agent: {prompt: "Review {ctx.doc}. Reply with passed/notes.", schema: Review, output: review}
  - transform: {value: "ctx.review.passed and 'OK' or 'NEEDS WORK'", output: verdict}

Data flow between steps

Every step's expression / template evaluates against exactly the same two top-level things — there is no third form and no bare-name shortcut:

  • ctx — every named store accumulated so far in the current scope, as a flat namespace. A step that declared output: X makes it readable, from every later step in scope, as ctx.X — never bare X.
  • pipe — the immediately preceding step's own result in the current scope, unnamed and ephemeral. Read as bare pipe (not ctx.pipe — pipe is a top-level context key, not a named store). A step with no output: still produces pipe data for whichever step runs next; it just isn't durably named.

So output: X does two things at once: it becomes pipe for the very next step in the same sequence, and it is durably written to ctx.X, visible to every later step in scope (not only the immediate next one) until the scope ends.

A worked trace — three steps, showing exactly what ctx and pipe hold at each boundary:

steps:
  - transform: {value: "ctx.name + '!'", output: greeting}   # step 0
  - tool: {name: shout, args: {text: !expr pipe}, output: shouted}  # step 1
  - transform: {value: "ctx.shouted.text + ' (done)'", output: final}   # step 2

seeded with ctx = {name: "Reyn"}:

Before step ctx pipe
0 {name: "Reyn"} null (no prior step)
1 {name: "Reyn", greeting: "Reyn!"} "Reyn!" (step 0's result)
2 {name: "Reyn", greeting: "Reyn!", shouted: {text: "REYN!"}} {text: "REYN!"} (step 1's result — see tool step results below)
(after step 2) {..., final: "REYN! (done)"} "REYN! (done)"

Step 1 reads step 0's result two ways that both work: bare pipe (its immediate predecessor) or ctx.greeting (the durable store) — they hold the same value right after step 0, but only ctx.greeting stays reachable once a step runs in between.

Scoping exceptions (see Formal grammar § Structural invariants below for the normative statement):

  • for_each/parallel branches each evaluate against an isolated copy of the outer ctx taken at branch-start — a write inside a branch never leaks back to the outer scope or to a sibling branch.
  • A call/match callee's ctx is built only from the names bound by pass: — a name not bound there is invisible to the callee, ctx shortcut or not. Each pass: entry is an explicit {NAME: EXPR} mapping: EXPR is an R1 expression evaluated against the caller's current full context (ctx/pipe/item/acc — whatever is in scope, exactly like transform.value), and the result is bound to NAME in the callee's ctx (see the for_each/fold sections below and Data flow between steps for a worked example). A failing expression fails the step, naming the entry.
  • fold's do and for_each's do add extra top-level keys to their own context (item/acc for fold, item for for_each) alongside ctx and pipe — see their own sections below.

Forwarding a loop variable into a sub-pipeline. item/acc are top-level context keys, not named stores — so an agent step's {item} prompt reads them directly, and a call/match step used as do: reaches them the same way any pass: entry reaches anything: its EXPR is evaluated against the current do-scope context, so item/acc are just more names in scope:

pipeline: outer
steps:
  - for_each:
      over: ctx.suspects
      on_error: abort
      do:
        call:
          pipeline: interrogate
          pass:
            suspect: item
      collect: {transform: {value: "pipe"}}
      output: verdicts

Here interrogate is a sibling pipeline co-located in the same file (a dot-less call target resolves to a same-file sibling; a cross-file target uses the qualified other_key.name form — see Pipeline registration). It reads the current suspect as ctx.suspect — pass: {suspect: item} evaluated the bare-path expression item against the for_each scope's context (which carries item alongside ctx/pipe) and bound the result to suspect in the callee's ctx. fold works the same way, and its do-scope additionally carries acc (the running accumulator), reachable the same way: pass: {running: acc}.

pipeline: document keys

Key Required Meaning
pipeline yes The declared local name. Registered globally as {entry-key}.{name} (namespacing is always on) and referenced by a call/match step's target — see Pipeline registration. Must not contain . (reserved separator).
description no Human-readable summary; surfaced to the LLM alongside the name when a registered pipeline is listed as a pipeline__<name> catalog action. Defaults to empty.
steps yes Non-empty list of steps, executed in order (see Step kinds and Primitives).

input, defaults, and refine are part of the pipeline design's fuller grammar but have no runtime yet — a document using them fails to parse with an explicit "not yet supported" error rather than being silently ignored.

Formal grammar

The EBNF below is the canonical, current grammar — derived directly from parse_pipeline_dsl (src/reyn/core/pipeline/parser.py), not from an earlier design proposal. It covers exactly what the parser accepts today: a definition conforming to it parses cleanly; a violation is rejected. Mapping keys are unordered in YAML — the linear order below is for readability, not a positional requirement. NAME is a bare identifier-like string; EXPR is an R1 expression source string; TPL is an agent.prompt template string ({ctx.dotted.path} / {pipe} interpolation, not R1).

Document      ::= YamlDoc ("---" YamlDoc)*        (* exactly one PipelineDoc across the whole text *)
YamlDoc       ::= SchemaDoc | PipelineDoc

SchemaDoc     ::= "schema:" NAME "fields:" FieldMap
PipelineDoc   ::= "pipeline:" NAME
                  ("description:" STRING)?
                  "steps:" Step+

Step          ::= "transform:" TransformBody
                 | "tool:"      ToolBody
                 | "agent:"     AgentBody
                 | "call:"      CallBody
                 | "match:"     MatchBody
                 | "fold:"      FoldBody
                 | "for_each:"  ForEachBody
                 | "parallel:"  ParallelBody

TransformBody ::= "{" "value:" EXPR ["output:" NAME] "}"

ToolBody      ::= "{" "name:" STRING
                      ["args:" ArgMap]
                      ["schema:" NAME]
                      ["on_error:" OnError]        (* optional — omitted = no isError check (pre-#3130 behavior) *)
                      ["output:" NAME] "}"
ArgMap        ::= "{" (KEY ":" ArgValue ("," KEY ":" ArgValue)*)? "}"
ArgValue      ::= LITERAL | "!expr" EXPR        (* !expr only as the WHOLE value, never nested *)

AgentBody     ::= "{" "prompt:" TPL
                      ["identity:" NAME]
                      ["capabilities:" "{" "tools:" "[" NAME* "]" "}"]
                      ["schema:" NAME]
                      ["model:" NAME]
                      ["output:" NAME] "}"

CallBody      ::= "{" "pipeline:" NAME            (* static literal, never EXPR *)
                      ["pass:" "{" (NAME ":" EXPR)* "}"]
                      ["output:" NAME] "}"

MatchBody     ::= "{" "on:" EXPR
                      "cases:" "{" (LABEL ":" MatchTarget)+ "}"
                      ["default:" MatchTarget]
                      ["output:" NAME] "}"
MatchTarget   ::= "{" "pipeline:" NAME ["pass:" "{" (NAME ":" EXPR)* "}"] "}"

FoldBody      ::= "{" [ListSource]
                      "init:" EXPR
                      "do:" Step
                      "output:" NAME              (* required, unlike call's *)
                      ["max_items:" INT] "}"

ForEachBody   ::= "{" [ListSource]
                      ["max_parallel:" INT]
                      "on_error:" OnError          (* required — no default *)
                      "do:" Step
                      "collect:" Step
                      ["output:" NAME] "}"

ParallelBody  ::= "{" ["on_error:" OnError]        (* optional — defaults to "abort" *)
                      "branches:" "{" (NAME ":" Step)+ "}"
                      "collect:" Step
                      ["output:" NAME] "}"

ListSource    ::= "over:" EXPR | "items:" "[" LITERAL* "]"   (* mutually exclusive *)
OnError       ::= "continue" | "abort" | "retry(" INT ")"

FieldMap      ::= "{" (NAME ":" FieldType)+ "}"
FieldType     ::= "{" "type:" ("bool" | "string") "}"
                 | "{" "type:" "number" ["minimum:" LITERAL] ["maximum:" LITERAL] "}"
                 | "{" "type:" "enum" "values:" "[" LITERAL+ "]" "}"
                 | "{" "type:" "list" "of:" FieldType "}"     (* 'of' non-list; no lists-of-lists *)
                 | "{" "type:" "object" "fields:" FieldMap "}"
                 | "{" "type:" "ref" "schema:" NAME "}"

Structural invariants the grammar alone doesn't show (enforced by the parser and executor, not just documented convention):

  • A call/match/fold's do/for_each's do/collect/parallel's branch/collect is a full nested Step — any step kind, including another compositional primitive.
  • call, match's case/default, and parallel's branches all name a static literal pipeline/step target — never a runtime expression. Only match.on and for_each/fold's over are runtime-evaluated.
  • pass: is the only channel a call/match callee's context is built from — each entry's EXPR is evaluated against the caller's context and bound to that entry's NAME; a NAME not bound by any entry is invisible to the callee (see Data flow between steps).
  • for_each/parallel branches each get an isolated copy of the outer named stores — no sibling communication between concurrent items/branches (see Data flow between steps).
  • !expr is the only way a tool argument becomes a resolved expression instead of a literal — nesting it inside a list/mapping value is a parse error, not a silent no-op.

Step kinds

Every step is a single-key mapping naming its kind. Three are linear leaf steps — they read the context, do one piece of work, and produce a result:

transform

A pure step: value is evaluated as an R1 expression against the current context (ctx/pipe — see Data flow between steps); the result becomes this step's pipe data (and, if output is set, is also written to that named store).

- transform: {value: "'Hello, ' + ctx.name + '!'", output: greeting}
Key Required Meaning
value yes An R1 expression source.
output no Named store to write the result to.

tool

A side-effecting step: dispatches name with args through the same qualified-action-routing-then-bare-lookup a live invoke_action call uses — so a tool step can name either a qualified action (read_file) or a bare registered tool name (web_search).

- tool: {name: web_search, args: {query: !expr ctx.brief, limit: 5}, output: results}
Key Required Meaning
name yes The tool/action name (literal string).
args no Mapping of argument name → value. Each value is a literal unless tagged !expr (see Literals vs !expr below).
schema no A registered schema name the result must conform to (verify: schema — see Schemas). Non-conformance fails the step (checked against the RAW tool result, before the text/structured reduction below).
on_error no (#3130) One of continue, abort, or retry(N) — see tool.on_error below. Omitted is a distinct state, not a synonym for abort: it preserves the pre-#3130 behavior (only a raised exception fails the step; a canonical-error result — meta.isError — passes through unchecked, e.g. for a schema:-gated preflight probe to inspect downstream).
output no Named store to write the result to.

tool.on_error

By default (the key omitted) a tool step only fails on a raised exception — a tool that returns a canonical-error result (meta.isError: true, e.g. an MCP tool's isError branch) returns normally; reacting to it requires a downstream schema:-gated preflight step. on_error (#3130) lets a single tool step react to that natively, mirroring for_each/parallel's on_error — same three values, same retry(n) regex, same retry-safety model (a retry re-invokes the tool: any internal side effect it has may re-fire, same caveat as for_each's on_error: retry(n)):

  • abort — a raised exception OR a canonical-error result fails the step (PipelineExecutionError), naming the tool's own error text.
  • continue — the step does not fail; instead the failure is bound to output as a typed error-envelope — the SAME {text, structured, meta: {isError: true}} shape a canonical-error tool result already takes (a raised exception is rendered into that same shape). This is not the for_each-style "drop the item" — a single step's output is a NAMED binding downstream steps reference by name, so dropping it would leave an unbound-variable reference; binding a discriminable error-envelope instead keeps ctx.<output>.meta.isError inspectable downstream, symmetric with the success shape.
  • retry(N) — re-run the whole step (dispatch + schema: validation) up to N more times; if it still fails, falls through to abort (there is no combined "retry then continue" value, matching for_each).
- tool: {name: fetch_url, args: {url: !expr ctx.link}, on_error: continue, output: page}
- transform: {value: "get(ctx.page, 'meta.isError', false) ? 'unreachable' : ctx.page.text", output: body}

tool step results

A tool step's result — what lands as pipe and ctx.<output> — is always the flat shape {text: ..., structured: ..., meta: ...}, uniform across every tool kind (the same shape the chat side exposes to an LLM):

  • text — the tool's canonical string body (empty string when the tool has no text-shaped result).
  • structured — present only when the tool produced non-text data; absent entirely (no key) otherwise. A tool result dict with no recognized shape is wrapped whole as structured (nothing is ever lost).
  • meta — the tool's signal fields: small, high-signal values the producer deliberately kept out of the body because they change what the caller does next. Present only when the tool emitted any; absent entirely (no key) otherwise — which is the common case, since most successful results carry no signal.

This conversion is shape-only — it never truncates, offloads, or caps a value the way the chat-side tool-result path does; a pipeline step's ctx retains the full value for downstream steps to consume programmatically. Read a tool's text body via ctx.<name>.text and its structured payload via ctx.<name>.structured (or a field of it, e.g. ctx.hits.structured.count).

Reading meta safely. Because meta is absent when a producer emits no signal, a bare ctx.<name>.meta.<field> path raises on such a producer (bare paths are not safe navigation — see The R1 expression language). That is deliberate: a step that depends on a signal fails loudly rather than silently reading a default. Use get(...) when the signal is genuinely optional:

- tool: {name: embed, args: {texts: !expr ctx.chunks}, output: embedded}
# the model that ACTUALLY produced the vectors (an `embedding_model` input may
# be a model-CLASS alias like "standard"; meta.model is what it resolved to):
- transform: {value: "ctx.embedded.meta.model", output: resolved_model}
# optional signals — default rather than raise:
- transform: {value: "get(ctx.embedded, 'meta.cost_usd', null)", output: spend}

What a given tool puts in meta is that tool's own contract; common examples: embed → model / total_tokens / cost_usd / priced; exec → a nonzero returncode (a zero exit is not signal, so no key); an MCP call → isError. Reach for meta when you need what a result cost or how it was produced — as opposed to structured, which is the result itself.

To run a sandboxed command from a pipeline, use a tool step naming exec directly:

- tool: {name: exec, args: {argv: !expr "['ls', ctx.dir]"}, output: listing}

The previous step's pipe data can be threaded to the process's STDIN via the stdin_pipe arg (args: {argv: [...], stdin_pipe: !expr pipe}) — the same JSON-encoded-STDIN / STDOUT-becomes-result shape the removed shell: step used to wire up automatically. (#3226 Phase 1+2 removed the pipeline DSL's shell: step — thin sugar that ran /bin/sh -c <command>, the sole shell-injection surface in the codebase — without adding a new step kind: a tool: {name: exec, ...} step already covers argv-based exec in full.)

Literals vs !expr

A tool argument value is a literal — passed through to the tool exactly as written — unless it is tagged with the YAML tag !expr:

args: {query: !expr ctx.brief, limit: !expr "ctx.n + 1", label: "a plain string"}

query and limit are R1 expression sources, resolved against the step's ctx/pipe context at run time (see Data flow between steps); label is the literal string "a plain string". !expr is only honored as the whole value of an argument — one hiding inside a nested list or mapping is a parse error, so there is no ambiguity between "a literal that happens to look like an expression" and "an expression."

transform.value is always an R1 expression (no !expr tag needed — there is no literal form for a transform step). An agent step's prompt is never an R1 expression — see below.

agent

An LLM-driven leaf step: prompt (a template string) is interpolated against the current context and run as one turn in an ephemeral session, capability-narrowed to capabilities composed with the invoker's own per-session narrowing — never instead of it — under identity (or the invoker's own identity if omitted).

- agent: {prompt: "Summarize: {ctx.doc}", capabilities: {tools: [read_file]}, schema: Summary, output: summary}
Key Required Meaning
prompt yes A template string — {ctx.dotted.path} / {pipe} references are interpolated (values only, no operators — this is string interpolation, not an R1 expression).
identity no The agent identity to run under. Defaults to the run's invoker. A registered pipeline may name any identity; an inline, agent-generated pipeline may only name the invoker's own identity — naming another agent's identity is rejected by the static-analysis gate as a capability escalation (see Ad-hoc inline launch).
capabilities no {tools: [NAME*]} — narrows the ephemeral session's tool surface. Restrict-only: a pipeline step can never exceed the invoker's own envelope. The two narrowings are composed most-restrictive-wins — denies union, allow-lists intersect, and an omitted allow-list means "no restriction from this side", so omitting capabilities leaves the invoker's own allow-list in force rather than clearing it.
schema no Same verify: schema semantics as tool, applied to the parsed JSON reply.
output no Named store to write the result to.

Every agent step, wherever it is reached (top-level or fanned out inside a for_each), charges the run's shared spawn budget — see Safety caps.

Agent-step exception surfacing (#187 / #2732)

An agent step's leaf session (spawn_ephemeral_session, always mode="ephemeral") drives its router loop inside Session._handle_inbox_text. (#3595 step 1: the prompt rides the inbox as TurnOrigin.AGENT_STEP ("agent_step"), not "user", and _run_turn_body routes that kind straight to this shared turn body — so a prompt that happens to start with / is content the model reads, never a slash command the OS executes. #3595 S5 then deleted the operator entry that used to sit above the body (_handle_user_message, whose only remaining content was the slash short-circuit): interpreting text as a command is CLIENT work now, so no inbox kind reaches a dispatch at all.) Two hardening passes changed how a mid-work failure there reaches the pipeline executor:

#187 B1 — full exception surfaced, not just classified. Earlier, a router-loop exception (e.g. the LLM call raising after litellm exhausts its own retries) was caught, classified into a short outbox summary string, and otherwise silently swallowed. For interactive chat this is fine — the summary IS the reply. For an autonomous run-once agent step, it meant the step ended mid-edit with no diagnosable trace: no logged LLM response, req=resp+1, nothing to point a fix at. The fix logs the full traceback (logger.exception) AND emits a router_loop_terminated_by_exception audit event (chain_id, error_type, repr(exc)[:500], and — #4381 stage 1 — cause, the deepest __cause__ in the exception chain's type name, so reyn's own wrapper type (e.g. ContextOverflowError) never hides what actually happened underneath it) — both kept unconditionally, on top of the (unchanged) classified outbox summary.

#2732 — the classified-summary path silently produced a successful-looking empty answer for agent steps. The catch-all except Exception in _handle_inbox_text (in _handle_user_message until #3595 split the body out, and the whole of it since S5 deleted that entry) is intentionally broad (any LLM-call or router-loop exception, not only credential errors) — that breadth predates #2732. What #2732 fixed is what happens after classification for an ephemeral session: spawn_ephemeral_session hardcodes mode="ephemeral", and run_agent_step's join only reads kind="agent" outbox messages — so a kind="error" classified message was silently dropped, run_agent_step returned "" with no exception, and executor.py's except AgentStepError never fired. The pipeline step looked like it had succeeded with an empty answer.

The fix re-raises the classified error as AgentStepError, but only when self._ephemeral is true — an unconditional re-raise would break the interactive chat loop, which relies on this method returning normally after queuing the error reply to the outbox. It re-raises the base AgentStepError (not an LLM-specific subclass) because this catch-all also covers non-LLM exceptions; from exc preserves the chain (already retained by the traceback log + audit event above, for both branches).

Compositional primitives

Five primitives compose steps into non-linear control flow — the full Appendix-B set, all supported today.

call — sub-pipeline

Synchronously runs a registered sub-pipeline by static name and threads its final output out as this step's result.

- call:
    pipeline: validate_doc
    pass:
      doc: ctx.doc
      rules: ctx.rules
    output: validation
Key Required Meaning
pipeline yes A static literal pipeline name — never a runtime expression. Dot-less = a same-file sibling ({entry-key}.name); dotted = a global (other_key.name) — see target resolution. An unresolved/unregistered target fails at load or step time.
pass no A flat {NAME: EXPR} mapping. The callee's context is built fresh from only these bindings — a NAME not bound by any entry is structurally invisible to the callee. Each entry's EXPR is an R1 expression evaluated against the caller's current context (ctx/pipe/item/acc — whatever is in scope, exactly like transform.value), and the result is bound to NAME in the callee's ctx (see Data flow between steps). A failing expression fails the step, naming the entry.
output no Named store to write the callee's final result to.

The callee's first step receives the caller's pipe data at the call site; the callee's own final step output becomes this call step's result. A callee failure fails the call step.

match — runtime-selected sub-pipeline

Evaluates on to a value, selects the case whose label string-equals it, and runs that case's target exactly like a call step.

- match:
    on: "ctx.review.passed"
    cases:
      "True": {pipeline: report_pass, pass: {review: ctx.review}}
      "False": {pipeline: report_fail, pass: {review: ctx.review}}
    default: {pipeline: report_unknown}
    output: report
Key Required Meaning
on yes An R1 expression evaluated against the current context; its stringified result selects a case label.
cases yes Non-empty mapping of LABEL: {pipeline, pass?} — each target a static literal name, exactly like call (same pass: flat NAME -> R1-EXPRESSION mapping).
default no {pipeline, pass?} run when no case label matches. A step with no matching case and no default fails.
output no Named store to write the selected callee's result to.

Every case/default target is a static literal — the runtime value only ever selects a label, never a target directly.

fold — sequential accumulator

Walks a list in order, threading an accumulator through a repeated do step. do's context extends the usual ctx/pipe (see Data flow between steps) with two extra top-level keys, item and acc — see the table below.

- fold:
    over: ctx.items
    init: "0"
    do: {transform: {value: "acc + item"}}
    output: total
    max_items: 1000
Key Required Meaning
init yes An R1 expression evaluated once, before the first iteration, seeding acc.
do yes A single step re-invoked once per list item, in a context of {ctx, pipe, item, acc} — item is the current element, acc the running accumulator; do's return value becomes the next acc.
output yes Named store for the final acc (a fold's whole point is producing a named result — required, unlike call's optional output).
over no* An R1 expression resolving to the list to walk.
items no* A static literal list.
max_items no Caps the walk to the first N elements (a longer source is silently truncated, never an error).

* over and items are mutually exclusive; if neither is given, the list falls back to the step's incoming pipe data. Item failure fails the whole fold. There is no collect (unlike for_each) — each item's result depends on the accumulated state of the ones before it, so there is nothing to collect independently.

item/acc are reachable beyond do's own step: a do: {call: {pipeline: X, pass: {current: item}}} (or pass: {running: acc}) forwards the current element (or the running accumulator) into a call/match sub-pipeline, the same way an agent do's {item}/{acc} prompt reference already could.

for_each — concurrent fan-out

Runs do over each list item as an isolated concurrent sub-scope, then runs collect once over the ordered results. See Data flow between steps for the isolation rule this section's do context relies on (each item gets its own ctx copy — writes never leak between items or back to the outer scope).

- for_each:
    over: ctx.reviewers
    max_parallel: 4
    on_error: "retry(2)"
    do: {agent: {prompt: "Review as {item}: {ctx.doc}", schema: Review}}
    collect: {transform: {value: "pipe"}}
    output: reviews
Key Required Meaning
do yes A step run once per item, in a context of {ctx, pipe, item} — ctx is an isolated copy of the outer named stores (no sibling visibility between items), pipe is this step's own incoming pipe data held constant across every item.
collect yes A step run once, after the fan-out, over the ordered list of surviving item results (its pipe context). Its result is this step's overall result.
on_error yes One of continue (a failed item is dropped from the results, never re-run on resume), abort (a failed item cancels the still-pending items and fails the whole step), or retry(N) (re-run the failed item up to N more times, then fall back to abort).
over no* Same as fold.
items no* Same as fold.
max_parallel no Caps live concurrency (a Semaphore). Omitted, defaults to a conservative finite value — never unbounded by omission.
output no Named store to write collect's result to.

* over/items are mutually exclusive, falling back to incoming pipe data like fold. There is no item-level acc (that is fold-only) — an item cannot see any other item's result.

item is reachable beyond do's own step the same way fold's item/acc are: do: {call: {pipeline: X, pass: {current: item}}} forwards the current element into a call/match sub-pipeline used as do:.

parallel — heterogeneous named-branch fan-out

for_each's heterogeneous sibling: instead of fanning one do step out over a runtime-sized list, parallel fans a static, finite set of distinct named branches out concurrently, then runs collect once over the named map of their results. Same isolation rule as for_each — see Data flow between steps: each branch gets its own ctx copy, and collect's pipe is the whole {branch_name: result} map, not any one branch's result directly.

- parallel:
    on_error: "abort"
    branches:
      security: {agent: {prompt: "Security-review {ctx.doc}", schema: Review}}
      style: {agent: {prompt: "Style-review {ctx.doc}", schema: Review}}
    collect: {transform: {value: "{security: pipe.security, style: pipe.style}"}}
    output: reviews
Key Required Meaning
branches yes A non-empty {NAME: Step} mapping — each branch is its own, independently-shaped step (a different kind/config per name), unlike for_each's one do re-invoked per item. Every branch runs concurrently; the branch count itself is the concurrency bound (no max_parallel — the set is statically finite).
collect yes A step run once, after every branch lands, over the named map {branch_name: result} (not an ordered list, unlike for_each). Its result is this step's overall result.
on_error no One of continue, abort (the default when omitted — unlike for_each, where on_error is required), or retry(N) — same semantics as for_each's on_error. A continue-dropped branch's key is absent from collect's named map.
output no Named store to write collect's result to.

When at least one branch actually dropped (on_error: continue), collect's named map ALSO carries a reserved __branch_errors__ entry — {branch_name: error_text} for every dropped branch, the failing branch's own error text (e.g. a verify: schema failure names the tool's own error/content when the result carried one, not just which field mismatched) — so collect can report why a branch failed, not just that it did (a schema-gated reachability probe like the builtin RAG ingest pipeline's X1 pre-flight reads get(pipe, "__branch_errors__.<name>", "") for this). Absent entirely when no branch dropped — a collect step that never loses a branch sees no shape change. A branch may not be named __branch_errors__ (parse error) since it would collide with this reserved key.

Each branch's context is {ctx, pipe} — ctx an isolated copy of the outer named stores, pipe this step's own incoming pipe data held constant across every branch. There is no item/acc (those are for_each/fold-only) and no sibling visibility between branches.

The R1 expression language

transform.value, a tool argument tagged !expr, and match.on all resolve against the same small, total expression language (R1) — a purpose-built tree-walking interpreter, not a general scripting language and not a code-execution sandbox. It has no recursion, no user-defined functions, no unbounded loops (every combinator iterates one already-materialized list exactly once), no IO, and no eval/exec.

expr           ::= or_expr
or_expr        ::= and_expr ("or" and_expr)*
and_expr       ::= not_expr ("and" not_expr)*
not_expr       ::= "not" not_expr | comparison
comparison     ::= additive (cmp_op additive)?
additive       ::= multiplicative (("+" | "-") multiplicative)*
multiplicative ::= unary (("*" | "/") unary)*
unary          ::= "-" unary | primary
primary        ::= NUMBER | STRING | "true" | "false" | "null"
                  | "(" expr ")"
                  | "[" (expr ("," expr)*)? "]"
                  | "{" (IDENT ":" expr ("," IDENT ":" expr)*)? "}"
                  | combinator
                  | path
combinator     ::= "map" "(" expr "," lambda ")"
                  | "filter" "(" expr "," lambda ")"
                  | "all" "(" expr "," lambda ")"
                  | "any" "(" expr "," lambda ")"
                  | "find" "(" expr "," lambda ")"
                  | "count" "(" expr ")"
                  | "sum" "(" expr ")"
                  | "join" "(" expr "," expr ")"
                  | "get" "(" expr "," STRING ("," expr)? ")"
                  | "parse_json" "(" expr ")"
lambda         ::= IDENT "->" expr        (* only valid as a combinator's own argument *)
path           ::= IDENT ("." IDENT)*
cmp_op         ::= "==" | "!=" | "<" | ">" | "<=" | ">="

Literals: true / false / null, integers, floats, single- or double-quoted strings.

Field refs: a dotted path against the context, e.g. ctx.review.passed or bare pipe. A missing path or a non-mapping intermediate segment raises — bare paths are not safe navigation; use get(...) for that (below).

Operators: and / or / not; comparisons == != < > <= >= (</>/<=/>= require two numbers or two strings; ==/!= work on anything); arithmetic + - * / (numeric; + also concatenates strings and lists). Division by zero raises.

Combinators — the only call-like syntax the grammar has, a fixed closed set:

Combinator Signature Meaning
map map(list, item -> expr) Transform each element.
filter filter(list, item -> expr) Keep elements where the lambda is true.
all all(list, item -> expr) True iff every element satisfies the lambda.
any any(list, item -> expr) True iff some element satisfies the lambda.
find find(list, item -> expr) First matching element, or null.
count count(list) Element count.
sum sum(list) Numeric sum.
join join(list, sep) String-join.
get get(base, "dotted.path", default?) Safe navigation — unlike a bare Path, never raises on a missing path; returns default (or null) instead.
parse_json parse_json(string) Decode a JSON string into its value (object/array/string/number/bool/null). Raises if the argument is not a string or is not valid JSON — there is no safe/default-returning variant.

A lambda (item -> expr) is only ever valid as the direct argument of map/filter/all/any/find — it is not a value that can be assigned or passed around, and naming anything outside this fixed combinator set as a function call is a parse error.

Example expressions: "'Hello, ' + ctx.name + '!'", "ctx.n + 1", "all(ctx.reviews, r -> r.passed)".

An agent step's prompt is a different mechanism: a template string where {ctx.dotted.path} / {pipe} references are interpolated as plain values — not R1 expressions, no operators inside the braces.

Schemas — verify: schema

A schema names a nested, monomorphic type: a set of fields, each a scalar (bool/string/number), an enum, a typed list (its element type, of, is mandatory — no untyped lists, and lists-of-lists are not allowed), a nested inline object, or a ref to another registered schema (a recursive-reference cycle across the registered set is rejected at registration time).

schema: Review
fields:
  passed: {type: bool}
  notes: {type: string}
  tags: {type: list, of: {type: string}}

A number field may additionally carry minimum: and/or maximum: (inclusive bounds; #2963):

schema: Score
fields:
  score: {type: number, minimum: 0.0, maximum: 1.0}

Before this, {type: number} was the only way to describe a numeric field — a rubric prompt saying "score from 0.0 to 1.0" had nothing enforcing it in the schema itself, so a model answering 85 on a 0-100 scale against a score >= 0.6 threshold check passed unchallenged (the range lived only as a prompt request, not a validated constraint). minimum/maximum are checked at schema-registration time (minimum must not exceed maximum; both, if given, must themselves be numbers) and enforced at value-validation time — a value outside the (inclusive) bounds is an out_of_range error, alongside missing_required / type_mismatch / enum_invalid / unresolved_ref / unknown_type. Scoped to number only: bool/string have no natural notion of a range, and a string length cap or list/array element-count cap is a separate, not-yet-demonstrated need. As the "0062" note just below describes, an agent step's schema: converts to the provider's response_format — so minimum/maximum reach generation-time constraint too, not just post-hoc validation.

A tool/agent step's schema: NAME key names a registered schema its result (or, for agent, its parsed JSON reply) must conform to — non-conformance fails the step. Schemas declared in the same DSL document set (standalone schema: documents) are what makes this possible for an ad-hoc inline pipeline too, since its schemas travel with the same definition string.

agent step structured output (0062). An agent step's schema: NAME does more than validate — it constrains generation: the ephemeral session's answer turn is issued with a provider-side response_format built from the named schema, so the model is asked for schema-shaped JSON directly rather than free-forming text the OS then hopes parses. The parsed value is still validated afterwards (belt-and-suspenders) and bound to output. This needs a model that supports structured output; an unsupported model, a provider-rejected schema, or exhausted-re-prompt non-conformance each fail the step with a distinct, diagnosable error rather than silently falling back to free-form text.

An agent step also accepts model: NAME, an optional model-class override for that step's ephemeral session (e.g. model: strong for a step that needs a more capable model than the pipeline's default):

- kind: agent
  prompt: "Review {ctx.doc}."
  model: strong
  schema: Review
  output: review

Invocation

One tool, run_pipeline, launches a pipeline. Proposal 0067 P7 unified the prior four launch verbs (run_pipeline, run_pipeline_async, run_pipeline_inline, run_pipeline_inline_async, 0 aliases kept) into this single verb with three orthogonal parameters:

Parameter Values Selects
name= xor definition= a registered pipeline name / an ad-hoc DSL string Registered vs. inline launch
collect= "attached" (default) | "async" Sync-blocking vs. fire-and-forget
on_settle= "deliver" (default) | "<pipeline name>" | "drop" What happens to the result on settle — P4's delivery vocabulary; accepted but ignored for collect="attached", since the result is already returned in-band

Every combination converges on the same execution: a launch spawns a dedicated PipelineExecutorDriver session and the pipeline runs inside it (see Driver-as-session) — run_pipeline never runs a pipeline inline on the caller's own turn.

Registered launch

run_pipeline(name, input?, collect?, on_settle?) looks a pipeline up by its registered name (see Pipeline registration). input seeds the pipeline's initial named context (ctx.*) for its first step; omit it for a pipeline that needs no seed input. A name that isn't registered fails clearly.

Sync vs async

  • Sync (collect="attached", the default): the caller attaches to the driver-session's run and blocks until it reaches a terminal state, reading the result back in-band ({status: "ok", data: {run_id, output, named_stores}} on success; on failure/cancellation, the standard dispatch-error shape {status: "error", error: {kind, message}} — kind is pipeline_failed or pipeline_cancelled (run_id is folded into message, not a separate field) so router_loop.feedback() renders it as Error (<kind>): <message> like every other tool error, #2649). Live pipeline_step_started / pipeline_step_completed audit-events stream to the caller for the run's duration (what a TUI live view renders), and a cooperative Ctrl-C stops the run cleanly at the next step boundary. If the attach itself is interrupted by a crash, the run is not lost — it is handed to the same recovery path async uses, and the result arrives later as an inbox message instead ({status: "started", data: {run_id}}). on_settle= is accepted but ignored in this mode.
  • Async (collect="async"): returns {status: "started", data: {run_id}} immediately; once the run reaches a terminal state, the result is handled per on_settle= — delivered as a [pipeline] inbox message ("deliver", the default), routed into another pipeline (a pipeline name), or discarded ("drop").

Ad-hoc inline launch

run_pipeline(definition, input?, collect?, on_settle?) — passing definition= instead of name= — takes a pipeline DSL string the calling agent generates at run time — the same Appendix-B grammar as a registered pipeline file, including any schema: documents the definition's own steps reference. name= and definition= are mutually exclusive; exactly one must be given. There is no pre-registration: the string is parsed and run through a static-analysis gate before anything is spawned, so a bad definition fails clearly and spawns nothing:

  1. The definition parses.
  2. Every step schema: reference resolves within the definition's own schemas.
  3. Every tool step's name resolves to a registered tool or qualified action.
  4. (Partly structural, partly runtime-enforced) the driver-session spawns under the invoker's own identity — which carries the identity-keyed layers of the envelope (the agent's own permissions, its topology capability_profile bindings, the _delegate floor) — and is handed the invoker's per-session narrowing, which is keyed by session id and so does not follow from identity. A tool step's dispatch then re-checks that narrowing at run time, because it executes outside the router loop whose gates cover every other tool path.
  5. No tool step launches a pipeline or delegates — nesting is call-only.
  6. Inline-only: an agent step's identity, if set, must equal the invoker's own identity. A registered pipeline is exempt from this check (a trusted registrant deliberately chose the identity); an inline, agent-generated one naming a different identity is a capability escalation and is rejected.

An inline run is crash-recoverable identically to a registered one — its full parsed definition (including its schemas) is persisted into the work-order, so recovery never needs to re-parse or look anything up.

Safety caps

Two operator-set caps in reyn.yaml's safety.spawn block bound a pipeline run's fan-out, threaded into every run/resume call:

# reyn.yaml
safety:
  spawn:
    max_pipeline_fan_out_depth: 5   # default
    max_pipeline_spawns: 100        # default
Key Default Meaning
max_pipeline_fan_out_depth 5 Maximum nesting depth of for_each fan-out scopes (a top-level for_each is depth 1; a for_each inside another's do/collect is depth 2; …). A for_each that would exceed this fails the step rather than spawning. 0 = unlimited.
max_pipeline_spawns 100 Maximum number of ephemeral sessions one pipeline run may spawn across all its agent steps — top-level or fanned out via for_each. A per-run monotonic counter; a spawn past the cap fails the step. 0 = unlimited.

Both default to conservative finite values — a run is never unbounded by omission. Neither cap is reachable by an LLM at run time; both are operator-set and restart-only.

Security

See Pipeline registration § Security: launching a pipeline, in any name=/definition=/collect= combination, sits on the same HIGH-severity, spawn-adjacent capability floor as delegating to another agent. A context narrowed by the untrusted-content floor or an unbound delegate's floor cannot launch a pipeline, registered or inline.

Grammar (for generation)

A compact, self-contained block for an agent authoring a pipeline definition at run time (e.g. for run_pipeline(definition=...)) — the grammar plus the rules that don't fall out of the grammar alone, plus one canonical example. This section stands on its own; it does not assume the prose above has been read.

Grammar — same EBNF as Formal grammar above, repeated here for convenience:

Document      ::= YamlDoc ("---" YamlDoc)*        (* exactly one PipelineDoc total *)
YamlDoc       ::= SchemaDoc | PipelineDoc
SchemaDoc     ::= "schema:" NAME "fields:" FieldMap
PipelineDoc   ::= "pipeline:" NAME ("description:" STRING)? "steps:" Step+

Step          ::= "transform:" "{" "value:" EXPR ["output:" NAME] "}"
                 | "tool:"     "{" "name:" STRING ["args:" ArgMap] ["schema:" NAME]
                                    ["on_error:" OnError] ["output:" NAME] "}"
                 | "agent:"    "{" "prompt:" TPL ["identity:" NAME]
                                    ["capabilities:" "{" "tools:" "[" NAME* "]" "}"]
                                    ["schema:" NAME] ["model:" NAME] ["output:" NAME] "}"
                 | "call:"     "{" "pipeline:" NAME ["pass:" "{" (NAME ":" EXPR)* "}"] ["output:" NAME] "}"
                 | "match:"    "{" "on:" EXPR "cases:" "{" (LABEL ":" MatchTarget)+ "}"
                                    ["default:" MatchTarget] ["output:" NAME] "}"
                 | "fold:"     "{" [ListSource] "init:" EXPR "do:" Step "output:" NAME
                                    ["max_items:" INT] "}"
                 | "for_each:" "{" [ListSource] ["max_parallel:" INT] "on_error:" OnError
                                    "do:" Step "collect:" Step ["output:" NAME] "}"
                 | "parallel:" "{" ["on_error:" OnError] "branches:" "{" (NAME ":" Step)+ "}"
                                    "collect:" Step ["output:" NAME] "}"

MatchTarget   ::= "{" "pipeline:" NAME ["pass:" "{" (NAME ":" EXPR)* "}"] "}"
ArgMap        ::= "{" (KEY ":" ArgValue ("," KEY ":" ArgValue)*)? "}"
ArgValue      ::= LITERAL | "!expr" EXPR
ListSource    ::= "over:" EXPR | "items:" "[" LITERAL* "]"
OnError       ::= "continue" | "abort" | "retry(" INT ")"
FieldMap      ::= "{" (NAME ":" FieldType)+ "}"
FieldType     ::= "{type: bool}" | "{type: string}"
                 | "{type: number[, minimum: LITERAL][, maximum: LITERAL]}"
                 | "{type: enum, values: [" LITERAL+ "]}"
                 | "{type: list, of:" FieldType "}"
                 | "{type: object, fields:" FieldMap "}"
                 | "{type: ref, schema:" NAME "}"
EXPR          ::= (* see The R1 expression language above *)
TPL           ::= (* a string with {ctx.dotted.path} / {pipe} interpolation, values only *)

Hard rules (violating any of these is either a parse error or a run-time step failure — never a silent wrong result):

  1. call's pipeline:, match's case/default pipeline:, and every parallel branch's step: only pipeline: targets in call/match are ever a static literal name — never an expression. The runtime-evaluated match.on only ever selects a case label, never a target directly.
  2. !expr marks a tool argument as an R1 expression; everything else is a literal, passed through untouched. Do not write {ctx.x} inside an unmarked argument expecting interpolation — that only works for agent.prompt (TPL), and only there.
  3. pass: is the only way a call/match callee sees anything from the caller's scope — every entry is an explicit {NAME: EXPR} mapping (no bare-NAME shorthand). EXPR is evaluated against the caller's current full context (ctx/pipe/item/acc, whatever is in scope) and bound to NAME in the callee's ctx. A NAME with no entry is invisible to the callee, not silently inherited — see Data flow between steps.
  4. for_each.on_error is required — state continue/abort/retry(n) explicitly. parallel.on_error is optional and defaults to abort. tool.on_error (#3130) is also optional, but — unlike parallel — omitted is a distinct state from "abort": omitted means no canonical-error (meta.isError) check runs at all (only a raised exception fails the step, the pre-#3130 behavior); an explicit abort additionally fails the step on a canonical-error tool result. See tool.on_error.
  5. A for_each/parallel item or branch cannot see any other item's or branch's result — only collect sees the merged set (an ordered list for for_each, a {branch_name: result} map for parallel).
  6. fold.output is required (a fold's entire point is a named accumulated result); every other step kind's output is optional.
  7. Every agent step is capability-narrowed to at most the invoking session's own envelope — naming a wider capabilities set than the invoker has does not grant it. In an inline (agent-generated, not file-registered) definition, an agent step's identity must be omitted or equal to the invoker's own — naming any other identity is rejected.
  8. !expr may only be the entire value of an args/command entry — never nested inside a list or mapping value.

One canonical example (all three step kinds, one primitive, one schema):

schema: Review
fields:
  passed: {type: bool}
  notes: {type: string}
---
pipeline: review_and_report
description: Review a document and summarize the verdict.
steps:
  - agent:
      prompt: "Review {ctx.doc}. Reply with passed (bool) and notes (string)."
      schema: Review
      output: review
  - transform:
      value: "ctx.review.passed and 'OK' or 'NEEDS WORK'"
      output: verdict
  - tool:
      name: exec
      args: {argv: !expr "['echo', ctx.verdict]"}
      output: shouted