From aae4ebe199a6f928b221ba20b626432160cda184 Mon Sep 17 00:00:00 2001
From: qhd <1766646056@qq.com>
Date: Sat, 22 Aug 2026 08:28:04 +0800
Subject: [PATCH] =?UTF-8?q?fix:=20=E4=BF=AE=E5=A4=8D=E5=B7=A5=E4=BD=9C?=
 =?UTF-8?q?=E6=B5=81=E6=89=A7=E8=A1=8C=E7=8A=B6=E6=80=81=E6=AF=94=E8=BE=83?=
 =?UTF-8?q?=E4=B8=8E=E8=90=BD=E5=BA=93=E9=80=BB=E8=BE=91?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- 解引用状态指针按值比较，避免执行记录误显示为运行中
- panic 时兜底写入失败状态，防止记录卡在 Running
- 保存上下文提前至异步执行入口，确保断连后仍能落库
- 移除结果文件地址前缀拼接，使用原始 URL
---
 workflow/service/flow/flow_ws_exec.go       | 31 ++++++++++++++-------
 workflow/service/session/session_service.go | 19 +++++++------
 2 files changed, 32 insertions(+), 18 deletions(-)

diff --git a/workflow/service/flow/flow_ws_exec.go b/workflow/service/flow/flow_ws_exec.go
index a6f4e48..54d3770 100644
--- a/workflow/service/flow/flow_ws_exec.go
+++ b/workflow/service/flow/flow_ws_exec.go
@@ -118,17 +118,27 @@ func handleExecute(ctx context.Context, conn *wsCommon.WsConnection, payload int
 
 	// 异步执行工作流（直接 goroutine，不依赖上游 workerPool 二次排队）
 	go func() {
+		// 落库用不带取消的 ctx（保留 request 值），保证前端终止/断连后记录仍能写入
+		saveCtx := context.WithoutCancel(ctx)
+		start := time.Now()
+		var execId int64
+		var execErr error
+		recorded := false
 		defer func() {
 			if r := recover(); r != nil {
 				glog.Errorf(execCtx, "workflow panic: %v", r)
+				execErr = fmt.Errorf("工作流异常: %v", r)
+				// panic 被 recover 吞掉时，正常路径的 recordWorkflow 不会执行，
+				// 在此兜底把失败状态落库，避免 exec_workflow 记录卡在 Running
+				if !recorded && !g.IsEmpty(execId) {
+					recordWorkflow(saveCtx, execId, time.Since(start), execErr)
+					recorded = true
+				}
 				_ = writeJSON(conn, &wsCommon.WsPushMsg{Type: "error", Message: "工作流异常", Error: fmt.Sprintf("%v", r)})
 			}
 		}()
 		defer conn.SetMeta("execCancel", nil)
 
-		// 落库用不带取消的 ctx（保留 request 值），保证前端终止/断连后记录仍能写入
-		saveCtx := context.WithoutCancel(ctx)
-
 		// 会话落库：前端 sessionId 对应会话已存在则复用，否则按流程名新建
 		flowName := defaultSessionName
 		if flowUser, e := flowDao.FlowUserDao.Get(saveCtx, &flowDto.GetFlowUserReq{Id: execPayload.FlowId}); e == nil && flowUser != nil && flowUser.FlowName != "" {
@@ -139,20 +149,19 @@ func handleExecute(ctx context.Context, conn *wsCommon.WsConnection, payload int
 			_ = writeJSON(conn, &wsCommon.WsPushMsg{Type: "error", Message: "工作流会话创建失败", Error: fmt.Sprintf("%v", e)})
 		}
 
-		start := time.Now()
-
 		reporter := &wsProgressReporter{conn: conn}
 		progressCtx := context.WithValue(execCtx, wsProgressCtxKey{}, reporter)
 
 		_ = writeJSON(conn, &wsCommon.WsPushMsg{Type: "ack", Message: fmt.Sprintf("开始执行工作流(共 %d 个节点)", len(execPayload.FlowContent.Nodes))})
 
-		execId, err := executeOrResume(progressCtx, conn, execPayload)
+		execId, execErr = executeOrResume(progressCtx, conn, execPayload)
 		if !g.IsEmpty(execId) {
 			glog.Infof(saveCtx, "工作流执行完成，execId: %v", execId)
-			recordWorkflow(saveCtx, execId, time.Since(start), err)
+			recordWorkflow(saveCtx, execId, time.Since(start), execErr)
+			recorded = true
 		}
-		if err != nil {
-			_ = writeJSON(conn, &wsCommon.WsPushMsg{Type: "error", Message: "工作流执行失败", Error: err.Error()})
+		if execErr != nil {
+			_ = writeJSON(conn, &wsCommon.WsPushMsg{Type: "error", Message: "工作流执行失败", Error: execErr.Error()})
 			return
 		}
 		// 成功：把本次执行保存的结果文件路径（exec_workflow_result）一并推给前端
@@ -290,7 +299,9 @@ func execute(ctx context.Context, conn *wsCommon.WsConnection, execId int64, sta
 			return
 		}
 	} else {
-		if status == flow.FlowExecutionStatusFailed.Code() {
+		// FlowExecutionStatus 是 *int8 别名，Code() 返回包级指针，直接 == 是地址比较恒为 false，
+		// 需解引用按值比较，否则复用失败记录时不会重置为 Running、也不更新 RequestParams
+		if status != nil && *status == *flow.FlowExecutionStatusFailed.Code() {
 			_, err = sessionDao.ExecWorkflowDao.Update(ctx, &sessionDto.UpdateWorkflowReq{
 				Id:            execId,
 				NodeGroupId:   nodeGroupId,
diff --git a/workflow/service/session/session_service.go b/workflow/service/session/session_service.go
index 23c722b..abb6274 100644
--- a/workflow/service/session/session_service.go
+++ b/workflow/service/session/session_service.go
@@ -190,12 +190,16 @@ func chatExecVO(c *entity.ExecChat) *sessionDto.VOSessionInfoResult {
 
 func workflowExecVO(w *entity.ExecWorkflow, resultFileUrl string) *sessionDto.VOSessionInfoResult {
 	status := 1
-	if w.Status == flow.FlowExecutionStatusFailed.Code() {
-		status = resultStatusFailed
-	} else if w.Status == flow.FlowExecutionStatusSuccess.Code() {
-		status = resultStatusSuccess
-	} else if w.Status == flow.FlowExecutionStatusCancel.Code() {
-		status = resultStatusCancel
+	// FlowExecutionStatus 是 *int8 别名，Code() 返回包级指针，直接 == 是地址比较恒为 false，
+	// 需解引用按值比较，否则所有执行记录在前端都会误显示为"运行中"
+	if w.Status != nil {
+		if *w.Status == *flow.FlowExecutionStatusFailed.Code() {
+			status = resultStatusFailed
+		} else if *w.Status == *flow.FlowExecutionStatusSuccess.Code() {
+			status = resultStatusSuccess
+		} else if *w.Status == *flow.FlowExecutionStatusCancel.Code() {
+			status = resultStatusCancel
+		}
 	}
 	return &sessionDto.VOSessionInfoResult{
 		Id:            w.Id,
@@ -265,7 +269,6 @@ func (s *sessionService) ResultList(ctx context.Context, req *sessionDto.ListWor
 		execsByDate[date] = append(execsByDate[date], e)
 	}
 
-	prefix := res.ImgAddressPrefix
 	for _, d := range dates {
 		execList := execsByDate[d]
 		if len(execList) == 0 {
@@ -283,7 +286,7 @@ func (s *sessionService) ResultList(ctx context.Context, req *sessionDto.ListWor
 				if rf.ResultFileUrl == "" {
 					continue
 				}
-				content := prefix + rf.ResultFileUrl
+				content := rf.ResultFileUrl
 				ext := flowService.GetFileTypeByPath(content)
 				suffix := outputItemSuffix(ext)
 				suffixCount[suffix]++