xyflow change to flowgraam

internal/ai/runtime/workflow/executor.go

@@ -112,7 +112,7 @@ type conditionEvaluation struct {
 
 func (e *Executor) Execute(ctx context.Context, input Input) (*Result, error) {
 	state := newRunState(input)
-	currentID := strings.TrimSpace(input.Definition.EntryNodeID)
+	currentID := state.startNodeID()
 	if currentID == "" {
 		return nil, fmt.Errorf("workflow entry node is required")
 	}
@@ -247,11 +247,20 @@ func newRunState(input Input) *runState {
 		}
 	}
 	for _, edge := range input.Definition.Edges {
-		state.outgoing[edge.Source] = append(state.outgoing[edge.Source], edge)
+		state.outgoing[strings.TrimSpace(edge.SourceNodeID)] = append(state.outgoing[strings.TrimSpace(edge.SourceNodeID)], edge)
 	}
 	return state
 }
 
+func (s *runState) startNodeID() string {
+	for _, node := range s.nodesByID {
+		if strings.TrimSpace(node.Type) == workflowregistry.NodeTypeStart {
+			return strings.TrimSpace(node.ID)
+		}
+	}
+	return ""
+}
+
 func (e *Executor) executeNode(ctx context.Context, state *runState, node dsl.Node) error {
 	switch node.Type {
 	case workflowregistry.NodeTypeStart:
@@ -587,19 +596,19 @@ func (e *Executor) executePrepareTicketDraft(ctx context.Context, state *runStat
 	input := graphs.PrepareTicketDraftInput{
 		Issue: issue,
 	}
-	if title := strings.TrimSpace(readStringConfig(node.Config, "title")); title != "" {
+	if title := strings.TrimSpace(readStringConfig(node.Data.Config, "title")); title != "" {
 		input.Title = title
 	}
-	if description := strings.TrimSpace(readStringConfig(node.Config, "description")); description != "" {
+	if description := strings.TrimSpace(readStringConfig(node.Data.Config, "description")); description != "" {
 		input.Description = description
 	}
-	if impact := strings.TrimSpace(readStringConfig(node.Config, "impact")); impact != "" {
+	if impact := strings.TrimSpace(readStringConfig(node.Data.Config, "impact")); impact != "" {
 		input.Impact = impact
 	}
-	if expectedOutcome := strings.TrimSpace(readStringConfig(node.Config, "expectedOutcome")); expectedOutcome != "" {
+	if expectedOutcome := strings.TrimSpace(readStringConfig(node.Data.Config, "expectedOutcome")); expectedOutcome != "" {
 		input.ExpectedOutcome = expectedOutcome
 	}
-	if currentAttempt := strings.TrimSpace(readStringConfig(node.Config, "currentAttempt")); currentAttempt != "" {
+	if currentAttempt := strings.TrimSpace(readStringConfig(node.Data.Config, "currentAttempt")); currentAttempt != "" {
 		input.CurrentAttempt = currentAttempt
 	}
 	args, err := json.Marshal(input)
@@ -632,20 +641,20 @@ func (e *Executor) executeAnalyzeConversation(ctx context.Context, state *runSta
 	input := graphs.AnalyzeConversationInput{
 		ObservedIssue: userMessage,
 	}
-	if strings.TrimSpace(readStringConfig(node.Config, "goal")) != "" {
-		input.Goal = strings.TrimSpace(readStringConfig(node.Config, "goal"))
+	if strings.TrimSpace(readStringConfig(node.Data.Config, "goal")) != "" {
+		input.Goal = strings.TrimSpace(readStringConfig(node.Data.Config, "goal"))
 	}
-	if readBoolConfig(node.Config, "needTicket") {
+	if readBoolConfig(node.Data.Config, "needTicket") {
 		input.NeedTicket = true
 	}
-	if readBoolConfig(node.Config, "needHumanHandoff") {
+	if readBoolConfig(node.Data.Config, "needHumanHandoff") {
 		input.NeedHumanHandoff = true
 	}
-	if readBoolConfig(node.Config, "needQualityCheck") {
+	if readBoolConfig(node.Data.Config, "needQualityCheck") {
 		input.NeedQualityCheck = true
 	}
-	if strings.TrimSpace(readStringConfig(node.Config, "additionalContext")) != "" {
-		input.AdditionalContext = strings.TrimSpace(readStringConfig(node.Config, "additionalContext"))
+	if strings.TrimSpace(readStringConfig(node.Data.Config, "additionalContext")) != "" {
+		input.AdditionalContext = strings.TrimSpace(readStringConfig(node.Data.Config, "additionalContext"))
 	}
 	args, err := json.Marshal(input)
 	if err != nil {
@@ -670,7 +679,7 @@ func (e *Executor) executeAnalyzeConversation(ctx context.Context, state *runSta
 }
 
 func (e *Executor) executeHandoffToHuman(state *runState, node dsl.Node) error {
-	if _, hasConfirmedInput := node.Inputs["confirmed"]; hasConfirmedInput && !truthy(state.resolveInput(node, "confirmed")) {
+	if _, hasConfirmedInput := node.Data.InputsValues["confirmed"]; hasConfirmedInput && !truthy(state.resolveInput(node, "confirmed")) {
 		state.setNodeVars(node.ID, map[string]any{
 			"handoffId":  int64(0),
 			"reason":     strings.TrimSpace(toString(state.resolveInput(node, "reason"))),
@@ -751,7 +760,7 @@ func (e *Executor) executeAnswerabilityGate(state *runState, node dsl.Node) erro
 }
 
 func (e *Executor) executeLLMReply(ctx context.Context, state *runState, node dsl.Node) error {
-	if staticReply := strings.TrimSpace(readStringConfig(node.Config, "staticReply")); staticReply != "" {
+	if staticReply := strings.TrimSpace(readStringConfig(node.Data.Config, "staticReply")); staticReply != "" {
 		state.setNodeVars(node.ID, map[string]any{"replyText": staticReply})
 		return nil
 	}
@@ -761,10 +770,10 @@ func (e *Executor) executeLLMReply(ctx context.Context, state *runState, node ds
 	}
 	knowledgeItems := toString(state.resolveInput(node, "knowledgeItems"))
 	systemPrompt := strings.TrimSpace(state.input.AIAgent.SystemPrompt)
-	if prompt := strings.TrimSpace(readStringConfig(node.Config, "prompt")); prompt != "" {
+	if prompt := strings.TrimSpace(readStringConfig(node.Data.Config, "prompt")); prompt != "" {
 		systemPrompt = strings.TrimSpace(systemPrompt + "\n\n" + prompt)
 	}
-	if _, declaresKnowledge := node.Inputs["knowledgeItems"]; declaresKnowledge && len(utils.SplitInt64s(state.input.AIAgent.KnowledgeIDs)) > 0 && !hasItems(state.resolveInput(node, "knowledgeItems")) {
+	if _, declaresKnowledge := node.Data.InputsValues["knowledgeItems"]; declaresKnowledge && len(utils.SplitInt64s(state.input.AIAgent.KnowledgeIDs)) > 0 && !hasItems(state.resolveInput(node, "knowledgeItems")) {
 		state.setNodeVars(node.ID, map[string]any{"replyText": workflowKnowledgeFallbackReply(state.input.AIAgent)})
 		return nil
 	}
@@ -798,11 +807,11 @@ func (s *runState) nextNodeID(sourceNodeID string) (string, bool, error) {
 	}
 	node := s.nodesByID[sourceNodeID]
 	if strings.TrimSpace(node.Type) != workflowregistry.NodeTypeCondition {
-		return strings.TrimSpace(edges[0].Target), true, nil
+		return strings.TrimSpace(edges[0].TargetNodeID), true, nil
 	}
 	config := dsl.ConditionConfig{}
-	if len(node.Config) > 0 {
-		if err := json.Unmarshal(node.Config, &config); err != nil {
+	if len(node.Data.Config) > 0 {
+		if err := json.Unmarshal(node.Data.Config, &config); err != nil {
 			return "", false, fmt.Errorf("invalid condition node config: %w", err)
 		}
 	}
@@ -864,11 +873,13 @@ func (s *runState) evaluateConditionBranch(sourceNodeID string, branch dsl.Condi
 		evaluation.Matched = true
 		return true, evaluation, nil
 	}
-	left := s.resolveSelector(condition.Left)
 	operator := strings.TrimSpace(condition.Operator)
+	var left any
 	if condition.Left != nil {
-		evaluation.SourceNodeID = strings.TrimSpace(condition.Left.NodeID)
-		evaluation.SourceField = strings.TrimSpace(condition.Left.Field)
+		left = s.resolveValue(*condition.Left)
+		evaluation.SourceNodeID, evaluation.SourceField, _ = condition.Left.Ref()
+		evaluation.SourceNodeID = strings.TrimSpace(evaluation.SourceNodeID)
+		evaluation.SourceField = strings.TrimSpace(evaluation.SourceField)
 	}
 	evaluation.Operator = operator
 	evaluation.LeftValue = left
@@ -909,8 +920,11 @@ func (s *runState) evaluateConditionBranch(sourceNodeID string, branch dsl.Condi
 
 func (s *runState) edgeIDForTarget(sourceNodeID string, targetNodeID string) string {
 	for _, edge := range s.outgoing[sourceNodeID] {
-		if strings.TrimSpace(edge.Target) == targetNodeID {
-			return strings.TrimSpace(edge.ID)
+		if strings.TrimSpace(edge.TargetNodeID) == targetNodeID {
+			if edge.SourcePortID != "" {
+				return strings.TrimSpace(edge.SourcePortID)
+			}
+			return strings.TrimSpace(edge.SourceNodeID + "->" + edge.TargetNodeID)
 		}
 	}
 	return ""
@@ -921,27 +935,27 @@ func (s *runState) setNodeVars(nodeID string, values map[string]any) {
 }
 
 func (s *runState) resolveInput(node dsl.Node, inputName string) any {
-	selector, ok := node.Inputs[inputName]
+	value, ok := node.Data.InputsValues[inputName]
 	if !ok {
 		return nil
 	}
-	return s.resolveSelector(&selector)
+	return s.resolveValue(value)
 }
 
 func (s *runState) nodeInputPreview(node dsl.Node) map[string]any {
-	inputs := make(map[string]any, len(node.Inputs))
-	for name, selector := range node.Inputs {
-		inputs[name] = s.resolveSelector(&selector)
+	inputs := make(map[string]any, len(node.Data.InputsValues))
+	for name, value := range node.Data.InputsValues {
+		inputs[name] = s.resolveValue(value)
 	}
 	ret := map[string]any{
 		"inputs": inputs,
 	}
-	if len(node.Config) > 0 {
+	if len(node.Data.Config) > 0 {
 		var cfg any
-		if err := json.Unmarshal(node.Config, &cfg); err == nil {
+		if err := json.Unmarshal(node.Data.Config, &cfg); err == nil {
 			ret["config"] = cfg
 		} else {
-			ret["config"] = string(node.Config)
+			ret["config"] = string(node.Data.Config)
 		}
 	}
 	return ret
@@ -969,15 +983,28 @@ func workflowPreviewJSON(value any) string {
 	return string(raw[:maxPreviewBytes])
 }
 
-func (s *runState) resolveSelector(selector *dsl.VariableSelector) any {
-	if selector == nil {
+func (s *runState) resolveValue(value dsl.Value) any {
+	switch value.Type {
+	case dsl.ValueTypeRef:
+		nodeID, field, ok := value.Ref()
+		if !ok {
+			return nil
+		}
+		fields := s.vars[strings.TrimSpace(nodeID)]
+		if fields == nil {
+			return nil
+		}
+		return fields[strings.TrimSpace(field)]
+	case dsl.ValueTypeConstant:
+		return value.ConstantContent
+	case dsl.ValueTypeTemplate:
+		if len(value.Content) > 0 {
+			return value.Content[0]
+		}
+		return nil
+	default:
 		return nil
 	}
-	fields := s.vars[strings.TrimSpace(selector.NodeID)]
-	if fields == nil {
-		return nil
-	}
-	return fields[strings.TrimSpace(selector.Field)]
 }
 
 func readStringConfig(raw json.RawMessage, key string) string {

internal/ai/runtime/workflow/executor_test.go

@@ -435,286 +435,254 @@ func findNodeTrace(items []NodeTrace, nodeID string) *NodeTrace {
 }
 
 func emptyKnowledgeReplyDefinition() dsl.Definition {
-	return dsl.Definition{
-		SchemaVersion: 1,
-		EntryNodeID:   "start_1",
-		Nodes: []dsl.Node{
-			{ID: "start_1", Type: workflowregistry.NodeTypeStart, Name: "Start"},
-			{ID: "reply_1", Type: workflowregistry.NodeTypeLLMReply, Name: "Reply", Inputs: map[string]dsl.VariableSelector{
-				"userMessage":    {NodeID: "start_1", Field: "userMessage"},
-				"knowledgeItems": {NodeID: "missing_retrieve", Field: "items"},
-			}},
-			{ID: "send_1", Type: workflowregistry.NodeTypeSendReply, Name: "Send", Inputs: map[string]dsl.VariableSelector{
-				"replyText": {NodeID: "reply_1", Field: "replyText"},
-			}},
-			{ID: "end_1", Type: workflowregistry.NodeTypeEnd, Name: "End"},
+	return wfTestDefinition(
+		[]dsl.Node{
+			wfTestNode("start_1", workflowregistry.NodeTypeStart, "Start", nil, nil),
+			wfTestNode("reply_1", workflowregistry.NodeTypeLLMReply, "Reply", map[string]dsl.Value{
+				"userMessage":    dsl.RefValue("start_1", "userMessage"),
+				"knowledgeItems": dsl.RefValue("missing_retrieve", "items"),
+			}, nil),
+			wfTestNode("send_1", workflowregistry.NodeTypeSendReply, "Send", wfTestInputs("replyText", "reply_1", "replyText"), nil),
+			wfTestNode("end_1", workflowregistry.NodeTypeEnd, "End", nil, nil),
 		},
-		Edges: []dsl.Edge{
-			{ID: "edge_start_reply", Source: "start_1", Target: "reply_1"},
-			{ID: "edge_reply_send", Source: "reply_1", Target: "send_1"},
-			{ID: "edge_send_end", Source: "send_1", Target: "end_1"},
+		[]dsl.Edge{
+			wfTestEdge("start_1", "reply_1", "edge_start_reply"),
+			wfTestEdge("reply_1", "send_1", "edge_reply_send"),
+			wfTestEdge("send_1", "end_1", "edge_send_end"),
 		},
-	}
+	)
 }
 
 func policyFirstWorkflowDefinition() dsl.Definition {
-	return dsl.Definition{
-		SchemaVersion: 1,
-		EntryNodeID:   "start_1",
-		Nodes: []dsl.Node{
-			{ID: "start_1", Type: workflowregistry.NodeTypeStart, Name: "Start"},
-			{ID: "understanding_1", Type: workflowregistry.NodeTypeConversationUnderstanding, Name: "Understanding", Inputs: map[string]dsl.VariableSelector{
-				"userMessage": {NodeID: "start_1", Field: "userMessage"},
-			}},
-			{ID: "policy_1", Type: workflowregistry.NodeTypeReplyPolicy, Name: "Policy", Inputs: map[string]dsl.VariableSelector{
-				"userMessage":    {NodeID: "start_1", Field: "userMessage"},
-				"messageIntent":  {NodeID: "understanding_1", Field: "messageIntent"},
-				"answerScope":    {NodeID: "understanding_1", Field: "answerScope"},
-				"riskSignals":    {NodeID: "understanding_1", Field: "riskSignals"},
-				"knowledgeItems": {NodeID: "retrieve_1", Field: "items"},
-			}},
-			{ID: "policy_route_1", Type: workflowregistry.NodeTypeCondition, Name: "Policy Route", Config: mustMarshalWorkflowTestConfig(dsl.ConditionConfig{Branches: []dsl.ConditionBranch{
-				{ID: "direct", Name: "Direct", TargetNodeID: "send_direct_1", Condition: &dsl.Condition{
-					Left:     &dsl.VariableSelector{NodeID: "policy_1", Field: "action"},
-					Operator: "eq",
-					Right:    "direct_reply",
-				}},
-				{ID: "knowledge", Name: "Knowledge", TargetNodeID: "retrieve_end", Condition: &dsl.Condition{
-					Left:     &dsl.VariableSelector{NodeID: "policy_1", Field: "action"},
-					Operator: "eq",
-					Right:    "retrieve_knowledge",
-				}},
+	return wfTestDefinition(
+		[]dsl.Node{
+			wfTestNode("start_1", workflowregistry.NodeTypeStart, "Start", nil, nil),
+			wfTestNode("understanding_1", workflowregistry.NodeTypeConversationUnderstanding, "Understanding", wfTestInputs("userMessage", "start_1", "userMessage"), nil),
+			wfTestNode("policy_1", workflowregistry.NodeTypeReplyPolicy, "Policy", map[string]dsl.Value{
+				"userMessage":    dsl.RefValue("start_1", "userMessage"),
+				"messageIntent":  dsl.RefValue("understanding_1", "messageIntent"),
+				"answerScope":    dsl.RefValue("understanding_1", "answerScope"),
+				"riskSignals":    dsl.RefValue("understanding_1", "riskSignals"),
+				"knowledgeItems": dsl.RefValue("retrieve_1", "items"),
+			}, nil),
+			wfTestNode("policy_route_1", workflowregistry.NodeTypeCondition, "Policy Route", nil, dsl.ConditionConfig{Branches: []dsl.ConditionBranch{
+				wfTestConditionBranch("direct", "Direct", "send_direct_1", "policy_1", "action", "eq", "direct_reply"),
+				wfTestConditionBranch("knowledge", "Knowledge", "retrieve_end", "policy_1", "action", "eq", "retrieve_knowledge"),
 				{ID: "default", Name: "Default", TargetNodeID: "end_1", Default: true},
-			}})},
-			{ID: "send_direct_1", Type: workflowregistry.NodeTypeSendReply, Name: "Send Direct", Inputs: map[string]dsl.VariableSelector{
-				"replyText": {NodeID: "policy_1", Field: "replyText"},
-			}},
-			{ID: "retrieve_end", Type: workflowregistry.NodeTypeEnd, Name: "Retrieve"},
-			{ID: "end_1", Type: workflowregistry.NodeTypeEnd, Name: "End"},
+			}}),
+			wfTestNode("send_direct_1", workflowregistry.NodeTypeSendReply, "Send Direct", wfTestInputs("replyText", "policy_1", "replyText"), nil),
+			wfTestNode("retrieve_end", workflowregistry.NodeTypeEnd, "Retrieve", nil, nil),
+			wfTestNode("end_1", workflowregistry.NodeTypeEnd, "End", nil, nil),
 		},
-		Edges: []dsl.Edge{
-			{ID: "edge_start_understanding", Source: "start_1", Target: "understanding_1"},
-			{ID: "edge_understanding_policy", Source: "understanding_1", Target: "policy_1"},
-			{ID: "edge_policy_route", Source: "policy_1", Target: "policy_route_1"},
-			{ID: "edge_policy_direct", Source: "policy_route_1", Target: "send_direct_1"},
-			{ID: "edge_policy_knowledge", Source: "policy_route_1", Target: "retrieve_end"},
-			{ID: "edge_policy_default", Source: "policy_route_1", Target: "end_1"},
-			{ID: "edge_send_direct_end", Source: "send_direct_1", Target: "end_1"},
+		[]dsl.Edge{
+			wfTestEdge("start_1", "understanding_1", "edge_start_understanding"),
+			wfTestEdge("understanding_1", "policy_1", "edge_understanding_policy"),
+			wfTestEdge("policy_1", "policy_route_1", "edge_policy_route"),
+			wfTestEdge("policy_route_1", "send_direct_1", "edge_policy_direct"),
+			wfTestEdge("policy_route_1", "retrieve_end", "edge_policy_knowledge"),
+			wfTestEdge("policy_route_1", "end_1", "edge_policy_default"),
+			wfTestEdge("send_direct_1", "end_1", "edge_send_direct_end"),
 		},
-	}
+	)
 }
 
 func conditionalReplyDefinition() dsl.Definition {
-	return dsl.Definition{
-		SchemaVersion: 1,
-		EntryNodeID:   "start_1",
-		Nodes: []dsl.Node{
-			{ID: "start_1", Type: workflowregistry.NodeTypeStart, Name: "Start"},
-			{ID: "condition_1", Type: workflowregistry.NodeTypeCondition, Name: "Route", Config: mustMarshalWorkflowTestConfig(dsl.ConditionConfig{Branches: []dsl.ConditionBranch{
-				{ID: "vip", Name: "VIP", TargetNodeID: "vip_reply", Condition: &dsl.Condition{
-					Left:     &dsl.VariableSelector{NodeID: "start_1", Field: "userMessage"},
-					Operator: "eq",
-					Right:    "vip",
-				}},
+	return wfTestDefinition(
+		[]dsl.Node{
+			wfTestNode("start_1", workflowregistry.NodeTypeStart, "Start", nil, nil),
+			wfTestNode("condition_1", workflowregistry.NodeTypeCondition, "Route", nil, dsl.ConditionConfig{Branches: []dsl.ConditionBranch{
+				wfTestConditionBranch("vip", "VIP", "vip_reply", "start_1", "userMessage", "eq", "vip"),
 				{ID: "default", Name: "Default", TargetNodeID: "normal_reply", Default: true},
-			}})},
-			{ID: "vip_reply", Type: workflowregistry.NodeTypeLLMReply, Name: "VIP", Config: []byte(`{"staticReply":"VIP reply"}`)},
-			{ID: "normal_reply", Type: workflowregistry.NodeTypeLLMReply, Name: "Normal", Config: []byte(`{"staticReply":"Normal reply"}`)},
-			{ID: "send_vip", Type: workflowregistry.NodeTypeSendReply, Name: "Send VIP", Inputs: map[string]dsl.VariableSelector{
-				"replyText": {NodeID: "vip_reply", Field: "replyText"},
-			}},
-			{ID: "send_normal", Type: workflowregistry.NodeTypeSendReply, Name: "Send Normal", Inputs: map[string]dsl.VariableSelector{
-				"replyText": {NodeID: "normal_reply", Field: "replyText"},
-			}},
-			{ID: "end_1", Type: workflowregistry.NodeTypeEnd, Name: "End"},
+			}}),
+			wfTestNode("vip_reply", workflowregistry.NodeTypeLLMReply, "VIP", nil, map[string]any{"staticReply": "VIP reply"}),
+			wfTestNode("normal_reply", workflowregistry.NodeTypeLLMReply, "Normal", nil, map[string]any{"staticReply": "Normal reply"}),
+			wfTestNode("send_vip", workflowregistry.NodeTypeSendReply, "Send VIP", wfTestInputs("replyText", "vip_reply", "replyText"), nil),
+			wfTestNode("send_normal", workflowregistry.NodeTypeSendReply, "Send Normal", wfTestInputs("replyText", "normal_reply", "replyText"), nil),
+			wfTestNode("end_1", workflowregistry.NodeTypeEnd, "End", nil, nil),
 		},
-		Edges: []dsl.Edge{
-			{ID: "edge_start_condition", Source: "start_1", Target: "condition_1"},
-			{ID: "edge_condition_vip", Source: "condition_1", Target: "vip_reply"},
-			{ID: "edge_condition_default", Source: "condition_1", Target: "normal_reply"},
-			{ID: "edge_vip_send", Source: "vip_reply", Target: "send_vip"},
-			{ID: "edge_normal_send", Source: "normal_reply", Target: "send_normal"},
-			{ID: "edge_send_vip_end", Source: "send_vip", Target: "end_1"},
-			{ID: "edge_send_normal_end", Source: "send_normal", Target: "end_1"},
+		[]dsl.Edge{
+			wfTestEdge("start_1", "condition_1", "edge_start_condition"),
+			wfTestEdge("condition_1", "vip_reply", "edge_condition_vip"),
+			wfTestEdge("condition_1", "normal_reply", "edge_condition_default"),
+			wfTestEdge("vip_reply", "send_vip", "edge_vip_send"),
+			wfTestEdge("normal_reply", "send_normal", "edge_normal_send"),
+			wfTestEdge("send_vip", "end_1", "edge_send_vip_end"),
+			wfTestEdge("send_normal", "end_1", "edge_send_normal_end"),
 		},
-	}
+	)
 }
 
 func createTicketWorkflowDefinition() dsl.Definition {
-	return dsl.Definition{
-		SchemaVersion: 1,
-		EntryNodeID:   "start_1",
-		Nodes: []dsl.Node{
-			{ID: "start_1", Type: workflowregistry.NodeTypeStart, Name: "Start"},
-			{ID: "draft_1", Type: workflowregistry.NodeTypePrepareTicketDraft, Name: "Draft", Inputs: map[string]dsl.VariableSelector{
-				"issue": {NodeID: "start_1", Field: "userMessage"},
-			}},
-			{ID: "prompt_1", Type: workflowregistry.NodeTypeLLMReply, Name: "Prompt", Config: []byte(`{"staticReply":"请确认创建工单"}`)},
-			{ID: "confirm_1", Type: workflowregistry.NodeTypeHumanConfirm, Name: "Confirm", Inputs: map[string]dsl.VariableSelector{
-				"prompt": {NodeID: "prompt_1", Field: "replyText"},
-			}},
-			{ID: "confirm_route_1", Type: workflowregistry.NodeTypeCondition, Name: "Confirm Route", Config: mustMarshalWorkflowTestConfig(dsl.ConditionConfig{Branches: []dsl.ConditionBranch{
-				{ID: "yes", Name: "Yes", TargetNodeID: "create_ticket_1", Condition: &dsl.Condition{
-					Left:     &dsl.VariableSelector{NodeID: "confirm_1", Field: "confirmed"},
-					Operator: "is_true",
-				}},
+	return wfTestDefinition(
+		[]dsl.Node{
+			wfTestNode("start_1", workflowregistry.NodeTypeStart, "Start", nil, nil),
+			wfTestNode("draft_1", workflowregistry.NodeTypePrepareTicketDraft, "Draft", wfTestInputs("issue", "start_1", "userMessage"), nil),
+			wfTestNode("prompt_1", workflowregistry.NodeTypeLLMReply, "Prompt", nil, map[string]any{"staticReply": "请确认创建工单"}),
+			wfTestNode("confirm_1", workflowregistry.NodeTypeHumanConfirm, "Confirm", wfTestInputs("prompt", "prompt_1", "replyText"), nil),
+			wfTestNode("confirm_route_1", workflowregistry.NodeTypeCondition, "Confirm Route", nil, dsl.ConditionConfig{Branches: []dsl.ConditionBranch{
+				wfTestConditionBranch("yes", "Yes", "create_ticket_1", "confirm_1", "confirmed", "is_true", nil),
 				{ID: "default", Name: "Cancel", TargetNodeID: "cancel_end", Default: true},
-			}})},
-			{ID: "create_ticket_1", Type: workflowregistry.NodeTypeCreateTicket, Name: "Create Ticket", Inputs: map[string]dsl.VariableSelector{
-				"ticketDraft": {NodeID: "draft_1", Field: "ticketDraft"},
-				"confirmed":   {NodeID: "confirm_1", Field: "confirmed"},
-			}},
-			{ID: "end_1", Type: workflowregistry.NodeTypeEnd, Name: "End"},
-			{ID: "cancel_end", Type: workflowregistry.NodeTypeEnd, Name: "Cancel"},
+			}}),
+			wfTestNode("create_ticket_1", workflowregistry.NodeTypeCreateTicket, "Create Ticket", map[string]dsl.Value{
+				"ticketDraft": dsl.RefValue("draft_1", "ticketDraft"),
+				"confirmed":   dsl.RefValue("confirm_1", "confirmed"),
+			}, nil),
+			wfTestNode("end_1", workflowregistry.NodeTypeEnd, "End", nil, nil),
+			wfTestNode("cancel_end", workflowregistry.NodeTypeEnd, "Cancel", nil, nil),
 		},
-		Edges: []dsl.Edge{
-			{ID: "edge_start_draft", Source: "start_1", Target: "draft_1"},
-			{ID: "edge_draft_prompt", Source: "draft_1", Target: "prompt_1"},
-			{ID: "edge_prompt_confirm", Source: "prompt_1", Target: "confirm_1"},
-			{ID: "edge_confirm_route", Source: "confirm_1", Target: "confirm_route_1"},
-			{ID: "edge_confirm_create", Source: "confirm_route_1", Target: "create_ticket_1"},
-			{ID: "edge_confirm_cancel", Source: "confirm_route_1", Target: "cancel_end"},
-			{ID: "edge_create_end", Source: "create_ticket_1", Target: "end_1"},
+		[]dsl.Edge{
+			wfTestEdge("start_1", "draft_1", "edge_start_draft"),
+			wfTestEdge("draft_1", "prompt_1", "edge_draft_prompt"),
+			wfTestEdge("prompt_1", "confirm_1", "edge_prompt_confirm"),
+			wfTestEdge("confirm_1", "confirm_route_1", "edge_confirm_route"),
+			wfTestEdge("confirm_route_1", "create_ticket_1", "edge_confirm_create"),
+			wfTestEdge("confirm_route_1", "cancel_end", "edge_confirm_cancel"),
+			wfTestEdge("create_ticket_1", "end_1", "edge_create_end"),
 		},
-	}
+	)
 }
 
 func humanConfirmWorkflowDefinition() dsl.Definition {
-	return dsl.Definition{
-		SchemaVersion: 1,
-		EntryNodeID:   "start_1",
-		Nodes: []dsl.Node{
-			{ID: "start_1", Type: workflowregistry.NodeTypeStart, Name: "Start"},
-			{ID: "prompt_1", Type: workflowregistry.NodeTypeLLMReply, Name: "Prompt", Config: []byte(`{"staticReply":"请确认创建工单"}`)},
-			{ID: "confirm_1", Type: workflowregistry.NodeTypeHumanConfirm, Name: "Confirm", Inputs: map[string]dsl.VariableSelector{
-				"prompt": {NodeID: "prompt_1", Field: "replyText"},
-			}},
-			{ID: "confirm_route_1", Type: workflowregistry.NodeTypeCondition, Name: "Confirm Route", Config: mustMarshalWorkflowTestConfig(dsl.ConditionConfig{Branches: []dsl.ConditionBranch{
-				{ID: "yes", Name: "Yes", TargetNodeID: "end_1", Condition: &dsl.Condition{
-					Left:     &dsl.VariableSelector{NodeID: "confirm_1", Field: "confirmed"},
-					Operator: "is_true",
-				}},
+	return wfTestDefinition(
+		[]dsl.Node{
+			wfTestNode("start_1", workflowregistry.NodeTypeStart, "Start", nil, nil),
+			wfTestNode("prompt_1", workflowregistry.NodeTypeLLMReply, "Prompt", nil, map[string]any{"staticReply": "请确认创建工单"}),
+			wfTestNode("confirm_1", workflowregistry.NodeTypeHumanConfirm, "Confirm", wfTestInputs("prompt", "prompt_1", "replyText"), nil),
+			wfTestNode("confirm_route_1", workflowregistry.NodeTypeCondition, "Confirm Route", nil, dsl.ConditionConfig{Branches: []dsl.ConditionBranch{
+				wfTestConditionBranch("yes", "Yes", "end_1", "confirm_1", "confirmed", "is_true", nil),
 				{ID: "default", Name: "Cancel", TargetNodeID: "cancel_end", Default: true},
-			}})},
-			{ID: "end_1", Type: workflowregistry.NodeTypeEnd, Name: "End"},
-			{ID: "cancel_end", Type: workflowregistry.NodeTypeEnd, Name: "Cancel"},
+			}}),
+			wfTestNode("end_1", workflowregistry.NodeTypeEnd, "End", nil, nil),
+			wfTestNode("cancel_end", workflowregistry.NodeTypeEnd, "Cancel", nil, nil),
 		},
-		Edges: []dsl.Edge{
-			{ID: "edge_start_prompt", Source: "start_1", Target: "prompt_1"},
-			{ID: "edge_prompt_confirm", Source: "prompt_1", Target: "confirm_1"},
-			{ID: "edge_confirm_route", Source: "confirm_1", Target: "confirm_route_1"},
-			{ID: "edge_confirm_yes", Source: "confirm_route_1", Target: "end_1"},
-			{ID: "edge_confirm_cancel", Source: "confirm_route_1", Target: "cancel_end"},
+		[]dsl.Edge{
+			wfTestEdge("start_1", "prompt_1", "edge_start_prompt"),
+			wfTestEdge("prompt_1", "confirm_1", "edge_prompt_confirm"),
+			wfTestEdge("confirm_1", "confirm_route_1", "edge_confirm_route"),
+			wfTestEdge("confirm_route_1", "end_1", "edge_confirm_yes"),
+			wfTestEdge("confirm_route_1", "cancel_end", "edge_confirm_cancel"),
 		},
-	}
+	)
 }
 
 func prepareTicketDraftWorkflowDefinition() dsl.Definition {
-	return dsl.Definition{
-		SchemaVersion: 1,
-		EntryNodeID:   "start_1",
-		Nodes: []dsl.Node{
-			{ID: "start_1", Type: workflowregistry.NodeTypeStart, Name: "Start"},
-			{ID: "draft_1", Type: workflowregistry.NodeTypePrepareTicketDraft, Name: "Draft", Inputs: map[string]dsl.VariableSelector{
-				"issue": {NodeID: "start_1", Field: "userMessage"},
-			}},
-			{ID: "draft_route_1", Type: workflowregistry.NodeTypeCondition, Name: "Draft Route", Config: mustMarshalWorkflowTestConfig(dsl.ConditionConfig{Branches: []dsl.ConditionBranch{
-				{ID: "ready", Name: "Ready", TargetNodeID: "ready_end", Condition: &dsl.Condition{
-					Left:     &dsl.VariableSelector{NodeID: "draft_1", Field: "ticketDraft"},
-					Operator: "exists",
-				}},
+	return wfTestDefinition(
+		[]dsl.Node{
+			wfTestNode("start_1", workflowregistry.NodeTypeStart, "Start", nil, nil),
+			wfTestNode("draft_1", workflowregistry.NodeTypePrepareTicketDraft, "Draft", wfTestInputs("issue", "start_1", "userMessage"), nil),
+			wfTestNode("draft_route_1", workflowregistry.NodeTypeCondition, "Draft Route", nil, dsl.ConditionConfig{Branches: []dsl.ConditionBranch{
+				wfTestConditionBranch("ready", "Ready", "ready_end", "draft_1", "ticketDraft", "exists", nil),
 				{ID: "default", Name: "Default", TargetNodeID: "default_end", Default: true},
-			}})},
-			{ID: "ready_end", Type: workflowregistry.NodeTypeEnd, Name: "Ready"},
-			{ID: "default_end", Type: workflowregistry.NodeTypeEnd, Name: "Default"},
+			}}),
+			wfTestNode("ready_end", workflowregistry.NodeTypeEnd, "Ready", nil, nil),
+			wfTestNode("default_end", workflowregistry.NodeTypeEnd, "Default", nil, nil),
 		},
-		Edges: []dsl.Edge{
-			{ID: "edge_start_draft", Source: "start_1", Target: "draft_1"},
-			{ID: "edge_draft_route", Source: "draft_1", Target: "draft_route_1"},
-			{ID: "edge_draft_ready", Source: "draft_route_1", Target: "ready_end"},
-			{ID: "edge_draft_default", Source: "draft_route_1", Target: "default_end"},
+		[]dsl.Edge{
+			wfTestEdge("start_1", "draft_1", "edge_start_draft"),
+			wfTestEdge("draft_1", "draft_route_1", "edge_draft_route"),
+			wfTestEdge("draft_route_1", "ready_end", "edge_draft_ready"),
+			wfTestEdge("draft_route_1", "default_end", "edge_draft_default"),
 		},
-	}
+	)
 }
 
 func analyzeConversationWorkflowDefinition() dsl.Definition {
-	return dsl.Definition{
-		SchemaVersion: 1,
-		EntryNodeID:   "start_1",
-		Nodes: []dsl.Node{
-			{ID: "start_1", Type: workflowregistry.NodeTypeStart, Name: "Start"},
-			{ID: "analyze_1", Type: workflowregistry.NodeTypeAnalyzeConversation, Name: "Analyze", Inputs: map[string]dsl.VariableSelector{
-				"userMessage": {NodeID: "start_1", Field: "userMessage"},
-			}},
-			{ID: "analyze_route_1", Type: workflowregistry.NodeTypeCondition, Name: "Analyze Route", Config: mustMarshalWorkflowTestConfig(dsl.ConditionConfig{Branches: []dsl.ConditionBranch{
-				{ID: "handoff", Name: "Handoff", TargetNodeID: "handoff_end", Condition: &dsl.Condition{
-					Left:     &dsl.VariableSelector{NodeID: "analyze_1", Field: "needHumanHandoff"},
-					Operator: "is_true",
-				}},
+	return wfTestDefinition(
+		[]dsl.Node{
+			wfTestNode("start_1", workflowregistry.NodeTypeStart, "Start", nil, nil),
+			wfTestNode("analyze_1", workflowregistry.NodeTypeAnalyzeConversation, "Analyze", wfTestInputs("userMessage", "start_1", "userMessage"), nil),
+			wfTestNode("analyze_route_1", workflowregistry.NodeTypeCondition, "Analyze Route", nil, dsl.ConditionConfig{Branches: []dsl.ConditionBranch{
+				wfTestConditionBranch("handoff", "Handoff", "handoff_end", "analyze_1", "needHumanHandoff", "is_true", nil),
 				{ID: "default", Name: "Default", TargetNodeID: "default_end", Default: true},
-			}})},
-			{ID: "handoff_end", Type: workflowregistry.NodeTypeEnd, Name: "Handoff"},
-			{ID: "default_end", Type: workflowregistry.NodeTypeEnd, Name: "Default"},
+			}}),
+			wfTestNode("handoff_end", workflowregistry.NodeTypeEnd, "Handoff", nil, nil),
+			wfTestNode("default_end", workflowregistry.NodeTypeEnd, "Default", nil, nil),
 		},
-		Edges: []dsl.Edge{
-			{ID: "edge_start_analyze", Source: "start_1", Target: "analyze_1"},
-			{ID: "edge_analyze_route", Source: "analyze_1", Target: "analyze_route_1"},
-			{ID: "edge_analyze_handoff", Source: "analyze_route_1", Target: "handoff_end"},
-			{ID: "edge_analyze_default", Source: "analyze_route_1", Target: "default_end"},
+		[]dsl.Edge{
+			wfTestEdge("start_1", "analyze_1", "edge_start_analyze"),
+			wfTestEdge("analyze_1", "analyze_route_1", "edge_analyze_route"),
+			wfTestEdge("analyze_route_1", "handoff_end", "edge_analyze_handoff"),
+			wfTestEdge("analyze_route_1", "default_end", "edge_analyze_default"),
 		},
-	}
+	)
 }
 
 func handoffWorkflowDefinition() dsl.Definition {
-	return dsl.Definition{
-		SchemaVersion: 1,
-		EntryNodeID:   "start_1",
-		Nodes: []dsl.Node{
-			{ID: "start_1", Type: workflowregistry.NodeTypeStart, Name: "Start"},
-			{ID: "handoff_1", Type: workflowregistry.NodeTypeHandoffToHuman, Name: "Handoff", Inputs: map[string]dsl.VariableSelector{
-				"reason": {NodeID: "start_1", Field: "userMessage"},
-			}},
-			{ID: "handoff_route_1", Type: workflowregistry.NodeTypeCondition, Name: "Handoff Route", Config: mustMarshalWorkflowTestConfig(dsl.ConditionConfig{Branches: []dsl.ConditionBranch{
-				{ID: "assigned", Name: "Assigned", TargetNodeID: "assigned_end", Condition: &dsl.Condition{
-					Left:     &dsl.VariableSelector{NodeID: "handoff_1", Field: "decision"},
-					Operator: "eq",
-					Right:    string(services.HandoffDecisionAssigned),
-				}},
+	return wfTestDefinition(
+		[]dsl.Node{
+			wfTestNode("start_1", workflowregistry.NodeTypeStart, "Start", nil, nil),
+			wfTestNode("handoff_1", workflowregistry.NodeTypeHandoffToHuman, "Handoff", wfTestInputs("reason", "start_1", "userMessage"), nil),
+			wfTestNode("handoff_route_1", workflowregistry.NodeTypeCondition, "Handoff Route", nil, dsl.ConditionConfig{Branches: []dsl.ConditionBranch{
+				wfTestConditionBranch("assigned", "Assigned", "assigned_end", "handoff_1", "decision", "eq", string(services.HandoffDecisionAssigned)),
 				{ID: "default", Name: "Default", TargetNodeID: "default_end", Default: true},
-			}})},
-			{ID: "assigned_end", Type: workflowregistry.NodeTypeEnd, Name: "Assigned"},
-			{ID: "default_end", Type: workflowregistry.NodeTypeEnd, Name: "Default"},
+			}}),
+			wfTestNode("assigned_end", workflowregistry.NodeTypeEnd, "Assigned", nil, nil),
+			wfTestNode("default_end", workflowregistry.NodeTypeEnd, "Default", nil, nil),
 		},
-		Edges: []dsl.Edge{
-			{ID: "edge_start_handoff", Source: "start_1", Target: "handoff_1"},
-			{ID: "edge_handoff_route", Source: "handoff_1", Target: "handoff_route_1"},
-			{ID: "edge_handoff_assigned", Source: "handoff_route_1", Target: "assigned_end"},
-			{ID: "edge_handoff_default", Source: "handoff_route_1", Target: "default_end"},
+		[]dsl.Edge{
+			wfTestEdge("start_1", "handoff_1", "edge_start_handoff"),
+			wfTestEdge("handoff_1", "handoff_route_1", "edge_handoff_route"),
+			wfTestEdge("handoff_route_1", "assigned_end", "edge_handoff_assigned"),
+			wfTestEdge("handoff_route_1", "default_end", "edge_handoff_default"),
+		},
+	)
+}
+
+func handoffAfterConfirmationWorkflowDefinition() dsl.Definition {
+	return wfTestDefinition(
+		[]dsl.Node{
+			wfTestNode("start_1", workflowregistry.NodeTypeStart, "Start", nil, nil),
+			wfTestNode("prompt_1", workflowregistry.NodeTypeLLMReply, "Prompt", nil, map[string]any{"staticReply": "请确认转人工"}),
+			wfTestNode("confirm_1", workflowregistry.NodeTypeHumanConfirm, "Confirm", wfTestInputs("prompt", "prompt_1", "replyText"), nil),
+			wfTestNode("handoff_1", workflowregistry.NodeTypeHandoffToHuman, "Handoff", map[string]dsl.Value{
+				"reason":    dsl.RefValue("start_1", "userMessage"),
+				"confirmed": dsl.RefValue("confirm_1", "confirmed"),
+			}, nil),
+			wfTestNode("end_1", workflowregistry.NodeTypeEnd, "End", nil, nil),
+		},
+		[]dsl.Edge{
+			wfTestEdge("start_1", "prompt_1", "edge_start_prompt"),
+			wfTestEdge("prompt_1", "confirm_1", "edge_prompt_confirm"),
+			wfTestEdge("confirm_1", "handoff_1", "edge_confirm_handoff"),
+			wfTestEdge("handoff_1", "end_1", "edge_handoff_end"),
+		},
+	)
+}
+
+func wfTestDefinition(nodes []dsl.Node, edges []dsl.Edge) dsl.Definition {
+	return dsl.Definition{SchemaVersion: dsl.SchemaVersion, Nodes: nodes, Edges: edges}
+}
+
+func wfTestNode(id string, nodeType string, title string, inputs map[string]dsl.Value, config any) dsl.Node {
+	return dsl.Node{
+		ID:   id,
+		Type: nodeType,
+		Meta: dsl.NodeMeta{Position: dsl.Position{X: 0, Y: 0}},
+		Data: dsl.NodeData{
+			Title:        title,
+			InputsValues: inputs,
+			Config:       mustMarshalWorkflowTestConfig(config),
 		},
 	}
 }
 
-func handoffAfterConfirmationWorkflowDefinition() dsl.Definition {
-	return dsl.Definition{
-		SchemaVersion: 1,
-		EntryNodeID:   "start_1",
-		Nodes: []dsl.Node{
-			{ID: "start_1", Type: workflowregistry.NodeTypeStart, Name: "Start"},
-			{ID: "prompt_1", Type: workflowregistry.NodeTypeLLMReply, Name: "Prompt", Config: []byte(`{"staticReply":"请确认转人工"}`)},
-			{ID: "confirm_1", Type: workflowregistry.NodeTypeHumanConfirm, Name: "Confirm", Inputs: map[string]dsl.VariableSelector{
-				"prompt": {NodeID: "prompt_1", Field: "replyText"},
-			}},
-			{ID: "handoff_1", Type: workflowregistry.NodeTypeHandoffToHuman, Name: "Handoff", Inputs: map[string]dsl.VariableSelector{
-				"reason":    {NodeID: "start_1", Field: "userMessage"},
-				"confirmed": {NodeID: "confirm_1", Field: "confirmed"},
-			}},
-			{ID: "end_1", Type: workflowregistry.NodeTypeEnd, Name: "End"},
-		},
-		Edges: []dsl.Edge{
-			{ID: "edge_start_prompt", Source: "start_1", Target: "prompt_1"},
-			{ID: "edge_prompt_confirm", Source: "prompt_1", Target: "confirm_1"},
-			{ID: "edge_confirm_handoff", Source: "confirm_1", Target: "handoff_1"},
-			{ID: "edge_handoff_end", Source: "handoff_1", Target: "end_1"},
+func wfTestInputs(name string, nodeID string, field string) map[string]dsl.Value {
+	return map[string]dsl.Value{name: dsl.RefValue(nodeID, field)}
+}
+
+func wfTestEdge(source string, target string, id string) dsl.Edge {
+	return dsl.Edge{SourceNodeID: source, TargetNodeID: target, SourcePortID: id}
+}
+
+func wfTestConditionBranch(id string, name string, targetNodeID string, nodeID string, field string, operator string, right any) dsl.ConditionBranch {
+	return dsl.ConditionBranch{
+		ID:           id,
+		Name:         name,
+		TargetNodeID: targetNodeID,
+		Condition: &dsl.Condition{
+			Left:     &dsl.Value{Type: dsl.ValueTypeRef, Content: []string{nodeID, field}},
+			Operator: operator,
+			Right:    right,
 		},
 	}
 }