fix: harden desktop monitoring recovery

server/internal/tenant/app/monitoring_callback_service.go

@@ -40,6 +40,7 @@ const (
 	// reports skip writes and only refresh the row at this interval so dashboards
 	// do not depend on per-heartbeat database churn.
 	monitoringHeartbeatAccessSnapshotRefreshAfter = 10 * time.Minute
+	monitoringRuntimeUnknownMaxDispatchAttempts   = 3
 )
 
 var monitoringCitationMarkerPattern = regexp.MustCompile(`(?i)\[\s*citation\s*:\s*\d+\s*\]`)
@@ -194,6 +195,15 @@ type MonitoringCancelTaskResponse struct {
 	TaskStatus string `json:"task_status"`
 }
 
+func isMonitoringInfrastructureCancelReason(reason *string) bool {
+	switch strings.TrimSpace(optionalStringValue(reason)) {
+	case "desktop_infra_unavailable", "playwright_cdp_unavailable", "playwright_cdp_recovery":
+		return true
+	default:
+		return false
+	}
+}
+
 type monitoringInstallationIdentity struct {
 	ID          string
 	TenantID    int64
@@ -250,6 +260,7 @@ type monitoringCollectTask struct {
 	TriggerSource      string
 	QuestionText       string
 	TargetClientID     sql.NullString
+	DispatchAttempts   int
 }
 
 type monitoringDesktopExecutionLease struct {
@@ -258,6 +269,7 @@ type monitoringDesktopExecutionLease struct {
 	TargetClientID      uuid.UUID
 	LeaseTokenHash      []byte
 	InterruptGeneration int
+	ActiveAttemptID     pgtype.UUID
 }
 
 type monitoringLeasedTaskRow struct {
@@ -945,7 +957,24 @@ func (s *MonitoringCallbackService) skipTaskForInstallation(
 	}
 
 	skipReason := normalizeSkipReason(req.SkipReason)
-	skipOutcome := monitoringSkipOutcomeForReason(skipReason)
+	skipOutcome := monitoringSkipOutcomeForReason(skipReason, task.DispatchAttempts, task.TriggerSource)
+	errorMessage := strings.TrimSpace(optionalStringValue(req.ErrorMessage))
+	requestPayloadJSON := []byte(`{}`)
+	if skipReason == "runtime_unknown" {
+		var marshalErr error
+		requestPayloadJSON, marshalErr = json.Marshal(map[string]any{
+			"runtime_unknown": map[string]any{
+				"reason":            skipReason,
+				"message":           errorMessage,
+				"dispatch_attempts": task.DispatchAttempts,
+				"max_attempts":      monitoringRuntimeUnknownMaxDispatchAttempts,
+				"retryable":         skipOutcome.Requeue,
+			},
+		})
+		if marshalErr != nil {
+			return nil, response.ErrInternal(50041, "payload_encode_failed", "failed to encode monitoring task skip payload")
+		}
+	}
 
 	tx, err := s.monitoringPool.Begin(ctx)
 	if err != nil {
@@ -953,16 +982,48 @@ func (s *MonitoringCallbackService) skipTaskForInstallation(
 	}
 	defer tx.Rollback(ctx)
 
-	if _, err := tx.Exec(ctx, `
-		UPDATE monitoring_collect_tasks
-		SET status = $2,
-		    callback_received_at = NOW(),
-		    completed_at = $3,
-		    skip_reason = $4,
-		    error_message = $5,
-		    updated_at = NOW()
-		WHERE id = $1
-	`, task.ID, skipOutcome.TaskStatus, completedAt, nullableStringPtr(skipOutcome.StoredSkipReason), nullableString(req.ErrorMessage)); err != nil {
+	if skipOutcome.Requeue {
+		if _, err := tx.Exec(ctx, `
+			UPDATE monitoring_collect_tasks
+			SET status = 'pending',
+			    lease_token_hash = NULL,
+			    leased_to_executor = NULL,
+			    leased_at = NULL,
+			    lease_expires_at = NULL,
+			    callback_received_at = NULL,
+			    completed_at = NULL,
+			    skip_reason = NULL,
+			    dispatch_after = NOW() + $2::interval,
+			    error_message = NULLIF($3, ''),
+			    request_payload_json = (
+					CASE
+						WHEN request_payload_json IS NULL OR jsonb_typeof(request_payload_json) <> 'object'
+							THEN '{}'::jsonb
+						ELSE request_payload_json
+					END
+				) || $4::jsonb,
+			    updated_at = NOW()
+			WHERE id = $1
+		`, task.ID, formatPgInterval(skipOutcome.RetryAfter), errorMessage, requestPayloadJSON); err != nil {
+			return nil, response.ErrInternal(50041, "task_update_failed", "failed to requeue monitoring task after runtime unknown")
+		}
+	} else if _, err := tx.Exec(ctx, `
+			UPDATE monitoring_collect_tasks
+			SET status = $2,
+			    callback_received_at = NOW(),
+			    completed_at = $3,
+			    skip_reason = $4,
+			    error_message = $5,
+			    request_payload_json = (
+					CASE
+						WHEN request_payload_json IS NULL OR jsonb_typeof(request_payload_json) <> 'object'
+							THEN '{}'::jsonb
+						ELSE request_payload_json
+					END
+				) || $6::jsonb,
+			    updated_at = NOW()
+			WHERE id = $1
+		`, task.ID, skipOutcome.TaskStatus, completedAt, nullableStringPtr(skipOutcome.StoredSkipReason), nullableString(req.ErrorMessage), requestPayloadJSON); err != nil {
 		return nil, response.ErrInternal(50041, "task_update_failed", "failed to update monitoring task skip status")
 	}
 
@@ -976,29 +1037,56 @@ func (s *MonitoringCallbackService) skipTaskForInstallation(
 
 	_ = s.touchInstallation(ctx, installation.ID)
 
-	completedAtText := completedAt.Format(time.RFC3339)
+	var completedAtText *string
+	if !skipOutcome.Requeue {
+		text := completedAt.Format(time.RFC3339)
+		completedAtText = &text
+	}
 	return &MonitoringSkipTaskResponse{
 		TaskID:      task.ID,
 		TaskStatus:  skipOutcome.TaskStatus,
 		SkipReason:  skipReason,
-		CompletedAt: &completedAtText,
+		CompletedAt: completedAtText,
 	}, nil
 }
 
 type monitoringSkipOutcome struct {
 	TaskStatus       string
 	StoredSkipReason string
+	Requeue          bool
+	RetryAfter       time.Duration
 }
 
-func monitoringSkipOutcomeForReason(skipReason string) monitoringSkipOutcome {
+func monitoringSkipOutcomeForReason(skipReason string, dispatchAttempts int, triggerSource string) monitoringSkipOutcome {
 	switch strings.TrimSpace(skipReason) {
 	case "runtime_unknown":
+		if strings.EqualFold(strings.TrimSpace(triggerSource), "manual") {
+			return monitoringSkipOutcome{TaskStatus: "failed"}
+		}
+		if dispatchAttempts < monitoringRuntimeUnknownMaxDispatchAttempts {
+			return monitoringSkipOutcome{
+				TaskStatus: "pending",
+				Requeue:    true,
+				RetryAfter: monitoringRuntimeUnknownRetryAfter(dispatchAttempts),
+			}
+		}
 		return monitoringSkipOutcome{TaskStatus: "failed"}
 	default:
 		return monitoringSkipOutcome{TaskStatus: "skipped", StoredSkipReason: strings.TrimSpace(skipReason)}
 	}
 }
 
+func monitoringRuntimeUnknownRetryAfter(dispatchAttempts int) time.Duration {
+	switch {
+	case dispatchAttempts <= 0:
+		return 30 * time.Second
+	case dispatchAttempts == 1:
+		return 2 * time.Minute
+	default:
+		return 5 * time.Minute
+	}
+}
+
 func nullableStringPtr(value string) *string {
 	trimmed := strings.TrimSpace(value)
 	if trimmed == "" {
@@ -1045,15 +1133,23 @@ func (s *MonitoringCallbackService) CancelTaskForDesktopClient(
 		if marshalErr != nil {
 			return nil, response.ErrInternal(50041, "payload_encode_failed", "failed to encode phase2 desktop cancel payload")
 		}
-		taskRepo := repository.NewDesktopTaskRepository(s.businessPool)
-		canceledTask, cancelErr := taskRepo.CancelByLease(ctx, desktopExecution.DesktopTaskID, desktopExecution.LeaseTokenHash, errorJSON)
-		if cancelErr != nil {
-			if errors.Is(cancelErr, pgx.ErrNoRows) {
-				return nil, response.ErrConflict(40941, "task_status_invalid", "phase2 monitor desktop task is no longer active")
+		if isMonitoringInfrastructureCancelReason(req.Reason) {
+			requeuedTask, requeueErr := s.requeueDesktopMonitorTaskAfterInfrastructureCancel(ctx, desktopExecution, errorJSON)
+			if requeueErr != nil {
+				return nil, requeueErr
 			}
-			return nil, response.ErrInternal(50041, "desktop_task_cancel_failed", "failed to cancel phase2 monitor desktop task lease")
+			desktopTaskCanceled = requeuedTask
+		} else {
+			taskRepo := repository.NewDesktopTaskRepository(s.businessPool)
+			canceledTask, cancelErr := taskRepo.CancelByLease(ctx, desktopExecution.DesktopTaskID, desktopExecution.LeaseTokenHash, errorJSON)
+			if cancelErr != nil {
+				if errors.Is(cancelErr, pgx.ErrNoRows) {
+					return nil, response.ErrConflict(40941, "task_status_invalid", "phase2 monitor desktop task is no longer active")
+				}
+				return nil, response.ErrInternal(50041, "desktop_task_cancel_failed", "failed to cancel phase2 monitor desktop task lease")
+			}
+			desktopTaskCanceled = canceledTask
 		}
-		desktopTaskCanceled = canceledTask
 	} else if task.Status != "leased" {
 		return nil, response.ErrConflict(40941, "task_status_invalid", "task status does not allow lease cancel")
 	}
@@ -1066,11 +1162,11 @@ func (s *MonitoringCallbackService) CancelTaskForDesktopClient(
 		    leased_at = NULL,
 		    lease_expires_at = NULL,
 		    callback_received_at = NULL,
-		    dispatch_after = NOW(),
+		    dispatch_after = CASE WHEN $3::boolean THEN NOW() + interval '30 seconds' ELSE NOW() END,
 		    error_message = $2,
 		    updated_at = NOW()
 		WHERE id = $1
-	`, task.ID, nullableString(req.Reason)); err != nil {
+	`, task.ID, nullableString(req.Reason), isMonitoringInfrastructureCancelReason(req.Reason)); err != nil {
 		return nil, response.ErrInternal(50041, "task_update_failed", "failed to cancel monitoring task lease")
 	}
 
@@ -1106,6 +1202,72 @@ func (s *MonitoringCallbackService) CancelTaskForDesktopClient(
 	}, nil
 }
 
+func (s *MonitoringCallbackService) requeueDesktopMonitorTaskAfterInfrastructureCancel(
+	ctx context.Context,
+	lease *monitoringDesktopExecutionLease,
+	errorJSON []byte,
+) (*repository.DesktopTask, error) {
+	if s == nil || s.businessPool == nil || lease == nil {
+		return nil, response.ErrInternal(50041, "desktop_task_requeue_failed", "failed to requeue phase2 monitor desktop task lease")
+	}
+
+	tx, err := s.businessPool.Begin(ctx)
+	if err != nil {
+		return nil, response.ErrInternal(50041, "desktop_task_requeue_begin_failed", "failed to start phase2 monitor desktop task requeue")
+	}
+	defer tx.Rollback(ctx)
+
+	repo := repository.NewDesktopTaskRepository(tx)
+	row := tx.QueryRow(ctx, `
+		UPDATE desktop_tasks
+		SET status = 'queued',
+		    error = $3,
+		    result = NULL,
+		    active_attempt_id = NULL,
+		    lease_token_hash = NULL,
+		    lease_expires_at = NULL,
+		    interrupted_at = COALESCE(interrupted_at, NOW()),
+		    interrupt_reason = COALESCE(interrupt_reason, 'desktop_infra_unavailable'),
+		    enqueued_at = NOW(),
+		    updated_at = NOW()
+		WHERE desktop_id = $1
+		  AND lease_token_hash = $2
+		  AND kind = 'monitor'
+		  AND status = 'in_progress'
+		RETURNING `+desktopTaskRepositoryReturningColumns,
+		lease.DesktopTaskID,
+		lease.LeaseTokenHash,
+		errorJSON,
+	)
+	task, scanErr := scanRepositoryDesktopTask(row)
+	if scanErr != nil {
+		if errors.Is(scanErr, pgx.ErrNoRows) {
+			return nil, response.ErrConflict(40941, "task_status_invalid", "phase2 monitor desktop task is no longer active")
+		}
+		return nil, response.ErrInternal(50041, "desktop_task_requeue_failed", "failed to requeue phase2 monitor desktop task lease")
+	}
+
+	if lease.ActiveAttemptID.Valid {
+		attemptID, convErr := uuid.FromBytes(lease.ActiveAttemptID.Bytes[:])
+		if convErr != nil {
+			return nil, response.ErrInternal(50041, "desktop_task_attempt_decode_failed", "failed to decode phase2 monitor desktop task attempt")
+		}
+		if finishErr := repo.FinishAttempt(ctx, repository.FinishDesktopTaskAttemptParams{
+			TaskID:      task.DesktopID,
+			AttemptID:   attemptID,
+			FinalStatus: literalStringPtr("aborted"),
+			Error:       errorJSON,
+		}); finishErr != nil {
+			return nil, response.ErrInternal(50041, "desktop_task_attempt_finish_failed", "failed to finish phase2 monitor desktop task attempt")
+		}
+	}
+
+	if err := tx.Commit(ctx); err != nil {
+		return nil, response.ErrInternal(50041, "desktop_task_requeue_commit_failed", "failed to commit phase2 monitor desktop task requeue")
+	}
+	return task, nil
+}
+
 func (s *MonitoringCallbackService) ProcessQueuedTaskResult(ctx context.Context, envelope monitoringTaskResultEnvelope) (*MonitoringTaskResultResponse, error) {
 	if envelope.TaskID <= 0 {
 		return nil, response.ErrBadRequest(40041, "invalid_task_id", "task id is required")
@@ -1393,6 +1555,7 @@ func (s *MonitoringCallbackService) loadCollectTask(ctx context.Context, tenantI
 			completed_at,
 			skip_reason,
 			trigger_source,
+			COALESCE(dispatch_attempts, 0) AS dispatch_attempts,
 			COALESCE(target_client_id::text, '') AS target_client_id,
 			COALESCE(NULLIF(t.question_text_snapshot, ''), (
 				SELECT question_text_snapshot
@@ -1430,6 +1593,7 @@ func (s *MonitoringCallbackService) loadCollectTask(ctx context.Context, tenantI
 		&item.CompletedAt,
 		&item.SkipReason,
 		&item.TriggerSource,
+		&item.DispatchAttempts,
 		&item.TargetClientID,
 		&item.QuestionText,
 	); err != nil {
@@ -1464,6 +1628,7 @@ func (s *MonitoringCallbackService) loadCollectTaskByID(ctx context.Context, tas
 			completed_at,
 			skip_reason,
 			trigger_source,
+			COALESCE(dispatch_attempts, 0) AS dispatch_attempts,
 			COALESCE(target_client_id::text, '') AS target_client_id,
 			COALESCE(NULLIF(t.question_text_snapshot, ''), (
 				SELECT question_text_snapshot
@@ -1500,6 +1665,7 @@ func (s *MonitoringCallbackService) loadCollectTaskByID(ctx context.Context, tas
 		&item.CompletedAt,
 		&item.SkipReason,
 		&item.TriggerSource,
+		&item.DispatchAttempts,
 		&item.TargetClientID,
 		&item.QuestionText,
 	); err != nil {
@@ -1518,23 +1684,25 @@ func (s *MonitoringCallbackService) loadActiveDesktopMonitorExecutionLease(
 ) (*monitoringDesktopExecutionLease, error) {
 	item := &monitoringDesktopExecutionLease{}
 	if err := s.businessPool.QueryRow(ctx, `
-		SELECT
-			desktop_id,
-			status,
-			target_client_id,
-			lease_token_hash,
-			interrupt_generation
+			SELECT
+				desktop_id,
+				status,
+				target_client_id,
+				lease_token_hash,
+				active_attempt_id,
+				interrupt_generation
 		FROM desktop_tasks
 		WHERE kind = 'monitor'
 		  AND monitor_task_id = $1
 		  AND status IN ('queued', 'in_progress')
 		ORDER BY created_at DESC, id DESC
 		LIMIT 1
-	`, monitorTaskID).Scan(
+		`, monitorTaskID).Scan(
 		&item.DesktopTaskID,
 		&item.Status,
 		&item.TargetClientID,
 		&item.LeaseTokenHash,
+		&item.ActiveAttemptID,
 		&item.InterruptGeneration,
 	); err != nil {
 		if err == pgx.ErrNoRows {