Workflows
A workflow composes several agents into ordered steps — one YAML file that orchestrates the run and passes data from each step to the next.
What is a workflow
Where an agent is a single AI worker, a workflow is several of them composed into an ordered process. It lives in one YAML file and describes which specialist owns each stage, where the review and approval boundaries are, and how data flows from one step to the next.
A workflow can:
- run linear steps in order;
- fan one agent out across many items with
for_each:; - fan many specialist agents out over the same subject with
parallel:; - run structured review panels with an optional fix loop using
panel:; - route down one of two arms with
branch:; - call an SCM, issue-tracker, or CI tool directly — no agent, no tokens — with
action:; - pause for a human gate with
approval:, and notify or clean up around it; - start manually, on cron, or from an event trigger.
Step prompts are rendered with minijinja templates against the workflow's inputs, prior step outputs,
and optional issue or event context — that is how data moves between steps. Workflow files resolve from
~/.rupu/workflows/<name>.yaml (global) and
<project>/.rupu/workflows/<name>.yaml (project-local), with
project-local files shadowing global ones by name:.
for_each: step fans out over a list, then later steps consume the aggregate.The YAML format
A workflow is a YAML document with a handful of top-level keys. Only name and
a non-empty steps array are required; everything else is optional. The
frontmatter is parsed strictly — extraneous fields inside blocks such as trigger:
are rejected at parse time.
| Key | Type | Required | Meaning |
|---|---|---|---|
name | string | yes | Workflow identifier used by rupu workflow run <name>. |
description | string | no | Human-readable summary. |
trigger | object | no | How the workflow starts: manual (default), cron, or event. |
inputs | map | no | Typed runtime inputs (type, required, default, enum). |
defaults | object | no | Workflow-wide defaults — e.g. continue_on_error: true inherited by steps. |
contracts | object | no | Named structured outputs validated against a JSON Schema. |
autoflow | object | no | Autonomous-ownership metadata for rupu autoflow (see Autoflows). |
notifyIssue | bool | no | Auto-comment back only when the run target is an issue. |
steps | array<Step> | yes | The ordered step list. An empty array is invalid. |
Inputs are declared once and supplied at run time. Each input takes a type of
string, int, or bool; an
optional required flag; an optional default that must
match the type; and an optional enum of allowed stringified values.
inputs: phase: type: string required: true retries: type: int default: 3
Step kinds
Every step has a unique id and exactly one execution shape. The shapes are
mutually exclusive and inferred from which block the step sets — a linear step, a
for_each: fan-out, a parallel: multi-agent fan-out,
a panel: review, a branch: routing decision, an
action: connector call, a run: command, or a
standalone approval: gate. Not every step runs an agent:
branch:, action:, run:,
and gate steps are deterministic and cost no tokens.
| Shape | Defining fields | What it does |
|---|---|---|
linear | agent + prompt | One agent runs one prompt; its output feeds the next step. The basic shape. |
for_each | for_each (+ agent, prompt) | Renders a list and runs the same agent once per item, with max_parallel concurrency. |
parallel | parallel (list of sub-steps) | Different specialists run concurrently over the subject. Each sub-step has its own id / agent / prompt; the parent must not set agent / prompt. |
panel | panel (panelists + subject) | Several reviewer agents emit structured findings over one subject, with an optional gate review/fix loop. |
branch | branch (condition + then / else) | Evaluates a condition and skips the arm that wasn’t taken. No agent, no tokens. |
action | action (+ with) | Calls one SCM / issue / CI tool from the MCP catalog directly — a deterministic API call, no agent, no tokens. |
run | run (cmd + args) | Executes a declared command as an argv vector and binds its output. No language model, no tokens. Composes with for_each. |
approval gate | approval alone | A step whose only job is the human decision. Pauses the run, and can notify, auto-approve, route a timeout, or run cleanup on reject. |
These fields apply to any step regardless of shape:
| Key | Applies to | Meaning |
|---|---|---|
id | all | Unique within the workflow. |
actions | all | Action-protocol allowlist — not a tool allowlist. Use [] unless you intentionally use the action protocol. |
when | all | Minijinja expression reduced to truthy / falsy; falsy values are empty string, false, 0, no, off. |
continue_on_error | all | Tolerate failure and keep going; inherits the workflow default. |
max_parallel | for_each, parallel, panel | Concurrency cap (at least 1). |
approval | all | Human pause before the step dispatches (checked after when:). On a step with no other shape it becomes a gate node of its own. |
The sections below cover the shapes you reach for day to day. For the complete field-by-field spec — every key, its type, its defaults, and the parse-time rules — see the Authoring reference.
Conditional steps with when:
when: is rendered as a template and then reduced to a boolean. If it is falsy
the step is skipped (and steps.<id>.skipped becomes true downstream).
when: "{{ steps.review.success }}" when: "{{ steps.panel.max_severity == 'critical' }}"
Conditional routing with branch:
Where when: gates one step at a time, branch: routes
the run down one of two arms. The step evaluates condition — a template
expression, truthy by the same rules as when: — then marks every step id in the
arm that wasn’t taken to be skipped as the run walks past them. Steps in neither arm are the
rejoin point and always run.
- id: triage agent: classifier prompt: "Is this report a security issue or a docs issue? Answer with one word." - id: route branch: condition: "{{ 'security' in steps.triage.output }}" then: [deep_scan] # runs when the condition is truthy else: [doc_fix] # runs when it is falsy - id: deep_scan agent: security-reviewer prompt: "Audit the reported behaviour end to end." - id: doc_fix agent: writer prompt: "Correct the documentation." - id: report # in neither arm — the rejoin, always runs agent: writer prompt: "{{ steps.deep_scan.output }}{{ steps.doc_fix.output }}"
A branch step sets no agent / prompt and no other
shape. Both then and else default to an empty list.
The rules are checked at parse time: every listed id must exist, must be declared after the branch
step (routing is forward-only), and no id may appear in both arms or name the branch itself. The lists are
the complete membership of each arm — if an arm contains a nested branch, list that nested
branch’s steps too. when: is not allowed on a branch step; express the
condition in the branch instead. Downstream, {{ steps.route.output }}
is the arm that was taken (then or else), and a
skipped step’s output renders as the empty string — which is what makes the rejoin above work.
Connector steps with action:
An action: step calls one tool from rupu’s
MCP tool catalog — open a pull request, comment on an issue,
dispatch a CI pipeline — directly. There is no agent and no model call: the parameters you
write are the request that gets made, so the step is deterministic, instant, and costs no tokens. Use it
wherever you would otherwise have asked an agent to make one predictable API call.
- id: open_pr action: scm.prs.create with: owner: Section9Labs repo: rupu title: "fix: {{ inputs.title }}" body: "{{ steps.implement.output }}" head: "rupu/issue-{{ issue.number }}" base: main
with: values are minijinja-rendered at execution time, so they can reference
inputs, earlier steps, and the issue or event context. The tool’s JSON response binds to
{{ steps.open_pr.output }}, and
when: / continue_on_error: behave as on any step. An
optional platform: or tracker: key inside
with: picks a specific connector; omit it and the configured default is used.
Only string values are templated — a parameter the tool declares as a number or a boolean takes a literal
(draft: true, number: 412) and passes through as it
is written.
with: key are
validated at parse time against the static tool catalog — an unknown tool, an unknown parameter, or a
missing required one is a workflow error, not a run-time surprise, and no credentials are needed to lint it.
At run time the call goes through the same in-process tool layer and the same permission gating agents use:
in readonly mode a Write-class action step fails with an explicit message, and the
executed call is recorded in the run’s audit trail.
Deterministic commands with run:
Reach for run: when a step must produce an exact value —
scoring a benchmark, validating a fixture, rendering a report, a preflight check. No language model is
involved: the runner executes the command you declared and binds its output. Same answer every time, no
tokens spent.
- id: score run: cmd: python3 args: ["tools/score_fixture.py", "{{ item.fixture }}", "{{ item.candidate }}"] cwd: "{{ inputs.cybermark_root }}" env: { CYBERBENCH_RELEASE_KEY: "{{ inputs.release_key }}" } parse: json # json | lines | raw (default: raw) timeout_seconds: 300 allow_exit_codes: [0] # default [0]; anything else fails the step
There is no shell. cmd and args are
handed to the OS as an argv vector — no pipes, no redirection, no globbing, no word splitting. Each
args entry renders independently and becomes exactly one argument, so a rendered
value containing ;, &&, or
$(...) arrives as literal text instead of executing. If you genuinely need shell
features, invoke a shell yourself — cmd: sh,
args: ["-c", "..."] — visible in the workflow file, and your decision rather than
something the engine does to every step behind your back.
Structured output is indexable through {{ steps.<id>.json }}
under parse: json or parse: lines;
{{ steps.<id>.output }} stays a string for every step kind, and
.stdout, .stderr,
.exit_code and .duration_ms are bound too. Under
parse: json, stdout that is not valid JSON fails the step rather than
quietly falling back to the raw string.
run: composes with for_each: and
max_parallel:, which is how you score N items concurrently — per-unit results land
in steps.<id>.results in declared order regardless of finish order.
distribute: is not supported and is rejected at parse time: a
run: step executes on the coordinator, and silently dropping it would let you
believe work was spread across a fleet when it never left the local host.
run: executes commands, so it is off by
default. Set [workflow] run_step_enabled = true in your rupu config, and
optionally narrow it with run_step_allowlist = ["python3", "make"] (matched on
basename, so /bin/bash and bash gate alike). With the
toggle off, a workflow containing a run: step fails — it does not skip
the step, because a benchmark that quietly omitted its scoring step would report a plausible-looking but
meaningless number. readonly mode refuses run: steps
outright, and bypass cannot override the workspace opt-in.
For the field-by-field spec — every key, its type, its default, the output bindings, and the parse-time rules — see the Authoring reference.
Human gates with approval:
When a step sets approval.required: true, the run pauses before that step
dispatches. Resume with rupu workflow approve <run-id> or reject with
rupu workflow reject <run-id> --reason "...".
approval: required: true prompt: | About to deploy {{ inputs.tag }}. Approve? timeout_seconds: 3600
That inline form is still supported. A step that sets approval: and
nothing else — no agent, no prompt, no fan-out — is a gate node: a step of its own
whose only job is the decision, with its own id, its own place in the graph, and richer options.
| Gate field | Meaning |
|---|---|
prompt | What the approver reads, rendered with the usual template context. |
auto_approve | Template expression evaluated when the gate is reached; truthy resolves the gate without pausing. Gate nodes only. |
timeout_seconds | How long the gate waits before it expires. |
on_timeout | approve, reject, or fail (the default) when the gate expires. Requires timeout_seconds. |
notify | Connector actions fired best-effort when the gate opens — e.g. comment on the issue that a human is needed. A failed hook is logged, never blocks the pause. |
on_reject | Inline cleanup steps run after a reject, before the run ends. Plain agent steps and action steps only — no nested gates and no fan-out. |
- id: merge_gate approval: prompt: | {{ steps.review.max_severity }} findings. Approve to open the PR. auto_approve: "{{ steps.review.max_severity == 'low' }}" timeout_seconds: 86400 on_timeout: reject notify: # fired when the gate opens - action: issues.comment with: project: Section9Labs/rupu number: 412 body: "Approval needed at the merge gate for {{ inputs.tag }}." on_reject: # cleanup after a reject, then the run ends - id: file_followup action: issues.create with: project: Section9Labs/rupu title: "Merge gate rejected for {{ inputs.tag }}" body: "{{ steps.review.output }}"
A gate node always pauses unless auto_approve resolves it
(required: is only meaningful on the legacy inline form). Its output is a small
JSON record — the decision
(approved / rejected), how it was reached
(via: human, auto, or
timeout), the reason, and a timestamp — so a later step can read
{{ steps.merge_gate.decision }}. Everything nested under
notify and on_reject is validated at parse time with
the same rules as any other action or step.
The panel: gate loop
A panel: step runs its panelists over the rendered
subject; each panelist's final message must contain a parseable JSON object with
a findings array. An optional gate turns the panel into
a review/fix loop:
| Gate field | Required | Meaning |
|---|---|---|
until_no_findings_at_severity_or_above | yes | Severity threshold to clear: low, medium, high, or critical. |
fix_with | yes | Agent that addresses findings between passes; its output becomes the next pass's subject. |
max_iterations | yes | Cap on panel passes (at least 1). The loop stops when the gate clears or this is reached. |
Run steps across the fleet
Any step can execute on a remote host instead of the local machine. There are two placements plus an opt-in file sync — omit them all and everything runs locally, exactly as before.
Fan out a for_each across hosts
A distribute: block spreads a for_each step's units round-robin across the named hosts; results aggregate back like a local fan-out.
- id: review_each agent: code-reviewer for_each: "{{ inputs.files }}" distribute: hosts: [gpu-box, build-box] # round-robin across fleet hosts prompt: "Review {{ item }}."
for_each-only; hosts must be non-empty; each unit is attributed to its host; failures honor continue_on_error.
Pin a single step to a host
A host: on a linear step (agent + prompt) runs just that step on one host — good for a step that needs a specific machine (a GPU box, a licensed toolchain).
- id: heavy_analysis agent: analyzer host: gpu-box prompt: "Analyze {{ steps.collect.output }}."
host: is valid only on a linear step (not for_each/parallel/panel) and must be non-empty.
Let a remote step touch the repo — workspace: sync
By default a remote step is self-contained: it sees only its rendered prompt and prior steps' string outputs, never your files. To let a remote step read and write real project files, add workspace: sync — the coordinator ships the workspace to the host (auto git-or-tar), the agent works on it, and the changed files are propagated back so downstream steps see them.
- id: apply_fix agent: developer host: build-box workspace: sync # file-mutating step, run remotely prompt: "Apply the change and run the tests."
workspace: sync is only valid on a remote step (one with host: or distribute:); the git-or-tar mode is detected automatically, no config needed.Template expressions
Prompts, when: conditions, and for_each: lists are
rendered with minijinja. Missing variables render as empty strings. The most common references are:
| Expression | Resolves to |
|---|---|
{{ inputs.<name> }} | A runtime input value. |
{{ steps.<id>.output }} | The final output string of an earlier step. |
{{ steps.<id>.success }} | Whether that step completed successfully. |
{{ steps.<id>.skipped }} | Whether that step was skipped by when:. |
{{ steps.<id>.results }} | Per-item (for_each) or per-sub-step (parallel) outputs. |
{{ steps.<id>.sub_results.<sub_id>.output }} | A named output from a parallel: sub-step (also .success). |
{{ steps.<id>.findings }} | Aggregated panel findings (also .max_severity, .iterations, .resolved). |
{{ item }} · {{ loop.index }} | The current item and 1-based index inside a for_each: (also loop.index0, loop.length, loop.first, loop.last). |
read_file('<path>') | The contents of a file a prior step wrote, resolved against the run's working directory. Fails loudly if missing. |
When the workflow is invoked against an issue target, an issue.* context is also
available (issue.number, issue.title,
issue.labels, and more); event-triggered workflows get the payload under
event.*.
for_each: JSON-parses any
rendered value that starts with [, have an upstream step write a clean JSON
array file and feed it with for_each: "{{ read_file('reports/items.json') }}".
That decouples control flow from how terse the agent was in chat.
A complete example
This is the investigate-then-fix workflow that ships as a rupu template — a
two-step linear bug fix where the second step consumes the first step's output via a template expression.
name: investigate-then-fix description: Two-step bug fix — investigate, then propose minimal edit. steps: - id: investigate agent: fix-bug actions: [] prompt: | Investigate the bug described by: {{ inputs.prompt }} Stop without making edits. Report the root cause as text. - id: propose agent: fix-bug actions: [] prompt: | Based on this investigation: {{ steps.investigate.output }} Propose and apply the minimal fix.
The fan-out shapes follow the same skeleton. A for_each: step adds a
for_each list and references {{ item }};
a later step then folds the results with a loop over
steps.review_each.results:
- id: review_each agent: code-reviewer actions: [] for_each: "{{ inputs.files }}" max_parallel: 4 prompt: | Review file {{ item }} ({{ loop.index }} of {{ loop.length }}). - id: summarize agent: writer actions: [] prompt: | Combine these per-file reviews into one summary. {% for r in steps.review_each.results %} {{ r }} {% endfor %}
Generate one from a description
You can also describe the pipeline in plain language and let rupu draft the YAML. create generates the workflow with a model — running a validate → repair loop against the real schema so the output parses — then opens it for you to refine. The same flow lives in the control plane’s authoring UI (and its visual editor).
# AI-draft a workflow from a one-line brief rupu workflow create nightly-audit \ --describe "Run a parallel security + performance review panel over the diff, then summarize" # or scaffold an empty workflow to edit yourself (omit --describe) rupu workflow create my-workflow
Running a workflow
List and inspect the workflows rupu can see, then run one:
# list every resolvable workflow (project + global scope) rupu workflow list # show one workflow's resolved definition rupu workflow show investigate-then-fix # run it, passing typed inputs rupu workflow run investigate-then-fix --input prompt="NPE on empty cart checkout"
Pass each declared input with a repeated --input key=value flag. If the
workflow also takes an issue target, supply it positionally before the inputs:
rupu workflow run <name> <issue-ref> --input .... When a step requires
approval:, the run pauses; resume it with
rupu workflow approve <run-id> or reject with
rupu workflow reject <run-id> --reason "...".