Merge pull request #476 from chaitin/fix/handle-processing-duplicate-vmready

fix: 修复 Agent 重连导致运行中任务异常变为 error

Author: yokowu

backend/biz/host/handler/v1/internal.go

@@ -29,17 +29,17 @@ import (
 
 // InternalHostHandler 处理 taskflow 回调的 host/VM 相关接口
 type InternalHostHandler struct {
-	logger          *slog.Logger
-	repo            domain.HostRepo
-	teamRepo        domain.TeamHostRepo
-	redis           *redis.Client
-	cache           *cache.Cache
-	hook            domain.InternalHook // 可选，由内部项目通过 WithInternalHook 注入
-	taskLifecycle   *lifecycle.Manager[uuid.UUID, consts.TaskStatus, lifecycle.TaskMetadata]
-	hostUsecase     domain.HostUsecase
-	taskConns       *ws.TaskConn
-	projectUsecase  domain.ProjectUsecase
-	tokenProvider   *gituc.TokenProvider
+	logger         *slog.Logger
+	repo           domain.HostRepo
+	teamRepo       domain.TeamHostRepo
+	redis          *redis.Client
+	cache          *cache.Cache
+	hook           domain.InternalHook // 可选，由内部项目通过 WithInternalHook 注入
+	taskLifecycle  *lifecycle.Manager[uuid.UUID, consts.TaskStatus, lifecycle.TaskMetadata]
+	hostUsecase    domain.HostUsecase
+	taskConns      *ws.TaskConn
+	projectUsecase domain.ProjectUsecase
+	tokenProvider  *gituc.TokenProvider
 }
 
 func NewInternalHostHandler(i *do.Injector) (*InternalHostHandler, error) {
@@ -346,6 +346,9 @@ func (h *InternalHostHandler) VmReady(c *web.Context, req taskflow.VirtualMachin
 
 	for _, t := range vm.Edges.Tasks {
 		h.logger.With("task", t).DebugContext(c.Request().Context(), "vm-ready")
+		if t.Status == consts.TaskStatusProcessing {
+			continue
+		}
 
 		if t.Kind == consts.TaskTypeReview && t.SubType == consts.TaskSubTypePrReview {
 		} else {

backend/biz/task/usecase/task.go

@@ -139,7 +139,9 @@ func (a *TaskUsecase) Info(ctx context.Context, user *domain.User, id uuid.UUID)
 				case types.ConditionTypeFailed:
 					vm.Status = taskflow.VirtualMachineStatusOffline
 				case types.ConditionTypeHibernated:
-					vm.Status = taskflow.VirtualMachineStatusHibernated
+					if strings.ToLower(strings.TrimSpace(cond.Reason)) == "hibernated" {
+						vm.Status = taskflow.VirtualMachineStatusHibernated
+					}
 				case types.ConditionTypeReady:
 					if time.Since(time.Unix(vm.CreatedAt, 0)) > 2*time.Minute {
 						vm.Status = taskflow.VirtualMachineStatusOffline

backend/pkg/lifecycle/taskhook.go

@@ -58,6 +58,7 @@ func (h *TaskHook) OnStateChange(ctx context.Context, id uuid.UUID, from, to con
 
 func (h *TaskHook) withError(ctx context.Context, id, uid uuid.UUID, fn func() error) {
 	if err := fn(); err != nil {
+		h.logger.With("error", err, "task_id", id).ErrorContext(ctx, "failed to handle processing")
 		if err := h.taskLifecycle.Transition(ctx, id, consts.TaskStatusError, TaskMetadata{
 			TaskID: id,
 			UserID: uid,
@@ -78,6 +79,16 @@ func (h *TaskHook) handleError(ctx context.Context, id, uid uuid.UUID) error {
 
 func (h *TaskHook) handleProcessing(ctx context.Context, id uuid.UUID, metadata TaskMetadata) error {
 	h.withError(ctx, id, metadata.UserID, func() error {
+		// 从 DB 查询当前任务状态，如果已经是 processing 说明是 Agent 重连触发的重复 vm-ready，跳过
+		t, err := h.repo.GetByID(ctx, id)
+		if err != nil {
+			return fmt.Errorf("failed to get task: %w", err)
+		}
+		if t.Status == consts.TaskStatusProcessing {
+			h.logger.With("task_id", id).InfoContext(ctx, "task already processing, skipping (likely agent reconnect)")
+			return nil
+		}
+
 		reqKey := fmt.Sprintf("task:create_req:%s", id.String())
 		val, err := h.redis.Get(ctx, reqKey).Result()
 		if err != nil {