fix: address all PR #274 review comments - precision, deep copy, string sort, context cancellation

Co-authored-by: intel352 <77607+intel352@users.noreply.github.com>

Author: Copilot

docs/plans/2026-03-06-fan-out-fan-in.md

@@ -1,7 +1,5 @@
 # Fan-Out / Fan-In / Map-Reduce Implementation Plan
 
-> **For Claude:** REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task.
-
 **Goal:** Add parallel execution capabilities (`step.parallel`, concurrent `step.foreach`) and collection template functions (`sum`, `pluck`, `groupBy`, etc.) to the workflow engine.
 
 **Architecture:** Concurrency is opt-in at the step level — the pipeline executor stays sequential. `step.parallel` spawns goroutines for fixed branches; `step.foreach` gets an optional worker pool. Each goroutine operates on a deep-copied PipelineContext, eliminating shared mutable state. 10 collection template functions are added for inline aggregation.

lsp/document.go

@@ -74,17 +74,17 @@ const (
 
 // PositionContext describes what context the cursor is in within the document.
 type PositionContext struct {
-	Section          SectionKind
-	ModuleType       string // if inside a modules[] item config, the type value
-	StepType         string // if inside a pipeline step config, the step type value
-	FieldName        string // the field name at the cursor
-	InTemplate       bool   // cursor is inside {{ }}
-	DependsOn        bool   // cursor is in a dependsOn array value
-	PipelineName     string // name of the pipeline containing the cursor (if any)
-	CurrentStepName  string // name of the step containing the cursor (if any)
-	TemplatePath     *TemplateExprPath // parsed template expression at cursor, if InTemplate
-	Line             int
-	Character        int
+	Section         SectionKind
+	ModuleType      string            // if inside a modules[] item config, the type value
+	StepType        string            // if inside a pipeline step config, the step type value
+	FieldName       string            // the field name at the cursor
+	InTemplate      bool              // cursor is inside {{ }}
+	DependsOn       bool              // cursor is in a dependsOn array value
+	PipelineName    string            // name of the pipeline containing the cursor (if any)
+	CurrentStepName string            // name of the step containing the cursor (if any)
+	TemplatePath    *TemplateExprPath // parsed template expression at cursor, if InTemplate
+	Line            int
+	Character       int
 }
 
 // ContextAt analyses the document content at the given (zero-based) line and

mcp/tools.go

@@ -1236,9 +1236,9 @@ func (s *Server) handleInferPipelineContext(_ context.Context, req mcp.CallToolR
 	// Infer outputs for each preceding step.
 	stepReg := schema.GetStepSchemaRegistry()
 	type stepContext struct {
-		Name    string                   `json:"name"`
-		Type    string                   `json:"type"`
-		Outputs []schema.InferredOutput  `json:"outputs"`
+		Name    string                  `json:"name"`
+		Type    string                  `json:"type"`
+		Outputs []schema.InferredOutput `json:"outputs"`
 	}
 	stepsCtx := make([]stepContext, 0, len(precedingSteps))
 	for _, step := range precedingSteps {

module/auth_m2m_test.go

@@ -1270,7 +1270,7 @@ func TestM2M_Introspect_AllowOthers_BearerMissingClaim(t *testing.T) {
 	m := NewM2MAuthModule("m2m", "this-is-a-valid-secret-32-bytes!", time.Hour, "test-issuer")
 	m.RegisterClient(M2MClient{ClientID: "client-a", ClientSecret: "secret-a-long-enough!", Scopes: []string{"read"}}) //nolint:gosec // test credential
 	m.RegisterClient(M2MClient{ClientID: "client-b", ClientSecret: "secret-b-long-enough!", Scopes: []string{"read"}}) //nolint:gosec // test credential
-	m.SetIntrospectPolicy(true, "", "role", "admin") // role=admin required
+	m.SetIntrospectPolicy(true, "", "role", "admin")                                                                   // role=admin required
 
 	tokenA := issueTestToken(t, m, "client-a", "secret-a-long-enough!")
 	tokenB := issueTestToken(t, m, "client-b", "secret-b-long-enough!")

module/database_partitioned_test.go

@@ -197,11 +197,11 @@ func TestIsSupportedPartitionDriver(t *testing.T) {
 // testMultiPartitionManager extends testPartitionManager with MultiPartitionManager support.
 type testMultiPartitionManager struct {
 	testPartitionManager
-	configs                 []PartitionConfig
-	ensureForKeyCalledWith  []struct{ key, value string }
-	syncForKeyCalledWith    []string
-	ensureForKeyErr         error
-	syncForKeyErr           error
+	configs                []PartitionConfig
+	ensureForKeyCalledWith []struct{ key, value string }
+	syncForKeyCalledWith   []string
+	ensureForKeyErr        error
+	syncForKeyErr          error
 }
 
 func (m *testMultiPartitionManager) PartitionConfigs() []PartitionConfig { return m.configs }
@@ -403,10 +403,10 @@ func TestPartitionedDatabase_PartitionConfigs_DeepCopiesTables(t *testing.T) {
 func TestPartitionedDatabase_BackwardCompat_SinglePartition(t *testing.T) {
 	// Old-style config without Partitions field must behave exactly as before.
 	cfg := PartitionedDatabaseConfig{
-		Driver:       "pgx",
-		PartitionKey: "tenant_id",
-		Tables:       []string{"forms", "submissions"},
-		PartitionType: PartitionTypeList,
+		Driver:              "pgx",
+		PartitionKey:        "tenant_id",
+		Tables:              []string{"forms", "submissions"},
+		PartitionType:       PartitionTypeList,
 		PartitionNameFormat: "{table}_{tenant}",
 	}
 	pd := NewPartitionedDatabase("db", cfg)

module/pipeline_step_foreach.go

@@ -197,11 +197,25 @@ func (s *ForEachStep) executeConcurrent(ctx context.Context, pc *PipelineContext
 	var firstErr error
 	var errOnce sync.Once
 
+outer:
 	for i, item := range items {
-		wg.Add(1)
+		// Stop launching new goroutines if the context is already cancelled
+		// (e.g., from a previous fail_fast error or external cancellation).
+		if branchCtx.Err() != nil {
+			break outer
+		}
+
 		i, item := i, item
 
-		sem <- struct{}{} // acquire semaphore slot
+		// Acquire a semaphore slot, but respect context cancellation so we don't
+		// block indefinitely when fail_fast has already cancelled the context.
+		select {
+		case sem <- struct{}{}: // acquired slot; fall through to launch goroutine
+		case <-branchCtx.Done():
+			break outer
+		}
+
+		wg.Add(1)
 		go func() {
 			defer wg.Done()
 			defer func() { <-sem }() // release slot

module/pipeline_step_foreach_test.go

@@ -4,7 +4,9 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"sync/atomic"
 	"testing"
+	"time"
 
 	"github.com/CrisisTextLine/modular"
 )
@@ -19,6 +21,24 @@ func (s *foreachFailStep) Execute(_ context.Context, _ *PipelineContext) (*StepR
 	return nil, errors.New("step failed")
 }
 
+// foreachSlowStep is a PipelineStep that sleeps for a fixed duration then succeeds.
+// It is used in timing-based concurrency tests to verify that items actually run
+// in parallel rather than sequentially.
+type foreachSlowStep struct {
+	stepName string
+	delay    time.Duration
+}
+
+func (s *foreachSlowStep) Name() string { return s.stepName }
+func (s *foreachSlowStep) Execute(ctx context.Context, _ *PipelineContext) (*StepResult, error) {
+	select {
+	case <-time.After(s.delay):
+		return &StepResult{Output: map[string]any{"done": true}}, nil
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	}
+}
+
 // buildTestForEachStep creates a ForEachStep with a fresh StepRegistry for testing.
 // It registers a simple "step.set" factory so sub-steps can be built.
 func buildTestForEachStep(t *testing.T, name string, config map[string]any) (PipelineStep, error) {
@@ -545,45 +565,58 @@ func TestForEachStep_AppPassedToSubStep(t *testing.T) {
 }
 
 func TestForEachStep_ConcurrentExecution(t *testing.T) {
-	// 5 items each taking 100ms, concurrency=5 — should complete in ~100ms not 500ms
+	// 5 items each taking 50ms, concurrency=5 — should complete in ~50ms not 250ms.
+	// We use a custom slow step that sleeps to ensure actual concurrency is tested.
+	const itemDelay = 50 * time.Millisecond
+	const numItems = 5
+
 	registry := NewStepRegistry()
-	registry.Register("step.set", NewSetStepFactory())
+	registry.Register("step.slow", func(name string, cfg map[string]any, app modular.Application) (PipelineStep, error) {
+		return &foreachSlowStep{stepName: name, delay: itemDelay}, nil
+	})
 
 	factory := NewForEachStepFactory(func() *StepRegistry { return registry })
 	step, err := factory("par-foreach", map[string]any{
 		"collection":  "items",
 		"item_var":    "item",
-		"concurrency": 5,
+		"concurrency": numItems, // full concurrency — all items run in parallel
 		"step": map[string]any{
 			"name": "process",
-			"type": "step.set",
-			"values": map[string]any{
-				"processed": "true",
-			},
+			"type": "step.slow",
 		},
 	}, nil)
 	if err != nil {
 		t.Fatal(err)
 	}
 
-	items := make([]any, 5)
+	items := make([]any, numItems)
 	for i := range items {
 		items[i] = map[string]any{"id": i}
 	}
 	pc := NewPipelineContext(map[string]any{"items": items}, nil)
 
+	start := time.Now()
 	result, err := step.Execute(context.Background(), pc)
+	elapsed := time.Since(start)
 	if err != nil {
 		t.Fatal(err)
 	}
 
 	results := result.Output["results"].([]any)
-	if len(results) != 5 {
-		t.Fatalf("expected 5 results, got %d", len(results))
+	if len(results) != numItems {
+		t.Fatalf("expected %d results, got %d", numItems, len(results))
 	}
 	count := result.Output["count"]
-	if count != 5 {
-		t.Fatalf("expected count=5, got %v", count)
+	if count != numItems {
+		t.Fatalf("expected count=%d, got %v", numItems, count)
+	}
+
+	// With full concurrency the wall-clock time should be roughly one item's delay.
+	// Allow 3× headroom for slow CI environments; reject if it took as long as sequential.
+	maxExpected := itemDelay * 3
+	if elapsed > time.Duration(numItems)*itemDelay {
+		t.Fatalf("concurrent execution took %v; expected <%v (sequential would be %v)",
+			elapsed, maxExpected, time.Duration(numItems)*itemDelay)
 	}
 }
 
@@ -706,8 +739,8 @@ func TestForEachStep_ConcurrencyZeroIsSequential(t *testing.T) {
 		"item_var":    "item",
 		"concurrency": 0,
 		"step": map[string]any{
-			"name": "s",
-			"type": "step.set",
+			"name":   "s",
+			"type":   "step.set",
 			"values": map[string]any{"ok": "true"},
 		},
 	}, nil)
@@ -755,3 +788,70 @@ func TestForEachStep_ForeachMapNotSetWhenConflict(t *testing.T) {
 		t.Fatalf("execute error: %v", execErr)
 	}
 }
+
+func TestForEachStep_ConcurrentFailFastStopsEarly(t *testing.T) {
+	// With fail_fast and slow remaining items, context cancellation should prevent
+	// the producer from launching unnecessary goroutines after the first error.
+	const itemDelay = 100 * time.Millisecond
+	registry := NewStepRegistry()
+	// first item fails immediately; remaining items sleep
+	callCount := int32(0)
+	registry.Register("step.slow_or_fail", func(name string, cfg map[string]any, app modular.Application) (PipelineStep, error) {
+		return &foreachSlowOrFailStep{stepName: name, delay: itemDelay, callCount: &callCount}, nil
+	})
+
+	factory := NewForEachStepFactory(func() *StepRegistry { return registry })
+	step, err := factory("ff-early-stop", map[string]any{
+		"collection":     "items",
+		"item_var":       "item",
+		"concurrency":    1, // 1 worker so cancellation stops the queue
+		"error_strategy": "fail_fast",
+		"step": map[string]any{
+			"name": "work",
+			"type": "step.slow_or_fail",
+		},
+	}, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// 10 items; with concurrency=1 the producer should stop after the first failure.
+	items := make([]any, 10)
+	for i := range items {
+		items[i] = map[string]any{"id": i}
+	}
+	pc := NewPipelineContext(map[string]any{"items": items}, nil)
+
+	start := time.Now()
+	_, err = step.Execute(context.Background(), pc)
+	elapsed := time.Since(start)
+	if err == nil {
+		t.Fatal("expected error with fail_fast")
+	}
+	// If context cancellation works, we should not execute all 10 slow items.
+	// Sequential execution of all 10 would take ~1s; early stop should be much faster.
+	if elapsed >= time.Duration(len(items))*itemDelay {
+		t.Fatalf("fail_fast did not stop early: took %v (expected less than %v)", elapsed, time.Duration(len(items))*itemDelay)
+	}
+}
+
+// foreachSlowOrFailStep fails on the first call and sleeps on subsequent calls.
+type foreachSlowOrFailStep struct {
+	stepName  string
+	delay     time.Duration
+	callCount *int32
+}
+
+func (s *foreachSlowOrFailStep) Name() string { return s.stepName }
+func (s *foreachSlowOrFailStep) Execute(ctx context.Context, _ *PipelineContext) (*StepResult, error) {
+	n := atomic.AddInt32(s.callCount, 1)
+	if n == 1 {
+		return nil, fmt.Errorf("first item fails immediately")
+	}
+	select {
+	case <-time.After(s.delay):
+		return &StepResult{Output: map[string]any{"done": true}}, nil
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	}
+}

module/pipeline_step_parallel.go

@@ -171,29 +171,54 @@ func (s *ParallelStep) Execute(ctx context.Context, pc *PipelineContext) (*StepR
 	}, nil
 }
 
-// buildParallelChildContext creates a deep copy of the PipelineContext for a parallel branch.
-// Each branch gets its own isolated copy so goroutines don't share mutable state.
+// deepCopyValue recursively copies maps and slices so that goroutines operating
+// on different branches cannot mutate each other's data through shared references.
+// Primitive values (bool, string, numbers, nil) are returned as-is since they
+// are immutable in Go.
+func deepCopyValue(v any) any {
+	switch val := v.(type) {
+	case map[string]any:
+		cp := make(map[string]any, len(val))
+		for k, v2 := range val {
+			cp[k] = deepCopyValue(v2)
+		}
+		return cp
+	case []any:
+		cp := make([]any, len(val))
+		for i, v2 := range val {
+			cp[i] = deepCopyValue(v2)
+		}
+		return cp
+	default:
+		// Primitive values are safe to share.
+		return v
+	}
+}
+
+// buildParallelChildContext creates a true deep copy of the PipelineContext for a
+// parallel branch. Each branch gets its own isolated copy so goroutines cannot
+// race on nested maps or slices.
 func buildParallelChildContext(parent *PipelineContext) *PipelineContext {
 	childTrigger := make(map[string]any, len(parent.TriggerData))
 	for k, v := range parent.TriggerData {
-		childTrigger[k] = v
+		childTrigger[k] = deepCopyValue(v)
 	}
 
 	childMeta := make(map[string]any, len(parent.Metadata))
 	for k, v := range parent.Metadata {
-		childMeta[k] = v
+		childMeta[k] = deepCopyValue(v)
 	}
 
 	childCurrent := make(map[string]any, len(parent.Current))
 	for k, v := range parent.Current {
-		childCurrent[k] = v
+		childCurrent[k] = deepCopyValue(v)
 	}
 
 	childOutputs := make(map[string]map[string]any, len(parent.StepOutputs))
 	for k, v := range parent.StepOutputs {
 		out := make(map[string]any, len(v))
 		for k2, v2 := range v {
-			out[k2] = v2
+			out[k2] = deepCopyValue(v2)
 		}
 		childOutputs[k] = out
 	}

module/pipeline_step_parallel_test.go

@@ -325,3 +325,30 @@ func TestParallelStep_DefaultErrorStrategy(t *testing.T) {
 		t.Fatal("expected error with default fail_fast strategy")
 	}
 }
+
+func TestBuildParallelChildContext_DeepCopy(t *testing.T) {
+	// Verify that buildParallelChildContext performs a true deep copy so that
+	// mutations to nested maps/slices in a child context don't affect the parent.
+	nested := map[string]any{"inner": "original"}
+	slice := []any{"a", "b"}
+	parent := &PipelineContext{
+		TriggerData: map[string]any{"data": nested},
+		Current:     map[string]any{"list": slice},
+		Metadata:    map[string]any{},
+		StepOutputs: map[string]map[string]any{},
+	}
+
+	child := buildParallelChildContext(parent)
+
+	// Mutate the child's nested map — parent should be unaffected.
+	child.TriggerData["data"].(map[string]any)["inner"] = "mutated"
+	if parent.TriggerData["data"].(map[string]any)["inner"] != "original" {
+		t.Fatal("deep copy failed: mutation of child TriggerData nested map affected parent")
+	}
+
+	// Mutate the child's slice — parent should be unaffected.
+	child.Current["list"].([]any)[0] = "z"
+	if parent.Current["list"].([]any)[0] != "a" {
+		t.Fatal("deep copy failed: mutation of child Current slice affected parent")
+	}
+}