diff --git a/workflow/service/flow/flow_ws_exec.go b/workflow/service/flow/flow_ws_exec.go
index 0a3cb76..3f1b6eb 100644
--- a/workflow/service/flow/flow_ws_exec.go
+++ b/workflow/service/flow/flow_ws_exec.go
@@ -156,6 +156,33 @@ func handleExecute(ctx context.Context, conn *wsCommon.WsConnection, payload int
 		_ = writeJSON(conn, &wsCommon.WsPushMsg{Type: "ack", Message: fmt.Sprintf("开始执行工作流(共 %d 个节点)", len(execPayload.FlowContent.Nodes))})
 
 		execId, execErr = executeOrResume(progressCtx, conn, execPayload)
+		if errors.Is(execErr, errExecAlreadyRunning) {
+			return // 运行中/已触发恢复：不写终态（状态由后台恢复收敛）
+		}
+		// in-process 重试：程序报错且未耗尽 → retry_count++ 落库（保持 status=1，前端不闪失败）→ 续跑
+		retryCount := 0
+		for execErr != nil && execId > 0 && shouldRetry(execErr) && retryCount < execMaxRetryCount {
+			retryCount++
+			if err := sessionDao.ExecWorkflowDao.UpdateRetry(saveCtx, execId, 1, retryCount); err != nil {
+				glog.Errorf(saveCtx, "重试标记落库失败 execId=%d: %v", execId, err)
+				break
+			}
+			glog.Infof(saveCtx, "工作流执行失败，自动重试 %d/%d，execId=%d: %v", retryCount, execMaxRetryCount, execId, execErr)
+			_, execErr = reExecute(progressCtx, execId)
+		}
+		// 错误分类落库：用户取消 → retryable=0 不重试；
+		// 程序报错（非取消）→ retryable=1，retry_count 已记（重试循环内递增；UpdateRetry 失败 break 时也为当前值）
+		if execErr != nil {
+			if errors.Is(execErr, context.Canceled) {
+				_ = sessionDao.ExecWorkflowDao.UpdateRetry(saveCtx, execId, 0, 0)
+			} else {
+				_ = sessionDao.ExecWorkflowDao.UpdateRetry(saveCtx, execId, 1, retryCount)
+				if retryCount >= execMaxRetryCount {
+					// 终局清理（Task 5 裁定）：重试耗尽，exec 永久失败，清 flow_async_task 孤儿缓存
+					_ = flowDao.FlowAsyncTaskDao.DeleteByExecution(saveCtx, execId)
+				}
+			}
+		}
 		if !g.IsEmpty(execId) {
 			glog.Infof(saveCtx, "工作流执行完成，execId: %v", execId)
 			recordWorkflow(saveCtx, execId, time.Since(start), execErr)
@@ -283,6 +310,19 @@ func executeOrResume(ctx context.Context, conn *wsCommon.WsConnection, req *sess
 		return 0, fmt.Errorf("查询最近工作流执行记录失败: %v", err)
 	}
 	if lastExec != nil {
+		if *lastExec.Status == *flow.FlowExecutionStatusRunning.Code() {
+			// status=1：可能正在跑（本节点或其它节点后台恢复）或僵尸遗留。
+			// 心跳新鲜 → 真在跑，不新建避免双跑；心跳陈旧 → 僵尸，触发后台恢复。
+			// 都不新建执行，恢复在后台完成，完成后状态自然收敛（spec §9）。
+			nowMs := time.Now().UnixMilli()
+			if lastExec.LastHeartbeat < nowMs-int64(heartbeatStaleAfter/time.Millisecond) {
+				_ = writeJSON(conn, &wsCommon.WsPushMsg{Type: "round_start", Message: "检测到未完成执行，正在恢复", Data: map[string]interface{}{"id": lastExec.Id}})
+				go recoverExecution(context.WithoutCancel(ctx), lastExec.Id)
+			} else {
+				_ = writeJSON(conn, &wsCommon.WsPushMsg{Type: "round_start", Message: "工作流正在执行中", Data: map[string]interface{}{"id": lastExec.Id}})
+			}
+			return lastExec.Id, errExecAlreadyRunning
+		}
 		if *lastExec.Status == *flow.FlowExecutionStatusFailed.Code() && flowContentEqual(lastExec.RequestParams, req.FlowContent) {
 			_ = writeJSON(conn, &wsCommon.WsPushMsg{Type: "round_start", Message: "运行开始", Data: map[string]interface{}{
 				"id": lastExec.Id,
@@ -322,6 +362,7 @@ func execute(ctx context.Context, conn *wsCommon.WsConnection, execId int64, sta
 			NodeGroupId:   nodeGroupId,
 			Status:        flow.FlowExecutionStatusRunning.Code(),
 			RequestParams: req.FlowContent,
+			LastHeartbeat: time.Now().UnixMilli(),
 		})
 		if err != nil || g.IsEmpty(execId) {
 			glog.Errorf(ctx, "工作流执行记录创建失败: %v", err)
@@ -332,12 +373,12 @@ func execute(ctx context.Context, conn *wsCommon.WsConnection, execId int64, sta
 		// 需解引用按值比较，否则复用失败记录时不会重置为 Running、也不更新 RequestParams
 		if status != nil && *status == *flow.FlowExecutionStatusFailed.Code() {
 			glog.Infof(ctx, "工作流断点续跑execute，execId: %v", execId)
-			_, err = sessionDao.ExecWorkflowDao.Update(ctx, &sessionDto.UpdateWorkflowReq{
-				Id:            execId,
-				NodeGroupId:   nodeGroupId,
-				Status:        flow.FlowExecutionStatusRunning.Code(),
-				RequestParams: req.FlowContent,
-			})
+			err = sessionDao.ExecWorkflowDao.ResetRunning(ctx, execId, nodeGroupId)
+			if err != nil {
+				return
+			}
+			// 复用失败记录时参数可能已变：把新参数落库，供后续 reExecute/恢复例程按记录参数续跑
+			_, err = sessionDao.ExecWorkflowDao.Update(ctx, &sessionDto.UpdateWorkflowReq{Id: execId, RequestParams: req.FlowContent})
 			if err != nil {
 				return
 			}
@@ -360,6 +401,8 @@ func execute(ctx context.Context, conn *wsCommon.WsConnection, execId int64, sta
 	_ = writeJSON(conn, &wsCommon.WsPushMsg{Type: "round_start", Message: "运行开始", Data: map[string]interface{}{
 		"id": execId,
 	}})
+	stop := startHeartbeat(ctx, execId)
+	defer stop()
 	err = BuildExecution(ctx, true, req.FlowId, execId, nodeGroupId, conn.SessionId, req.FlowContent)
 	if err != nil {
 		return
@@ -374,14 +417,12 @@ func reExecute(ctx context.Context, execWorkflowId int64) (id int64, err error)
 		return
 	}
 	var nodeGroupId = uuid.NewString()
-	_, err = sessionDao.ExecWorkflowDao.Update(ctx, &sessionDto.UpdateWorkflowReq{
-		Id:          flowInfo.Id,
-		NodeGroupId: nodeGroupId,
-		Status:      flow.FlowExecutionStatusRunning.Code(),
-	})
+	err = sessionDao.ExecWorkflowDao.ResetRunning(ctx, flowInfo.Id, nodeGroupId)
 	if err != nil {
 		return
 	}
+	stop := startHeartbeat(ctx, flowInfo.Id)
+	defer stop()
 	err = BuildExecution(ctx, false, flowInfo.FlowId, flowInfo.Id, nodeGroupId, flowInfo.SessionId, flowInfo.RequestParams)
 	if err != nil {
 		return
@@ -468,5 +509,7 @@ func BuildExecution(ctx context.Context, forceNewRun bool, flowId, executionId i
 	_ = flowDao.FlowCheckpointDao.Delete(ctx, gconv.String(executionId))
 	// 清理该执行段结果，下次执行无残留（失败则保留，供 reExecute 复用）
 	_ = flowDao.FlowSegmentResultDao.DeleteByExecution(ctx, executionId)
+	// 清理该执行异步任务缓存，下次执行无残留（失败则保留，供恢复复用）
+	_ = flowDao.FlowAsyncTaskDao.DeleteByExecution(ctx, executionId)
 	return
 }