v2: sub-workflow node + parallel waves (batch fan-out phase 1)

[xBalbinus]

Summary

Phase 1 of the batch fan-out design (spec included as the first commit): the dag/v1 interpreter cannot express "a small DAG per item" — a foreach body is one inline node, and even sibling tool nodes execute sequentially. This PR adds the two pieces that make the per-customer batch shape expressible.

Sub-workflow node. A new workflow node type ({ workflowId, input }) starts a child run of the referenced definition and parks until it settles, valid at top level and as a foreach body. Each batch item becomes a real run: per-node status, reruns, and the runs list double as the batch tracker. The child records parentRunId/parentNodeId/parentIteration in its RunParams (jsonb — no migration). Dispatch is replay-safe: the child run id derives from the checkpoint PK (sha256(runId:nodeId:iteration)), the intent checkpoint persists it before creation, and createRun is insert-if-absent. Guards: exact-owner match on the reference (team sharing waits for RBAC Phase B), nesting depth 1 enforced at dispatch and against the resolved definition, loud failure when the host wires no resolveWorkflow. A completed child's output is its stop node's declared output, else a map of leaf-node results. Cancel propagates to parked children as the same durable cancel signal terminate() writes; settle-time wake of the parent lives in the interpreter, with the host sweep as the lost-wake backstop (new { kind: 'run' } wait condition).

Parallel waves. The interpreter executes a wave's runnable nodes concurrently (bounded, 5) instead of a sequential for...of await. Same-wave nodes never feed each other — the template context and runnable set are computed before the wave, and each executor writes only its own checkpoint row — so this changes wall-clock, not semantics. Outcomes still aggregate in definition order.

Engine port change is minimal: one optional resolveWorkflow method, implemented api-side over workflow_definitions. Starting the child needs no engine call — a run starts by existing in the store with a requested wake, which is what LocalRunHost.start does.

Testing

• New suite packages/workflow/src/nodes/workflow-call.test.ts (13): drive-level lifecycle (start/park/settle/output/failure), replay convergence on the derived child id, depth + unresolved + unwired guards, cancel propagation, foreach fan-out over two child runs, a concurrency-overlap proof for parallel waves, and validator coverage.
• Full @valet/workflow suite 272 passed (no regressions from the wave change); api workflow suites 150 passed; PG store conformance 59 passed; root pnpm typecheck clean.

e2e scorecard

