Skip to content

feat(workflow): bounded loops, parallel for_each and list expressions - #1148

Draft
SaulMoro wants to merge 18 commits into
Tencent:mainfrom
SaulMoro:feat/1079-loops-for-each
Draft

SaulMoro wants to merge 18 commits into
Tencent:mainfrom
SaulMoro:feat/1079-loops-for-each

Conversation

@SaulMoro

@SaulMoro SaulMoro commented Oct 15, 2026 •

Copy link
Copy Markdown
Collaborator

Summary

Workflows can now repeat work and fan it out. A step loop runs a range of steps again, a workflow loop runs every step round after round, break ends a workflow early, and for_each runs a step once per item. A driver that lists the attempts it runs in running gets several leases at once, and --driver local uses that to run up to workflows.max_parallel commands together.

What a workflow author can write:

# workflow, top level or inline
loop: { max_iterations: 30, while: "<expr>" }                  # run every step again, round after round

# step
loop: { back_to: review, while: "<expr>", max_iterations: 3 }  # run back_to through this step again
break: "steps.check.outputs.state == 'passed'"                 # on success, end this workflow as success
for_each: { items: inputs.packages, max_parallel: 2 }          # instances build[0], build[1], ... each reads item
if: any(steps.build.results, .outcome == 'failed') && length(steps.review.passes) < 3
#   new reads: steps.<id>.passes, steps.<id>.results, item, [n], < <= > >=, length any all filter

# local config
workflows: { max_parallel: 4 }                                 # --driver local runs up to 4 commands at once

What one next call does:

next(run, { driver, running?, canTakeLease? })
├─ settle, phase 1: everything that needs no driver
│  ├─ step loops: run the range again, stop, or fail loop_exhausted
│  ├─ skips, end-only steps, finished sub-workflows and for_each steps
│  ├─ apply the end or break reached first (lowest settledAt), cancel the leases still open under it
│  └─ workflow loops: run another round, stop, or fail loop_exhausted
├─ run has ended                         → done { outcome, end?, failure? }
├─ own open lease missing from running   → that lease again
├─ canTakeLease is false                 → wait
├─ settle, phase 2: first ready step     → lease { step, attempt, revision, pass }
└─ every ready step is leased            → wait { reason: first open lease }

Every data answer also carries cancelled, the attempts in running that are no longer this driver's open leases.
A driver that sends no running gets its open lease back on every call, as today.

Review commit by commit:

Commits What they deliver
a0a516b, e4ffa3a, cffbb2d, a64ed7d Refactors with no behavior change. Read checks move to reads.ts, step paths get one parser, and step state becomes a zod union with transitions. A run file written before this change is still read.
6978410, 8be3fde Parallel leases. next settles in two phases and takes running and canTakeLease. report checks each report against its own lease, and ends are ordered by settledAt.
62b1714 The cancelled status, lease_cancelled, and cancelled in every next answer.
d1a91c9, a0057d8 Expressions gain [n], < <= > >=, and length, any, all, filter.
f2ecbf7, e28e862 Step loops, steps.<id>.passes, and reads of a later step's previous pass.
92cfbbd Workflow loops and break.
5a41539, a4b99da The guide splits into a base and an advanced page in English and Chinese, then documents loops and local parallel commands.
b998e8c --driver local runs up to workflows.max_parallel commands.
ca40965, dfea7fa for_each, item, and results with their docs, after a refactor that lists a step's expressions in one place.
0d738ae The real-CLI e2e test below, as src/__tests__/e2e/workflow-loop-fan-out.test.ts.

Type of Change

  • Bug fix (non-breaking change that fixes an issue)
  • New feature (non-breaking change that adds functionality)
  • Breaking change (fix or feature causing existing behavior to change)
  • Documentation only
  • Refactor / internal cleanup

Test Plan

Both runs below use a fresh git repo with an isolated HOME, teamai init --self --agent claude --force, and this workflow. tick.js bumps a counter file and prints {"count":N}. barrier.js writes <ME>.ready and exits 0 once <OTHER>.ready exists, or 1 after 3 s. record.js writes PASSES and RESULTS to record.json.

schemaVersion: v1alpha1
name: loop-and-fan
description: Count to three in a loop, meet two items at a barrier, then record both.
inputs:
  pairs: [{ me: string, other: string }]
