Fix BatchFlow sequential execution and improve clarity

- Fix bug where sequential batch flows shared state between iterations
- Rename BatchParams to FlowInputs for better clarity
- Add documentation explaining FlowInputs purpose
- Enhance README with detailed batch flow example
- Ensure each flow iteration gets its own isolated SharedStore

Breaking change: BatchParams renamed to FlowInputs

🤖 Generated with [opencode](https://opencode.ai)

Co-Authored-By: opencode <[email protected]>

Author: ezynda3

README.md

@@ -208,29 +208,55 @@ func (n *RateLimitedNode) Exec(ctx context.Context, prepResult any) (any, error)
 Process multiple items concurrently:
 
 ```go
-// Create a flow factory
+// Simple batch node for processing items
+processFunc := func(ctx context.Context, item any) (any, error) {
+    // Process each item
+    return fmt.Sprintf("processed: %v", item), nil
+}
+
+batchNode := flyt.NewBatchNode(processFunc, true) // true for concurrent
+shared.Set("items", []string{"item1", "item2", "item3"})
+```
+
+### Batch Flows
+
+Run the same flow multiple times with different parameters:
+
+```go
+// Create a flow factory - returns a new flow instance for each iteration
 flowFactory := func() *flyt.Flow {
-    node := flyt.NewNode(
+    validateNode := flyt.NewNode(
+        flyt.WithPrepFunc(func(ctx context.Context, shared *flyt.SharedStore) (any, error) {
+            // Each flow has its own SharedStore with merged FlowInputs
+            userID, _ := shared.Get("user_id")
+            email, _ := shared.Get("email")
+            return map[string]any{"user_id": userID, "email": email}, nil
+        }),
         flyt.WithExecFunc(func(ctx context.Context, prepResult any) (any, error) {
-            // Process individual item
-            return processItem(prepResult)
+            data := prepResult.(map[string]any)
+            // Process user data
+            return processUser(data), nil
         }),
     )
-    return flyt.NewFlow(node)
+    return flyt.NewFlow(validateNode)
 }
 
-// Define batch parameters
-batchFunc := func(ctx context.Context, shared *flyt.SharedStore) ([]flyt.BatchParams, error) {
-    return []flyt.BatchParams{
-        {"id": 1, "data": "item1"},
-        {"id": 2, "data": "item2"},
-        {"id": 3, "data": "item3"},
+// Define input parameters for each flow iteration
+// Each FlowInputs map is merged into that flow's isolated SharedStore
+batchFunc := func(ctx context.Context, shared *flyt.SharedStore) ([]flyt.FlowInputs, error) {
+    // Could fetch from database, API, etc.
+    return []flyt.FlowInputs{
+        {"user_id": 1, "email": "[email protected]"},
+        {"user_id": 2, "email": "[email protected]"},
+        {"user_id": 3, "email": "[email protected]"},
     }, nil
 }
 
 // Create and run batch flow
 batchFlow := flyt.NewBatchFlow(flowFactory, batchFunc, true) // true for concurrent
 err := batchFlow.Run(ctx, shared)
+
+// Each flow runs in isolation with its own SharedStore containing the FlowInputs
 ```
 
 ### Nested Flows

batch.go

@@ -195,11 +195,12 @@ func (n *batchNode) Post(ctx context.Context, shared *SharedStore, prepResult, e
 	return DefaultAction, nil
 }
 
-// BatchParams holds parameters for a batch iteration
-type BatchParams map[string]any
+// FlowInputs holds input parameters for a flow iteration in batch processing.
+// These parameters are merged into each flow's isolated SharedStore.
+type FlowInputs map[string]any
 
-// BatchFlowFunc returns parameters for each batch iteration
-type BatchFlowFunc func(ctx context.Context, shared *SharedStore) ([]BatchParams, error)
+// BatchFlowFunc returns input parameters for each flow iteration in batch processing
+type BatchFlowFunc func(ctx context.Context, shared *SharedStore) ([]FlowInputs, error)
 
 // NewBatchFlow creates a flow that runs multiple times with different parameters.
 // The flowFactory must create new flow instances for concurrent execution to avoid
@@ -271,7 +272,7 @@ func (n *batchFlowNode) Exec(ctx context.Context, prepResult any) (any, error) {
 		return nil, fmt.Errorf("batchFlowNode: exec failed: invalid prepResult type %T, expected map[string]any", prepResult)
 	}
 
-	batchParams, ok := data["batchParams"].([]BatchParams)
+	batchParams, ok := data["batchParams"].([]FlowInputs)
 	if !ok {
 		return nil, fmt.Errorf("batchFlowNode: exec failed: invalid batchParams type in prepResult")
 	}
@@ -352,10 +353,6 @@ func (n *batchFlowNode) Exec(ctx context.Context, prepResult any) (any, error) {
 		}
 	} else {
 		// Run flows sequentially
-		// Create a shared store for sequential execution
-		seqShared := NewSharedStore()
-		seqShared.Merge(sharedData)
-
 		for i, params := range batchParams {
 			// Check context
 			if err := ctx.Err(); err != nil {
@@ -365,10 +362,12 @@ func (n *batchFlowNode) Exec(ctx context.Context, prepResult any) (any, error) {
 			// Create a new flow instance for each iteration
 			flow := n.flowFactory()
 
-			// Merge params into shared for the flow execution
-			seqShared.Merge(params)
+			// Create isolated shared store for each iteration
+			iterShared := NewSharedStore()
+			iterShared.Merge(sharedData)
+			iterShared.Merge(params)
 
-			if err := flow.Run(ctx, seqShared); err != nil {
+			if err := flow.Run(ctx, iterShared); err != nil {
 				return nil, fmt.Errorf("batchFlowNode: flow %d failed: %w", i, err)
 			}
 		}

batch_test.go

@@ -198,10 +198,10 @@ func TestBatchFlowConcurrency(t *testing.T) {
 	}
 
 	// Batch function returns parameters for each iteration
-	batchFunc := func(ctx context.Context, shared *SharedStore) ([]BatchParams, error) {
-		params := make([]BatchParams, 5)
+	batchFunc := func(ctx context.Context, shared *SharedStore) ([]FlowInputs, error) {
+		params := make([]FlowInputs, 5)
 		for i := range params {
-			params[i] = BatchParams{"index": i}
+			params[i] = FlowInputs{"index": i}
 		}
 		return params, nil
 	}
@@ -247,8 +247,8 @@ func TestBatchFlowCustomCountKey(t *testing.T) {
 	}
 
 	// Batch function returns parameters
-	batchFunc := func(ctx context.Context, shared *SharedStore) ([]BatchParams, error) {
-		return []BatchParams{
+	batchFunc := func(ctx context.Context, shared *SharedStore) ([]FlowInputs, error) {
+		return []FlowInputs{
 			{"id": 1},
 			{"id": 2},
 			{"id": 3},