diff --git a/docs/superpowers/specs/2026-08-25-video-segment-resume-design.md b/docs/superpowers/specs/2026-08-25-video-segment-resume-design.md deleted file mode 100644 index 9470637..0000000 --- a/docs/superpowers/specs/2026-08-25-video-segment-resume-design.md +++ /dev/null @@ -1,127 +0,0 @@ -# 视频节点段级断点续跑设计 - -日期：2026-08-25 -状态：已与用户确认（架构/数据流/错误处理/测试） - -## 1. 背景与目标 - -工作流执行生成视频时，因模型单段时长限制，长视频被拆成多段（如 5 段）分别生成。当前失败粒度是**节点级**：任意一段失败即整个节点失败，重新执行时该节点全部段重新生成，浪费资源且失败可能重复。 - -**目标**：把失败重试粒度细化到**段级**—— - -- 执行内：每段生成失败后自动重试 1 次（共 2 次尝试） -- 跨执行：重新执行（reExecute）时只重新生成失败的段，已成功的段复用 -- 全部段齐备后按段序拼接，保证顺序 - -## 2. 现状分析 - -### 2.1 分段生成 - -- 视频生成节点统一走 `ModelLambda`（`workflow/service/flow/lambda_node.go:53`） -- `preTool=split_shots_pipeline` 前置处理器把长视频拆成 N 段，每段产出 FLAT 模型请求参数，含 `segment_index`（`split_shots_pipeline.go`） -- `ModelLambda` 并发调用模型，`results[i]` 按段序对齐（`lambda_node.go:77-109`） - -### 2.2 拼接 - -- 视频模型多段输出自动调用 `concat_videos`（`media.go`）按**列表顺序**拼接 -- `collectSegmentResults` 按 `outputRes` 列表顺序取各段 URL（`media.go:229-251`），不依赖排序 - -### 2.3 失败与重执行语义 - -- **任何段失败 → `ModelLambda` `return nil, errs[i]`**（`lambda_node.go:93-95`）→ 节点失败 → 工作流失败 -- `executeOrResume`（`flow_ws_exec.go:279`）判定：同会话+同工作流最近一次执行失败 + 本次参数与上次一致 → `reExecute`（Eino checkpoint **节点级**续跑）；否则 `execute`（`WithForceNewRun` 全新跑） -- `reExecute` 复用同一条 exec 记录（`execution_id` 不变），更新 `nodeGroupId` -- Eino checkpoint（`DbCheckPointStore`）只存节点级状态（`CompletedNodes`/`SavedFlowInput`），**不存段结果** - -## 3. 设计 - -### 3.1 架构与组件 - -#### 组件 A：`flow_segment_result` 表 + entity + DAO - -只存**成功**段的 URL（失败段不落库 → 下次执行"表里无该段"即视为需重新生成）。 - -``` -flow_segment_result - id 雪花 - execution_id int64 // exec_workflow 执行ID；reExecute 复用同一条 - node_id string // 视频生成节点 ID - segment_index int // 段序 - video_url string // 已生成成功的视频地址 - + 基础字段（tenant/creator/时间） -``` - -唯一键：`(execution_id, node_id, segment_index)`。 - -DAO 方法： -- `Save(ctx, execId, nodeId, segmentIndex, videoURL) error` — 段成功后落库（INSERT ON CONFLICT DO UPDATE） -- `ListByNode(ctx, execId, nodeId) (map[int]string, error)` — 返回 `segment_index → video_url` -- `DeleteByExecution(ctx, execId) error` — 清理指定执行的段结果，两个时机：**全新执行（forceNewRun）清旧段**、**工作流执行成功后清理** - -#### 组件 B：`FlowExecutionInput` 加 `ForceNewRun bool` - -区分"全新执行"与"断点续跑"。`reExecute` 复用同一条 exec 记录、`execution_id` 不变，单靠 id 无法区分，需显式标志。 - -- `BuildExecution`（`flow_ws_exec.go:392`）从 `forceNewRun` 参数写入 `execInput.ForceNewRun` -- `ModelLambda` 读 `nodeInput.Global.ForceNewRun`： - - `true`（全新 execute）→ 先 `DeleteByExecution`，全部段重新生成 - - `false`（reExecute）→ 复用已成功段，只生成缺失段 - -#### 组件 C：`ModelLambda` 分段视频场景改造 - -仅在 `len(paramsList) > 1` 且为视频模型时启用，非分段场景零影响。 - -``` -1. 拆段后（各段含 segment_index） -2. 若 !ForceNewRun：ListByNode → 已成功段集合 -3. 需生成段 = 全部分段 − 已成功段 -4. 并行调模型生成缺失段；每段失败自动重试 1 次 -5. 每段成功立即 Save 落库（执行中断也能保留） -6. 仍有失败段 → 节点失败（成功段已落库，供下次 reExecute 复用） -7. 全部成功 → 复用段 + 新生成段按 segment_index 升序合并 → 现有 concat 逻辑按序拼接 -``` - -顺序保证关键点：`concat_videos.collectSegmentResults` 按**列表顺序**拼接（`media.go:229`），故 `ModelLambda` 输出 `outputRes` 必须按 `segment_index` 升序排好，复用与新生混合也遵循该序。 - -### 3.2 数据流走查（以 5 段、第 4 段失败为例） - -#### 首次执行（`ForceNewRun=true`，全新 execute） - -1. `executeOrResume` → `execute` → `BuildExecution(forceNewRun=true)` 写入标志 -2. `ModelLambda`：拆段得 5 份参数 → `DeleteByExecution` 清旧 → 需生成全部 5 段 -3. 并行调模型，每段失败重试 1 次；每段成功立即 `Save(execId, nodeId, idx, url)` -4. 段 4 两次均失败 → 节点失败 → 工作流失败（段 1/2/3/5 已落库），exec_workflow 记失败 -5. 若全部段成功 → 节点成功 → 工作流继续 → `BuildExecution` 成功返回时删 checkpoint 并 `DeleteByExecution` 清理段结果 - -#### 重新执行（`ForceNewRun=false`，reExecute 断点续跑） - -1. `executeOrResume` 判定上次失败+参数一致 → `reExecute`（复用同一条 exec 记录） -2. `ModelLambda`：拆段仍得 5 段 → `!ForceNewRun` → `ListByNode` 得 `{1:url1, 2:url2, 3:url3, 5:url5}` -3. 需生成段 = 仅段 4，只调模型生成它（失败重试 1 次）→ 成功 → `Save(…, 4, url4)` -4. 合并 1-5 按段序 → `concat_videos` 按序拼接 → 节点成功，后续节点继续 -5. 工作流成功 → `BuildExecution` 返回时删 checkpoint + `DeleteByExecution` 清理段结果（下次全新执行无残留） - -### 3.3 错误处理与边界 - -| 场景 | 行为 | -|---|---| -| 拼接失败（media 挂） | 段结果已落库；节点失败，重跑复用段、只重拼接 | -| 重跑时某段仍失败 | 重试 1 次后仍失败 → 节点失败；成功段持续累积，可再点执行 | -| 参数变化 | `flowContentEqual` 判定不一致 → 走 `execute`（forceNewRun）→ 删段结果、全生成 | -| 段数变化 | reExecute 要求参数一致，段数恒定，安全 | -| 非分段/非视频节点 | 不查表、不落库、不重试，原逻辑零影响 | -| 并行段并发写表 | 每段 index 独立，唯一键 `(execution_id, node_id, segment_index)`，无冲突 | -| 手动中断/断连 | 已成功段已落库，重跑自动复用 | -| 重试次数 | 每段失败重试 1 次（共 2 次尝试），参数化可调 | -| 工作流执行成功 | `BuildExecution` 成功返回 → 删 checkpoint 并 `DeleteByExecution` 清理该执行段结果，下次执行全新生成 | - -### 3.4 测试 - -- **DAO 单测**：`Save` / `ListByNode` / `DeleteByExecution` -- **ModelLambda 段级逻辑单测**（mock 模型调用）：只调缺失段、失败重试 1 次、顺序合并 -- **集成走查**：5 段流程第 4 段失败 → 重跑只生成第 4 段 → 拼接顺序正确 -- **回归**：单段/非视频节点行为不变 - -## 4. 开放问题 - -- 已解决：段结果表数据保留策略 = **`forceNewRun` 清理旧段 + 工作流执行成功后清理**（`BuildExecution` 成功路径与删 checkpoint 并列调用 `DeleteByExecution`）；执行失败则保留，供 reExecute 复用。无残留、无需定时清理。 \ No newline at end of file diff --git a/workflow/dao/flow/flow_segment_result_dao_test.go b/workflow/dao/flow/flow_segment_result_dao_test.go deleted file mode 100644 index 16068b1..0000000 --- a/workflow/dao/flow/flow_segment_result_dao_test.go +++ /dev/null @@ -1,113 +0,0 @@ -package flow - -import ( - "context" - "os" - "testing" - "time" - - "gitea.redpowerfuture.com/red-future/common/beans" - - // 空导入：pgsql 驱动。main.go 里已有，但 go test ./workflow/dao/flow/ 单测包不经 main， - // 不加则报 "cannot find database driver for specified database type pgsql" - _ "github.com/gogf/gf/contrib/drivers/pgsql/v2" -) - -// TestFlowSegmentResultDao 段结果 DAO 集成测试：Save / ListByNode / DeleteByExecution。 -// 依赖本地 black_deacon + 用户信息注入；默认跳过，设置 AI_AGENT_TEST_DB=1 时运行。 -func TestFlowSegmentResultDao(t *testing.T) { - if os.Getenv("AI_AGENT_TEST_DB") == "" { - t.Skip("设置 AI_AGENT_TEST_DB=1 且本地 black_deacon 可用时运行") - } - // gfdb Insert/Select 钩子从 ctx 读用户信息补 tenant_id/creator 并追加租户过滤 - ctx := context.WithValue(context.Background(), "user", &beans.User{UserName: "unit_test", TenantId: 999999}) - execId := time.Now().UnixMilli() - nodeId := "seg_test_node" - - dao := FlowSegmentResultDao - if err := dao.Save(ctx, execId, nodeId, 1, "video_url", "http://host/u1.mp4"); err != nil { - t.Fatalf("Save 段1失败: %v", err) - } - if err := dao.Save(ctx, execId, nodeId, 2, "video_url", "http://host/u2.mp4"); err != nil { - t.Fatalf("Save 段2失败: %v", err) - } - - // 唯一键冲突 → 更新不报错 - if err := dao.Save(ctx, execId, nodeId, 2, "video_url", "http://host/u2-new.mp4"); err != nil { - t.Fatalf("Save 段2(冲突更新)失败: %v", err) - } - - m, err := dao.ListByNode(ctx, execId, nodeId) - if err != nil { - t.Fatalf("ListByNode失败: %v", err) - } - if len(m) != 2 { - t.Fatalf("期望 2 段，实际 %d", len(m)) - } - if m[1].URL != "http://host/u1.mp4" || m[2].URL != "http://host/u2-new.mp4" { - t.Fatalf("段内容不符: %+v", m) - } - if m[2].Key != "video_url" { - t.Fatalf("段2 key 不符: %s", m[2].Key) - } - - if err := dao.DeleteByExecution(ctx, execId); err != nil { - t.Fatalf("DeleteByExecution失败: %v", err) - } - m, err = dao.ListByNode(ctx, execId, nodeId) - if err != nil { - t.Fatalf("删除后 ListByNode失败: %v", err) - } - if len(m) != 0 { - t.Fatalf("删除后应无段结果，实际 %d", len(m)) - } -} - -// TestFlowSegmentResultDaoReuseAfterCleanup 硬删→重存的"续跑复用"保证。 -// DeleteByExecution 必须物理删除：实体嵌入 beans.SQLBaseDO（含 deleted_at），若走 -// gfdb Model.Delete() 会退化为软删（UPDATE ... SET deleted_at=now()），而后续 Save -// 的 ON CONFLICT ... DO UPDATE SET 不含 deleted_at（OmitNil 丢弃 nil），软删行无法复活， -// ListByNode 永远看不到 → 每次重执行都全量重生成。本测试验证：清理（物理删两行）后 -// 对同一 (execution_id, node_id, segment_index) 重存段2，仅段2可见（len==1）且为更新后 URL。 -func TestFlowSegmentResultDaoReuseAfterCleanup(t *testing.T) { - if os.Getenv("AI_AGENT_TEST_DB") == "" { - t.Skip("设置 AI_AGENT_TEST_DB=1 且本地 black_deacon 可用时运行") - } - ctx := context.WithValue(context.Background(), "user", &beans.User{UserName: "unit_test", TenantId: 999999}) - execId := time.Now().UnixMilli() - nodeId := "seg_test_cleanup" - dao := FlowSegmentResultDao - - // 物理残留由硬清理（docker exec psql DELETE FROM）兜底 - t.Cleanup(func() { _ = dao.DeleteByExecution(ctx, execId) }) - - if err := dao.Save(ctx, execId, nodeId, 1, "video_url", "http://host/u1-cleanup.mp4"); err != nil { - t.Fatalf("Save 段1失败: %v", err) - } - if err := dao.Save(ctx, execId, nodeId, 2, "video_url", "http://host/u2-cleanup.mp4"); err != nil { - t.Fatalf("Save 段2失败: %v", err) - } - - // 物理删除整条执行（DELETE FROM ... WHERE execution_id=?） - if err := dao.DeleteByExecution(ctx, execId); err != nil { - t.Fatalf("DeleteByExecution失败: %v", err) - } - - // 清理后重新保存段2（同一 exec/node/idx，不同 URL）： - // 物理删除已移除该行，Save 走纯 INSERT，段2 在 ListByNode 中可见且为更新后的 URL - if err := dao.Save(ctx, execId, nodeId, 2, "video_url", "http://host/u2-cleanup-new.mp4"); err != nil { - t.Fatalf("Save 段2(清理后重存)失败: %v", err) - } - - m, err := dao.ListByNode(ctx, execId, nodeId) - if err != nil { - t.Fatalf("ListByNode失败: %v", err) - } - // 期望：段1 已物理删除，仅段2 可见（len==1）且为更新后的 URL - if len(m) != 1 { - t.Fatalf("硬删后重存应只剩段2(len=1)，实际 len=%d: %+v", len(m), m) - } - if m[2].URL != "http://host/u2-cleanup-new.mp4" { - t.Fatalf("段2 应为更新后的 URL，实际: %+v", m[2]) - } -} diff --git a/workflow/model/dto/flow/flow_user_dto.go b/workflow/model/dto/flow/flow_user_dto.go index 8d64dce..ef4357f 100644 --- a/workflow/model/dto/flow/flow_user_dto.go +++ b/workflow/model/dto/flow/flow_user_dto.go @@ -34,6 +34,7 @@ type UpdateFlowUserReq struct { NodeInputParams []*entity.FlowNode `json:"nodeInputParams" description:"节点输入参数"` AccessLevel flow.FlowUserAccessLevel `json:"accessLevel" description:"访问权限：1私有，2团队，3公开"` SourceFlowTemplateId int64 `json:"sourceFlowTemplateId" description:"来源流程模板ID"` + SubFlows []UpdateFlowUserReq `json:"subFlows" description:"子流程"` } type DeleteFlowUserReq struct { diff --git a/workflow/model/entity/flow_user.go b/workflow/model/entity/flow_user.go index 758fce6..86cb830 100644 --- a/workflow/model/entity/flow_user.go +++ b/workflow/model/entity/flow_user.go @@ -65,7 +65,7 @@ type ValueSource struct { // SubFlowConfig 子流程节点配置 type SubFlowConfig struct { - WorkflowId int64 `json:"workflowId"` + WorkflowId int64 `json:"workflowId,string"` WorkflowName string `json:"workflowName"` Fields []map[string]any `json:"fields"` MaxConcurrency int `json:"maxConcurrency"` // 子流程并发数 diff --git a/workflow/service/flow/flow_user_service.go b/workflow/service/flow/flow_user_service.go index 858b21f..4507936 100644 --- a/workflow/service/flow/flow_user_service.go +++ b/workflow/service/flow/flow_user_service.go @@ -2,6 +2,7 @@ package flow import ( "ai-agent/workflow/consts/flow" + "ai-agent/workflow/consts/public" flowDao "ai-agent/workflow/dao/flow" flowDto "ai-agent/workflow/model/dto/flow" "ai-agent/workflow/model/entity" @@ -9,7 +10,9 @@ import ( "context" "gitea.redpowerfuture.com/red-future/common/beans" + "gitea.redpowerfuture.com/red-future/common/db/gfdb" "gitea.redpowerfuture.com/red-future/common/utils" + "github.com/gogf/gf/v2/database/gdb" "github.com/gogf/gf/v2/frame/g" "github.com/gogf/gf/v2/util/gconv" ) @@ -65,15 +68,35 @@ func (s *flowUserService) Update(ctx context.Context, req *flowDto.UpdateFlowUse } if !g.IsEmpty(get) { - id, err = flowDao.FlowUserDao.Insert(ctx, &flowDto.CreateFlowUserReq{ - FlowName: req.FlowName, - Description: req.Description, - FlowContent: req.FlowContent, - NodeInputParams: req.NodeInputParams, - SourceFlowTemplateId: get.Id, + // 模版 → 用户流程拷贝（含递归子流程拷贝）整体放一个事务：任一拷贝失败则整体回滚， + // 避免主流程已落库而子流程缺失/引用错乱的脏数据。 + // Transaction 会把 tx 注入回调 ctx，回调内用该 ctx 调 DAO 即自动进入事务。 + txErr := gfdb.DB(ctx, public.DbNameBlackDeacon).Transaction(ctx, func(ctx context.Context, tx gdb.TX) error { + // 子流程引用重写：模版里的 sub_flow 节点指向子模版/子流程 id， + // 拷贝成用户自己的流程时要把子工作流也复制一份（子流程可能再嵌套子流程，递归）， + // 并把主流程 SubConfig.WorkflowId 改写为新拷贝的 id + subIdMap, copyErr := copySubFlows(ctx, req.SubFlows) + if copyErr != nil { + return copyErr + } + rewriteSubFlowWorkflowIds(req.FlowContent, subIdMap) + // SubConfig 改写后重新提取节点参数，保证落库的 NodeInputParams 与 FlowContent 一致 + req.NodeInputParams = ExtractFlowNodeFrom(req.FlowContent) + newId, insertErr := flowDao.FlowUserDao.Insert(ctx, &flowDto.CreateFlowUserReq{ + FlowName: req.FlowName, + Description: req.Description, + FlowContent: req.FlowContent, + NodeInputParams: req.NodeInputParams, + SourceFlowTemplateId: get.Id, + }) + if insertErr != nil { + return insertErr + } + id = newId + return nil }) - if err != nil { - return + if txErr != nil { + return nil, txErr } } else { _, err = flowDao.FlowUserDao.Update(ctx, req) @@ -196,3 +219,58 @@ func (s *flowUserService) List(ctx context.Context, req *flowDto.ListFlowUserReq return } + +// copySubFlows 递归拷贝子工作流（子流程可能再包含子流程），返回 旧id→新id 映射。 +// 映射键是子工作流在来源（模版）中的 id，即父流程 sub_flow 节点 SubConfig.WorkflowId 指向的值； +// 拷贝时把子工作流落为当前用户自己的流程（SourceFlowTemplateId 记录来源）。 +// 同一子流程被多个节点引用时只拷贝一份，避免产生孤儿副本。 +func copySubFlows(ctx context.Context, subs []flowDto.UpdateFlowUserReq) (map[int64]int64, error) { + idMap := make(map[int64]int64) + for i := range subs { + sub := subs[i] + if _, done := idMap[sub.Id]; done { + continue + } + nestedMap, err := copySubFlows(ctx, sub.SubFlows) + if err != nil { + return nil, err + } + for k, v := range nestedMap { + idMap[k] = v + } + // 改写本子流程对更深层子流程的引用后，再落库为新的用户流程 + rewriteSubFlowWorkflowIds(sub.FlowContent, idMap) + var nodeInputParams []*entity.FlowNode + if sub.FlowContent != nil { + nodeInputParams = ExtractFlowNodeFrom(sub.FlowContent) + } + newId, err := flowDao.FlowUserDao.Insert(ctx, &flowDto.CreateFlowUserReq{ + FlowName: sub.FlowName, + Description: sub.Description, + FlowContent: sub.FlowContent, + NodeInputParams: nodeInputParams, + SourceFlowTemplateId: sub.Id, + }) + if err != nil { + return nil, err + } + idMap[sub.Id] = newId + } + return idMap, nil +} + +// rewriteSubFlowWorkflowIds 把 flowContent 中 sub_flow 节点的 WorkflowId 按 idMap 改写为新拷贝的流程 id +func rewriteSubFlowWorkflowIds(flowContent *entity.FlowInfo, idMap map[int64]int64) { + if flowContent == nil { + return + } + for i := range flowContent.Nodes { + n := &flowContent.Nodes[i] + if n.SubConfig == nil || n.SubConfig.WorkflowId == 0 { + continue + } + if newId, ok := idMap[n.SubConfig.WorkflowId]; ok { + n.SubConfig.WorkflowId = newId + } + } +} diff --git a/workflow/service/flow/lambda_node.go b/workflow/service/flow/lambda_node.go index 7bb77d3..d37b651 100644 --- a/workflow/service/flow/lambda_node.go +++ b/workflow/service/flow/lambda_node.go @@ -77,12 +77,13 @@ func ModelLambda(ctx context.Context, input any) (any, error) { var totalCost float64 if len(paramsList) > 1 { // 段级续跑仅在"多段 + 视频模型"启用；非视频分段（批量文本等）走原逻辑零影响。 - // 段级复用/落库还要求各段带互不重复的 segment_index（目前仅 split_shots_pipeline 产出）： - // 视频节点若被批量拆分（如 split_batch_model_params），paramsList>1 但无 segment_index， - // 此时不得启用段级续跑，否则续跑会拿同一段结果拼出 N 段重复视频（静默损坏）。 - // segVideo 为 false 时走既有非段级合并路径（全量生成、不落库、不复用），全新执行行为不变， + // 段身份 = 列表位置（0-based）：paramsList 顺序即段序，concat 按列表顺序拼接； + // 位置互不重复且跨 reExecute 稳定（参数一致 → 段数/顺序不变）。不依赖 params 里的 + // segment_index——真实链路（上游 split_shots_pipeline 转写 → 下游 split_segment 按 + // __segment_fields 拆分，invokePreTool 剥离 __ 内部键）下 paramsList 只有模型参数。 + // segVideo=false 时走既有非段级合并路径（全量生成、不落库、不复用），全新执行行为不变， // 后续自动 concat 判断（独立的 isVideoModel 调用）仍正常执行。 - segVideo := isVideoModel(ctx, nodeInput.Config.ModelConfig.ModelId) && hasDistinctSegmentIndex(paramsList) + segVideo := isVideoModel(ctx, nodeInput.Config.ModelConfig.ModelId) // 续跑（!ForceNewRun）时读取该节点已成功段；全新执行不查（BuildExecution 已清旧段），saved 为 nil → 全量重生成 var saved map[int]entity.SegmentRef @@ -98,6 +99,7 @@ func ModelLambda(ctx context.Context, input any) (any, error) { results := make([][]map[string]any, len(paramsList)) tokenRes := make([]*gateway.ModelCallRes, len(paramsList)) errs := make([]error, len(paramsList)) + saveErrs := make([]error, len(paramsList)) isInference := make([]bool, len(paramsList)) var wg sync.WaitGroup for i, params := range paramsList { @@ -114,31 +116,29 @@ func ModelLambda(ctx context.Context, input any) (any, error) { break } } + // 每段成功立即落库：该段刚成功即持久化，其他段仍在跑/重试时已成功段也不丢； + // 后续段失败或进程崩溃（panic/OOM/kill）时，已完成段已在库中，reExecute 可直接复用 + if segVideo && errs[i] == nil { + for _, rec := range results[i] { + key := media.FindVideoKey(rec) + url := media.FindVideoURL(ctx, rec) + if key == "" || url == "" { + continue + } + if err := flowDao.FlowSegmentResultDao.Save(ctx, nodeInput.Global.ExecutionId, nodeInput.Config.Id, idxList[i], key, url); err != nil { + saveErrs[i] = err + } + } + } }(i, params) } wg.Wait() - // 每段成功立即落库：即使后续某段失败导致节点失败，成功段也保留供下次续跑复用 - if segVideo { - for i := range results { - if !needGen[i] || errs[i] != nil { - continue - } - for _, rec := range results[i] { - key := media.FindVideoKey(rec) - url := media.FindVideoURL(ctx, rec) - if key == "" || url == "" { - continue - } - if err := flowDao.FlowSegmentResultDao.Save(ctx, nodeInput.Global.ExecutionId, nodeInput.Config.Id, idxList[i], key, url); err != nil { - return nil, err - } - } - } - } - - // 仍有失败段 → 节点失败（成功段已落库，供下次 reExecute 复用） + // 仍有失败段或落库失败 → 节点失败（成功段已立即落库，供下次 reExecute 复用） for i := range results { + if saveErrs[i] != nil { + return nil, saveErrs[i] + } if needGen[i] && errs[i] != nil { return nil, errs[i] } diff --git a/workflow/service/flow/lambda_segment_resume.go b/workflow/service/flow/lambda_segment_resume.go index 52d1798..92a6fff 100644 --- a/workflow/service/flow/lambda_segment_resume.go +++ b/workflow/service/flow/lambda_segment_resume.go @@ -4,42 +4,24 @@ import ( "sort" "ai-agent/workflow/model/entity" - - "github.com/gogf/gf/v2/util/gconv" ) // segmentGenerateMaxAttempts 视频段生成最大尝试次数（失败自动重试 1 次，共 2 次尝试），参数化可调 const segmentGenerateMaxAttempts = 2 -// hasDistinctSegmentIndex 判断各段参数是否都带 segment_index 且互不重复。 -// 段级续跑依赖 segment_index 作为段的稳定身份；缺失或重复（如视频节点误用批量拆分）时 -// 不得启用复用/落库，否则续跑会拿同一段结果拼出 N 段重复视频（静默损坏）。 -func hasDistinctSegmentIndex(paramsList []map[string]any) bool { - seen := make(map[int]bool, len(paramsList)) - for _, params := range paramsList { - v, ok := params["segment_index"] - if !ok { - return false - } - idx := gconv.Int(v) - if seen[idx] { - return false - } - seen[idx] = true - } - return len(paramsList) > 0 -} - -// planSegmentResume 段级续跑决策：把 paramsList 各段映射到"是否需重新生成"。 -// savedMap 为该节点已成功段（段序号 → {key,url}）；段在表中缺失或地址为空则需重新生成。 -// 返回值与 paramsList 对齐。全新执行（savedMap 为 nil/空）时全部需生成。 +// planSegmentResume 段级续跑决策：段身份取列表位置（0-based，paramsList 顺序即段序）， +// 把各段映射到"是否需重新生成"。savedMap 为该节点已成功段（段序号 → {key,url}）； +// 段在表中缺失或地址为空则需重新生成。返回值与 paramsList 对齐。 +// 全新执行（savedMap 为 nil/空）时全部需生成。 +// 不用 params["segment_index"] 作为段身份：真实链路（上游 split_shots_pipeline 转写 → +// 下游 split_segment 按 __segment_fields 拆分，invokePreTool 剥离 __ 内部键）下 +// paramsList 只有模型参数；列表位置互不重复、顺序即段序、参数一致时跨 reExecute 稳定。 func planSegmentResume(paramsList []map[string]any, savedMap map[int]entity.SegmentRef) (idxList []int, needGen []bool) { idxList = make([]int, len(paramsList)) needGen = make([]bool, len(paramsList)) - for i, params := range paramsList { - idx := gconv.Int(params["segment_index"]) - idxList[i] = idx - ref, ok := savedMap[idx] + for i := range paramsList { + idxList[i] = i + ref, ok := savedMap[i] needGen[i] = !ok || ref.URL == "" } return diff --git a/workflow/service/flow/lambda_segment_resume_test.go b/workflow/service/flow/lambda_segment_resume_test.go deleted file mode 100644 index a8df876..0000000 --- a/workflow/service/flow/lambda_segment_resume_test.go +++ /dev/null @@ -1,124 +0,0 @@ -package flow - -import ( - "testing" - - "ai-agent/workflow/model/entity" -) - -// hasDistinctSegmentIndex：各段都带 segment_index 且互不重复 → 允许段级续跑 -func TestHasDistinctSegmentIndex(t *testing.T) { - distinct := []map[string]any{{"segment_index": 1}, {"segment_index": 2}, {"segment_index": 3}} - if !hasDistinctSegmentIndex(distinct) { - t.Fatal("互不重复的 segment_index 应返回 true") - } - missing := []map[string]any{{"segment_index": 1}, {"other": 2}} - if hasDistinctSegmentIndex(missing) { - t.Fatal("缺失 segment_index 应返回 false") - } - duplicate := []map[string]any{{"segment_index": 1}, {"segment_index": 1}} - if hasDistinctSegmentIndex(duplicate) { - t.Fatal("重复 segment_index 应返回 false") - } - if hasDistinctSegmentIndex(nil) { - t.Fatal("空列表应返回 false") - } - if hasDistinctSegmentIndex([]map[string]any{}) { - t.Fatal("空列表(非 nil)应返回 false") - } -} - -// planSegmentResume：段序号从 params 的 segment_index 读出；已成功段复用，缺失/空地址需生成 -func TestPlanSegmentResume(t *testing.T) { - paramsList := []map[string]any{ - {"segment_index": 1}, {"segment_index": 2}, {"segment_index": 3}, {"segment_index": 4}, {"segment_index": 5}, - } - saved := map[int]entity.SegmentRef{ - 1: {Key: "video_url", URL: "u1"}, 2: {Key: "video_url", URL: "u2"}, - 3: {Key: "video_url", URL: "u3"}, 5: {Key: "video_url", URL: "u5"}, - } - idxList, needGen := planSegmentResume(paramsList, saved) - if len(idxList) != 5 || len(needGen) != 5 { - t.Fatalf("长度不符: idxList=%v needGen=%v", idxList, needGen) - } - want := []bool{false, false, false, true, false} // 仅段4需重新生成 - for i := range want { - if needGen[i] != want[i] { - t.Fatalf("needGen[%d]=%v 期望 %v", i, needGen[i], want[i]) - } - if idxList[i] != i+1 { - t.Fatalf("idxList[%d]=%d 期望 %d", i, idxList[i], i+1) - } - } -} - -// planSegmentResume：无已成功段（全新执行）→ 全部需生成 -func TestPlanSegmentResumeEmptySaved(t *testing.T) { - paramsList := []map[string]any{{"segment_index": 0}, {"segment_index": 1}} - _, needGen := planSegmentResume(paramsList, nil) - for i := range needGen { - if !needGen[i] { - t.Fatalf("needGen[%d] 期望 true（无复用）", i) - } - } -} - -// mergeSegmentOutputs：复用段 + 新生段按段序号升序，顺序即拼接顺序 -func TestMergeSegmentOutputs(t *testing.T) { - idxList := []int{1, 2, 3, 4, 5} - needGen := []bool{false, false, false, true, false} - newRes := make([][]map[string]any, 5) - newRes[3] = []map[string]any{{"video_url": "new4"}} - saved := map[int]entity.SegmentRef{ - 1: {Key: "video_url", URL: "u1"}, 2: {Key: "video_url", URL: "u2"}, - 3: {Key: "video_url", URL: "u3"}, 5: {Key: "video_url", URL: "u5"}, - } - got := mergeSegmentOutputs(idxList, needGen, newRes, saved) - if len(got) != 5 { - t.Fatalf("合并后应有 5 段，实际 %d: %v", len(got), got) - } - wantURLs := []string{"u1", "u2", "u3", "new4", "u5"} - for i, rec := range got { - if rec["video_url"] != wantURLs[i] { - t.Fatalf("第 %d 段=%v 期望 %s", i, rec, wantURLs[i]) - } - } -} - -// mergeSegmentOutputs：复用段 key 保持模型原字段，重建记录与新生记录 key 不一致也不影响顺序 -func TestMergeSegmentOutputsMixedKeys(t *testing.T) { - idxList := []int{1, 2, 3} - needGen := []bool{false, true, false} - newRes := make([][]map[string]any, 3) - newRes[1] = []map[string]any{{"video_oss_url": "new2"}} - saved := map[int]entity.SegmentRef{ - 1: {Key: "video_url", URL: "u1"}, 3: {Key: "file_url", URL: "u3"}, - } - got := mergeSegmentOutputs(idxList, needGen, newRes, saved) - if len(got) != 3 { - t.Fatalf("合并后应有 3 段，实际 %d: %v", len(got), got) - } - if got[0]["video_url"] != "u1" || got[1]["video_oss_url"] != "new2" || got[2]["file_url"] != "u3" { - t.Fatalf("合并结果不符: %v", got) - } -} - -// mergeSegmentOutputs：paramsList 乱序时仍按段序号升序输出（顺序保证） -func TestMergeSegmentOutputsUnorderedParams(t *testing.T) { - idxList := []int{3, 1, 2} - needGen := []bool{true, false, true} - newRes := make([][]map[string]any, 3) - newRes[0] = []map[string]any{{"video_url": "new3"}} - newRes[2] = []map[string]any{{"video_url": "new2"}} - saved := map[int]entity.SegmentRef{1: {Key: "video_url", URL: "u1"}} - got := mergeSegmentOutputs(idxList, needGen, newRes, saved) - if len(got) != 3 { - t.Fatalf("合并后应有 3 段，实际 %d: %v", len(got), got) - } - wantURLs := []string{"u1", "new2", "new3"} - for i, rec := range got { - if rec["video_url"] != wantURLs[i] { - t.Fatalf("第 %d 段=%v 期望 %s（应段序升序）", i, rec, wantURLs[i]) - } - } -}