steps:
  - id: tick
    command: node tick.js
    output:
      count: number
    loop: { back_to: tick, while: "steps.tick.outputs.count < 3", max_iterations: 5 }
  - id: meet
    needs: [tick]
    for_each: inputs.pairs
    command: node barrier.js
    env:
      ME: "{{ item.me }}"
      OTHER: "{{ item.other }}"
  - id: record
    needs: [tick, meet]
    if: length(steps.tick.passes) == 3 && length(steps.meet.results) == 2
    command: node record.js
    env:
      PASSES: "{{ steps.tick.passes }}"
      RESULTS: "{{ steps.meet.results }}"

Before

CLI built from the base commit 27a026f. Pull warns and keeps no runnable version, so the run never starts. The output leaves out the info lines of pull and three more warnings, one for each new read in record.

$ teamai pull
⚠ [workflows] workflows/loop-and-fan.yaml:11: field "steps[0].loop": workflows/loop-and-fan.yaml uses `loop`, which this teamai does not know. Update teamai.
⚠ [workflows] workflows/loop-and-fan.yaml:14: field "steps[1].for_each": workflows/loop-and-fan.yaml uses `for_each`, which this teamai does not know. Update teamai.
⚠ [workflows] workflows/loop-and-fan.yaml has no runnable version
exit 0
$ teamai workflow start loop-and-fan --driver local --inputs '{"pairs":[{"me":"left","other":"right"},{"me":"right","other":"left"}]}'
{"error":{"code":"workflow_not_found","message":"Workflow \"loop-and-fan\" has no runnable version. Check workflows/loop-and-fan.yaml and pull again."}}
exit 1

After

CLI built from this branch (npm run build, teamai linked to it). The local config's workflows.max_parallel is set before each start.

$ teamai pull
exit 0

# local config: workflows.max_parallel: 2
$ teamai workflow start loop-and-fan --driver local --inputs '{"pairs":[{"me":"left","other":"right"},{"me":"right","other":"left"}]}'
{"data":{"done":{"run":"r30bd434e","outcome":"succeeded"},"cancelled":[]}}
exit 0 after 200 ms
$ cat ticks.txt
3
$ cat record.json
{"passes":[{"pass":1,"outcome":"success","outputs":{"count":1}},{"pass":2,"outcome":"success","outputs":{"count":2}},{"pass":3,"outcome":"success","outputs":{"count":3}}],"results":[{"item":{"me":"left","other":"right"},"outcome":"success"},{"item":{"me":"right","other":"left"},"outcome":"success"}]}

# local config: workflows.max_parallel: 1
$ teamai workflow start loop-and-fan --driver local --inputs '{"pairs":[{"me":"left","other":"right"},{"me":"right","other":"left"}]}'
{"data":{"done":{"run":"r750ddd8e","outcome":"failed"},"cancelled":[]}}
exit 1 after 3300 ms
$ cat ticks.txt
3
$ cat record.json
{"passes":[{"pass":1,"outcome":"success","outputs":{"count":1}},{"pass":2,"outcome":"success","outputs":{"count":2}},{"pass":3,"outcome":"success","outputs":{"count":3}}],"results":[{"item":{"me":"left","other":"right"},"outcome":"failed"},{"item":{"me":"right","other":"left"},"outcome":"success"}]}

$ pgrep -fl "tick.js|barrier.js|record.js"
(none)

At 2 the loop ran three passes and both items overlapped, so the run succeeded. At 1, meet[0] waited 3 s alone and failed. meet[1] still ran and succeeded, so a failed item cancelled no sibling. record ran in both cases because its if reads meet. src/__tests__/e2e/workflow-loop-fan-out.test.ts covers the same two cases.

The checkboxes are ticked when the PR leaves draft. CI runs on every push until then.

  • npx tsc --noEmit passes
  • npm run lint passes
  • npx vitest run passes
  • Added/updated tests for the change

Merge Danger

Door: Two-way. Each CLI reads the other's run files. This branch reads a run file written before it (fixture src/__tests__/fixtures/workflow/run-before-loops.json). The base CLI reads a run file this branch wrote, ignores the new fields, and continues the run. Checked by starting runs with this branch and calling workflow next on them with the base CLI:

Run started on this branch Base CLI
Two plain steps, a reported Leases b.
loop: { back_to: tick, max_iterations: 2 }, pass 2 leased Hands back the pass 2 lease, then answers done with succeeded after its report. This branch ends the same run failed, because tick would run a third time.
fail-fast, test cancelled at the broken end Answers done with failed and broken. A report for the cancelled attempt gets stale_revision instead of lease_cancelled.

