feat(workflow): lambda 节点接入统一异步恢复入口

workflow/service/flow/lambda_node.go

@@ -111,7 +111,7 @@ func ModelLambda(ctx context.Context, input any) (any, error) {
 				defer wg.Done()
 				// 视频段每段失败自动重试 1 次（共 2 次尝试）；非视频保持单次调用
 				for attempt := 0; attempt < segmentGenerateMaxAttempts; attempt++ {
-					results[i], tokenRes[i], isInference[i], errs[i] = ModelCallResultLambda(ctx, nodeInput.Config.ModelConfig.ModelId, nodeInput.Global.SessionId, params, nodeInput.Config.Prompt)
+					results[i], tokenRes[i], isInference[i], errs[i] = ModelCallResultLambda(ctx, nodeInput.Config.ModelConfig.ModelId, nodeInput.Global.SessionId, params, nodeInput.Config.Prompt, nodeInput.Global.ExecutionId, nodeInput.Config.Id, idxList[i])
 					if errs[i] == nil || !segVideo {
 						break
 					}
@@ -162,7 +162,7 @@ func ModelLambda(ctx context.Context, input any) (any, error) {
 		}
 	} else {
 		for _, params := range paramsList {
-			res, modelRes, _, err := ModelCallResultLambda(ctx, nodeInput.Config.ModelConfig.ModelId, nodeInput.Global.SessionId, params, nodeInput.Config.Prompt)
+			res, modelRes, _, err := ModelCallResultLambda(ctx, nodeInput.Config.ModelConfig.ModelId, nodeInput.Global.SessionId, params, nodeInput.Config.Prompt, nodeInput.Global.ExecutionId, nodeInput.Config.Id, flowDao.FlowAsyncSegSentinel)
 			if err != nil {
 				return nil, err
 			}

workflow/service/flow/lambda_node_util.go

@@ -64,7 +64,7 @@ func Notify(taskId string, result any) {
 // ModelCallResultLambda 调用模型并返回输出内容列表，同时回传本次调用的 token/费用（*gateway.ModelCallRes）
 // 与是否推理模型（供 ModelLambda 决定分批结果是否拼接），供调用方（ModelLambda）累计写入节点执行记录
 // token_info，最后由汇总节点聚合到 exec_workflow。
-func ModelCallResultLambda(ctx context.Context, modelId int64, sessionId string, modelRequestParams map[string]any, prompt string) ([]map[string]any, *gateway.ModelCallRes, bool, error) {
+func ModelCallResultLambda(ctx context.Context, modelId int64, sessionId string, modelRequestParams map[string]any, prompt string, execId int64, nodeId string, segIdx int) ([]map[string]any, *gateway.ModelCallRes, bool, error) {
 	modelInfo, err := gateway.GetModelInfoById(ctx, &gateway.GetModelInfoByIdReq{ModelId: modelId})
 	if err != nil {
 		return nil, nil, false, fmt.Errorf("获取模型配置失败: %w", err)
@@ -84,8 +84,8 @@ func ModelCallResultLambda(ctx context.Context, modelId int64, sessionId string,
 	}
 	// 推理模型：分批调用结果需拼接为单个字段，模型类型仅网关配置携带，此处顺带判断
 	isInference := modelInfo.ModelManage.ModelType != nil && *modelInfo.ModelManage.ModelType == model.TypeInference
-	// 异步模型 msgTopic 由 gateway.ModelCallResult 在为空时自动生成（唯一、带业务标识），调用方无需管理
-	responseParams, err := gateway.ModelCallResult(ctx, modelId, modelInfo.ModelManage.ResponseType, sessionId, modelRequestParams, businessParams)
+	// 统一异步入口：提交落库 flow_async_task，崩溃恢复重订阅 msg_topic 拿回结果（同步模型直接调用，不落库）
+	responseParams, err := AsyncModelCallWithRecovery(ctx, execId, nodeId, segIdx, modelId, modelInfo.ModelManage.ResponseType, sessionId, modelRequestParams, businessParams)
 	if err != nil {
 		return nil, nil, false, err
 	}

workflow/service/flow/lambda_segment_resume_test.go

@@ -0,0 +1,33 @@
+package flow
+
+import (
+	"testing"
+
+	"ai-agent/workflow/model/entity"
+)
+
+func TestPlanSegmentResumePositionBased(t *testing.T) {
+	params := []map[string]any{{"a": 1}, {"a": 2}, {"a": 3}}
+	saved := map[int]entity.SegmentRef{0: {Key: "video", URL: "url0"}, 2: {Key: "video", URL: "url2"}}
+	idxList, needGen := planSegmentResume(params, saved)
+	if len(idxList) != 3 || len(needGen) != 3 {
+		t.Fatalf("长度应等于段数")
+	}
+	if needGen[0] || !needGen[1] || needGen[2] {
+		t.Fatalf("位置0/2已保存应复用，位置1应生成: %v", needGen)
+	}
+}
+
+func TestMergeSegmentOutputsOrder(t *testing.T) {
+	idxList := []int{0, 1, 2}
+	needGen := []bool{false, true, false}
+	newRes := [][]map[string]any{nil, {{"video": "new1"}}, nil}
+	saved := map[int]entity.SegmentRef{0: {Key: "video", URL: "url0"}, 2: {Key: "video", URL: "url2"}}
+	merged := mergeSegmentOutputs(idxList, needGen, newRes, saved)
+	if len(merged) != 3 {
+		t.Fatalf("应合并 3 条: %v", merged)
+	}
+	if merged[0]["video"] != "url0" || merged[1]["video"] != "new1" || merged[2]["video"] != "url2" {
+		t.Fatalf("合并顺序应为 url0,new1,url2: %v", merged)
+	}
+}