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:;
• run as a graph instead of a list — fork concurrent tracks with split:, gather them with join:, wire edges with next: / depends_on:, and repeat a subgraph with bounded loops:;
• call a connector or findings tool directly — no agent, no tokens — with action:;
• pause for a human gate with approval:, and notify or clean up around it;
• place a step, or a for_each: fan-out, on a remote host with host: / distribute:;
• hand the agents a step launches extra skills with skills: / preload_skills:;
• 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:. A .yml extension works everywhere a workflow is looked up or listed (rupu workflow run / list, autoflows, the control plane); when one directory holds both <name>.yaml and <name>.yml, the .yaml file wins.

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.

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.

These fields apply to any step regardless of shape:

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.

Tool steps with action:

An action: step calls one tool from rupu’s tool catalog — open a pull request, comment on an issue, dispatch a CI pipeline, record or query a finding — directly. There is no agent and no model turn: the parameters you write are the request that gets made, under the run’s permission mode. Use it wherever you would otherwise have asked an agent to make one predictable call. The tools an action step may name are the connector tools (scm.*, issues.*, github.*, gitlab.*) and the findings tools (findings.record, findings.verify, findings.query, findings.tag); anything else — bash, dispatch, coverage.mark, board.post — is a parse error.

  - 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

String values in with: are minijinja-rendered at execution time, so they can reference inputs, earlier steps, and the issue or event context. An optional platform: or tracker: key inside with: picks a specific connector; omit it and the configured default is used. A parameter the tool declares as a number or boolean is typed: a literal with stray quotes (number: "7") is rejected at parse time, a whole-leaf template ("{{ n }}") is rendered and then parsed against the declared type, and a partial template ("issue-{{ n }}") against a typed field is a runtime error rather than a guess.

The tool’s return value binds to {{ steps.open_pr.output }} as a JSON string, so pull fields out of it with the fromjson filter — {{ (steps.fetch.output | fromjson).title }}, not steps.fetch.output.title. steps.<id>.success says whether the call succeeded, and when: / continue_on_error: behave as on any step. An action step may not carry a non-empty actions: list: its tool is already explicit, so there is nothing to narrow.

  - id: tagged
    action: findings.query
    with: { q: "tag:needs-poc severity>=high", all: true }
  - id: poc
    agent: poc-writer
    for_each: "{{ (steps.tagged.output | fromjson).rows | tojson }}"
    max_parallel: 4
    prompt: |
      Write a proof of concept for finding {{ item.id }}: {{ item.title }}
      ({{ item.location }}). When it works, tag it `has-poc` with findings.tag.

That is the fan-out pattern for findings: findings.query with all: true returns every match unpaged, | tojson hands for_each an exact JSON array, and each item is a slim row (id, title, severity, scope, location, concern_id, tags, declared_at, run_id). An action: findings.record step records a finding under the step’s findings_profile; see Coverage.

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.

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.

  - id: merge_gate
    approval:
      required: true
      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 pauses unless auto_approve resolves it (every documented gate sets required: true, as the inline form does). Its output is a small JSON record — the decision (approved / rejected), how it was reached (via: human, auto, or timeout), the approver, the reason, and a timestamp — and the decision is pre-parsed as {{ steps.merge_gate.decision }}, no fromjson needed. Everything nested under notify and on_reject is validated at parse time with the same rules as any other action or step.