A rollback loses no run file. A run that uses loops and is continued by an older CLI runs without its loop.

Blast Radius: Contained. Workflows are v1alpha1 and ship only in the 0.28.0 betas, and npm latest is 0.27.0.

Who What changes
External drivers A driver that sends no running gets the same leases and waits. Answers gain cancelled on every data answer, pass on every lease, and failure on a done that no end names, such as a top-level loop_exhausted. A driver that rejects unknown fields must accept them.
--driver local Unchanged at the default max_parallel: 1. Above 1, commands share the working tree, run without a terminal, and Ctrl-C, SIGTERM, SIGQUIT, or SIGHUP stops every command.
Workflow authors New fields and reads. A step without for_each whose text quotes {{ item }} no longer passes pull.
Runs in progress A run started before this change continues on it. A run started on it and continued by an older CLI loses its loops, as above.

Related Issues

Closes #1079. The plan is in this comment.

Notes for Reviewers

  • Breaking change is ticked for one case. The base CLI accepts prompt: "Explain what {{ item }} means in a Jinja loop." in a step without for_each. This branch rejects it with reads item, which is bound only in the templates of a for_each step, and the file has no runnable version. ${{ ... }} still keeps its braces as written.
  • Not run on Windows, where taskkill and signal re-raise apply with max_parallel above 1. CI has no Windows job. Linux runs in CI.
  • History may be rebased until the PR leaves draft.

@SaulMoro
SaulMoro force-pushed the feat/1079-loops-for-each branch 5 times, most recently from 3fed576 to 709cf64 Compare October 15, 2026 13:31
readScope, readDiagnostics, splitReads, conditionDiagnostics and
endDiagnostics move unchanged, with the types only they use. Loop and
for_each read rules will extend them there instead of growing parse.ts.

The YAML and diagnostic helpers both files use move unchanged to
parse-context.ts, so parse.ts and reads.ts do not import each other.
parseStepPath reads an instance path such as a.b[2].c into step ids and
item indexes. statesIn, localId and locateStep go through it, and statesIn
no longer counts a name[n] item as a step of the enclosing workflow. No
run holds an item yet, so behavior is unchanged.
StepState is now a discriminated union on status, defined once as a zod
schema in step-state.ts with its type inferred. Each status carries only
its own fields, including the ones loops and for_each will fill. The
transitions skip, lease, release, start, succeed and fail are the only
writers of step fields; stopped() and settled() move beside them and
switch over every status.

readRun parses steps through the schema. A run file from 0.22.0 reads
with settledAt 0, outcome taken from status where the engine settled the
step, and no keys its status does not have. A fixture written by the
0.22.0 CLI covers a succeeded command, an end-only step inside an inline
workflow, and a released step, and the run continues to the end.

One visible change: a released report now leaves a clean pending step.
Before, the step kept the outcome, exit code and message of the attempt
it gave back, which read as the result of a run that never finished. The
report envelope still answers outcome released.
The inline workflow test matched the stored steps loosely, so a stray
field such as the enclosing inputs on the step would pass. It now
compares the stored step exactly.
next takes two optional payload fields: running, the attempts of its
open leases a driver still runs, and canTakeLease. settleScope settles
to a fixpoint and starts over on any change. First it settles what
needs no driver: skips, end-only steps, and every running workflow step
whose own workflow has a result. Then it applies the reached end. Only
then does next hand back an open lease the driver did not list, and
only a driver that can take a lease gets the first ready step. A
running workflow step whose steps have nothing to lease no longer
blocks the steps after it, and with every ready step leased, next
answers wait instead of run_unreadable. Whatever settling changed is
saved, a wait included.

Every next answer lists in cancelled the running attempts that are no
longer open leases of the driver. A driver that sends neither field
gets its open lease back first, as before. Two engine tests compared a
whole done envelope and now include the empty cancelled list.

parseDriverPayload becomes parseNextPayload, since it parses the whole
next payload, and rejects running or canTakeLease of the wrong type.
…by settledAt

A report is now checked against the revision its own lease was handed
out at, not the run revision, so the report of a parallel lease does
not make another open lease stale_revision. An attempt that is not an
open lease keeps the older answers: stale_revision when its revision is
not the run's, otherwise wrong_attempt.

