feat(drawer): parallelism inside for_each — concurrent iteration (#155)

Adds Parallelism int to kind=for_each. 0 or 1 keeps the existing
serial fail-fast semantics. N > 1 spawns a worker pool so N
iterations run concurrently:

  {
    "kind":        "for_each",
    "over":        "${scrape-list.output.urls}",
    "as":          "url",
    "parallelism": 10,
    "steps":       [ ... ]
  }

## Correctness story

- **Ordering:** results[i] maps to items[i] regardless of
  completion order. Downstream refs like
  ${step.output.results.0.item} stay deterministic.
- **Stop-on-failure:** the parallel path allocates a sub-context
  (workerCtx). First failure cancels it so in-flight iterations
  abort promptly instead of finishing work after the step has
  already been declared failed. The recorded failure is the
  lowest-indexed one so traces are reproducible.
- **Continue-on-failure:** wg.Wait() still runs every scheduled
  iteration to completion; failures are marked per-item; the
  overall step completes ok with partial failure markers.
- **Race-free:** each worker owns its slot in results[]/errs[]
  (preallocated by index); only first-failure bookkeeping is
  mutex-protected. `go test ./... -race` clean.
- **Resource bounds:** buffered semaphore of size Parallelism caps
  inflight goroutines + child processes. No implicit 1:1 mapping
  between iterations and processes running simultaneously.

## Implementation

- Serial path was inlined for its fast-return case (avoids
  channel/goroutine overhead for small N). Parallel path uses
  chan+WaitGroup (no x/sync dependency).
- Per-iter body extracted into runForEachIter so serial and
  parallel share one implementation of per-iteration semantics.

## set extended

New STEP.parallelism=N path addressable from the CLI.

## Tests

- TestForEach_ParallelismSpeedsUpRuns — wall-clock assertion:
  4×400ms iterations with parallelism=4 completes in <1.2s
  (serial baseline is ~1.6s). Catches regressions to serial.
- TestForEach_ParallelismStopOnFirstFailure — cancels in-flight
  workers on first failure; step reports FOREACH_ITEM_FAILED.
- TestForEach_ParallelismContinueCollectsAll — all 4 iterations
  run; 2 odd-indexed failures marked per-item; overall step ok.

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: bobakemamian

cmd/drawer.go

@@ -52,6 +52,7 @@ Usage:
   buttons drawer NAME set STEP.args.FIELD=value       literal or ${ref} into a step arg
   buttons drawer NAME set STEP.over=EXPR              for_each: the array to iterate
   buttons drawer NAME set STEP.as=NAME                for_each: loop variable name
+  buttons drawer NAME set STEP.parallelism=N          for_each: max concurrent iterations (0/1 = serial)
   buttons drawer NAME set STEP.from=EXPR              aggregate: input array
   buttons drawer NAME set STEP.pluck=EXPR             aggregate: per-item expression
   buttons drawer NAME set STEP.steps.N.args.F=value   reach a nested step's arg

docs/cli/buttons_drawer.md

@@ -24,6 +24,7 @@ Usage:
   buttons drawer NAME set STEP.args.FIELD=value       literal or ${ref} into a step arg
   buttons drawer NAME set STEP.over=EXPR              for_each: the array to iterate
   buttons drawer NAME set STEP.as=NAME                for_each: loop variable name
+  buttons drawer NAME set STEP.parallelism=N          for_each: max concurrent iterations (0/1 = serial)
   buttons drawer NAME set STEP.from=EXPR              aggregate: input array
   buttons drawer NAME set STEP.pluck=EXPR             aggregate: per-item expression
   buttons drawer NAME set STEP.steps.N.args.F=value   reach a nested step's arg

docs/schemas/drawer.schema.json

@@ -208,6 +208,10 @@
           "description": "CEL expression producing the array to iterate over (kind=for_each)",
           "type": "string"
         },
