From f0f8724bd9cce6832e055badb32dc24857d8d864 Mon Sep 17 00:00:00 2001
From: qhd <1766646056@qq.com>
Date: Mon, 24 Aug 2026 16:04:45 +0800
Subject: [PATCH] =?UTF-8?q?fix:=20=E4=BF=AE=E5=A4=8D=E5=B9=B6=E8=A1=8C?=
 =?UTF-8?q?=E5=88=86=E6=94=AF=E7=BB=AD=E8=B7=91=E8=BE=93=E5=87=BA=E4=B8=A2?=
 =?UTF-8?q?=E5=A4=B1=E5=8F=8A=E6=A8=A1=E5=9E=8B=E7=A9=BA=E6=8C=87=E9=92=88?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

合并续跑时各分支独立的 ConfigMap 副本，避免汇合节点输出丢失；并对模型配置缺失和空 responseType 增加保护，防止空指针崩溃。
---
 gateway/model.go                            |  4 +-
 workflow/service/flow/flow_graph_builder.go | 41 +++++++++++++++++++--
 workflow/service/flow/lambda_node_util.go   |  5 +++
 3 files changed, 44 insertions(+), 6 deletions(-)

diff --git a/gateway/model.go b/gateway/model.go
index 4f75125..e1391c9 100644
--- a/gateway/model.go
+++ b/gateway/model.go
@@ -138,7 +138,7 @@ func ModelCallResult(ctx context.Context, modelId int64, responseType model.Resp
 	// 异步模型必须绑定消息主题接收结果：自动生成唯一主题，
 	// 带业务标识（bizName/modelId/sessionId）便于排查，每次调用唯一避免并发串结果
 	msgTopic := ""
-	if *responseType == *model.ResponseTypeAsync.Code() {
+	if responseType != nil && *responseType == *model.ResponseTypeAsync.Code() {
 		msgTopic = modelCallTopic(g.Cfg().MustGet(ctx, "server.name").String(), modelId, sessionId)
 	}
 
@@ -166,7 +166,7 @@ func ModelCallResult(ctx context.Context, modelId int64, responseType model.Resp
 		return nil, fmt.Errorf("创建模型任务失败:%v", res.ErrorMsg)
 	}
 	// 3. 订阅模型结果（异步模型）
-	if *responseType == *model.ResponseTypeAsync.Code() {
+	if responseType != nil && *responseType == *model.ResponseTypeAsync.Code() {
 		resultCh := make(chan *ModelCallRes, 1)
 		errCh := make(chan error, 1)
 
diff --git a/workflow/service/flow/flow_graph_builder.go b/workflow/service/flow/flow_graph_builder.go
index 9910901..651fbed 100644
--- a/workflow/service/flow/flow_graph_builder.go
+++ b/workflow/service/flow/flow_graph_builder.go
@@ -17,14 +17,47 @@ import (
 // BuildGraph 根据 FlowInfo 构建完整的 Eino Graph 拓扑
 func BuildGraph(ctx context.Context, flowContent *entity.FlowInfo) ([]entity.FlowNode, *compose.Graph[any, any]) {
 	// 注册自定义合并函数：处理 *flowDto.FlowExecutionInput 类型合并
-	// 由于 ConfigMap 是 map 引用类型，所有并行分支修改已经写入共享内存
-	// 直接返回第一个实例即可，所有修改都已经可见
 	compose.RegisterValuesMergeFunc(func(values []*flowDto.FlowExecutionInput) (*flowDto.FlowExecutionInput, error) {
 		if len(values) == 0 {
 			return nil, nil
 		}
-		// 返回第一个实例，ConfigMap 是指针，所有修改都已经写入共享数据结构
-		return values[0], nil
+		// 首次运行所有并行分支共享同一个 ConfigMap 指针，直接返回 values[0] 即可。
+		// 但续跑（ReExecute）时各分支从 checkpoint 反序列化出独立的 ConfigMap 副本，
+		// 只返回 values[0] 会丢失其他分支写入的 OutputResult（用户实测：node-8 成功的结果
+		// 在汇合节点 node-7 变 null）。以第一个为基底，把其余分支中缺失的节点输出合并进来。
+		base := values[0]
+		for _, v := range values[1:] {
+			if v == nil {
+				continue
+			}
+			for nodeId, cfg := range v.ConfigMap {
+				if cfg == nil {
+					continue
+				}
+				baseCfg, ok := base.ConfigMap[nodeId]
+				if !ok || baseCfg == nil {
+					base.ConfigMap[nodeId] = cfg
+					continue
+				}
+				if len(baseCfg.OutputResult) == 0 && len(cfg.OutputResult) > 0 {
+					baseCfg.OutputResult = cfg.OutputResult
+				}
+			}
+			// 合并已执行节点列表（按 NodeId 去重），续跑时被恢复分支的进度不丢失
+			for _, en := range v.ExecutedNodes {
+				dup := false
+				for _, b := range base.ExecutedNodes {
+					if b.NodeId == en.NodeId {
+						dup = true
+						break
+					}
+				}
+				if !dup {
+					base.ExecutedNodes = append(base.ExecutedNodes, en)
+				}
+			}
+		}
+		return base, nil
 	})
 
 	graph := compose.NewGraph[any, any](
diff --git a/workflow/service/flow/lambda_node_util.go b/workflow/service/flow/lambda_node_util.go
index 7ad2f6d..3dd5e4a 100644
--- a/workflow/service/flow/lambda_node_util.go
+++ b/workflow/service/flow/lambda_node_util.go
@@ -69,6 +69,11 @@ func ModelCallResultLambda(ctx context.Context, modelId int64, sessionId string,
 	if err != nil {
 		return nil, nil, false, fmt.Errorf("获取模型配置失败: %w", err)
 	}
+	// model-gateway 对不存在的模型返回 modelManage=null（HTTP 仍 200），零值 struct 里 ResponseType 是 nil 指针，
+	// 直接传入 ModelCallResult 会在 *responseType 处 panic，这里提前报业务错误
+	if g.IsEmpty(modelInfo.ModelManage.Id) {
+		return nil, nil, false, fmt.Errorf("模型配置不存在: modelId=%d", modelId)
+	}
 	businessParams := make(map[string]any)
 	if !g.IsEmpty(prompt) {
 		if modelInfo.ModelManage.ModelType != nil && *modelInfo.ModelManage.ModelType == model.TypeVideo {