A step the engine settles takes the latest settledAt of what settled
it: an end-only step, of the steps it needs and reads; a workflow step,
of its workflow's winning end, or else of its own steps. When a
workflow has reached more than one end, the one with the lowest
settledAt wins, and the step declared first on a tie. The winner no
longer depends on declaration order or on when next runs between
reports.
An end reached in a workflow now stops everything under it that has not
settled, its sub-workflows included: a leased step becomes cancelled,
keeps its attempt, and the attempt goes to run.cancelledAttempts; a
running workflow step becomes cancelled; a pending step is skipped. A
cancelled step counts as stopped and settled, and expressions read its
outcome as cancelled.

A report of a cancelled attempt is answered lease_cancelled, before the
driver check. Since next derives cancelled after settling, the answer
that applies the end, and every later one, done included, lists the
attempts the driver must stop.
A path in an `if`, a template or a workflow output can now read an
array element, as in `steps.review.outputs.files[0].path`,
`inputs.labels[1]` or `{{ with.refs[0] }}`, and repeat it, as in
`matrix[0][1]`. An index out of range, or into a value that is not an
array, reads null.

Pull walks an index through the items of a declared array and rejects
an index on a value declared as anything else. It also rejects an index
that is not a whole number from 0, an index right after steps.<id>, and
one after outcome or end, each with what to write instead. Messages
show the path as written.
An expression can compare numbers with <, <=, > and >=, and call
length(list), any(list, predicate), all(list, predicate) and
filter(list, predicate). A predicate reads each element of the list as
`.`, a field of it as `.name`, and an index into it as `.name[0]`; in a
nested predicate `.` is the innermost element. A comparison of anything
but two numbers is false, and a function whose list is not an array
gives null. Comparisons bind tighter than == and !=, and ! still
applies to a whole equality, as before.

Pull rejects what the types rule out: a comparison operand that is not
a number, a list argument that is not an array, and a field a predicate
reads that the items of its list do not declare. Parsing the if catches
literals and function results; the read checks catch paths, against the
types their own workflow declares. Paths inside list arguments and
predicates count as reads, so a step waits for them, and an end compared
inside a predicate is checked like any other. A template still reads
exactly one path.
A step with `loop: { back_to, max_iterations, while }` runs its range
again after it succeeds: back_to, the step, and every step between them.
Each run archives the range's current pass in `passes` and starts the
next pass id; a workflow step's children move to the run's archive, and
a new child continues the pass ids of that path. A step outside the
range that needs or reads a step inside it waits until the loop ends.
When `while` still holds after max_iterations runs, or a loop without
`while` would start one run too many, the owner fails `loop_exhausted`
with the outputs of its latest pass.

`steps.<id>.passes` reads `{ pass, outcome, outputs, end }` for every
settled pass, and pull types it from the step's output. A lease names
its `pass`. Pull rejects, with the code `bad_loop`, a malformed loop, a
back_to the owner does not need, a loop on a step with `end`, two loops
sharing a step, and steps that would wait for each other.

Four expression and template assertions on the "must read" message now
list `steps.<id>.passes`, because it is a readable step field.
A step may read a step after it in its own loop's range: one that needs
it, directly or not. The read sees that step's latest archived pass, so
it reads nothing on the first pass and never waits for the step. A step
inside an inline workflow reads through the step that runs it. Pull and
the engine classify each read through the same rule, so a read of any
other later step is still rejected.
A workflow file or an inline workflow may declare `loop: { while,
max_iterations }`. When a round reaches no end and every step has
settled, `while` decides: false ends the workflow with the round's
outcome and outputs, and otherwise every step runs again in the next
round, with the next pass id. After max_iterations rounds the workflow
fails `loop_exhausted` with the outputs of its last round; at the top
level the run keeps that failure and `done` shows it. Any end reached
in a round stops the loop; a failed step alone does not. The round is
kept on the step that runs the workflow, or on the run at the top.

`break: <expr>` on a step is evaluated when the step succeeds. When it
holds, the workflow, or its loop, ends as success, and what still runs
under it is cancelled like after an end.

Pull rejects, with the code `bad_loop`, a malformed workflow loop or
break, a `while` that reads anything but the workflow's own steps and
inputs, a step loop in a looping workflow, and a break inside a step
loop's range.

