fix: fail expired monitor desktop leases

server/cmd/scheduler/main.go

@@ -114,10 +114,10 @@ func main() {
 	desktopTaskService := tenantapp.NewDesktopTaskService(app.DB, app.MonitoringDB, app.RabbitMQ, app.Logger).
 		WithCache(app.Cache).
 		WithRedis(app.Redis)
+	publishLeaseRecoveryWorker := internalscheduler.NewPublishLeaseRecoveryWorker(desktopTaskService, app.Logger)
 	monitoringDailyTaskWorker := tenantapp.NewMonitoringDailyTaskWorker(monitoringService, monitoringCallbackService, app.Logger)
 	monitoringResultRecoveryWorker := internalscheduler.NewMonitoringResultRecoveryWorker(app.MonitoringDB, monitoringCallbackService, app.Logger)
-	monitoringLeaseRecoveryWorker := internalscheduler.NewMonitoringLeaseRecoveryWorker(app.MonitoringDB, app.Logger)
+	monitoringLeaseRecoveryWorker := internalscheduler.NewMonitoringLeaseRecoveryWorker(app.MonitoringDB, desktopTaskService, app.Logger)
-	publishLeaseRecoveryWorker := internalscheduler.NewPublishLeaseRecoveryWorker(desktopTaskService, app.Logger)
 	monitoringReceivedInspectionWorker := internalscheduler.NewMonitoringReceivedInspectionWorker(app.MonitoringDB, app.Logger)
 	generationStateCheckWorker := generateworker.NewGenerationTaskStateCheckWorker(app.DB, app.Logger, app.Config().Generation).
 		WithConfigProvider(app.ConfigStore).

server/internal/scheduler/monitoring_lease_recovery_worker.go

@@ -11,22 +11,24 @@ import (
 )
 
 const (
-	defaultMonitoringLeaseRecoveryInterval = 30 * time.Minute
+	defaultMonitoringLeaseRecoveryInterval = 1 * time.Minute
 	defaultMonitoringLeaseRecoveryTimeout  = 30 * time.Second
 	defaultMonitoringLeaseRecoveryLimit    = 1000
 )
 
 type MonitoringLeaseRecoveryWorker struct {
 	monitoringPool *pgxpool.Pool
+	desktopTasks   *tenantapp.DesktopTaskService
 	logger         *zap.Logger
 	interval       time.Duration
 	timeout        time.Duration
 	limit          int
 }
 
-func NewMonitoringLeaseRecoveryWorker(monitoringPool *pgxpool.Pool, logger *zap.Logger) *MonitoringLeaseRecoveryWorker {
+func NewMonitoringLeaseRecoveryWorker(monitoringPool *pgxpool.Pool, desktopTasks *tenantapp.DesktopTaskService, logger *zap.Logger) *MonitoringLeaseRecoveryWorker {
 	return &MonitoringLeaseRecoveryWorker{
 		monitoringPool: monitoringPool,
+		desktopTasks:   desktopTasks,
 		logger:         logger,
 		interval:       defaultMonitoringLeaseRecoveryInterval,
 		timeout:        defaultMonitoringLeaseRecoveryTimeout,
@@ -89,8 +91,23 @@ func (w *MonitoringLeaseRecoveryWorker) runOnce(parent context.Context) {
 		return
 	}
 
-	if expiredCount > 0 && w.logger != nil {
+	var monitorResult tenantapp.MonitorLeaseRecoveryResult
-		w.logger.Info("monitoring lease recovery completed", zap.Int64("expired_task_count", expiredCount))
+	if w.desktopTasks != nil {
+		var recoverErr error
+		monitorResult, recoverErr = w.desktopTasks.RecoverExpiredMonitorTasks(ctx, w.limit)
+		if recoverErr != nil {
+			if w.logger != nil {
+				w.logger.Warn("monitor desktop lease recovery failed", zap.Error(recoverErr))
+			}
+			return
+		}
+	}
+
+	if (expiredCount > 0 || monitorResult.Failed > 0) && w.logger != nil {
+		w.logger.Info("monitoring lease recovery completed",
+			zap.Int64("expired_task_count", expiredCount),
+			zap.Int("failed_desktop_task_count", monitorResult.Failed),
+		)
 	}
 }
 
@@ -126,9 +143,18 @@ func (w *MonitoringLeaseRecoveryWorker) RunOnce(parent context.Context, run JobR
 	if err := tx.Commit(ctx); err != nil {
 		return map[string]any{"stage": "commit"}, err
 	}
+	var monitorResult tenantapp.MonitorLeaseRecoveryResult
+	if w.desktopTasks != nil {
+		var recoverErr error
+		monitorResult, recoverErr = w.desktopTasks.RecoverExpiredMonitorTasks(ctx, limit)
+		if recoverErr != nil {
+			return map[string]any{"stage": "recover_desktop_tasks"}, recoverErr
+		}
+	}
 	return map[string]any{
-		"expired_task_count": expiredCount,
+		"expired_task_count":           expiredCount,
-		"duration_ms":        time.Since(startedAt).Milliseconds(),
+		"failed_desktop_monitor_count": monitorResult.Failed,
-		"batch_size":         limit,
+		"duration_ms":                  time.Since(startedAt).Milliseconds(),
+		"batch_size":                   limit,
 	}, nil
 }

server/internal/tenant/app/desktop_task_service.go

@@ -1888,6 +1888,7 @@ type recoveredDesktopTaskLease struct {
 	Kind          string
 	Status        string
 	ActiveAttempt pgtype.UUID
+	MonitorTaskID pgtype.Int8
 	Attempts      int
 	SubmitStarted bool
 	ErrorJSON     []byte
@@ -1899,6 +1900,10 @@ type PublishLeaseRecoveryResult struct {
 	Uncertain int `json:"uncertain"`
 }
 
+type MonitorLeaseRecoveryResult struct {
+	Failed int `json:"failed"`
+}
+
 // resolvePublishRecoveryOutcome decides what to do with a publish task whose lease was lost.
 //
 // If the client had already entered the irreversible submit phase (publish_submit_started_at
@@ -1917,6 +1922,13 @@ func resolvePublishRecoveryOutcome(submitStarted bool, attempts int) (status str
 	}
 }
 
+func resolveMonitorRecoveryOutcome(mode desktopTaskRecoveryMode, attempts int) (status string, payloadKind string) {
+	if mode == desktopTaskRecoveryModeLeaseExpiry || attempts >= 2 {
+		return "failed", "monitor_final"
+	}
+	return "queued", "monitor"
+}
+
 func (s *DesktopTaskService) recoverClientTasksOnStartup(ctx context.Context, client *repository.DesktopClient) error {
 	return s.recoverClientTasks(ctx, client, desktopTaskRecoveryModeStartup)
 }
@@ -1950,7 +1962,7 @@ func (s *DesktopTaskService) recoverClientTasks(
 	if err != nil {
 		return response.ErrInternal(50195, "desktop_task_recovery_payload_failed", "failed to encode publish desktop task recovery payload")
 	}
-	monitorErrorJSON, monitorReason, _, err := desktopTaskRecoveryPayload(mode, "monitor")
+	monitorErrorJSON, _, _, err := desktopTaskRecoveryPayload(mode, "monitor")
 	if err != nil {
 		return response.ErrInternal(50195, "desktop_task_recovery_payload_failed", "failed to encode monitor desktop task recovery payload")
 	}
@@ -1983,6 +1995,7 @@ func (s *DesktopTaskService) recoverClientTasks(
 			&item.ActiveAttempt,
 			&item.Attempts,
 			&item.SubmitStarted,
+			&item.MonitorTaskID,
 		); scanErr != nil {
 			s.logWarn("desktop task recovery scan failed", scanErr, zap.String("client_id", client.ID.String()), zap.String("mode", string(mode)))
 			return response.ErrInternal(50198, "desktop_task_recovery_scan_failed", "failed to parse recovered desktop tasks")
@@ -1998,23 +2011,40 @@ func (s *DesktopTaskService) recoverClientTasks(
 	for index := range recovered {
 		item := &recovered[index]
 		if item.Kind == "monitor" {
+			finalStatus, monitorPayloadKind := resolveMonitorRecoveryOutcome(mode, item.Attempts)
+			monitorErrorForTask, monitorReasonForTask, _, payloadErr := desktopTaskRecoveryPayload(mode, monitorPayloadKind)
+			if payloadErr != nil {
+				return response.ErrInternal(50195, "desktop_task_recovery_payload_failed", "failed to encode monitor desktop task recovery payload")
+			}
+			if finalStatus == "failed" {
+				if item.MonitorTaskID.Valid {
+					failedMonitorTaskIDs, failErr := s.failMonitorCollectTasksForDesktopRecovery(ctx, []int64{item.MonitorTaskID.Int64}, monitorErrorForTask)
+					if failErr != nil {
+						return failErr
+					}
+					if len(failedMonitorTaskIDs) == 0 {
+						continue
+					}
+				}
+			}
 			if _, execErr := tx.Exec(ctx, `
 				UPDATE desktop_tasks
-				SET status = 'queued',
+				SET status = $2,
-				    error = $2,
+				    error = $3,
 				    active_attempt_id = NULL,
 				    lease_token_hash = NULL,
 				    lease_expires_at = NULL,
 				    interrupted_at = COALESCE(interrupted_at, NOW()),
-				    interrupt_reason = $3,
+				    interrupt_reason = $4,
-				    enqueued_at = NOW(),
+				    enqueued_at = CASE WHEN $2 = 'queued' THEN NOW() ELSE enqueued_at END,
 				    updated_at = NOW()
 				WHERE desktop_id = $1
-			`, item.TaskID, monitorErrorJSON, monitorReason); execErr != nil {
+			`, item.TaskID, finalStatus, monitorErrorForTask, monitorReasonForTask); execErr != nil {
 				s.logWarn("desktop monitor task recovery update failed", execErr, zap.String("task_id", item.TaskID.String()), zap.String("mode", string(mode)))
 				return response.ErrInternal(50197, "desktop_task_recovery_query_failed", "failed to select recoverable desktop tasks")
 			}
-			item.Status = "queued"
+			item.Status = finalStatus
+			item.ErrorJSON = monitorErrorForTask
 			continue
 		}
 
@@ -2080,8 +2110,16 @@ func (s *DesktopTaskService) recoverClientTasks(
 			errorJSON = publishErrorJSON
 		}
 		if item.Kind == "monitor" {
-			finalStatus = literalStringPtr("aborted")
+			if item.Status == "failed" {
-			errorJSON = monitorErrorJSON
+				finalStatus = literalStringPtr("failed")
+			} else {
+				finalStatus = literalStringPtr("aborted")
+			}
+			if len(item.ErrorJSON) > 0 {
+				errorJSON = item.ErrorJSON
+			} else {
+				errorJSON = monitorErrorJSON
+			}
 		}
 
 		if finishErr := repo.FinishAttempt(ctx, repository.FinishDesktopTaskAttemptParams{
@@ -2308,6 +2346,318 @@ func (s *DesktopTaskService) RecoverExpiredPublishTasks(ctx context.Context, lim
 	return result, nil
 }
 
+func (s *DesktopTaskService) RecoverExpiredMonitorTasks(ctx context.Context, limit int) (MonitorLeaseRecoveryResult, error) {
+	var result MonitorLeaseRecoveryResult
+	if s == nil || s.pool == nil {
+		return result, nil
+	}
+	if limit <= 0 {
+		limit = 1000
+	}
+
+	errorJSON, reason, _, err := desktopTaskRecoveryPayload(desktopTaskRecoveryModeLeaseExpiry, "monitor_final")
+	if err != nil {
+		return result, response.ErrInternal(50195, "desktop_task_recovery_payload_failed", "failed to encode monitor desktop task recovery payload")
+	}
+
+	tx, err := s.pool.Begin(ctx)
+	if err != nil {
+		return result, response.ErrInternal(50196, "desktop_task_recovery_begin_failed", "failed to start desktop task recovery")
+	}
+	defer tx.Rollback(ctx)
+
+	rows, err := tx.Query(ctx, expiredMonitorDesktopTasksCandidateSQL(), limit)
+	if err != nil {
+		s.logWarn("expired desktop monitor recovery query failed", err)
+		return result, response.ErrInternal(50197, "desktop_task_recovery_query_failed", "failed to select recoverable desktop tasks")
+	}
+	defer rows.Close()
+
+	recovered := make([]recoveredDesktopTaskLease, 0)
+	monitorTaskIDs := make([]int64, 0)
+	for rows.Next() {
+		var item recoveredDesktopTaskLease
+		if scanErr := rows.Scan(
+			&item.TaskID,
+			&item.WorkspaceID,
+			&item.Kind,
+			&item.Status,
+			&item.ActiveAttempt,
+			&item.Attempts,
+			&item.SubmitStarted,
+			&item.MonitorTaskID,
+		); scanErr != nil {
+			s.logWarn("expired desktop monitor recovery scan failed", scanErr)
+			return result, response.ErrInternal(50198, "desktop_task_recovery_scan_failed", "failed to parse recovered desktop tasks")
+		}
+		item.Status = "failed"
+		item.ErrorJSON = errorJSON
+		recovered = append(recovered, item)
+		if item.MonitorTaskID.Valid {
+			monitorTaskIDs = append(monitorTaskIDs, item.MonitorTaskID.Int64)
+		}
+	}
+	if err := rows.Err(); err != nil {
+		s.logWarn("expired desktop monitor recovery rows failed", err)
+		return result, response.ErrInternal(50198, "desktop_task_recovery_scan_failed", "failed to iterate recovered desktop tasks")
+	}
+
+	failedMonitorTaskIDs, err := s.failMonitorCollectTasksForDesktopRecovery(ctx, monitorTaskIDs, errorJSON)
+	if err != nil {
+		return result, err
+	}
+	failedMonitorTaskSet := int64Set(failedMonitorTaskIDs)
+	recoverableDesktopIDs := make([]uuid.UUID, 0, len(recovered))
+	recoverable := make([]recoveredDesktopTaskLease, 0, len(recovered))
+	for _, item := range recovered {
+		if !item.MonitorTaskID.Valid {
+			continue
+		}
+		if _, ok := failedMonitorTaskSet[item.MonitorTaskID.Int64]; !ok {
+			continue
+		}
+		recoverableDesktopIDs = append(recoverableDesktopIDs, item.TaskID)
+		recoverable = append(recoverable, item)
+	}
+	if len(recoverableDesktopIDs) == 0 {
+		return result, nil
+	}
+
+	updateRows, err := tx.Query(ctx, failExpiredMonitorDesktopTasksSQL(), recoverableDesktopIDs, errorJSON, reason)
+	if err != nil {
+		s.logWarn("expired desktop monitor recovery update failed", err)
+		return result, response.ErrInternal(50197, "desktop_task_recovery_query_failed", "failed to select recoverable desktop tasks")
+	}
+	defer updateRows.Close()
+
+	updatedDesktopIDs := make(map[uuid.UUID]struct{}, len(recoverableDesktopIDs))
+	for updateRows.Next() {
+		var desktopID uuid.UUID
+		if scanErr := updateRows.Scan(&desktopID); scanErr != nil {
+			s.logWarn("expired desktop monitor recovery update scan failed", scanErr)
+			return result, response.ErrInternal(50198, "desktop_task_recovery_scan_failed", "failed to parse recovered desktop tasks")
+		}
+		updatedDesktopIDs[desktopID] = struct{}{}
+	}
+	if err := updateRows.Err(); err != nil {
+		s.logWarn("expired desktop monitor recovery update rows failed", err)
+		return result, response.ErrInternal(50198, "desktop_task_recovery_scan_failed", "failed to iterate recovered desktop tasks")
+	}
+
+	repo := repository.NewDesktopTaskRepository(tx)
+	for _, item := range recoverable {
+		if _, ok := updatedDesktopIDs[item.TaskID]; !ok {
+			continue
+		}
+		if !item.ActiveAttempt.Valid {
+			continue
+		}
+
+		attemptID, convErr := uuid.FromBytes(item.ActiveAttempt.Bytes[:])
+		if convErr != nil {
+			s.logWarn("expired desktop monitor recovery attempt id decode failed", convErr, zap.String("task_id", item.TaskID.String()))
+			continue
+		}
+
+		if finishErr := repo.FinishAttempt(ctx, repository.FinishDesktopTaskAttemptParams{
+			TaskID:      item.TaskID,
+			AttemptID:   attemptID,
+			FinalStatus: literalStringPtr("failed"),
+			Error:       errorJSON,
+		}); finishErr != nil {
+			s.logWarn("expired desktop monitor recovery attempt finish failed", finishErr, zap.String("task_id", item.TaskID.String()))
+		}
+	}
+
+	if err := tx.Commit(ctx); err != nil {
+		return result, response.ErrInternal(50199, "desktop_task_recovery_commit_failed", "failed to commit desktop task recovery")
+	}
+
+	result.Failed = len(updatedDesktopIDs)
+
+	for _, item := range recoverable {
+		if _, ok := updatedDesktopIDs[item.TaskID]; !ok {
+			continue
+		}
+		task, getErr := s.repo.GetByDesktopID(ctx, item.TaskID, item.WorkspaceID)
+		if getErr != nil {
+			s.logWarn("expired desktop monitor recovery reload after commit failed", getErr, zap.String("task_id", item.TaskID.String()))
+			continue
+		}
+		s.publishTaskEvent(ctx, task, "task_completed")
+	}
+
+	if result.Failed > 0 && s.logger != nil {
+		s.logger.Info("expired desktop monitor tasks recovered",
+			zap.Int("failed", result.Failed),
+		)
+	}
+
+	return result, nil
+}
+
+func expiredMonitorDesktopTasksCandidateSQL() string {
+	return `
+		SELECT
+			t.desktop_id,
+			t.workspace_id,
+			t.kind,
+			t.status,
+			t.active_attempt_id,
+			t.attempts,
+			(t.publish_submit_started_at IS NOT NULL) AS submit_started,
+			t.monitor_task_id
+		FROM desktop_tasks AS t
+		WHERE t.kind = 'monitor'
+		  AND t.status = 'in_progress'
+		  AND t.monitor_task_id IS NOT NULL
+		  AND t.lease_expires_at IS NOT NULL
+		  AND t.lease_expires_at < NOW()
+		ORDER BY t.lease_expires_at ASC, t.updated_at ASC, t.desktop_id ASC
+		LIMIT $1
+		FOR UPDATE OF t SKIP LOCKED
+	`
+}
+
+func failExpiredMonitorDesktopTasksSQL() string {
+	return `
+		UPDATE desktop_tasks
+		SET status = 'failed',
+		    error = $2,
+		    active_attempt_id = NULL,
+		    lease_token_hash = NULL,
+		    lease_expires_at = NULL,
+		    interrupted_at = COALESCE(interrupted_at, NOW()),
+		    interrupt_reason = COALESCE(interrupt_reason, $3),
+		    updated_at = NOW()
+		WHERE desktop_id = ANY($1::uuid[])
+		  AND kind = 'monitor'
+		  AND status = 'in_progress'
+		RETURNING desktop_id
+	`
+}
+
+func (s *DesktopTaskService) failMonitorCollectTasksForDesktopRecovery(ctx context.Context, monitorTaskIDs []int64, errorJSON []byte) ([]int64, error) {
+	if s == nil || s.monitoringPool == nil || len(monitorTaskIDs) == 0 {
+		return nil, nil
+	}
+	monitorTaskIDs = uniqueInt64s(monitorTaskIDs)
+	message := desktopTaskRecoveryMessage(errorJSON)
+	requestPayloadJSON, err := json.Marshal(map[string]any{
+		"runtime_error": json.RawMessage(errorJSON),
+	})
+	if err != nil {
+		return nil, response.ErrInternal(50195, "desktop_task_recovery_payload_failed", "failed to encode monitor desktop task recovery payload")
+	}
+	rows, err := s.monitoringPool.Query(ctx, `
+		WITH input AS (
+			SELECT unnest($1::bigint[]) AS id
+		),
+		updated AS (
+			UPDATE monitoring_collect_tasks
+			SET status = 'failed',
+			    lease_token_hash = NULL,
+			    leased_to_executor = NULL,
+			    leased_at = NULL,
+			    lease_expires_at = NULL,
+			    callback_received_at = COALESCE(callback_received_at, NOW()),
+			    completed_at = COALESCE(completed_at, NOW()),
+			    skip_reason = NULL,
+			    error_message = COALESCE(NULLIF(error_message, ''), $2),
+			    request_payload_json = (
+					CASE
+						WHEN request_payload_json IS NULL OR jsonb_typeof(request_payload_json) <> 'object'
+							THEN '{}'::jsonb
+						ELSE request_payload_json
+					END
+				) || $3::jsonb,
+			    updated_at = NOW()
+			WHERE id IN (SELECT id FROM input)
+			  AND COALESCE(execution_owner, 'legacy') = 'desktop_tasks'
+			  AND callback_received_at IS NULL
+			  AND status IN ('pending', 'leased', 'expired')
+			RETURNING id
+		),
+		already_failed AS (
+			SELECT t.id
+			FROM monitoring_collect_tasks t
+			JOIN input i ON i.id = t.id
+			WHERE COALESCE(t.execution_owner, 'legacy') = 'desktop_tasks'
+			  AND t.status = 'failed'
+			  AND (
+				t.error_message = $2
+				OR t.request_payload_json #>> '{runtime_error,source}' = 'desktop_task_lease_expiry'
+				OR t.request_payload_json #>> '{runtime_error,reason}' = 'lease_expired'
+			  )
+		)
+		SELECT id FROM updated
+		UNION
+		SELECT id FROM already_failed
+	`, monitorTaskIDs, message, requestPayloadJSON)
+	if err != nil {
+		return nil, response.ErrInternal(50041, "task_update_failed", "failed to fail expired monitor desktop tasks")
+	}
+	defer rows.Close()
+
+	failedIDs := make([]int64, 0, len(monitorTaskIDs))
+	for rows.Next() {
+		var id int64
+		if scanErr := rows.Scan(&id); scanErr != nil {
+			return nil, response.ErrInternal(50041, "task_update_failed", "failed to parse failed monitor desktop tasks")
+		}
+		failedIDs = append(failedIDs, id)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, response.ErrInternal(50041, "task_update_failed", "failed to iterate failed monitor desktop tasks")
+	}
+	return failedIDs, nil
+}
+
+func uniqueInt64s(values []int64) []int64 {
+	if len(values) <= 1 {
+		return values
+	}
+	seen := make(map[int64]struct{}, len(values))
+	result := make([]int64, 0, len(values))
+	for _, value := range values {
+		if value == 0 {
+			continue
+		}
+		if _, ok := seen[value]; ok {
+			continue
+		}
+		seen[value] = struct{}{}
+		result = append(result, value)
+	}
+	return result
+}
+
+func int64Set(values []int64) map[int64]struct{} {
+	result := make(map[int64]struct{}, len(values))
+	for _, value := range values {
+		if value == 0 {
+			continue
+		}
+		result[value] = struct{}{}
+	}
+	return result
+}
+
+func desktopTaskRecoveryMessage(errorJSON []byte) string {
+	const fallback = "desktop monitor task recovered to failed"
+	if len(errorJSON) == 0 {
+		return fallback
+	}
+	var payload map[string]any
+	if err := json.Unmarshal(errorJSON, &payload); err != nil {
+		return fallback
+	}
+	if message, ok := payload["message"].(string); ok && strings.TrimSpace(message) != "" {
+		return strings.TrimSpace(message)
+	}
+	return fallback
+}
+
 func desktopTaskRecoverySelectQuery(mode desktopTaskRecoveryMode) string {
 	query := `
 		SELECT
@@ -2317,7 +2667,8 @@ func desktopTaskRecoverySelectQuery(mode desktopTaskRecoveryMode) string {
 				status,
 				active_attempt_id,
 				attempts,
-				publish_submit_started_at IS NOT NULL AS submit_started
+				publish_submit_started_at IS NOT NULL AS submit_started,
+				monitor_task_id
 			FROM desktop_tasks
 		WHERE target_client_id = $1
 		  AND workspace_id = $2
@@ -2341,11 +2692,14 @@ func desktopTaskRecoveryPayload(mode desktopTaskRecoveryMode, kind string) ([]by
 	source := "desktop_task_recovery"
 	isPublishFinal := kind == "publish_final"
 	isPublishUncertain := kind == "publish_uncertain"
+	isMonitorFinal := kind == "monitor_final"
 
 	if isPublishFinal {
 		message = fmt.Sprintf("desktop publish task exceeded %d attempts during recovery; task has been marked failed", desktopPublishMaxAttempts)
 	} else if isPublishUncertain {
 		message = "desktop publish task may have already been submitted to the platform before the lease was lost; kept as unknown for manual reconcile to avoid duplicate publishing"
+	} else if isMonitorFinal {
+		message = "desktop monitor task lost its lease; task has been marked failed to avoid repeated platform retries"
 	}
 
 	switch mode {
@@ -2355,6 +2709,8 @@ func desktopTaskRecoveryPayload(mode desktopTaskRecoveryMode, kind string) ([]by
 			message = fmt.Sprintf("desktop client restarted while the publish task was in progress; task exceeded %d attempts and has been marked failed", desktopPublishMaxAttempts)
 		} else if isPublishUncertain {
 			message = "desktop client restarted while the publish task was mid-submit; kept as unknown for manual reconcile to avoid duplicate publishing"
+		} else if isMonitorFinal {
+			message = "desktop client restarted while the monitor task was in progress; task has been marked failed after repeated recovery"
 		} else if kind == "monitor" {
 			message = "desktop client restarted while the monitor task was in progress; task has been re-queued"
 		} else {
@@ -2366,6 +2722,8 @@ func desktopTaskRecoveryPayload(mode desktopTaskRecoveryMode, kind string) ([]by
 			message = fmt.Sprintf("desktop client went offline while the publish task was in progress; task exceeded %d attempts and has been marked failed", desktopPublishMaxAttempts)
 		} else if isPublishUncertain {
 			message = "desktop client went offline while the publish task was mid-submit; kept as unknown for manual reconcile to avoid duplicate publishing"
+		} else if isMonitorFinal {
+			message = "desktop client went offline while the monitor task was in progress; task has been marked failed after repeated recovery"
 		} else if kind == "monitor" {
 			message = "desktop client went offline while the monitor task was in progress; task has been re-queued"
 		} else {
@@ -2377,8 +2735,8 @@ func desktopTaskRecoveryPayload(mode desktopTaskRecoveryMode, kind string) ([]by
 			message = fmt.Sprintf("desktop task lease expired before publish completion; task exceeded %d attempts and has been marked failed", desktopPublishMaxAttempts)
 		} else if isPublishUncertain {
 			message = "desktop task lease expired while the publish task was mid-submit; kept as unknown for manual reconcile to avoid duplicate publishing"
-		} else if kind == "monitor" {
+		} else if isMonitorFinal || kind == "monitor" {
-			message = "desktop task lease expired before monitor completion; task has been re-queued"
+			message = "desktop task lease expired before monitor completion; task has been marked failed"
 		} else {
 			message = "desktop task lease expired before publish completion; task has been re-queued for retry"
 		}

server/internal/tenant/app/desktop_task_service_test.go

@@ -23,6 +23,7 @@ func TestDesktopTaskRecoverySelectQueryMatchesRecoveredLeaseScanOrder(t *testing
 		"active_attempt_id",
 		"attempts",
 		"publish_submit_started_at",
+		"monitor_task_id",
 	}
 
 	if len(gotColumns) != len(wantColumns) {
@@ -100,6 +101,27 @@ func TestDesktopTaskRecoveryPayloadPublishFinalFailsAfterMaxAttempts(t *testing.
 	}
 }
 
+func TestDesktopTaskRecoveryPayloadMonitorLeaseExpiryFails(t *testing.T) {
+	t.Parallel()
+
+	payload, reason, message, err := desktopTaskRecoveryPayload(desktopTaskRecoveryModeLeaseExpiry, "monitor_final")
+	if err != nil {
+		t.Fatalf("desktopTaskRecoveryPayload returned error: %v", err)
+	}
+	if reason != "lease_expired" {
+		t.Fatalf("reason = %q, want lease_expired", reason)
+	}
+	if !strings.Contains(message, "marked failed") {
+		t.Fatalf("message = %q, want marked failed", message)
+	}
+	if strings.Contains(message, "re-queued") {
+		t.Fatalf("message must not requeue expired monitor tasks: %q", message)
+	}
+	if !strings.Contains(string(payload), "desktop_task_lease_expiry") {
+		t.Fatalf("payload missing source: %s", payload)
+	}
+}
+
 func TestReconcileNonRetryPublishesTaskReconciled(t *testing.T) {
 	t.Parallel()
 
@@ -140,6 +162,35 @@ func TestResolvePublishRecoveryOutcome(t *testing.T) {
 	}
 }
 
+func TestResolveMonitorRecoveryOutcome(t *testing.T) {
+	t.Parallel()
+
+	cases := []struct {
+		name       string
+		mode       desktopTaskRecoveryMode
+		attempts   int
+		wantStatus string
+		wantKind   string
+	}{
+		{"startup first attempt requeues", desktopTaskRecoveryModeStartup, 1, "queued", "monitor"},
+		{"disconnect first attempt requeues", desktopTaskRecoveryModeDisconnect, 1, "queued", "monitor"},
+		{"startup repeated recovery fails", desktopTaskRecoveryModeStartup, 2, "failed", "monitor_final"},
+		{"disconnect repeated recovery fails", desktopTaskRecoveryModeDisconnect, 2, "failed", "monitor_final"},
+		{"lease expiry fails immediately", desktopTaskRecoveryModeLeaseExpiry, 1, "failed", "monitor_final"},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			gotStatus, gotKind := resolveMonitorRecoveryOutcome(tc.mode, tc.attempts)
+			if gotStatus != tc.wantStatus || gotKind != tc.wantKind {
+				t.Fatalf("resolveMonitorRecoveryOutcome(%q, %d) = (%q, %q), want (%q, %q)",
+					tc.mode, tc.attempts, gotStatus, gotKind, tc.wantStatus, tc.wantKind)
+			}
+		})
+	}
+}
+
 func TestNormalizeDesktopTaskCompletionStatusForKindFailsMonitorUnknown(t *testing.T) {
 	t.Parallel()
 
@@ -151,6 +202,46 @@ func TestNormalizeDesktopTaskCompletionStatusForKindFailsMonitorUnknown(t *testi
 	}
 }
 
+func TestExpiredMonitorDesktopTasksCandidateSQLSelectsOnlyExpiredInProgressMonitorTasks(t *testing.T) {
+	t.Parallel()
+
+	normalized := normalizeSQLForDesktopTaskTest(expiredMonitorDesktopTasksCandidateSQL())
+	for _, fragment := range []string{
+		"t.kind = 'monitor'",
+		"t.status = 'in_progress'",
+		"t.monitor_task_id IS NOT NULL",
+		"t.lease_expires_at IS NOT NULL",
+		"t.lease_expires_at < NOW()",
+		"ORDER BY t.lease_expires_at ASC",
+		"FOR UPDATE OF t SKIP LOCKED",
+	} {
+		if !strings.Contains(normalized, fragment) {
+			t.Fatalf("expired monitor candidate query missing %q: %s", fragment, normalized)
+		}
+	}
+}
+
+func TestFailExpiredMonitorDesktopTasksSQLMarksTerminalFailure(t *testing.T) {
+	t.Parallel()
+
+	normalized := normalizeSQLForDesktopTaskTest(failExpiredMonitorDesktopTasksSQL())
+	for _, fragment := range []string{
+		"SET status = 'failed'",
+		"active_attempt_id = NULL",
+		"lease_token_hash = NULL",
+		"lease_expires_at = NULL",
+		"interrupt_reason = COALESCE(interrupt_reason, $3)",
+		"WHERE desktop_id = ANY($1::uuid[])",
+		"AND kind = 'monitor'",
+		"AND status = 'in_progress'",
+		"RETURNING desktop_id",
+	} {
+		if !strings.Contains(normalized, fragment) {
+			t.Fatalf("expired monitor update query missing %q: %s", fragment, normalized)
+		}
+	}
+}
+
 func TestClassifyLeaseErrorDefersQueuedMonitorTask(t *testing.T) {
 	t.Parallel()
 

server/migrations/20260623103000_add_monitor_task_lease_recovery_index.down.sql

@@ -0,0 +1 @@
+DROP INDEX CONCURRENTLY IF EXISTS idx_desktop_tasks_monitor_lease_recovery;

server/migrations/20260623103000_add_monitor_task_lease_recovery_index.up.sql

@@ -0,0 +1,6 @@
+CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_desktop_tasks_monitor_lease_recovery
+    ON desktop_tasks (lease_expires_at ASC, updated_at ASC, desktop_id ASC)
+    WHERE kind = 'monitor'
+      AND status = 'in_progress'
+      AND monitor_task_id IS NOT NULL
+      AND lease_expires_at IS NOT NULL;

server/migrations_ops/20260623103000_tune_monitoring_lease_recovery.down.sql

@@ -0,0 +1,5 @@
+UPDATE ops.scheduler_jobs
+SET interval_seconds = 1800,
+    config = config - 'suppress_empty_success_runs' - 'empty_success_metric_keys',
+    updated_at = NOW()
+WHERE job_key = 'monitoring_lease_recovery';

server/migrations_ops/20260623103000_tune_monitoring_lease_recovery.up.sql

@@ -0,0 +1,6 @@
+UPDATE ops.scheduler_jobs
+SET interval_seconds = 60,
+    description = '按过期租约索引限批回收超过租约时间仍未回调的监控任务，并释放卡住的桌面监控任务。',
+    config = config || '{"suppress_empty_success_runs":true,"empty_success_metric_keys":["expired_task_count","failed_desktop_monitor_count"]}'::jsonb,
+    updated_at = NOW()
+WHERE job_key = 'monitoring_lease_recovery';