fix: support using static values for all input (#218)

Author: shentongmartin

compose/field_mapping.go

@@ -686,19 +686,14 @@ func validateFieldMapping(predecessorType reflect.Type, successorType reflect.Ty
 		return nil, fmt.Errorf("static check fail: predecessor output type should be struct or map, actual: %v", predecessorType)
 	}
 
-	var (
-		predecessorFieldType, successorFieldType                         reflect.Type
-		err                                                              error
-		predecessorIntermediateInterface, successorIntermediateInterface bool
-	)
-
-	for _, mapping := range mappings {
-		predecessorFieldType, predecessorIntermediateInterface, err = checkAndExtractFieldType(splitFieldPath(mapping.from), predecessorType)
+	for i := range mappings {
+		mapping := mappings[i]
+		predecessorFieldType, predecessorIntermediateInterface, err := checkAndExtractFieldType(splitFieldPath(mapping.from), predecessorType)
 		if err != nil {
 			return nil, fmt.Errorf("static check failed for mapping %s: %w", mapping, err)
 		}
 
-		successorFieldType, successorIntermediateInterface, err = checkAndExtractFieldType(splitFieldPath(mapping.to), successorType)
+		successorFieldType, successorIntermediateInterface, err := checkAndExtractFieldType(splitFieldPath(mapping.to), successorType)
 		if err != nil {
 			return nil, fmt.Errorf("static check failed for mapping %s: %w", mapping, err)
 		}

compose/workflow.go

@@ -451,10 +451,7 @@ func (wf *Workflow[I, O]) compile(ctx context.Context, options *graphCompileOpti
 					return mergeValues(values)
 				},
 				transform: func(in streamReader) streamReader {
-					sr, sw := schema.Pipe[map[string]any](1)
-					sw.Send(value, nil)
-					sw.Close()
-
+					sr := schema.StreamReaderFromArray([]map[string]any{value})
 					newS, err := mergeValues([]any{in, packStreamReader(sr)})
 					if err != nil {
 						errSR, errSW := schema.Pipe[map[string]any](1)
@@ -467,11 +464,11 @@ func (wf *Workflow[I, O]) compile(ctx context.Context, options *graphCompileOpti
 				},
 			}
 
-			if _, ok := wf.g.handlerPreNode[n.key]; !ok {
-				wf.g.handlerPreNode[n.key] = []handlerPair{pair}
-			} else {
-				wf.g.handlerPreNode[n.key] = append([]handlerPair{pair}, wf.g.handlerPreNode[n.key]...)
+			for i := range paths {
+				wf.g.fieldMappingRecords[n.key] = append(wf.g.fieldMappingRecords[n.key], ToFieldPath(paths[i]))
 			}
+
+			wf.g.handlerPreNode[n.key] = []handlerPair{pair}
 		}
 	}
 

compose/workflow_test.go

@@ -899,6 +899,68 @@ func TestStaticValue(t *testing.T) {
 		_, err := wf.Compile(context.Background())
 		assert.ErrorContains(t, err, "two terminal field paths conflict for node 0: [prefilled], [prefilled]")
 	})
+
+	t.Run("all inputs are static values", func(t *testing.T) {
+		wf := NewWorkflow[string, map[string]any]()
+		wf.AddLambdaNode("0", InvokableLambda(func(ctx context.Context, in map[string]any) (output map[string]any, err error) {
+			return in, nil
+		})).
+			AddDependency(START).
+			SetStaticValue(FieldPath{"a", "b"}, "a_b").
+			SetStaticValue(FieldPath{"c", "d"}, "c_d").
+			SetStaticValue(FieldPath{"a", "d"}, "a_d")
+		wf.End().AddInput("0")
+		r, err := wf.Compile(context.Background())
+		assert.NoError(t, err)
+		out, err := r.Invoke(context.Background(), "hello")
+		assert.NoError(t, err)
+		assert.Equal(t, map[string]any{
+			"a": map[string]any{
+				"b": "a_b",
+				"d": "a_d",
+			},
+			"c": map[string]any{
+				"d": "c_d",
+			},
+		}, out)
+
+		type a struct {
+			B string
+			D string
+		}
+
+		type s struct {
+			A a
+			C map[string]any
+		}
+
+		wf1 := NewWorkflow[string, *s]()
+		wf1.AddLambdaNode("0", InvokableLambda(func(ctx context.Context, in map[string]any) (output map[string]any, err error) {
+			return in, nil
+		})).
+			AddDependency(START).
+			SetStaticValue(FieldPath{"A", "B"}, "a_b").
+			SetStaticValue(FieldPath{"C", "D"}, "c_d").
+			SetStaticValue(FieldPath{"A", "D"}, "a_d")
+		wf1.End().AddInput("0", MapFieldPaths(FieldPath{"A", "B"}, FieldPath{"A", "B"}),
+			MapFieldPaths(FieldPath{"A", "D"}, FieldPath{"A", "D"}),
+			MapFields("C", "C"))
+		r1, err := wf1.Compile(context.Background())
+		assert.NoError(t, err)
+		out1, err := r1.Stream(context.Background(), "hello")
+		assert.NoError(t, err)
+		outChunk, err := out1.Recv()
+		out1.Close()
+		assert.Equal(t, &s{
+			A: a{
+				B: "a_b",
+				D: "a_d",
+			},
+			C: map[string]any{
+				"D": "c_d",
+			},
+		}, outChunk)
+	})
 }
 
 func TestBranch(t *testing.T) {