fix(publish): prevent stuck publish queue and duplicate posting

The desktop publish queue could stall for a long time and end users assumed
the software was broken. Root cause: a hung adapter held its execution slot
with no wall-clock timeout while auto-renewing its lease forever, so the
server never reclaimed it and every queued task behind it stayed 等待发布.
Auto-recovery also risked silently re-posting a non-idempotent article.

Client (Electron):
- per-task wall-clock deadline + abort; progress-gated lease renewal that
  stops and aborts a stalled task instead of renewing it forever
- decouple the concurrency cap from CDP-induced CPU/memory pressure
  (admission gate instead of self-throttling collapse); 15s watchdog pump
- all adapter network I/O now has fetch timeouts and honors context.signal;
  bounded image-upload concurrency with per-image timeout
- surface live adapter progress, elapsed time, queue position and a
  working-vs-queued distinction in the publish view

Server (tenant-api):
- publish lease-recovery worker (every 3m) + supporting index: reclaim
  expired in_progress publish leases, requeue (<3 attempts) or terminal-fail
- 3-minute lease TTL with client-presence-gated extension; max 3 attempts

Idempotency (production-grade core):
- durable publish_submit_started_at marker, set before the irreversible
  platform submit POST; recovery and abort route a maybe-submitted task to
  unknown (manual reconcile, kept in the dedup set) instead of re-posting
- desktop UI requires explicit confirmation before retrying a possibly-
  already-published task

Verified: go build/vet/test (incl. resolvePublishRecoveryOutcome), vue-tsc,
vitest 141/141, gofmt all clean.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

server/cmd/scheduler/main.go

@@ -99,9 +99,13 @@ func main() {
 		app.Logger,
 		app.Config().Scheduler,
 	).WithKolGenerationService(kolGenerationSvc).WithConfigProvider(app.ConfigStore)
+	desktopTaskService := tenantapp.NewDesktopTaskService(app.DB, app.MonitoringDB, app.RabbitMQ, app.Logger).
+		WithCache(app.Cache).
+		WithRedis(app.Redis)
 	monitoringDailyTaskWorker := tenantapp.NewMonitoringDailyTaskWorker(monitoringService, monitoringCallbackService, app.Logger)
 	monitoringResultRecoveryWorker := internalscheduler.NewMonitoringResultRecoveryWorker(app.MonitoringDB, monitoringCallbackService, app.Logger)
 	monitoringLeaseRecoveryWorker := internalscheduler.NewMonitoringLeaseRecoveryWorker(app.MonitoringDB, 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).
@@ -141,6 +145,12 @@ func main() {
 				return monitoringLeaseRecoveryWorker.RunOnce(runCtx, jobCtx)
 			},
 		},
+		{
+			Key: "publish_lease_recovery",
+			Run: func(runCtx context.Context, jobCtx internalscheduler.JobRunContext) (map[string]any, error) {
+				return publishLeaseRecoveryWorker.RunOnce(runCtx, jobCtx)
+			},
+		},
 		{
 			Key: "monitoring_received_inspection",
 			Run: func(runCtx context.Context, jobCtx internalscheduler.JobRunContext) (map[string]any, error) {

server/internal/scheduler/publish_lease_recovery_worker.go

@@ -0,0 +1,114 @@
+package scheduler
+
+import (
+	"context"
+	"time"
+
+	"go.uber.org/zap"
+
+	tenantapp "github.com/geo-platform/tenant-api/internal/tenant/app"
+)
+
+const (
+	defaultPublishLeaseRecoveryInterval = 3 * time.Minute
+	defaultPublishLeaseRecoveryTimeout  = 30 * time.Second
+	defaultPublishLeaseRecoveryLimit    = 1000
+)
+
+type PublishLeaseRecoveryWorker struct {
+	service  *tenantapp.DesktopTaskService
+	logger   *zap.Logger
+	interval time.Duration
+	timeout  time.Duration
+	limit    int
+}
+
+func NewPublishLeaseRecoveryWorker(service *tenantapp.DesktopTaskService, logger *zap.Logger) *PublishLeaseRecoveryWorker {
+	return &PublishLeaseRecoveryWorker{
+		service:  service,
+		logger:   logger,
+		interval: defaultPublishLeaseRecoveryInterval,
+		timeout:  defaultPublishLeaseRecoveryTimeout,
+		limit:    defaultPublishLeaseRecoveryLimit,
+	}
+}
+
+func (w *PublishLeaseRecoveryWorker) Start(ctx context.Context) {
+	go w.Run(ctx)
+}
+
+func (w *PublishLeaseRecoveryWorker) Run(ctx context.Context) {
+	if w == nil || w.service == nil {
+		return
+	}
+	w.run(ctx)
+}
+
+func (w *PublishLeaseRecoveryWorker) run(ctx context.Context) {
+	w.runOnce(context.Background())
+
+	ticker := time.NewTicker(w.interval)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			w.runOnce(context.Background())
+		}
+	}
+}
+
+func (w *PublishLeaseRecoveryWorker) runOnce(parent context.Context) {
+	ctx, cancel := context.WithTimeout(parent, w.timeout)
+	defer cancel()
+
+	result, err := w.service.RecoverExpiredPublishTasks(ctx, w.limit)
+	if err != nil {
+		if w.logger != nil {
+			w.logger.Warn("publish lease recovery failed", zap.Error(err))
+		}
+		return
+	}
+
+	if (result.Requeued > 0 || result.Failed > 0 || result.Uncertain > 0) && w.logger != nil {
+		w.logger.Info("publish lease recovery completed",
+			zap.Int("requeued_task_count", result.Requeued),
+			zap.Int("failed_task_count", result.Failed),
+			zap.Int("uncertain_task_count", result.Uncertain),
+		)
+	}
+}
+
+func (w *PublishLeaseRecoveryWorker) RunOnce(parent context.Context, run JobRunContext) (map[string]any, error) {
+	if w == nil || w.service == nil {
+		return map[string]any{"skipped": true, "reason": "worker_not_ready"}, nil
+	}
+	startedAt := time.Now()
+	timeout := w.timeout
+	if run.Job != nil && run.Job.TimeoutSeconds > 0 {
+		timeout = time.Duration(run.Job.TimeoutSeconds) * time.Second
+	}
+	limit := w.limit
+	if run.Job != nil && run.Job.BatchSize != nil && *run.Job.BatchSize > 0 {
+		limit = *run.Job.BatchSize
+	}
+	if value := AsInt(run.Config, "batch_size", 0); value > 0 && (run.Job == nil || run.Job.BatchSize == nil) {
+		limit = value
+	}
+	ctx, cancel := context.WithTimeout(parent, timeout)
+	defer cancel()
+
+	result, err := w.service.RecoverExpiredPublishTasks(ctx, limit)
+	if err != nil {
+		return map[string]any{"stage": "recover"}, err
+	}
+	return map[string]any{
+		"requeued_task_count":  result.Requeued,
+		"failed_task_count":    result.Failed,
+		"uncertain_task_count": result.Uncertain,
+		"duration_ms":          time.Since(startedAt).Milliseconds(),
+		"batch_size":           limit,
+	}, nil
+}