+        "parallelism": {
+          "description": "Max concurrent iterations (kind=for_each); 0 or 1 = serial",
+          "type": "integer"
+        },
         "pluck": {
           "description": "CEL expression evaluated per item; 'item' is bound to the current entry (kind=aggregate)",
           "type": "string"

internal/drawer/entity.go

@@ -125,6 +125,13 @@ type Step struct {
 	// "fail-fast overall, but tolerate per-item errors here."
 	OnItemFailure string `json:"on_item_failure,omitempty" jsonschema:"enum=stop,enum=continue,description=How to react to a per-item failure (kind=for_each)"`
 
+	// Parallelism bounds how many iterations run concurrently. 0 or
+	// 1 means serial (same as before). N > 1 spawns a worker pool
+	// of size N. Ordering of the results array still matches the
+	// input order regardless of completion order — downstream refs
+	// like ${step.output.results.0.item} stay deterministic.
+	Parallelism int `json:"parallelism,omitempty" jsonschema:"description=Max concurrent iterations (kind=for_each); 0 or 1 = serial"`
+
 	// Steps are the nested steps executed per item (for_each) OR
 	// the steps under a matching case's body (switch). Each child
 	// context is layered on the outer one.

internal/drawer/executor.go

@@ -8,6 +8,7 @@ import (
 	"fmt"
 	"os"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/autonoco/buttons/internal/battery"
@@ -410,13 +411,12 @@ func (e *Executor) runDrawerStep(ctx context.Context, step *Step, ctxMap Context
 // its own child context with the outer context plus the loop
 // variable (step.As) bound to the current item.
 //
-// v1 is serial (no parallelism). Per-item failures stop the loop by
-// default; OnItemFailure="continue" records the error and moves on.
-// Output is {results: [...]} — one entry per iteration, each a map
-// of {step_id: step.Output} for the nested steps. Downstream refs
-// walk this via ${<for_each_id>.output.results.0.step_id.field}
-// (indexed access) but will typically be consumed by an aggregator
-// step in later stages.
+// Concurrency: Parallelism <= 1 runs serially (deterministic order,
+// fail-fast). Parallelism > 1 spawns a worker pool. Results are
+// still written in input order regardless of completion order so
+// downstream refs like ${step.output.results.0.item} stay stable.
+// OnItemFailure="stop" in parallel mode cancels in-flight workers
+// via a sub-context the moment the first failure is observed.
 func (e *Executor) runForEachStep(ctx context.Context, step *Step, ctxMap Context) (StepRun, error) {
 	sr := StepRun{ID: step.ID}
 
@@ -462,65 +462,153 @@ func (e *Executor) runForEachStep(ctx context.Context, step *Step, ctxMap Contex
 	}
 
 	continueOnFail := step.OnItemFailure == "continue"
-	// []any (not []map[string]any) so downstream ref resolvers and
-	// aggregate steps can walk it without type gymnastics.
-	results := make([]any, 0, len(items))
+	parallelism := step.Parallelism
+	if parallelism < 1 {
+		parallelism = 1
+	}
+
+	// Preallocated, index-addressable results so writers can drop
+	// their output at position i without a mutex — each worker owns
+	// exactly one index.
+	results := make([]any, len(items))
+	errs := make([]error, len(items))
+
+	// Parallelism == 1 keeps the serial, fail-fast semantics that
+	// stage 3 shipped. Inlined to avoid the channel/goroutine cost
+	// for small loops where serial was already fine.
+	if parallelism == 1 {
+		for i, item := range items {
+			results[i], errs[i] = e.runForEachIter(ctx, step, asName, item, i, ctxMap)
+			if errs[i] != nil && !continueOnFail {
+				sr.Status = "failed"
+				sr.Output = map[string]any{
+					"results":   trimNilTail(results, i+1),
+					"completed": i + 1,
+					"total":     len(items),
+				}
+				sr.Error = &StepError{
+					Code:        "FOREACH_ITEM_FAILED",
+					Message:     fmt.Sprintf("iteration %d failed: %v", i, errs[i]),
+					Remediation: "fix the failing nested step or set on_item_failure: continue to tolerate per-item errors",
+				}
+				return sr, errs[i]
+			}
+		}
+		sr.Status = "ok"
+		sr.Output = map[string]any{"results": results, "total": len(items)}
+		return sr, nil
+	}
+
+	// Parallel mode. workerCtx gets cancelled on first failure
+	// when OnItemFailure != "continue" so in-flight iterations
+	// abort promptly instead of finishing their work after we've
+	// already decided to fail the step.
+	workerCtx, cancel := context.WithCancel(ctx)
+	defer cancel()
+
+	// Bounded semaphore + sync primitives. A plain buffered chan
+	// doubles as the semaphore (acquire by send, release by recv)
+	// without pulling in x/sync.
+	sem := make(chan struct{}, parallelism)
+	var wg sync.WaitGroup
+	var failMu sync.Mutex
+	var firstFailIdx int = -1
+	var firstFailErr error
 
 	for i, item := range items {
-		// Per-iteration context: start from the outer ctxMap and
-		// overlay the loop variable so nested ${<as>.field} refs
-		// resolve. Don't mutate the outer map — each iter gets its
-		// own.
-		iterCtx := Context{}
-		for k, v := range ctxMap {
-			iterCtx[k] = v
+		// Respect early-cancel decisions made by previous workers.
+		if workerCtx.Err() != nil {
+			break
 		}
-		iterCtx[asName] = item
-
-		iterOut := map[string]any{}
-		var iterErr error
-		for _, nested := range step.Steps {
-			nestedStep := nested
-			stepRes, serr := e.runStep(ctx, nil, &nestedStep, iterCtx)
-			iterOut[nestedStep.ID] = map[string]any{
-				"status": stepRes.Status,
-				"output": stepRes.Output,
-				"error":  stepRes.Error,
+		wg.Add(1)
+		sem <- struct{}{}
+		go func(idx int, it any) {
+			defer wg.Done()
+			defer func() { <-sem }()
+			out, err := e.runForEachIter(workerCtx, step, asName, it, idx, ctxMap)
+			results[idx] = out
+			errs[idx] = err
+			if err != nil && !continueOnFail {
+				failMu.Lock()
+				if firstFailIdx == -1 || idx < firstFailIdx {
+					firstFailIdx = idx
+					firstFailErr = err
+				}
+				failMu.Unlock()
+				cancel()
 			}
-			// Expose the just-completed nested step's output so
-			// later nested steps in the same iteration can chain.
-			iterCtx[nestedStep.ID] = map[string]any{"output": stepRes.Output}
-			if serr != nil {
-				iterErr = serr
-				break
+		}(i, item)
+	}
+	wg.Wait()
+
+	if firstFailIdx != -1 {
+		completed := 0
+		for i := range results {
+			if results[i] != nil {
+				completed++
 			}
 		}
+		sr.Status = "failed"
+		sr.Output = map[string]any{
+			"results":   results,
+			"completed": completed,
+			"total":     len(items),
+		}
+		sr.Error = &StepError{
+			Code:        "FOREACH_ITEM_FAILED",
+			Message:     fmt.Sprintf("iteration %d failed: %v", firstFailIdx, firstFailErr),
+			Remediation: "fix the failing nested step or set on_item_failure: continue to tolerate per-item errors",
+		}
+		return sr, firstFailErr
+	}
 
-		results = append(results, map[string]any{
-			"index":  i,
-			"item":   item,
-			"steps":  iterOut,
-			"failed": iterErr != nil,
-		})
+	sr.Status = "ok"
+	sr.Output = map[string]any{"results": results, "total": len(items)}
+	return sr, nil
+}
 
-		if iterErr != nil && !continueOnFail {
-			sr.Status = "failed"
-			sr.Output = map[string]any{"results": results, "completed": i + 1, "total": len(items)}
-			sr.Error = &StepError{
-				Code:        "FOREACH_ITEM_FAILED",
-				Message:     fmt.Sprintf("iteration %d failed: %v", i, iterErr),
-				Remediation: "fix the failing nested step or set on_item_failure: continue to tolerate per-item errors",
-			}
-			return sr, iterErr
+// runForEachIter executes the nested Steps for one iteration against
+// a per-iteration context. Extracted so both the serial and parallel
+// paths share exactly one implementation of the per-iter semantics.
+func (e *Executor) runForEachIter(ctx context.Context, step *Step, asName string, item any, idx int, outerCtx Context) (map[string]any, error) {
+	iterCtx := Context{}
+	for k, v := range outerCtx {
+		iterCtx[k] = v
+	}
+	iterCtx[asName] = item
+
+	iterOut := map[string]any{}
+	var iterErr error
+	for _, nested := range step.Steps {
+		nestedStep := nested
+		stepRes, serr := e.runStep(ctx, nil, &nestedStep, iterCtx)
+		iterOut[nestedStep.ID] = map[string]any{
+			"status": stepRes.Status,
+			"output": stepRes.Output,
+			"error":  stepRes.Error,
+		}
+		iterCtx[nestedStep.ID] = map[string]any{"output": stepRes.Output}
+		if serr != nil {
+			iterErr = serr
+			break
 		}
 	}
+	return map[string]any{
+		"index":  idx,
+		"item":   item,
+		"steps":  iterOut,
+		"failed": iterErr != nil,
+	}, iterErr
+}
 
-	sr.Status = "ok"
-	sr.Output = map[string]any{
-		"results": results,
-		"total":   len(items),
+// trimNilTail is used by the serial path to return a results slice
+// that only includes actually-completed iterations, keeping the
+// shape compatible with the pre-parallelism stage 3 behavior.
+func trimNilTail(xs []any, upTo int) []any {
+	if upTo > len(xs) {
+		upTo = len(xs)
 	}
-	return sr, nil
+	return xs[:upTo]
 }
 
 // runSwitchStep handles kind=switch — an if/elif/else chain. Walks

internal/drawer/schema_embedded.json

@@ -208,6 +208,10 @@
           "description": "CEL expression producing the array to iterate over (kind=for_each)",
           "type": "string"
         },
+        "parallelism": {
+          "description": "Max concurrent iterations (kind=for_each); 0 or 1 = serial",
+          "type": "integer"
+        },
         "pluck": {
           "description": "CEL expression evaluated per item; 'item' is bound to the current entry (kind=aggregate)",
           "type": "string"

internal/drawer/service.go

@@ -351,10 +351,19 @@ func (s *Service) SetField(drawerName, stepID, field, value string) (*Drawer, er
 		d.Steps[idx].Duration = value
 	case "until":
 		d.Steps[idx].Until = value
+	case "parallelism":
+		var n int
+		if _, err := fmt.Sscanf(value, "%d", &n); err != nil {
+			return nil, &ServiceError{Code: "VALIDATION_ERROR", Message: fmt.Sprintf("parallelism must be an integer, got %q", value)}
+		}
+		if n < 0 {
+			return nil, &ServiceError{Code: "VALIDATION_ERROR", Message: "parallelism must be >= 0"}
+		}
+		d.Steps[idx].Parallelism = n
 	default:
 		return nil, &ServiceError{
 			Code:    "STEP_FIELD_UNKNOWN",
-			Message: fmt.Sprintf("unknown step field %q — allowed: over, as, on_item_failure, from, pluck, button, drawer, duration, until", field),
+			Message: fmt.Sprintf("unknown step field %q — allowed: over, as, on_item_failure, parallelism, from, pluck, button, drawer, duration, until", field),
 		}
 	}
 	d.UpdatedAt = time.Now().UTC()