The inline workflow key message now lists `loop`, which an inline
workflow accepts, so its cache test assertion changes with it.
The new Advanced workflows page, in English and Chinese, covers parallel
leases and the next payload (running, canTakeLease, cancelled), how an
end cancels open steps and answers lease_cancelled, which end wins when
reports reach two, and the full expression reference: indexes,
comparisons, length/any/all/filter, precedence, and the pull type checks.

The base page changes only where it was no longer true: one step at a
time now applies to the local driver, and the operator list gains the
comparisons, indexes, and functions. Both pages link to each other, and
the usage guide lists the new page.

A new test reads every yaml block in docs/guide/**/workflow*.md. Blocks
with top-level steps are synced together as one team repo through
syncWorkflowDefinitions and must have no diagnostics. Other blocks must
parse as YAML. Each page needs at least one complete workflow.
…rallel

The local driver now runs up to workflows.max_parallel commands at once
(local config, default 1, no flag) and sends them to next as running. A
leased prompt is held until no command runs or is still exiting, because
its read-only check compares the whole worktree before and after it.

When next lists an attempt as cancelled, the driver drops it if it was
held and never started, and stops it if it runs: SIGTERM, then SIGKILL
after 5 s, and it waits for the process to close. A stopping command is
not sent as running again. A lease_cancelled answer to a report is
discarded, and the run goes on. Every return path stops what still runs
and waits for it.

With max_parallel 1, commands spawn as before: in teamai's process
group, so the terminal's Ctrl-C reaches them and /dev/tty works. Above
1, each command runs detached in a process group of its own, and
SIGINT, SIGQUIT, SIGTERM or SIGHUP makes the driver stop every group,
wait for them, and exit with that signal. With nothing running, as
during a prompt, a signal ends teamai at once, as before. Windows uses
taskkill /T /F in place of the group kill.
…el commands

Add the step loop, pass, previous-pass read, and pull rejection rules, and
the workflow loop and break rules, with examples taken from real runs. Add
workflows.max_parallel to the parallel section, tell a driver how to treat a
reconcile-only wait, and show pass in each lease. The base page no longer
says that --driver local always runs one step at a time.
stepExpressions, next to templatedTexts, lists the expressions a step
evaluates: if, loop.while and break. readsOf and parseReads in the engine
and the read checks in parse.ts iterate over it instead of three
hand-written lists. parseReads and parse.ts key a table by each
expression, so a new one does not compile until both say how to read it.

Behavior is unchanged.
A step with for_each runs its action once per item of a list. for_each
takes an expression, or a map of items and max_parallel. When settling
reaches the step, the engine evaluates the list and adds an instance per
item right after it, build[0], build[1], and so on. A step inside the
workflow an item runs has an id such as build[2].verify. Each instance
is an ordinary lease, so a driver that lists its running attempts gets
the next item while the others run, up to max_parallel.

Each instance reads its item as item, in prompt, env and with, and in
every step of the inline workflow it runs. A with value that is only
{{ item }} keeps the item's JSON type. The step's own if and for_each
run once, before any item, and cannot read its items.

The step settles once every item has, as steps.<id>.results, a list of
item, outcome and outputs in item order. It fails if any item failed,
and a failed item cancels none of the others. An empty list succeeds at
once with results []. An item with a string id can be named in another
item's after: that item starts once they succeed, and is skipped when
one of them failed, was skipped or was cancelled. An unknown or
duplicate id fails the step with invalid_items, and a cycle with
item_cycle, which names it, before any item runs.

Pull rejects for_each with loop, end or break, a malformed for_each,
item where no item is bound, item in a command, a read of the outputs
of a for_each step, and a read of the results of a step without one.
Each for_each problem drops the kept copy (bad_loop), so an older copy
cannot run the step once.

The rules that decide the items live in fan-out.ts, the settling in
engine.ts. The advanced guide gains a for_each section and the item and
results values, in English and Chinese, and a conformance scenario
covers parallel items.
…al CLI

Runs a three-pass step loop, then two for_each items that wait for each
other, with --driver local. With workflows.max_parallel 2 the run
succeeds; with 1 the first item times out and the second still runs.
@SaulMoro
SaulMoro force-pushed the feat/1079-loops-for-each branch from 709cf64 to 905d964 Compare October 15, 2026 13:38
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

[feat] Add bounded loops and parallel for_each

1 participant