server/internal/tenant/app/desktop_task_service.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"fmt"
 	"strings"
 	"time"
 
@@ -24,6 +25,8 @@ import (
 	"github.com/geo-platform/tenant-api/internal/tenant/repository"
 )
 
+const desktopPublishMaxAttempts = 3
+
 type DesktopTaskService struct {
 	pool           *pgxpool.Pool
 	monitoringPool *pgxpool.Pool
@@ -107,6 +110,10 @@ type ExtendDesktopTaskRequest struct {
 	LeaseToken string `json:"lease_token" binding:"required"`
 }
 
+type MarkPublishSubmitStartedRequest struct {
+	LeaseToken string `json:"lease_token" binding:"required"`
+}
+
 type CompleteDesktopTaskRequest struct {
 	LeaseToken string         `json:"lease_token" binding:"required"`
 	Status     string         `json:"status" binding:"required,oneof=succeeded failed unknown"`
@@ -200,8 +207,10 @@ func (s *DesktopTaskService) Lease(ctx context.Context, client *repository.Deskt
 		task, err = s.leaseQueuedDesktopTaskByID(ctx, client, *taskID, leaseParams)
 	case req.Kind != nil && strings.TrimSpace(*req.Kind) == "monitor":
 		task, err = s.leaseNextQueuedMonitorTask(ctx, client, leaseParams)
+	case req.Kind != nil && strings.TrimSpace(*req.Kind) == "publish":
+		task, err = s.leaseNextQueuedPublishTask(ctx, client, leaseParams)
 	default:
-		task, err = s.repo.LeaseNextQueued(ctx, client.ID, req.Kind, leaseParams)
+		task, err = s.leaseNextQueuedPublishTask(ctx, client, leaseParams)
 	}
 	if err != nil {
 		if errors.Is(err, pgx.ErrNoRows) {
@@ -438,6 +447,54 @@ func (s *DesktopTaskService) leaseNextQueuedMonitorTask(
 	return scanRepositoryDesktopTask(row)
 }
 
+func (s *DesktopTaskService) leaseNextQueuedPublishTask(
+	ctx context.Context,
+	client *repository.DesktopClient,
+	params repository.DesktopTaskLeaseParams,
+) (*repository.DesktopTask, error) {
+	row := s.pool.QueryRow(ctx, `
+		WITH candidate AS (
+			SELECT dt.desktop_id
+			FROM desktop_tasks AS dt
+			WHERE dt.target_client_id = $1
+			  AND dt.kind = 'publish'
+			  AND dt.status = 'queued'
+			  AND dt.attempts < $4
+			  AND (
+				NOT EXISTS (
+					SELECT 1
+					FROM desktop_publish_jobs AS j
+					WHERE j.desktop_id = dt.job_id
+					  AND j.status <> 'queued'
+				)
+			  )
+			ORDER BY dt.lane_weight DESC,
+				dt.priority DESC,
+				COALESCE(dt.enqueued_at, dt.created_at) ASC,
+				dt.created_at ASC,
+				dt.desktop_id ASC
+			LIMIT 1
+			FOR UPDATE OF dt SKIP LOCKED
+		)
+		UPDATE desktop_tasks AS t
+		SET active_attempt_id = $2,
+		    lease_token_hash = $3,
+		    lease_expires_at = now() + interval '3 minutes',
+		    status = 'in_progress',
+		    attempts = t.attempts + 1,
+		    started_at = COALESCE(t.started_at, now()),
+		    updated_at = now()
+		FROM candidate
+		WHERE t.desktop_id = candidate.desktop_id
+		RETURNING `+desktopTaskRepositoryReturningColumns,
+		client.ID,
+		params.AttemptID,
+		params.LeaseTokenHash,
+		desktopPublishMaxAttempts,
+	)
+	return scanRepositoryDesktopTask(row)
+}
+
 func (s *DesktopTaskService) leaseQueuedDesktopTaskByID(
 	ctx context.Context,
 	client *repository.DesktopClient,
@@ -457,6 +514,7 @@ func (s *DesktopTaskService) leaseQueuedDesktopTaskByID(
 			WHERE dt.desktop_id = $1
 			  AND dt.workspace_id = $2
 			  AND dt.status = 'queued'
+			  AND (dt.kind <> 'publish' OR dt.attempts < $9)
 			  AND (
 				(
 					dt.kind = 'monitor'
@@ -478,9 +536,10 @@ func (s *DesktopTaskService) leaseQueuedDesktopTaskByID(
 		SET target_client_id = CASE WHEN t.kind = 'monitor' THEN $4 ELSE t.target_client_id END,
 		    active_attempt_id = $7,
 		    lease_token_hash = $8,
-		    lease_expires_at = now() + interval '10 minutes',
+		    lease_expires_at = now() + CASE WHEN t.kind = 'publish' THEN interval '3 minutes' ELSE interval '10 minutes' END,
 		    status = 'in_progress',
 		    attempts = t.attempts + 1,
+		    started_at = COALESCE(t.started_at, now()),
 		    updated_at = now()
 		FROM candidate
 		WHERE t.desktop_id = candidate.desktop_id
@@ -493,6 +552,7 @@ func (s *DesktopTaskService) leaseQueuedDesktopTaskByID(
 		accountIDs,
 		params.AttemptID,
 		params.LeaseTokenHash,
+		desktopPublishMaxAttempts,
 	)
 	return scanRepositoryDesktopTask(row)
 }
@@ -638,6 +698,12 @@ func (s *DesktopTaskService) Extend(ctx context.Context, client *repository.Desk
 	if err := s.dropStaleMonitorTasksForClient(ctx, client); err != nil {
 		return nil, err
 	}
+	if s.redis != nil {
+		presence := loadDesktopClientPresence(ctx, s.redis, []uuid.UUID{client.ID})
+		if presence != nil && !presence[client.ID] {
+			return nil, response.ErrConflict(40986, "desktop_client_offline", "desktop client presence has expired")
+		}
+	}
 
 	task, err := s.repo.ExtendLease(ctx, desktopID, HashDesktopClientToken(strings.TrimSpace(req.LeaseToken)))
 	if err != nil {
@@ -653,6 +719,34 @@ func (s *DesktopTaskService) Extend(ctx context.Context, client *repository.Desk
 	return &view, nil
 }
 
+// MarkPublishSubmitStarted records a durable intent marker right before the client performs the
+// irreversible platform submit POST. Lease-recovery uses it to avoid silently re-posting a
+// non-idempotent article: once submit may have started, recovery routes the task to 'unknown'
+// (manual reconcile) instead of auto-requeueing it. Best-effort — a stale/lost lease is a no-op
+// and must never block the in-flight publish.
+func (s *DesktopTaskService) MarkPublishSubmitStarted(ctx context.Context, client *repository.DesktopClient, taskID uuid.UUID, leaseToken string) error {
+	if s == nil || s.pool == nil || client == nil {
+		return nil
+	}
+	leaseToken = strings.TrimSpace(leaseToken)
+	if leaseToken == "" {
+		return response.ErrBadRequest(40001, "invalid_params", "lease_token is required")
+	}
+	if _, err := s.pool.Exec(ctx, `
+		UPDATE desktop_tasks
+		SET publish_submit_started_at = COALESCE(publish_submit_started_at, NOW()),
+		    updated_at = NOW()
+		WHERE desktop_id = $1
+		  AND target_client_id = $2
+		  AND kind = 'publish'
+		  AND status = 'in_progress'
+		  AND lease_token_hash = $3
+	`, taskID, client.ID, HashDesktopClientToken(leaseToken)); err != nil {
+		return response.ErrInternal(50200, "desktop_task_submit_marker_failed", "failed to mark publish submit started")
+	}
+	return nil
+}
+
 func (s *DesktopTaskService) Complete(ctx context.Context, client *repository.DesktopClient, desktopID uuid.UUID, req CompleteDesktopTaskRequest) (*DesktopTaskView, error) {
 	if client == nil {
 		return nil, response.ErrUnauthorized(40108, "desktop_client_missing", "desktop client context is required")
@@ -1320,6 +1414,33 @@ type recoveredDesktopTaskLease struct {
 	Kind          string
 	Status        string
 	ActiveAttempt pgtype.UUID
+	Attempts      int
+	SubmitStarted bool
+	ErrorJSON     []byte
+}
+
+type PublishLeaseRecoveryResult struct {
+	Requeued  int `json:"requeued"`
+	Failed    int `json:"failed"`
+	Uncertain int `json:"uncertain"`
+}
+
+// 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
+// is set), the article may already exist on the platform. Auto-requeueing would silently
+// re-post a non-idempotent article, so we route the task to 'unknown' (kept in the dedup
+// active set, surfaced for manual reconcile) instead. Tasks that never reached submit are
+// safe to auto-requeue while attempts remain, and give up to 'failed' once exhausted.
+func resolvePublishRecoveryOutcome(submitStarted bool, attempts int) (status string, payloadKind string) {
+	switch {
+	case submitStarted:
+		return "unknown", "publish_uncertain"
+	case attempts >= desktopPublishMaxAttempts:
+		return "failed", "publish_final"
+	default:
+		return "queued", "publish"
+	}
 }
 
 func (s *DesktopTaskService) recoverClientTasksOnStartup(ctx context.Context, client *repository.DesktopClient) error {
@@ -1347,6 +1468,14 @@ func (s *DesktopTaskService) recoverClientTasks(
 	if err != nil {
 		return response.ErrInternal(50195, "desktop_task_recovery_payload_failed", "failed to encode publish desktop task recovery payload")
 	}
+	publishFinalErrorJSON, _, _, err := desktopTaskRecoveryPayload(mode, "publish_final")
+	if err != nil {
+		return response.ErrInternal(50195, "desktop_task_recovery_payload_failed", "failed to encode publish desktop task recovery payload")
+	}
+	publishUncertainErrorJSON, _, _, err := desktopTaskRecoveryPayload(mode, "publish_uncertain")
+	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")
 	if err != nil {
 		return response.ErrInternal(50195, "desktop_task_recovery_payload_failed", "failed to encode monitor desktop task recovery payload")
@@ -1378,6 +1507,8 @@ func (s *DesktopTaskService) recoverClientTasks(
 			&item.Kind,
 			&item.Status,
 			&item.ActiveAttempt,
+			&item.Attempts,
+			&item.SubmitStarted,
 		); 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")
@@ -1389,6 +1520,7 @@ func (s *DesktopTaskService) recoverClientTasks(
 		return response.ErrInternal(50198, "desktop_task_recovery_scan_failed", "failed to iterate recovered desktop tasks")
 	}
 
+	publishOutcomes := make([]*desktopPublishSyncOutcome, 0)
 	for index := range recovered {
 		item := &recovered[index]
 		if item.Kind == "monitor" {
@@ -1412,22 +1544,49 @@ func (s *DesktopTaskService) recoverClientTasks(
 			continue
 		}
 
+		finalStatus, payloadKind := resolvePublishRecoveryOutcome(item.SubmitStarted, item.Attempts)
+		nextErrorJSON := publishErrorJSON
+		switch payloadKind {
+		case "publish_final":
+			nextErrorJSON = publishFinalErrorJSON
+		case "publish_uncertain":
+			nextErrorJSON = publishUncertainErrorJSON
+		}
 		if _, execErr := tx.Exec(ctx, `
 			UPDATE desktop_tasks
-			SET status = 'unknown',
-			    error = $2,
+			SET status = $2,
+			    error = $3,
+			    result = CASE WHEN $2 = 'queued' THEN NULL ELSE result END,
 			    active_attempt_id = NULL,
 			    lease_token_hash = NULL,
 			    lease_expires_at = NULL,
 			    interrupted_at = COALESCE(interrupted_at, NOW()),
-			    interrupt_reason = COALESCE(interrupt_reason, $3),
+			    interrupt_reason = COALESCE(interrupt_reason, $4),
+			    enqueued_at = CASE WHEN $2 = 'queued' THEN NOW() ELSE enqueued_at END,
 			    updated_at = NOW()
 			WHERE desktop_id = $1
-		`, item.TaskID, publishErrorJSON, publishReason); execErr != nil {
+		`, item.TaskID, finalStatus, nextErrorJSON, publishReason); execErr != nil {
 			s.logWarn("desktop publish 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 = "unknown"
+		item.Status = finalStatus
+		item.ErrorJSON = nextErrorJSON
+		// Only terminal 'failed' tasks finalize the publish_record; 'unknown' (may already be
+		// published) is left in the dedup active set for manual reconcile, 'queued' is requeued.
+		if finalStatus == "failed" {
+			task, getErr := repo.GetByDesktopID(ctx, item.TaskID, item.WorkspaceID)
+			if getErr != nil {
+				s.logWarn("desktop publish task recovery reload failed", getErr, 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")
+			}
+			publishOutcome, syncErr := syncDesktopPublishTaskState(ctx, tx, task)
+			if syncErr != nil {
+				return syncErr
+			}
+			if publishOutcome != nil {
+				publishOutcomes = append(publishOutcomes, publishOutcome)
+			}
+		}
 	}
 
 	for _, item := range recovered {
@@ -1442,7 +1601,10 @@ func (s *DesktopTaskService) recoverClientTasks(
 		}
 
 		finalStatus := literalStringPtr(item.Status)
-		errorJSON := publishErrorJSON
+		errorJSON := item.ErrorJSON
+		if len(errorJSON) == 0 {
+			errorJSON = publishErrorJSON
+		}
 		if item.Kind == "monitor" {
 			finalStatus = literalStringPtr("aborted")
 			errorJSON = monitorErrorJSON
@@ -1462,6 +1624,8 @@ func (s *DesktopTaskService) recoverClientTasks(
 		return response.ErrInternal(50199, "desktop_task_recovery_commit_failed", "failed to commit desktop task recovery")
 	}
 
+	s.afterRecoveredPublishOutcomes(ctx, publishOutcomes)
+
 	for _, item := range recovered {
 		task, getErr := s.repo.GetByDesktopID(ctx, item.TaskID, item.WorkspaceID)
 		if getErr != nil {
@@ -1470,7 +1634,7 @@ func (s *DesktopTaskService) recoverClientTasks(
 		}
 
 		eventType := "task_completed"
-		if item.Kind == "monitor" {
+		if item.Status == "queued" {
 			eventType = "task_available"
 		}
 		s.publishTaskEvent(ctx, task, eventType)
@@ -1488,15 +1652,199 @@ func (s *DesktopTaskService) recoverClientTasks(
 	return nil
 }
 
+func (s *DesktopTaskService) RecoverExpiredPublishTasks(ctx context.Context, limit int) (PublishLeaseRecoveryResult, error) {
+	var result PublishLeaseRecoveryResult
+	if s == nil || s.pool == nil {
+		return result, nil
+	}
+	if limit <= 0 {
+		limit = 1000
+	}
+
+	publishErrorJSON, publishReason, _, err := desktopTaskRecoveryPayload(desktopTaskRecoveryModeLeaseExpiry, "publish")
+	if err != nil {
+		return result, response.ErrInternal(50195, "desktop_task_recovery_payload_failed", "failed to encode publish desktop task recovery payload")
+	}
+	publishFinalErrorJSON, _, _, err := desktopTaskRecoveryPayload(desktopTaskRecoveryModeLeaseExpiry, "publish_final")
+	if err != nil {
+		return result, response.ErrInternal(50195, "desktop_task_recovery_payload_failed", "failed to encode publish desktop task recovery payload")
+	}
+	publishUncertainErrorJSON, _, _, err := desktopTaskRecoveryPayload(desktopTaskRecoveryModeLeaseExpiry, "publish_uncertain")
+	if err != nil {
+		return result, response.ErrInternal(50195, "desktop_task_recovery_payload_failed", "failed to encode publish 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, `
+		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
+		FROM desktop_tasks AS t
+		WHERE t.kind = 'publish'
+		  AND t.status = 'in_progress'
+		  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
+	`, limit)
+	if err != nil {
+		s.logWarn("expired desktop publish 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)
+	for rows.Next() {
+		var item recoveredDesktopTaskLease
+		if scanErr := rows.Scan(
+			&item.TaskID,
+			&item.WorkspaceID,
+			&item.Kind,
+			&item.Status,
+			&item.ActiveAttempt,
+			&item.Attempts,
+			&item.SubmitStarted,
+		); scanErr != nil {
+			s.logWarn("expired desktop publish recovery scan failed", scanErr)
+			return result, response.ErrInternal(50198, "desktop_task_recovery_scan_failed", "failed to parse recovered desktop tasks")
+		}
+		recovered = append(recovered, item)
+	}
+	if err := rows.Err(); err != nil {
+		s.logWarn("expired desktop publish recovery rows failed", err)
+		return result, response.ErrInternal(50198, "desktop_task_recovery_scan_failed", "failed to iterate recovered desktop tasks")
+	}
+
+	repo := repository.NewDesktopTaskRepository(tx)
+	publishOutcomes := make([]*desktopPublishSyncOutcome, 0)
+	for index := range recovered {
+		item := &recovered[index]
+		finalStatus, payloadKind := resolvePublishRecoveryOutcome(item.SubmitStarted, item.Attempts)
+		nextErrorJSON := publishErrorJSON
+		switch payloadKind {
+		case "publish_final":
+			nextErrorJSON = publishFinalErrorJSON
+		case "publish_uncertain":
+			nextErrorJSON = publishUncertainErrorJSON
+		}
+		if _, execErr := tx.Exec(ctx, `
+			UPDATE desktop_tasks
+			SET status = $2,
+			    error = $3,
+			    result = CASE WHEN $2 = 'queued' THEN NULL ELSE result END,
+			    active_attempt_id = NULL,
+			    lease_token_hash = NULL,
+			    lease_expires_at = NULL,
+			    interrupted_at = COALESCE(interrupted_at, NOW()),
+			    interrupt_reason = COALESCE(interrupt_reason, $4),
+			    enqueued_at = CASE WHEN $2 = 'queued' THEN NOW() ELSE enqueued_at END,
+			    updated_at = NOW()
+			WHERE desktop_id = $1
+			  AND status = 'in_progress'
+		`, item.TaskID, finalStatus, nextErrorJSON, publishReason); execErr != nil {
+			s.logWarn("expired desktop publish recovery update failed", execErr, zap.String("task_id", item.TaskID.String()))
+			return result, response.ErrInternal(50197, "desktop_task_recovery_query_failed", "failed to select recoverable desktop tasks")
+		}
+
+		item.Status = finalStatus
+		item.ErrorJSON = nextErrorJSON
+		switch finalStatus {
+		case "failed":
+			result.Failed++
+			task, getErr := repo.GetByDesktopID(ctx, item.TaskID, item.WorkspaceID)
+			if getErr != nil {
+				s.logWarn("expired desktop publish recovery reload failed", getErr, zap.String("task_id", item.TaskID.String()))
+				return result, response.ErrInternal(50197, "desktop_task_recovery_query_failed", "failed to select recoverable desktop tasks")
+			}
+			publishOutcome, syncErr := syncDesktopPublishTaskState(ctx, tx, task)
+			if syncErr != nil {
+				return result, syncErr
+			}
+			if publishOutcome != nil {
+				publishOutcomes = append(publishOutcomes, publishOutcome)
+			}
+		case "unknown":
+			// May already be published — keep as 'unknown' (still in the dedup active set) for
+			// manual reconcile. Do NOT sync to a terminal publish_record state and do NOT requeue.
+			result.Uncertain++
+		default:
+			result.Requeued++
+		}
+	}
+
+	for _, item := range recovered {
+		if !item.ActiveAttempt.Valid {
+			continue
+		}
+
+		attemptID, convErr := uuid.FromBytes(item.ActiveAttempt.Bytes[:])
+		if convErr != nil {
+			s.logWarn("expired desktop publish 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(item.Status),
+			Error:       item.ErrorJSON,
+		}); finishErr != nil {
+			s.logWarn("expired desktop publish 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")
+	}
+
+	s.afterRecoveredPublishOutcomes(ctx, publishOutcomes)
+
+	for _, item := range recovered {
+		task, getErr := s.repo.GetByDesktopID(ctx, item.TaskID, item.WorkspaceID)
+		if getErr != nil {
+			s.logWarn("expired desktop publish recovery reload after commit failed", getErr, zap.String("task_id", item.TaskID.String()))
+			continue
+		}
+		eventType := "task_completed"
+		if item.Status == "queued" {
+			eventType = "task_available"
+		}
+		s.publishTaskEvent(ctx, task, eventType)
+	}
+
+	if len(recovered) > 0 && s.logger != nil {
+		s.logger.Info("expired desktop publish tasks recovered",
+			zap.Int("requeued", result.Requeued),
+			zap.Int("failed", result.Failed),
+			zap.Int("uncertain", result.Uncertain),
+		)
+	}
+
+	return result, nil
+}
+
 func desktopTaskRecoverySelectQuery(mode desktopTaskRecoveryMode) string {
 	query := `
 		SELECT
 			desktop_id,
-			workspace_id,
-			kind,
-			status,
-			active_attempt_id
-		FROM desktop_tasks
+				workspace_id,
+				kind,
+				status,
+				active_attempt_id,
+				attempts,
+				publish_submit_started_at IS NOT NULL AS submit_started
+			FROM desktop_tasks
 		WHERE target_client_id = $1
 		  AND workspace_id = $2
 		  AND status = 'in_progress'
@@ -1517,36 +1865,62 @@ func desktopTaskRecoveryPayload(mode desktopTaskRecoveryMode, kind string) ([]by
 	reason := strings.TrimSpace(string(mode))
 	message := "desktop task was recovered after the client lost the active lease"
 	source := "desktop_task_recovery"
+	isPublishFinal := kind == "publish_final"
+	isPublishUncertain := kind == "publish_uncertain"
+
+	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"
+	}
 
 	switch mode {
 	case desktopTaskRecoveryModeStartup:
 		source = "desktop_client_startup"
-		if kind == "monitor" {
+		if isPublishFinal {
+			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 kind == "monitor" {
 			message = "desktop client restarted while the monitor task was in progress; task has been re-queued"
 		} else {
-			message = "desktop client restarted while the publish task was in progress; task has been moved to unknown for manual reconcile"
+			message = "desktop client restarted while the publish task was in progress; task has been re-queued for retry"
 		}
 	case desktopTaskRecoveryModeDisconnect:
 		source = "desktop_client_offline"
-		if kind == "monitor" {
+		if isPublishFinal {
+			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 kind == "monitor" {
 			message = "desktop client went offline while the monitor task was in progress; task has been re-queued"
 		} else {
-			message = "desktop client went offline while the publish task was in progress; task has been moved to unknown for manual reconcile"
+			message = "desktop client went offline while the publish task was in progress; task has been re-queued for retry"
 		}
 	case desktopTaskRecoveryModeLeaseExpiry:
 		source = "desktop_task_lease_expiry"
-		if kind == "monitor" {
+		if isPublishFinal {
+			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" {
 			message = "desktop task lease expired before monitor completion; task has been re-queued"
 		} else {
-			message = "desktop task lease expired before publish completion; task has been moved to unknown for manual reconcile"
+			message = "desktop task lease expired before publish completion; task has been re-queued for retry"
 		}
 	}
 
-	payload, err := marshalOptionalJSON(map[string]any{
+	fields := map[string]any{
 		"reason":  reason,
 		"message": message,
 		"source":  source,
-	})
+	}
+	if isPublishUncertain {
+		// Surfaced to the desktop UI so a manual retry of a maybe-already-published task asks for
+		// explicit confirmation instead of silently re-posting.
+		fields["publish_submit_uncertain"] = true
+	}
+	payload, err := marshalOptionalJSON(fields)
 	if err != nil {
 		return nil, "", "", err
 	}
@@ -1781,6 +2155,20 @@ func (s *DesktopTaskService) publishTaskEvent(ctx context.Context, task *reposit
 	publishDesktopTaskLifecycleEvent(ctx, s.messaging, s.logger, task, eventType)
 }
 
+func (s *DesktopTaskService) afterRecoveredPublishOutcomes(ctx context.Context, outcomes []*desktopPublishSyncOutcome) {
+	for _, outcome := range outcomes {
+		if outcome == nil {
+			continue
+		}
+		invalidateArticleCaches(ctx, s.cache, outcome.TenantID, &outcome.ArticleID)
+		if outcome.ArticleAlias != nil {
+			if syncErr := s.syncMonitoringArticleAlias(ctx, outcome.ArticleAlias); syncErr != nil {
+				s.logWarn("monitoring article alias sync failed after publish recovery", syncErr, zap.Int64("article_id", outcome.ArticleID))
+			}
+		}
+	}
+}
+
 func reconcileDesktopTaskEventType(_ string, status string) string {
 	if status == "retry" {
 		return "task_available"

server/internal/tenant/app/desktop_task_service_test.go

@@ -17,6 +17,8 @@ func TestDesktopTaskRecoverySelectQueryMatchesRecoveredLeaseScanOrder(t *testing
 		"kind",
 		"status",
 		"active_attempt_id",
+		"attempts",
+		"publish_submit_started_at",
 	}
 
 	if len(gotColumns) != len(wantColumns) {
@@ -55,6 +57,45 @@ func TestReconcileRetryPublishesTaskAvailableForAllDesktopTaskKinds(t *testing.T
 	}
 }
 
+func TestDesktopTaskRecoveryPayloadPublishLeaseExpiryRequeuesInsteadOfUnknown(t *testing.T) {
+	t.Parallel()
+
+	payload, reason, message, err := desktopTaskRecoveryPayload(desktopTaskRecoveryModeLeaseExpiry, "publish")
+	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, "re-queued for retry") {
+		t.Fatalf("message = %q, want re-queued for retry", message)
+	}
+	if strings.Contains(message, "unknown") {
+		t.Fatalf("message must not move publish recovery to unknown: %q", message)
+	}
+	if !strings.Contains(string(payload), "desktop_task_lease_expiry") {
+		t.Fatalf("payload missing source: %s", payload)
+	}
+}
+
+func TestDesktopTaskRecoveryPayloadPublishFinalFailsAfterMaxAttempts(t *testing.T) {
+	t.Parallel()
+
+	payload, _, message, err := desktopTaskRecoveryPayload(desktopTaskRecoveryModeLeaseExpiry, "publish_final")
+	if err != nil {
+		t.Fatalf("desktopTaskRecoveryPayload returned error: %v", err)
+	}
+	if !strings.Contains(message, "marked failed") {
+		t.Fatalf("message = %q, want marked failed", message)
+	}
+	if !strings.Contains(message, "3 attempts") {
+		t.Fatalf("message = %q, want max attempts", message)
+	}
+	if !strings.Contains(string(payload), "marked failed") {
+		t.Fatalf("payload missing final failure message: %s", payload)
+	}
+}
+
 func TestReconcileNonRetryPublishesTaskReconciled(t *testing.T) {
 	t.Parallel()
 
@@ -63,6 +104,38 @@ func TestReconcileNonRetryPublishesTaskReconciled(t *testing.T) {
 	}
 }
 
+func TestResolvePublishRecoveryOutcome(t *testing.T) {
+	t.Parallel()
+
+	cases := []struct {
+		name          string
+		submitStarted bool
+		attempts      int
+		wantStatus    string
+		wantKind      string
+	}{
+		{"pre-submit first attempt requeues", false, 1, "queued", "publish"},
+		{"pre-submit just below cap requeues", false, desktopPublishMaxAttempts - 1, "queued", "publish"},
+		{"pre-submit at cap fails terminally", false, desktopPublishMaxAttempts, "failed", "publish_final"},
+		{"pre-submit above cap fails terminally", false, desktopPublishMaxAttempts + 1, "failed", "publish_final"},
+		// Submit may have happened: never auto-requeue or silently fail — route to unknown
+		// (manual reconcile) regardless of remaining attempts, to avoid duplicate publishing.
+		{"submit started first attempt is uncertain", true, 1, "unknown", "publish_uncertain"},
+		{"submit started at cap is still uncertain", true, desktopPublishMaxAttempts, "unknown", "publish_uncertain"},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			gotStatus, gotKind := resolvePublishRecoveryOutcome(tc.submitStarted, tc.attempts)
+			if gotStatus != tc.wantStatus || gotKind != tc.wantKind {
+				t.Fatalf("resolvePublishRecoveryOutcome(%v, %d) = (%q, %q), want (%q, %q)",
+					tc.submitStarted, tc.attempts, gotStatus, gotKind, tc.wantStatus, tc.wantKind)
+			}
+		})
+	}
+}
+
 func recoverDesktopTaskSelectColumns(query string) []string {
 	re := regexp.MustCompile(`(?is)select\s+(.*?)\s+from\s+desktop_tasks`)
 	match := re.FindStringSubmatch(query)

server/internal/tenant/repository/generated/desktop_task.sql.go

@@ -420,7 +420,7 @@ func (q *Queries) CreateDesktopTaskAttempt(ctx context.Context, arg CreateDeskto
 
 const extendDesktopTaskLease = `-- name: ExtendDesktopTaskLease :one
 UPDATE desktop_tasks
-SET lease_expires_at = now() + interval '10 minutes',
+SET lease_expires_at = now() + CASE WHEN kind = 'publish' THEN interval '3 minutes' ELSE interval '10 minutes' END,
     updated_at = now()
 WHERE desktop_id = $1
   AND lease_token_hash = $2
@@ -564,6 +564,10 @@ WITH candidate AS (
         $4::text IS NULL
         OR dt.kind = $4::text
       )
+      AND (
+        dt.kind <> 'publish'
+        OR dt.attempts < 3
+      )
       AND (
         dt.kind <> 'publish'
         OR NOT EXISTS (
@@ -584,9 +588,10 @@ WITH candidate AS (
 UPDATE desktop_tasks AS t
 SET active_attempt_id = $1,
     lease_token_hash = $2,
-    lease_expires_at = now() + interval '10 minutes',
+    lease_expires_at = now() + CASE WHEN t.kind = 'publish' THEN interval '3 minutes' ELSE interval '10 minutes' END,
     status = 'in_progress',
     attempts = t.attempts + 1,
+    started_at = COALESCE(t.started_at, now()),
     updated_at = now()
 FROM candidate
 WHERE t.desktop_id = candidate.desktop_id
@@ -650,13 +655,18 @@ const leaseQueuedDesktopTaskByDesktopID = `-- name: LeaseQueuedDesktopTaskByDesk
 UPDATE desktop_tasks
 SET active_attempt_id = $1,
     lease_token_hash = $2,
-    lease_expires_at = now() + interval '10 minutes',
+    lease_expires_at = now() + CASE WHEN kind = 'publish' THEN interval '3 minutes' ELSE interval '10 minutes' END,
     status = 'in_progress',
     attempts = attempts + 1,
+    started_at = COALESCE(started_at, now()),
     updated_at = now()
 WHERE desktop_id = $3
   AND target_client_id = $4
   AND status = 'queued'
+  AND (
+    kind <> 'publish'
+    OR attempts < 3
+  )
 RETURNING id, desktop_id, job_id, tenant_id, workspace_id, target_account_id, target_client_id, platform_id, kind, payload, status, dedup_key, active_attempt_id, lease_token_hash, lease_expires_at, attempts, result, error, created_at, updated_at, priority, lane, lane_weight, source, scheduler_group_key, monitor_task_id, supersedes_task_id, control_flags, interrupt_generation, started_at, interrupted_at, interrupt_reason, enqueued_at
 `
 
@@ -754,7 +764,7 @@ SET status = CASE
     active_attempt_id = NULL,
     lease_token_hash = NULL,
     lease_expires_at = NULL,
-    attempts = attempts + CASE WHEN $1::text = 'retry' THEN 1 ELSE 0 END,
+    attempts = CASE WHEN $1::text = 'retry' THEN 0 ELSE attempts END,
     updated_at = now()
 WHERE desktop_id = $4
   AND workspace_id = $5

server/internal/tenant/repository/generated/desktop_task_sql_test.go

@@ -17,4 +17,29 @@ func TestLeaseNextQueuedDesktopTaskLocksOnlyDesktopTasks(t *testing.T) {
 	if !strings.Contains(query, "NOT EXISTS") {
 		t.Fatalf("lease query must filter non-queued publish jobs without joining the lock target; query:\n%s", query)
 	}
+	if !strings.Contains(query, "dt.attempts < 3") {
+		t.Fatalf("lease query must cap publish retries; query:\n%s", query)
+	}
+	if !strings.Contains(query, "interval '3 minutes'") {
+		t.Fatalf("lease query must use short publish lease ttl; query:\n%s", query)
+	}
+}
+
+func TestExtendDesktopTaskLeaseUsesShortPublishTTL(t *testing.T) {
+	query := extendDesktopTaskLease
+
+	if !strings.Contains(query, "CASE WHEN kind = 'publish' THEN interval '3 minutes' ELSE interval '10 minutes' END") {
+		t.Fatalf("extend query must keep publish lease ttl short; query:\n%s", query)
+	}
+}
+
+func TestReconcileDesktopTaskRetryResetsAttempts(t *testing.T) {
+	query := reconcileDesktopTask
+
+	if !strings.Contains(query, "attempts = CASE WHEN $1::text = 'retry' THEN 0 ELSE attempts END") {
+		t.Fatalf("retry reconcile must reset attempts so manual retry can be leased; query:\n%s", query)
+	}
+	if strings.Contains(query, "attempts + CASE") {
+		t.Fatalf("retry reconcile must not increment attempts; query:\n%s", query)
+	}
 }

server/internal/tenant/repository/generated/models.go

@@ -634,6 +634,93 @@ type MediaPlatform struct {
 	UpdatedAt   pgtype.Timestamptz `json:"updated_at"`
 }
 
+type MediaSupplyOrder struct {
+	ID                  int64              `json:"id"`
+	TenantID            int64              `json:"tenant_id"`
+	WorkspaceID         int64              `json:"workspace_id"`
+	BrandID             int64              `json:"brand_id"`
+	UserID              int64              `json:"user_id"`
+	ArticleID           pgtype.Int8        `json:"article_id"`
+	Supplier            string             `json:"supplier"`
+	ModelID             int32              `json:"model_id"`
+	Status              string             `json:"status"`
+	Title               string             `json:"title"`
+	ContentSnapshot     string             `json:"content_snapshot"`
+	Remark              pgtype.Text        `json:"remark"`
+	OrderBrand          pgtype.Text        `json:"order_brand"`
+	ExternalOrderID     pgtype.Text        `json:"external_order_id"`
+	ExternalOrderCode   pgtype.Text        `json:"external_order_code"`
+	SupplierStatus      pgtype.Text        `json:"supplier_status"`
+	CostTotalCents      int64              `json:"cost_total_cents"`
+	SellTotalCents      int64              `json:"sell_total_cents"`
+	WalletDebitCents    int64              `json:"wallet_debit_cents"`
+	WalletDebitedAt     pgtype.Timestamptz `json:"wallet_debited_at"`
+	WalletRefundedAt    pgtype.Timestamptz `json:"wallet_refunded_at"`
+	RequestPayloadJson  []byte             `json:"request_payload_json"`
+	ResponsePayloadJson []byte             `json:"response_payload_json"`
+	ErrorMessage        pgtype.Text        `json:"error_message"`
+	AttemptCount        int32              `json:"attempt_count"`
+	NextAttemptAt       pgtype.Timestamptz `json:"next_attempt_at"`
+	QueuedAt            pgtype.Timestamptz `json:"queued_at"`
+	SubmittedAt         pgtype.Timestamptz `json:"submitted_at"`
+	CompletedAt         pgtype.Timestamptz `json:"completed_at"`
+	CreatedAt           pgtype.Timestamptz `json:"created_at"`
+	UpdatedAt           pgtype.Timestamptz `json:"updated_at"`
+	DeletedAt           pgtype.Timestamptz `json:"deleted_at"`
+}
+
+type MediaSupplyOrderItem struct {
+	ID                   int64              `json:"id"`
+	OrderID              int64              `json:"order_id"`
+	ResourceID           int64              `json:"resource_id"`
+	SupplierResourceID   string             `json:"supplier_resource_id"`
+	PriceType            string             `json:"price_type"`
+	ResourceNameSnapshot string             `json:"resource_name_snapshot"`
+	LockedCostPriceCents int64              `json:"locked_cost_price_cents"`
+	LockedSellPriceCents int64              `json:"locked_sell_price_cents"`
+	Status               string             `json:"status"`
+	ExternalArticleUrl   pgtype.Text        `json:"external_article_url"`
+	RawJson              []byte             `json:"raw_json"`
+	CreatedAt            pgtype.Timestamptz `json:"created_at"`
+	UpdatedAt            pgtype.Timestamptz `json:"updated_at"`
+}
+
+type MediaSupplySyncJob struct {
+	ID           int64              `json:"id"`
+	Supplier     string             `json:"supplier"`
+	ModelID      pgtype.Int4        `json:"model_id"`
+	Status       string             `json:"status"`
+	RequestedBy  pgtype.Int8        `json:"requested_by"`
+	StartedAt    pgtype.Timestamptz `json:"started_at"`
+	FinishedAt   pgtype.Timestamptz `json:"finished_at"`
+	ErrorMessage pgtype.Text        `json:"error_message"`
+	AttemptCount int32              `json:"attempt_count"`
+	StatsJson    []byte             `json:"stats_json"`
+	CreatedAt    pgtype.Timestamptz `json:"created_at"`
+	UpdatedAt    pgtype.Timestamptz `json:"updated_at"`
+}
+
+type MediaSupplyUserWallet struct {
+	TenantID     int64              `json:"tenant_id"`
+	UserID       int64              `json:"user_id"`
+	BalanceCents int64              `json:"balance_cents"`
+	CreatedAt    pgtype.Timestamptz `json:"created_at"`
+	UpdatedAt    pgtype.Timestamptz `json:"updated_at"`
+}
+
+type MediaSupplyWalletLedger struct {
+	ID                int64              `json:"id"`
+	TenantID          int64              `json:"tenant_id"`
+	UserID            int64              `json:"user_id"`
+	OrderID           pgtype.Int8        `json:"order_id"`
+	DeltaCents        int64              `json:"delta_cents"`
+	BalanceAfterCents int64              `json:"balance_after_cents"`
+	Reason            string             `json:"reason"`
+	Note              pgtype.Text        `json:"note"`
+	CreatedBy         pgtype.Int8        `json:"created_by"`
+	CreatedAt         pgtype.Timestamptz `json:"created_at"`
+}
+
 type Plan struct {
 	ID              int64              `json:"id"`
 	PlanCode        string             `json:"plan_code"`
@@ -797,6 +884,45 @@ type ScheduleTask struct {
 	ConsecutiveFailures  int32              `json:"consecutive_failures"`
 }
 
+type SupplierMediaPriceOverride struct {
+	ID             int64              `json:"id"`
+	ResourceID     int64              `json:"resource_id"`
+	PriceType      string             `json:"price_type"`
+	SellPriceCents int64              `json:"sell_price_cents"`
+	Enabled        bool               `json:"enabled"`
+	UpdatedBy      pgtype.Int8        `json:"updated_by"`
+	CreatedAt      pgtype.Timestamptz `json:"created_at"`
+	UpdatedAt      pgtype.Timestamptz `json:"updated_at"`
+}
+
+type SupplierMediaResource struct {
+	ID                 int64              `json:"id"`
+	Supplier           string             `json:"supplier"`
+	SupplierResourceID string             `json:"supplier_resource_id"`
+	ModelID            int32              `json:"model_id"`
+	Name               string             `json:"name"`
+	Status             string             `json:"status"`
+	CostPriceCents     int64              `json:"cost_price_cents"`
+	CostPricesJson     []byte             `json:"cost_prices_json"`
+	SalePriceLabel     pgtype.Text        `json:"sale_price_label"`
+	ResourceUrl        pgtype.Text        `json:"resource_url"`
+	BaiduWeight        pgtype.Int4        `json:"baidu_weight"`
+	ResourceRemark     pgtype.Text        `json:"resource_remark"`
+	CustomerVisible    bool               `json:"customer_visible"`
+	ChannelType        pgtype.Text        `json:"channel_type"`
+	Region             pgtype.Text        `json:"region"`
+	InclusionEffect    pgtype.Text        `json:"inclusion_effect"`
+	LinkType           pgtype.Text        `json:"link_type"`
+	PublishRate        pgtype.Text        `json:"publish_rate"`
+	DeliverySpeed      pgtype.Text        `json:"delivery_speed"`
+	SupplierUpdatedAt  pgtype.Timestamptz `json:"supplier_updated_at"`
+	LastSyncedAt       pgtype.Timestamptz `json:"last_synced_at"`
+	RawJson            []byte             `json:"raw_json"`
+	CreatedAt          pgtype.Timestamptz `json:"created_at"`
+	UpdatedAt          pgtype.Timestamptz `json:"updated_at"`
+	DeletedAt          pgtype.Timestamptz `json:"deleted_at"`
+}
+
 type TaskRecord struct {
 	ID           int64              `json:"id"`
 	TenantID     int64              `json:"tenant_id"`

server/internal/tenant/repository/queries/desktop_task.sql

@@ -69,6 +69,10 @@ WITH candidate AS (
         sqlc.narg(kind)::text IS NULL
         OR dt.kind = sqlc.narg(kind)::text
       )
+      AND (
+        dt.kind <> 'publish'
+        OR dt.attempts < 3
+      )
       AND (
         dt.kind <> 'publish'
         OR NOT EXISTS (
@@ -89,9 +93,10 @@ WITH candidate AS (
 UPDATE desktop_tasks AS t
 SET active_attempt_id = sqlc.arg(attempt_id),
     lease_token_hash = sqlc.arg(lease_token_hash),
-    lease_expires_at = now() + interval '10 minutes',
+    lease_expires_at = now() + CASE WHEN t.kind = 'publish' THEN interval '3 minutes' ELSE interval '10 minutes' END,
     status = 'in_progress',
     attempts = t.attempts + 1,
+    started_at = COALESCE(t.started_at, now()),
     updated_at = now()
 FROM candidate
 WHERE t.desktop_id = candidate.desktop_id
@@ -101,18 +106,23 @@ RETURNING t.*;
 UPDATE desktop_tasks
 SET active_attempt_id = sqlc.arg(attempt_id),
     lease_token_hash = sqlc.arg(lease_token_hash),
-    lease_expires_at = now() + interval '10 minutes',
+    lease_expires_at = now() + CASE WHEN kind = 'publish' THEN interval '3 minutes' ELSE interval '10 minutes' END,
     status = 'in_progress',
     attempts = attempts + 1,
+    started_at = COALESCE(started_at, now()),
     updated_at = now()
 WHERE desktop_id = sqlc.arg(desktop_id)
   AND target_client_id = sqlc.arg(client_id)
   AND status = 'queued'
+  AND (
+    kind <> 'publish'
+    OR attempts < 3
+  )
 RETURNING *;
 
 -- name: ExtendDesktopTaskLease :one
 UPDATE desktop_tasks
-SET lease_expires_at = now() + interval '10 minutes',
+SET lease_expires_at = now() + CASE WHEN kind = 'publish' THEN interval '3 minutes' ELSE interval '10 minutes' END,
     updated_at = now()
 WHERE desktop_id = sqlc.arg(desktop_id)
   AND lease_token_hash = sqlc.arg(lease_token_hash)
@@ -186,7 +196,7 @@ SET status = CASE
     active_attempt_id = NULL,
     lease_token_hash = NULL,
     lease_expires_at = NULL,
-    attempts = attempts + CASE WHEN sqlc.arg(status)::text = 'retry' THEN 1 ELSE 0 END,
+    attempts = CASE WHEN sqlc.arg(status)::text = 'retry' THEN 0 ELSE attempts END,
     updated_at = now()
 WHERE desktop_id = sqlc.arg(desktop_id)
   AND workspace_id = sqlc.arg(workspace_id)

server/internal/tenant/transport/desktop_task_handler.go

@@ -111,6 +111,26 @@ func (h *DesktopTaskHandler) Extend(c *gin.Context) {
 	response.Success(c, data)
 }
 
+func (h *DesktopTaskHandler) MarkPublishSubmitStarted(c *gin.Context) {
+	taskID, err := uuid.Parse(c.Param("id"))
+	if err != nil {
+		response.Error(c, response.ErrBadRequest(40084, "invalid_desktop_task_id", "task id must be a uuid"))
+		return
+	}
+
+	var req app.MarkPublishSubmitStartedRequest
+	if err := c.ShouldBindJSON(&req); err != nil {
+		response.Error(c, response.ErrBadRequest(40001, "invalid_params", err.Error()))
+		return
+	}
+
+	if err := h.svc.MarkPublishSubmitStarted(c.Request.Context(), MustDesktopClient(c.Request.Context()), taskID, req.LeaseToken); err != nil {
+		response.Error(c, err)
+		return
+	}
+	response.Success(c, gin.H{"ok": true})
+}
+
 func (h *DesktopTaskHandler) Result(c *gin.Context) {
 	desktopID, err := uuid.Parse(c.Param("id"))
 	if err != nil {

server/internal/tenant/transport/router.go

@@ -66,6 +66,7 @@ func RegisterRoutes(app *bootstrap.App) {
 	desktopAuth.POST("/tasks/lease", desktopTaskHandler.Lease)
 	desktopAuth.POST("/tasks/:id/lease", desktopTaskHandler.Lease)
 	desktopAuth.POST("/tasks/:id/extend", desktopTaskHandler.Extend)
+	desktopAuth.POST("/tasks/:id/publish-submit-marker", desktopTaskHandler.MarkPublishSubmitStarted)
 	desktopAuth.POST("/tasks/:id/cancel", desktopTaskHandler.Cancel)
 	desktopAuth.POST("/tasks/:id/result", desktopTaskHandler.Result)
 	desktopAuth.POST("/publish-tasks/:id/retry", desktopTaskHandler.RetryPublish)

server/migrations/20260530100000_add_publish_task_lease_recovery_index.down.sql

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

server/migrations/20260530100000_add_publish_task_lease_recovery_index.up.sql

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

server/migrations/20260530101000_add_desktop_task_publish_submit_marker.down.sql

@@ -0,0 +1,2 @@
+ALTER TABLE desktop_tasks
+    DROP COLUMN IF EXISTS publish_submit_started_at;

server/migrations/20260530101000_add_desktop_task_publish_submit_marker.up.sql

@@ -0,0 +1,6 @@
+-- Durable "submit started" intent marker for desktop publish tasks.
+-- Set immediately before the irreversible platform submit POST so lease-recovery can tell
+-- "definitely not submitted yet" (safe to auto-requeue) apart from "may already be published"
+-- (route to 'unknown' / manual reconcile and never silently re-post a non-idempotent article).
+ALTER TABLE desktop_tasks
+    ADD COLUMN IF NOT EXISTS publish_submit_started_at TIMESTAMPTZ;

server/migrations_ops/20260530100000_add_publish_lease_recovery_scheduler_job.down.sql

@@ -0,0 +1,8 @@
+DELETE FROM ops.scheduler_job_triggers
+WHERE job_key = 'publish_lease_recovery';
+
+DELETE FROM ops.scheduler_job_runs
+WHERE job_key = 'publish_lease_recovery';
+
+DELETE FROM ops.scheduler_jobs
+WHERE job_key = 'publish_lease_recovery';

server/migrations_ops/20260530100000_add_publish_lease_recovery_scheduler_job.up.sql

@@ -0,0 +1,16 @@
+INSERT INTO ops.scheduler_jobs
+    (job_key, display_name, category, description, enabled, schedule_type, interval_seconds, timezone, timeout_seconds, batch_size, config)
+VALUES
+    ('publish_lease_recovery', '发布租约回收', 'publish', '按过期租约限批回收卡住的桌面发布任务，超过最大重试后转失败。', true, 'interval', 180, 'Asia/Shanghai', 30, 1000, '{}'::jsonb)
+ON CONFLICT (job_key) DO UPDATE
+SET display_name = EXCLUDED.display_name,
+    category = EXCLUDED.category,
+    description = EXCLUDED.description,
+    enabled = EXCLUDED.enabled,
+    schedule_type = EXCLUDED.schedule_type,
+    interval_seconds = EXCLUDED.interval_seconds,
+    timezone = EXCLUDED.timezone,
+    timeout_seconds = EXCLUDED.timeout_seconds,
+    batch_size = EXCLUDED.batch_size,
+    config = ops.scheduler_jobs.config || EXCLUDED.config,
+    updated_at = NOW();