diff --git a/workflow/model/dto/flow/flow_execution_dto.go b/workflow/model/dto/flow/flow_execution_dto.go index 98f4eb1..5330cb0 100644 --- a/workflow/model/dto/flow/flow_execution_dto.go +++ b/workflow/model/dto/flow/flow_execution_dto.go @@ -33,6 +33,7 @@ type FlowExecutionInput struct { ConfigMap map[string]*entity.FlowNode `json:"configMap"` SessionId string `json:"sessionId" dc:"会话ID"` ExecutedNodes []ExecutedNode `json:"executedNodes"` // 已执行节点列表,包含执行状态 + ForceNewRun bool `json:"forceNewRun" dc:"是否全新执行(false=断点续跑,视频节点段级复用已成功段)"` } // ExecutedNode 已执行节点记录,包含节点ID和执行状态 diff --git a/workflow/service/flow/flow_graph_util.go b/workflow/service/flow/flow_graph_util.go index 01bf692..d752443 100644 --- a/workflow/service/flow/flow_graph_util.go +++ b/workflow/service/flow/flow_graph_util.go @@ -39,6 +39,9 @@ func BuildNodeExecutionInput(ctx context.Context, input any, flowNode entity.Flo return nil, nil, fmt.Errorf("节点:%v 进程状态为空,节点入参参数为空", flowNode.Name) } } + // 续跑必定非全新执行:checkpoint 恢复的 SavedFlowInput 里 ForceNewRun 是上次(fresh)执行留下的 true, + // 不清则 ModelLambda 误走"清段重生成"而非复用已成功段 + execInput.ForceNewRun = false } else { var ok bool execInput, ok = input.(*flowDto.FlowExecutionInput) diff --git a/workflow/service/flow/flow_ws_exec.go b/workflow/service/flow/flow_ws_exec.go index 81c86d5..0a3cb76 100644 --- a/workflow/service/flow/flow_ws_exec.go +++ b/workflow/service/flow/flow_ws_exec.go @@ -421,6 +421,7 @@ func BuildExecution(ctx context.Context, forceNewRun bool, flowId, executionId i FlowId: flowId, ConfigMap: configMap, SessionId: sessionId, + ForceNewRun: forceNewRun, } var opts []compose.Option @@ -428,6 +429,13 @@ func BuildExecution(ctx context.Context, forceNewRun bool, flowId, executionId i if forceNewRun { opts = append(opts, compose.WithForceNewRun()) } + // 全新执行前清理该执行残留段结果:forceNewRun 复用同一条 exec 记录时可能留有旧参数生成的段, + // 不清则续跑会误复用。只按 execution_id 清理(放在图启动前,避免多视频节点互相误删) + if forceNewRun { + if err := flowDao.FlowSegmentResultDao.DeleteByExecution(ctx, executionId); err != nil { + return fmt.Errorf("清理段结果失败: %v", err) + } + } _, err = runGraph.Invoke(ctx, execInput, opts...) if err != nil { // 图执行被 ctx 取消(WS 断连/用户终止):返回 context.Canceled 语义,让 recordWorkflow @@ -458,5 +466,7 @@ func BuildExecution(ctx context.Context, forceNewRun bool, flowId, executionId i } // 清理断点数据 _ = flowDao.FlowCheckpointDao.Delete(ctx, gconv.String(executionId)) + // 清理该执行段结果,下次执行无残留(失败则保留,供 reExecute 复用) + _ = flowDao.FlowSegmentResultDao.DeleteByExecution(ctx, executionId) return }