fix: 修复并行分支续跑输出丢失及模型空指针

合并续跑时各分支独立的 ConfigMap 副本，避免汇合节点输出丢失；并对模型配置缺失和空 responseType 增加保护，防止空指针崩溃。

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)
 

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](

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 {