18 passed, 3 red rows, all explained: engine-unit is the documented Node-version trap (green under Node 22 — 538 passed); web-build was this change (the editor's exhaustive node-type maps require the new type) and is fixed in the second commit (web build green, 573 web tests pass); cli is a missing packages/api/node_modules/.bin/tsx bin link present in the untouched checkout too — environmental, unrelated. Docker/API-key suites skipped (daemon and key absent in the shell).

[valet-valet-turnkey-dev]

The PR is well-structured and the implementation is largely correct, but there are two real correctness issues and one security-relevant gap worth flagging.

1. executeWave has a race on the next counter — the pool pattern is not safe when workers are async. Two workers can read the same idx before either increments it because next++ is not atomic across await boundaries inside the same microtask loop. In practice Node.js is single-threaded so the read-and-increment of a plain JS let is synchronous and cannot interleave, but the while(true) { const idx = next++; ... await ... } shape looks racy and is worth a comment; more importantly, if a worker throws after incrementing next but before writing outcomes[idx], that slot stays undefined and the rejected check rethrows correctly — but the caller then gets a sparse array with at least one unset element at index idx. The throw rejected.reason path is reached, so the drive loop does abort, but any partially-written outcomes entries could leak if the caller ever inspected them. This is a latent correctness hole if executeWave's contract is ever relaxed.

2. settleFromChild skips putIntent before completeCheckpoint on the failure path — every other failure path in this file (the fail() helper, the depth guard, etc.) calls putIntent first to make the write idempotent. The failure branch inside settleFromChild (lines ~150-165 of workflow-call.ts) calls completeCheckpoint directly without a preceding putIntent. On re-entry after a crash between completeCheckpoint and the return, the executor will re-run the whole settleFromChild path and call completeCheckpoint again. Whether completeCheckpoint is idempotent on a second call with the same key is store-defined — if the in-memory store is idempotent it works, but this violates the intent→complete ordering the rest of the file enforces, and a stricter store implementation would reject it.

   • packages/workflow/src/nodes/workflow-call.ts around line 152 (the child.outcome !== 'completed' branch) — add await store.putIntent(...) before store.completeCheckpoint(...) to match the success path and the fail() helper.

3. computeChildOutput only looks at iteration === 0 checkpoints — byNode is built with if (cp.iteration === 0). If the child workflow contains a foreach node, its body checkpoints use higher iterations and the foreach's own completion checkpoint is at iteration 0 but cp.result reflects the aggregate. That's probably fine. However, the stop-node output branch looks for result.output inside the checkpoint result object; whether the stop executor stores output under that key depends on convention not visible in this diff. If the stop executor stores the output differently, computeChildOutput silently falls through to the leaf-node map. This is worth a test that actually asserts the stop-output path, not just the leaf-output path — the existing test (resumes with the child stop output) does cover this, but only for the in-memory store's exact shape.

4. wakeParentRun does an unconditional getRun on every settle — packages/workflow/src/interpreter.ts, wakeParentRun. This adds a store round-trip to every twoPhaseSettle and resumeTerminalize call, including for top-level runs that have no parent (the common case). The check is run?.params.parentRunId !== undefined, which is fast once the row is fetched, but getRun is called on a run that was just settled and is still in memory/cache. Low severity, but given this is on the hot path of every run settlement it is worth noting — passing parentRunId through from the already-loaded WorkflowRun would eliminate the extra fetch.

---

Created on behalf of Xiangan He xiangan@turnkey.io

[xBalbinus]

Checked all four against the code. No change made — reasoning below, happy to be argued out of any of it.

1. executeWave race on next — not a race. next++ is a single synchronous read-modify-write and the worker's first suspension point is the await on the line after the claim, so run-to-completion guarantees no index is taken twice. Verified with 500 items across 5 workers: no duplicate claims.

The sparse-array half can't reach a caller either. A slot is only left unwritten when its executor rejects, and throw rejected.reason runs before return outcomes. If an executor never settles, allSettled never resolves and the function never returns at all.

2. settleFromChild missing putIntent — the premise doesn't hold for this file. Both branches call completeCheckpoint bare, not just the failure one, and that matches the convention here: putIntent is used where no row exists yet (foreach.ts, submission-node.ts). settleFromChild is only reachable when a checkpoint carrying effects.childRunId already exists.

Idempotency is also contract-defined rather than store-defined — store.ts documents terminal rows as immutable, a same-attempt write as a no-op, and a stale attempt as a WorkflowFenceError, and the conformance suite pins it across both the in-memory and Postgres stores. Adding putIntent here would fence against the terminal row and abort the drive, so the suggested change would introduce the failure it's guarding against.

3. computeChildOutput / stop-node output key — the convention does hold. stop.ts persists { outcome, output, message } as the checkpoint result, which is exactly the key read here, and an existing test discriminates that branch from the leaf-map fallback. On the foreach case: a foreach's own checkpoint is always at iteration 0 and its result is the aggregate a consumer should see, and body checkpoints key off the body id, which validation forbids from colliding with a node id — so no lookup reaches them.

4. wakeParentRun extra getRun — fair, and real: one additional SELECT per run settlement, on a path that already issues beginTerminalize + settleRun + getRun. Threading run.params.parentRunId through from the already-loaded run is about six lines. Left out to keep this PR to the feature; happy to fold it in if you'd rather not carry it.