Gate decisions are path-scoped. In a graph workflow several gates can be parked at once — one per branch of a split:, say. Each decision applies only to its own gate: approving releases just the path behind that gate, and rejecting prunes only the steps reachable solely through it (running that gate's on_reject chain) while sibling paths carry on. Name the gate with --gate <step-id> once more than one is parked; the run stays awaiting_approval until every gate is resolved. Exactly one runner applies decisions: one recorded while the run is still executing elsewhere is handed to that runner, so an approve from the CLI and one from the control plane can never resume the same run twice.

# two gates parked on parallel paths — decide each one
rupu workflow approve run_01JC2X… --gate ship_api
rupu workflow reject  run_01JC2X… --gate ship_web --reason "bundle size regression"

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:

Messaging between the steps

Every workflow run has one message space, and every agent step and unit is offered the messaging tools by default — board.claim, board.release, board.post, board.read and msg.send — so steps can leave each other notes and the operator can message any of them while the run is live. The top-level messaging: key controls it: false turns it off for every step and unit and everything they launch, and { tools: [board.post, board.read, msg.send] } offers only the board.* / msg.* tools named. A placed unit gets the same setting on its host (rupu run --no-messaging, or one --messaging-tool per tool); a host whose rupu can’t honour it is refused the unit.

name: quiet-review
messaging: false         # no messaging tool for any step or unit, or anything they launch
steps:
  - id: review
    agent: reviewer
    prompt: Review crates/rupu-cp.

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."

What travels with a placed unit

A placed unit runs as rupu run <agent> on its host, and the coordinator launches it with everything the step resolved:

• Permission mode — the workflow run’s --mode, which outranks the agent’s permissionMode and the host’s own config. readonly and bypass travel as they are; ask (the default) travels as bypass, so --mode readonly is how you keep a placed unit from writing.
• Findings profile — the step’s findings_profile, else defaults.findings_profile, as rupu run --findings-profile; when neither is set the host resolves the agent’s findingsProfile from its own copy of the agent file. A host that can’t honour the profile refuses the launch — the unit never runs under a different one. The same goes for the run’s engagement profile ids (--engagement-profile, refused by a host that doesn’t advertise agent.engagement_profile).
• Skills — the coordinator resolves the unit’s skills (its own copy of the agent’s plus the step’s skills: / preload_skills:), packs each as a tar named by its content hash and ships only the ones the host lacks; the host verifies every archive and runs the unit with exactly those copies, so it needn’t have them installed. A host that doesn’t advertise agent.skill_bundle runs the unit with its own installed copies, and the run gets a warning naming the host. A step skill missing on the coordinator fails the unit before it is dispatched.
• Messaging — the workflow’s messaging: switch, as --no-messaging or one --messaging-tool per tool; a host that doesn’t advertise the capability is refused the unit rather than run it with messaging on.
• Not actions: — a non-empty actions: on a host: / distribute: step is rejected at parse time, since the narrowing can’t reach the remote agent.

A unit’s findings and coverage come back to the coordinator through the run’s coverage stream, and each placed-unit error or warning is prefixed host <id>:. On resume a placed step or unit is restarted from its prompt rather than continued from its transcript.

Template expressions

Prompts, when: conditions, and for_each: lists are rendered with minijinja. Missing variables render as empty strings. The most common references are:

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.* — event.id, event.vendor and event.repo.full_name at the top, with the vendor’s own object one level down under event.payload.* (event.payload.pull_request.merged, never event.pull_request.merged).

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 "...".

# run a draft without installing it, against a network engagement
rupu workflow run --file ./drafts/edge-sweep.yaml \
  --engagement-profile network \
  --input scope="10.20.0.0/24" --mode readonly

The live view

In a terminal, rupu workflow run and rupu workflow resume open a three-pane dashboard that reads as a workflow and keeps the whole run visible at once. A header carries the workflow, its codename, elapsed time and live meters (tokens, cost when the model is priced, findings and coverage when there are any); the body is split into three panes:

• Structure — the workflow as a DAG drawn with rails: forks for split: and parallel:, merges for join:, nested members for loops and panels, one glyph per step kind. A for_each step shows a density line (done / running / failed / waiting) and its live units. Settled steps fold into summary rows so a large run fits; running and parked steps never fold.
• Stream — the transcript of whatever is selected: a step, or after drilling in, one fan-out unit or sub-agent. Tool calls, thinking, replies and findings, as they happen. With nothing selected it follows the active step.
• Firehose — every agent in the run merged into one feed, each line tagged with its codename and, for placed work, its host. It is always on screen; an idle run shows waiting for activity… rather than an empty box.

 nightly-audit · slate-meadow · ◐ 4m12s · ⇡182k ⇣21k $0.94 · ⚑3
 ─ structure ──────────────┬─ stream · kestrel#12 ───────────────────
  ● collect ······· ✓ 18s  │ ▸ read_file  src/auth/session.rs
  ⊞ review · 31/48 ◐       │ ◇ thinking   "the token compare is not…"
    ✓29 ◐4 ✗2 ○13  ⚑3      │ ⚑ HIGH       timing-unsafe token compare
  ⏸ merge_gate · gate      ├─ live · 4 active ───────────────────────
  ○ summarize              │ kestrel#9   ▸ grep  "unwrap()"
                           │ kestrel#12  ⚑ HIGH  timing-unsafe compare
 tab pane · ↑↓ move · enter drill · ← back · / filter · Esc pause · q quit

In a narrow terminal the panes stack and tab cycles which one is shown. Pass --plain (or set RUPU_LIVE_VIEW=0) for the line printer; output that isn't a terminal always gets it. Either way the run ends with one completion summary built from the run's own records.

Codenames

Every run, step, fan-out unit and sub-agent gets a short human codename next to its id. The run is a crew (slate-meadow); each agent in it is a role named after an animal, numbered when there can be several (slate-meadow/kestrel#12 is the twelfth unit of a fan-out, and …/kestrel#12>wren#1 a sub-agent it dispatched). The same agent keeps the same role word across runs, so a name you recognise tells you which specialist did the work. Names are minted once and stored, and they appear in the live view, rupu workflow runs, transcripts, findings and the control plane. The rupu workflow commands that take a run id also accept its crew name (rupu workflow show-run slate-meadow); if two recent runs share it, the newest wins and rupu says so.

Pause, resume and recover

rupu workflow pause <run-id> stops a run at its next safe boundary and rupu workflow cancel <run-id> stops it outright. rupu workflow resume <run-id> takes a paused, failed, cancelled or rejected run back and re-runs only what didn't finish — and an agent that was cut off mid-flight (killed, crashed, out of memory) is continued from its transcript rather than started over. If its answer had in fact finished, that answer is reused with no model call. Pass --restart-interrupted to start interrupted steps over from their prompts instead. parallel: sub-steps, panel: members, steps inside a loops: subgraph and placed (host: / distribute:) units still restart. See Reliability for the full recovery model.