perf(admin-web): reduce redundant polling and cover active publish status

Unify list/detail poll intervals to 5s and skip refetch while a query is already in flight, replacing broad invalidateQueries cascades with targeted refetches. Poll now also triggers on active publish status via the new hasActivePublishStatus helper. Dedupe concurrent refreshBrands calls with a shared promise and skip the redundant AppShell refresh when already initialized. Gate publish-record loading on popover open. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
chore(admin-web): add favicon
2026-07-08 23:16:47 +08:00 · 2026-07-08 22:56:45 +08:00 · 2026-07-08 22:56:37 +08:00
19 changed files with 530 additions and 74 deletions
@@ -6,6 +6,11 @@
      name="viewport"
      content="width=device-width, initial-scale=1.0"
    />
+    <link
+      rel="icon"
+      href="/favicon.ico"
+      sizes="any"
+    />
    <title>省心推</title>
  </head>
  <body>
@@ -16,4 +21,3 @@
    ></script>
  </body>
 </html>
-
@@ -1,7 +1,7 @@
 <script setup lang="ts">
 import { CopyOutlined, LinkOutlined } from '@ant-design/icons-vue'
 import type { ArticleVersion, PublishRecord } from '@geo/shared-types'
-import { useQuery, useQueryClient } from '@tanstack/vue-query'
+import { useQuery } from '@tanstack/vue-query'
 import type { TableColumnsType } from 'ant-design-vue'
 import { message } from 'ant-design-vue'
 import { computed, ref, watch } from 'vue'
@@ -22,6 +22,7 @@ import {
  getPublishStatusMeta,
  getSourceTypeLabel,
  hasActiveGenerationStatus,
+  hasActivePublishStatus,
 } from '@/lib/display'
 import { useCompanyStore } from '@/stores/company'

@@ -34,7 +35,6 @@ const emit = defineEmits<{
  close: []
 }>()

-const queryClient = useQueryClient()
 const { t } = useI18n()
 const companyStore = useCompanyStore()
 const activeTab = ref('content')
@@ -43,6 +43,7 @@ const streamTitle = ref('')
 const streamMarkdown = ref('')
 const streamStatus = ref<string | null>(null)
 const streamFeatureEnabled = isGenerationStreamEnabled()
+const ARTICLE_DETAIL_POLL_INTERVAL_MS = 5000

 const detailQuery = useQuery({
  queryKey: computed(() => ['articles', 'detail', companyStore.currentBrandId, props.articleId]),
@@ -178,18 +179,30 @@ watch(
 )

 watch(
-  [() => props.open, () => props.articleId, () => detail.value?.generate_status],
-  ([open, articleId, generateStatus], _prev, onCleanup) => {
-    if (!open || !articleId || !hasActiveGenerationStatus(generateStatus)) {
+  [
+    () => props.open,
+    () => props.articleId,
+    () => detail.value?.generate_status,
+    () => detail.value?.publish_status,
+  ],
+  ([open, articleId, generateStatus, publishStatus], _prev, onCleanup) => {
+    const shouldPoll =
+      hasActiveGenerationStatus(generateStatus) || hasActivePublishStatus(publishStatus)
+
+    if (!open || !articleId || !shouldPoll) {
      return
    }

-    const timer = window.setInterval(
-      () => {
-        void refreshArticleData()
-      },
-      streamFeatureEnabled ? 5000 : 2000,
-    )
+    const timer = window.setInterval(() => {
+      if (
+        detailQuery.isFetching.value ||
+        versionsQuery.isFetching.value ||
+        publishRecordsQuery.isFetching.value
+      ) {
+        return
+      }
+      void refreshArticleData()
+    }, ARTICLE_DETAIL_POLL_INTERVAL_MS)

    onCleanup(() => {
      window.clearInterval(timer)
@@ -234,13 +247,7 @@ function resetStreamState(): void {
 }

 async function refreshArticleData(): Promise<void> {
-  await Promise.all([
-    detailQuery.refetch(),
-    versionsQuery.refetch(),
-    publishRecordsQuery.refetch(),
-    queryClient.invalidateQueries({ queryKey: ['articles'] }),
-    queryClient.invalidateQueries({ queryKey: ['workspace'] }),
-  ])
+  await Promise.all([detailQuery.refetch(), versionsQuery.refetch(), publishRecordsQuery.refetch()])
 }

 function resolvePublishLink(record: PublishRecord): string {
@@ -63,7 +63,9 @@ const publishRecordsQuery = useQuery({
    companyStore.currentBrandId,
    props.articleId,
  ]),
-  enabled: computed(() => shouldLoadRecords.value && Boolean(companyStore.currentBrandId)),
+  enabled: computed(
+    () => popoverOpen.value && shouldLoadRecords.value && Boolean(companyStore.currentBrandId),
+  ),
  queryFn: () => articlesApi.publishRecords(props.articleId as number),
  staleTime: 30 * 1000,
 })
@@ -17,7 +17,12 @@ import { articlesApi, promptRulesApi } from '@/lib/api'
 import { buildArticleClipboardContent, resolveArticleActionState } from '@/lib/article-list-actions'
 import { useArticleRegenerateAction } from '@/lib/article-regenerate-action'
 import { copyTextToClipboard } from '@/lib/clipboard'
-import { getGenerateStatusMeta, getPublishStatusMeta } from '@/lib/display'
+import {
+  getGenerateStatusMeta,
+  getPublishStatusMeta,
+  hasActiveGenerationStatus,
+  hasActivePublishStatus,
+} from '@/lib/display'
 import { formatError } from '@/lib/errors'
 import { formatPublishPlatformList } from '@/lib/publish-platforms'
 import { useCompanyStore } from '@/stores/company'
@@ -109,6 +114,7 @@ const listQuery = useQuery({
 })

 let pollingTimer: number | null = null
+const ARTICLE_LIST_POLL_INTERVAL_MS = 5000

 const deleteMutation = useMutation({
  mutationFn: (id: number) => articlesApi.remove(id),
@@ -205,8 +211,10 @@ function handleTableChange(nextPage: number, nextPageSize: number): void {
 function startPolling(): void {
  if (pollingTimer !== null) return
  pollingTimer = window.setInterval(() => {
-    void listQuery.refetch()
-  }, 3000)
+    if (!listQuery.isFetching.value) {
+      void listQuery.refetch()
+    }
+  }, ARTICLE_LIST_POLL_INTERVAL_MS)
 }

 function stopPolling(): void {
@@ -236,7 +244,9 @@ watch(
  () => listQuery.data.value?.items ?? [],
  (items) => {
    const hasActive = items.some(
-      (item) => item.generate_status === 'generating' || item.generate_status === 'running',
+      (item) =>
+        hasActiveGenerationStatus(item.generate_status) ||
+        hasActivePublishStatus(item.publish_status),
    )
    if (hasActive) {
      startPolling()
@@ -117,6 +117,7 @@ const columns = computed<TableColumnsType<InstantTaskItem>>(() => [
 ])

 let pollingTimer: number | null = null
+const TASK_LIST_POLL_INTERVAL_MS = 5000

 function applyFilters(): void {
  page.value = 1
@@ -153,8 +154,10 @@ function openTaskArticles(task: InstantTaskItem): void {
 function startPolling(): void {
  if (pollingTimer !== null) return
  pollingTimer = window.setInterval(() => {
-    void listQuery.refetch()
-  }, 3000)
+    if (!listQuery.isFetching.value) {
+      void listQuery.refetch()
+    }
+  }, TASK_LIST_POLL_INTERVAL_MS)
 }

 function stopPolling(): void {
@@ -235,12 +235,15 @@ function openTaskArticles(task: ScheduleTask): void {
 }

 let pollingTimer: number | null = null
+const TASK_LIST_POLL_INTERVAL_MS = 5000

 function startPolling(): void {
  if (pollingTimer !== null) return
  pollingTimer = window.setInterval(() => {
-    void listQuery.refetch()
-  }, 3000)
+    if (!listQuery.isFetching.value) {
+      void listQuery.refetch()
+    }
+  }, TASK_LIST_POLL_INTERVAL_MS)
 }

 function stopPolling(): void {
@@ -518,7 +518,7 @@ onMounted(() => {
  syncMobileState(mobileMediaQuery)
  mobileMediaQuery.addEventListener('change', syncMobileState)

-  if (!isMembershipBlocked.value) {
+  if (!isMembershipBlocked.value && !companyStore.initialized) {
    void companyStore.refreshBrands()
  }
 })
@@ -89,6 +89,18 @@ export function hasActiveGenerationStatus(status?: string | null): boolean {
  )
 }

+export function hasActivePublishStatus(status?: string | null): boolean {
+  const normalized = String(status ?? '')
+    .trim()
+    .toLowerCase()
+  return (
+    normalized === 'queued' ||
+    normalized === 'pending' ||
+    normalized === 'publishing' ||
+    normalized === 'running'
+  )
+}
+
 export function getPublishStatusMeta(status?: string | null): { label: string; color: string } {
  if (!status) {
    return { label: '--', color: 'default' }
@@ -15,6 +15,7 @@ export const useCompanyStore = defineStore('company', () => {
  const currentBrandId = ref<number | null>(readStoredCurrentBrandId())
  const loading = ref(false)
  const initialized = ref(false)
+  let refreshPromise: Promise<Brand[]> | null = null

  const currentBrand = computed(
    () => brands.value.find((brand) => brand.id === currentBrandId.value) ?? null,
@@ -56,15 +57,25 @@ export const useCompanyStore = defineStore('company', () => {
  }

  async function refreshBrands(): Promise<Brand[]> {
-    loading.value = true
-    try {
-      const nextBrands = normalizeBrandList(await brandsApi.list())
-      setBrands(nextBrands)
-      initialized.value = true
-      return nextBrands
-    } finally {
-      loading.value = false
+    if (refreshPromise) {
+      return refreshPromise
    }
+
+    loading.value = true
+    refreshPromise = brandsApi
+      .list()
+      .then((value) => {
+        const nextBrands = normalizeBrandList(value)
+        setBrands(nextBrands)
+        initialized.value = true
+        return nextBrands
+      })
+      .finally(() => {
+        loading.value = false
+        refreshPromise = null
+      })
+
+    return refreshPromise
  }

  function setBrands(value: Brand[] | null | undefined): void {
@@ -78,6 +89,7 @@ export const useCompanyStore = defineStore('company', () => {
    currentBrandId.value = null
    initialized.value = false
    loading.value = false
+    refreshPromise = null
    clearStoredCurrentBrandId()
  }

@@ -5,7 +5,7 @@ import { useMutation, useQuery, useQueryClient } from '@tanstack/vue-query'
 import type { TableColumnsType } from 'ant-design-vue'
 import { message } from 'ant-design-vue'
 import type { Dayjs } from 'dayjs'
-import { computed, reactive, ref } from 'vue'
+import { computed, onBeforeUnmount, reactive, ref, watch } from 'vue'
 import { useI18n } from 'vue-i18n'
 import { useRouter } from 'vue-router'

@@ -18,7 +18,7 @@ import { articlesApi } from '@/lib/api'
 import { buildArticleClipboardContent, resolveArticleActionState } from '@/lib/article-list-actions'
 import { useArticleRegenerateAction } from '@/lib/article-regenerate-action'
 import { copyTextToClipboard } from '@/lib/clipboard'
-import { getPublishStatusMeta } from '@/lib/display'
+import { getPublishStatusMeta, hasActivePublishStatus } from '@/lib/display'
 import { formatError } from '@/lib/errors'
 import { useCompanyStore } from '@/stores/company'

@@ -34,6 +34,8 @@ const selectedArticleId = ref<number | null>(null)
 const page = ref(1)
 const pageSize = ref(10)
 const createdRange = ref<[Dayjs, Dayjs] | null>(null)
+let articlePollingTimer: number | null = null
+const ARTICLE_LIST_POLL_INTERVAL_MS = 5000

 const draftFilters = reactive<{
  publish_status?: string
@@ -172,11 +174,51 @@ function openPreview(article: ArticleListItem): void {
  articleDrawerOpen.value = true
 }

+watch(
+  () => articleListQuery.data.value?.items || [],
+  (items: ArticleListItem[]) => {
+    const hasPublishingArticles = items.some((item) => hasActivePublishStatus(item.publish_status))
+
+    if (hasPublishingArticles) {
+      startArticlePolling()
+      return
+    }
+
+    stopArticlePolling()
+  },
+  { immediate: true },
+)
+
+onBeforeUnmount(() => {
+  stopArticlePolling()
+})
+
 function handleTableChange(nextPage: number, nextPageSize: number): void {
  page.value = nextPage
  pageSize.value = nextPageSize
 }

+function startArticlePolling(): void {
+  if (articlePollingTimer !== null) {
+    return
+  }
+
+  articlePollingTimer = window.setInterval(() => {
+    if (!articleListQuery.isFetching.value) {
+      void articleListQuery.refetch()
+    }
+  }, ARTICLE_LIST_POLL_INTERVAL_MS)
+}
+
+function stopArticlePolling(): void {
+  if (articlePollingTimer === null) {
+    return
+  }
+
+  window.clearInterval(articlePollingTimer)
+  articlePollingTimer = null
+}
+
 async function handleDelete(articleId: number): Promise<void> {
  await deleteMutation.mutateAsync(articleId)
 }
@@ -23,6 +23,7 @@ import {
  getGenerateStatusMeta,
  getPublishStatusMeta,
  hasActiveGenerationStatus,
+  hasActivePublishStatus,
 } from '@/lib/display'
 import { formatError } from '@/lib/errors'
 import { useCompanyStore } from '@/stores/company'
@@ -93,6 +94,7 @@ const articleListQuery = useQuery({
 })

 let articlePollingTimer: number | null = null
+const ARTICLE_LIST_POLL_INTERVAL_MS = 5000

 const deleteMutation = useMutation({
  mutationFn: (articleId: number) => articlesApi.remove(articleId),
@@ -180,8 +182,9 @@ watch(
    const hasGeneratingArticles = items.some((item) =>
      hasActiveGenerationStatus(item.generate_status),
    )
+    const hasPublishingArticles = items.some((item) => hasActivePublishStatus(item.publish_status))

-    if (hasGeneratingArticles) {
+    if (hasGeneratingArticles || hasPublishingArticles) {
      startArticlePolling()
      return
    }
@@ -245,9 +248,10 @@ function startArticlePolling(): void {
  }

  articlePollingTimer = window.setInterval(() => {
-    void queryClient.invalidateQueries({ queryKey: ['articles'] })
-    void articleListQuery.refetch()
-  }, 3000)
+    if (!articleListQuery.isFetching.value) {
+      void articleListQuery.refetch()
+    }
+  }, ARTICLE_LIST_POLL_INTERVAL_MS)
 }

 function stopArticlePolling(): void {
@@ -118,6 +118,7 @@ function inlineErrorPreview(messageText: string): string {

 function summaryForRecord(record: PublishRecord, status: string): string {
  const targetLabel = record.target_type === 'enterprise_site' ? '企业站点' : '目标平台'
+  const errorMessage = normalizeRecordErrorMessage(record.error_message, record.platform_id)
  switch (status) {
    case 'queued':
      return record.target_type === 'enterprise_site'
@@ -134,6 +135,9 @@ function summaryForRecord(record: PublishRecord, status: string): string {
        ? '本次企业站发布失败，请检查站点连接或发布错误。'
        : '本次发送失败，请检查账号状态或平台提示后重新提交。'
    case 'cancelled':
+      if (errorMessage && /等待通道|publish_queue_timeout/.test(errorMessage)) {
+        return '等待通道超时，系统已自动取消。'
+      }
      return '本次发布已取消。'
    case 'unknown':
    default:
@@ -42,6 +42,7 @@ import {
  getPublishStatusMeta,
  getTemplateMeta,
  hasActiveGenerationStatus,
+  hasActivePublishStatus,
 } from '@/lib/display'
 import { formatError } from '@/lib/errors'
 import { useCompanyStore } from '@/stores/company'
@@ -207,6 +208,7 @@ const displayedArticleTotal = computed(() =>
 )

 let articlePollingTimer: number | null = null
+const ARTICLE_LIST_POLL_INTERVAL_MS = 5000

 const deleteMutation = useMutation({
  mutationFn: (articleId: number) => articlesApi.remove(articleId),
@@ -418,8 +420,11 @@ watch(
    const hasGeneratingArticles = items.some((item: ArticleListItem) =>
      hasActiveGenerationStatus(item.generate_status),
    )
+    const hasPublishingArticles = items.some((item: ArticleListItem) =>
+      hasActivePublishStatus(item.publish_status),
+    )

-    if (hasGeneratingArticles) {
+    if (hasGeneratingArticles || hasPublishingArticles) {
      startArticlePolling()
      return
    }
@@ -444,10 +449,17 @@ function startArticlePolling(): void {
  }

  articlePollingTimer = window.setInterval(() => {
-    void queryClient.invalidateQueries({ queryKey: ['articles'] })
-    void queryClient.invalidateQueries({ queryKey: ['workspace', 'recent-articles'] })
-    void Promise.all([articleListQuery.refetch(), recentArticlesQuery.refetch()])
-  }, 3000)
+    if (shouldUseRecentFallback.value) {
+      if (!recentArticlesQuery.isFetching.value) {
+        void recentArticlesQuery.refetch()
+      }
+      return
+    }
+
+    if (!articleListQuery.isFetching.value) {
+      void articleListQuery.refetch()
+    }
+  }, ARTICLE_LIST_POLL_INTERVAL_MS)
 }

 function stopArticlePolling(): void {
@@ -31,7 +31,12 @@ import { articlesApi, tenantAccountsApi, workspaceApi } from '@/lib/api'
 import { buildArticleClipboardContent, resolveArticleActionState } from '@/lib/article-list-actions'
 import { useArticleRegenerateAction } from '@/lib/article-regenerate-action'
 import { copyTextToClipboard } from '@/lib/clipboard'
-import { formatDateTime, getTemplateMeta, hasActiveGenerationStatus } from '@/lib/display'
+import {
+  formatDateTime,
+  getTemplateMeta,
+  hasActiveGenerationStatus,
+  hasActivePublishStatus,
+} from '@/lib/display'
 import { formatError } from '@/lib/errors'
 import { isMediaPublishAccount } from '@/lib/publish-platforms'
 import { useCompanyStore } from '@/stores/company'
@@ -45,6 +50,7 @@ const articleDrawerOpen = ref(false)
 const selectedPublishArticleId = ref<number | null>(null)
 const publishModalOpen = ref(false)
 let recentArticlesPollingTimer: number | null = null
+const ARTICLE_LIST_POLL_INTERVAL_MS = 5000

 const overviewQuery = useQuery({
  queryKey: computed(() => ['workspace', 'overview', companyStore.currentBrandId]),
@@ -300,8 +306,9 @@ watch(
    const hasGeneratingArticles = items.some((item) =>
      hasActiveGenerationStatus(item.generate_status),
    )
+    const hasPublishingArticles = items.some((item) => hasActivePublishStatus(item.publish_status))

-    if (hasGeneratingArticles) {
+    if (hasGeneratingArticles || hasPublishingArticles) {
      startRecentArticlesPolling()
      return
    }
@@ -325,12 +332,11 @@ function startRecentArticlesPolling(): void {
      stopRecentArticlesPolling()
      return
    }
-    void queryClient.invalidateQueries({
-      queryKey: ['workspace', 'recent-articles', companyStore.currentBrandId],
-    })
-    void queryClient.invalidateQueries({ queryKey: ['articles'] })
+    if (recentArticlesQuery.isFetching.value) {
+      return
+    }
    void recentArticlesQuery.refetch()
-  }, 3000)
+  }, ARTICLE_LIST_POLL_INTERVAL_MS)
 }

 function stopRecentArticlesPolling(): void {
@@ -13,26 +13,36 @@ const (
 	defaultPublishLeaseRecoveryInterval = 3 * time.Minute
 	defaultPublishLeaseRecoveryTimeout  = 30 * time.Second
 	defaultPublishLeaseRecoveryLimit    = 1000
+	defaultPublishQueuedTaskTimeout     = 3 * 24 * time.Hour
 )

 type PublishLeaseRecoveryWorker struct {
-	service  *tenantapp.DesktopTaskService
-	logger   *zap.Logger
-	interval time.Duration
-	timeout  time.Duration
-	limit    int
+	service           *tenantapp.DesktopTaskService
+	logger            *zap.Logger
+	interval          time.Duration
+	timeout           time.Duration
+	limit             int
+	queuedTaskTimeout time.Duration
 }

 func NewPublishLeaseRecoveryWorker(service *tenantapp.DesktopTaskService, logger *zap.Logger) *PublishLeaseRecoveryWorker {
 	return &PublishLeaseRecoveryWorker{
-		service:  service,
-		logger:   logger,
-		interval: defaultPublishLeaseRecoveryInterval,
-		timeout:  defaultPublishLeaseRecoveryTimeout,
-		limit:    defaultPublishLeaseRecoveryLimit,
+		service:           service,
+		logger:            logger,
+		interval:          defaultPublishLeaseRecoveryInterval,
+		timeout:           defaultPublishLeaseRecoveryTimeout,
+		limit:             defaultPublishLeaseRecoveryLimit,
+		queuedTaskTimeout: defaultPublishQueuedTaskTimeout,
 	}
 }

+func (w *PublishLeaseRecoveryWorker) WithQueuedTaskTimeout(timeout time.Duration) *PublishLeaseRecoveryWorker {
+	if w != nil && timeout > 0 {
+		w.queuedTaskTimeout = timeout
+	}
+	return w
+}
+
 func (w *PublishLeaseRecoveryWorker) Start(ctx context.Context) {
 	go w.Run(ctx)
 }
@@ -72,15 +82,40 @@ func (w *PublishLeaseRecoveryWorker) runOnce(parent context.Context) {
 		return
 	}

-	if (result.Requeued > 0 || result.Failed > 0 || result.Uncertain > 0) && w.logger != nil {
+	queuedTaskTimeout := w.queuedTaskTimeout
+	if queuedTaskTimeout <= 0 {
+		queuedTaskTimeout = defaultPublishQueuedTaskTimeout
+	}
+	cancelledQueued, err := w.service.CancelStaleQueuedPublishTasks(ctx, queuedTaskTimeout, w.limit)
+	if err != nil {
+		if w.logger != nil {
+			w.logger.Warn("publish queued task timeout cleanup failed", zap.Error(err))
+		}
+		return
+	}
+	result.CancelledQueued = cancelledQueued
+
+	if (result.Requeued > 0 || result.Failed > 0 || result.Uncertain > 0 || result.CancelledQueued > 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),
+			zap.Int("cancelled_queued_task_count", result.CancelledQueued),
 		)
 	}
 }

+func publishQueuedTaskTimeoutFromRun(run JobRunContext) time.Duration {
+	timeout := AsDuration(run.Config, "queued_task_timeout", defaultPublishQueuedTaskTimeout)
+	if seconds := AsInt(run.Config, "queued_task_timeout_seconds", 0); seconds > 0 {
+		timeout = time.Duration(seconds) * time.Second
+	}
+	if timeout <= 0 {
+		return defaultPublishQueuedTaskTimeout
+	}
+	return timeout
+}
+
 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
@@ -104,11 +139,19 @@ func (w *PublishLeaseRecoveryWorker) RunOnce(parent context.Context, run JobRunC
 	if err != nil {
 		return map[string]any{"stage": "recover"}, err
 	}
+	queuedTimeout := publishQueuedTaskTimeoutFromRun(run)
+	cancelledQueued, err := w.service.CancelStaleQueuedPublishTasks(ctx, queuedTimeout, limit)
+	if err != nil {
+		return map[string]any{"stage": "cancel_stale_queued"}, err
+	}
+	result.CancelledQueued = cancelledQueued
 	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,
+		"requeued_task_count":         result.Requeued,
+		"failed_task_count":           result.Failed,
+		"uncertain_task_count":        result.Uncertain,
+		"cancelled_queued_task_count": result.CancelledQueued,
+		"queued_task_timeout_seconds": int64(queuedTimeout.Seconds()),
+		"duration_ms":                 time.Since(startedAt).Milliseconds(),
+		"batch_size":                  limit,
 	}, nil
 }
@@ -0,0 +1,42 @@
+package scheduler
+
+import (
+	"testing"
+	"time"
+)
+
+func TestPublishQueuedTaskTimeoutFromRunDefaultsToThreeDays(t *testing.T) {
+	t.Parallel()
+
+	got := publishQueuedTaskTimeoutFromRun(JobRunContext{})
+	if got != 3*24*time.Hour {
+		t.Fatalf("publishQueuedTaskTimeoutFromRun() = %s, want 72h", got)
+	}
+}
+
+func TestPublishQueuedTaskTimeoutFromRunReadsDurationConfig(t *testing.T) {
+	t.Parallel()
+
+	got := publishQueuedTaskTimeoutFromRun(JobRunContext{
+		Config: map[string]any{
+			"queued_task_timeout": "72h",
+		},
+	})
+	if got != 72*time.Hour {
+		t.Fatalf("publishQueuedTaskTimeoutFromRun(duration) = %s, want 72h", got)
+	}
+}
+
+func TestPublishQueuedTaskTimeoutFromRunReadsSecondsConfig(t *testing.T) {
+	t.Parallel()
+
+	got := publishQueuedTaskTimeoutFromRun(JobRunContext{
+		Config: map[string]any{
+			"queued_task_timeout":         "72h",
+			"queued_task_timeout_seconds": 3600,
+		},
+	})
+	if got != time.Hour {
+		t.Fatalf("publishQueuedTaskTimeoutFromRun(seconds) = %s, want 1h", got)
+	}
+}
@@ -1987,9 +1987,10 @@ type recoveredDesktopTaskLease struct {
 }

 type PublishLeaseRecoveryResult struct {
-	Requeued  int `json:"requeued"`
-	Failed    int `json:"failed"`
-	Uncertain int `json:"uncertain"`
+	Requeued        int `json:"requeued"`
+	Failed          int `json:"failed"`
+	Uncertain       int `json:"uncertain"`
+	CancelledQueued int `json:"cancelled_queued"`
 }

 type MonitorLeaseRecoveryResult struct {
@@ -2456,6 +2457,189 @@ func (s *DesktopTaskService) RecoverExpiredPublishTasks(ctx context.Context, lim
 	return result, nil
 }

+func staleQueuedPublishTasksCandidateSQL() string {
+	return `
+		SELECT
+			t.desktop_id,
+			t.workspace_id
+		FROM desktop_tasks AS t
+		JOIN desktop_publish_jobs AS j
+		  ON j.desktop_id = t.job_id
+		 AND j.tenant_id = t.tenant_id
+		 AND j.workspace_id = t.workspace_id
+		WHERE t.kind = 'publish'
+		  AND t.status = 'queued'
+		  AND GREATEST(
+		        COALESCE(t.enqueued_at, t.created_at),
+		        COALESCE(j.scheduled_at, t.created_at)
+		      ) < $1
+		ORDER BY COALESCE(t.enqueued_at, t.created_at) ASC,
+		         t.created_at ASC,
+		         t.desktop_id ASC
+		LIMIT $2
+		FOR UPDATE OF t SKIP LOCKED
+	`
+}
+
+func publishQueueTimeoutErrorPayload(timeout time.Duration) ([]byte, error) {
+	timeoutSeconds := int64(timeout.Seconds())
+	if timeoutSeconds < 0 {
+		timeoutSeconds = 0
+	}
+	message := fmt.Sprintf(
+		"发布任务在等待通道中超过 %s未被桌面端领取，系统已自动取消，避免任务无限期停留。请确认对应桌面端在线、账号登录正常且任务通道在线后，再重新发布。",
+		publishQueueTimeoutDisplayDuration(timeout),
+	)
+	return marshalOptionalJSON(map[string]any{
+		"code":                    "publish_queue_timeout",
+		"message":                 message,
+		"source":                  "publish_queue_timeout",
+		"queued_timeout_seconds":  timeoutSeconds,
+		"queued_timeout_duration": timeout.String(),
+	})
+}
+
+func publishQueueTimeoutDisplayDuration(timeout time.Duration) string {
+	if timeout <= 0 {
+		return "配置的时间"
+	}
+	minutes := int64(timeout.Round(time.Minute) / time.Minute)
+	if minutes <= 0 {
+		return "1 分钟"
+	}
+	const (
+		minutesPerHour = int64(60)
+		minutesPerDay  = int64(24 * 60)
+	)
+	if minutes%minutesPerDay == 0 {
+		return fmt.Sprintf("%d 天", minutes/minutesPerDay)
+	}
+	if minutes%minutesPerHour == 0 {
+		return fmt.Sprintf("%d 小时", minutes/minutesPerHour)
+	}
+	if minutes > minutesPerHour {
+		return fmt.Sprintf("%d 小时 %d 分钟", minutes/minutesPerHour, minutes%minutesPerHour)
+	}
+	return fmt.Sprintf("%d 分钟", minutes)
+}
+
+func (s *DesktopTaskService) CancelStaleQueuedPublishTasks(ctx context.Context, queuedFor time.Duration, limit int) (int, error) {
+	if s == nil || s.pool == nil || queuedFor <= 0 {
+		return 0, nil
+	}
+	if limit <= 0 {
+		limit = 1000
+	}
+	errorJSON, err := publishQueueTimeoutErrorPayload(queuedFor)
+	if err != nil {
+		return 0, response.ErrInternal(50195, "desktop_task_recovery_payload_failed", "failed to encode publish queue timeout payload")
+	}
+	staleBefore := time.Now().UTC().Add(-queuedFor)
+
+	tx, err := s.pool.Begin(ctx)
+	if err != nil {
+		return 0, response.ErrInternal(50196, "desktop_task_recovery_begin_failed", "failed to start desktop task recovery")
+	}
+	defer tx.Rollback(ctx)
+
+	rows, err := tx.Query(ctx, staleQueuedPublishTasksCandidateSQL(), staleBefore, limit)
+	if err != nil {
+		s.logWarn("stale queued publish task query failed", err)
+		return 0, response.ErrInternal(50197, "desktop_task_recovery_query_failed", "failed to select stale queued publish tasks")
+	}
+	defer rows.Close()
+
+	candidates := make([]recoveredDesktopTaskLease, 0)
+	for rows.Next() {
+		var item recoveredDesktopTaskLease
+		if scanErr := rows.Scan(&item.TaskID, &item.WorkspaceID); scanErr != nil {
+			s.logWarn("stale queued publish task scan failed", scanErr)
+			return 0, response.ErrInternal(50198, "desktop_task_recovery_scan_failed", "failed to parse stale queued publish task")
+		}
+		item.Kind = "publish"
+		item.Status = "aborted"
+		item.ErrorJSON = errorJSON
+		candidates = append(candidates, item)
+	}
+	if err := rows.Err(); err != nil {
+		s.logWarn("stale queued publish task rows failed", err)
+		return 0, response.ErrInternal(50198, "desktop_task_recovery_scan_failed", "failed to iterate stale queued publish tasks")
+	}
+
+	publishOutcomes := make([]*desktopPublishSyncOutcome, 0, len(candidates))
+	cancelled := 0
+	for index := range candidates {
+		item := &candidates[index]
+		row := tx.QueryRow(ctx, `
+			UPDATE desktop_tasks
+			SET status = 'aborted',
+			    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, 'publish_queue_timeout'),
+			    updated_at = NOW()
+			WHERE desktop_id = $1
+			  AND kind = 'publish'
+			  AND status = 'queued'
+			RETURNING desktop_id, tenant_id, workspace_id, platform_id, kind, payload, status, result, error
+		`, item.TaskID, errorJSON)
+
+		var updated repository.DesktopTask
+		if scanErr := row.Scan(
+			&updated.DesktopID,
+			&updated.TenantID,
+			&updated.WorkspaceID,
+			&updated.Platform,
+			&updated.Kind,
+			&updated.Payload,
+			&updated.Status,
+			&updated.Result,
+			&updated.Error,
+		); scanErr != nil {
+			if errors.Is(scanErr, pgx.ErrNoRows) {
+				continue
+			}
+			s.logWarn("stale queued publish task update failed", scanErr, zap.String("task_id", item.TaskID.String()))
+			return cancelled, response.ErrInternal(50197, "desktop_task_recovery_query_failed", "failed to cancel stale queued publish task")
+		}
+
+		publishOutcome, syncErr := syncDesktopPublishTaskState(ctx, tx, &updated)
+		if syncErr != nil {
+			return cancelled, syncErr
+		}
+		if publishOutcome != nil {
+			publishOutcomes = append(publishOutcomes, publishOutcome)
+		}
+		cancelled++
+	}
+
+	if err := tx.Commit(ctx); err != nil {
+		return cancelled, response.ErrInternal(50199, "desktop_task_recovery_commit_failed", "failed to commit desktop task recovery")
+	}
+
+	s.afterRecoveredPublishOutcomes(ctx, publishOutcomes)
+
+	for _, item := range candidates {
+		task, getErr := s.repo.GetByDesktopID(ctx, item.TaskID, item.WorkspaceID)
+		if getErr != nil {
+			s.logWarn("stale queued publish task reload after commit failed", getErr, zap.String("task_id", item.TaskID.String()))
+			continue
+		}
+		s.publishTaskEvent(ctx, task, "task_completed")
+	}
+
+	if cancelled > 0 && s.logger != nil {
+		s.logger.Info("stale queued publish tasks cancelled",
+			zap.Int("cancelled", cancelled),
+			zap.Duration("queued_for", queuedFor),
+		)
+	}
+
+	return cancelled, nil
+}
+
 func (s *DesktopTaskService) RecoverExpiredMonitorTasks(ctx context.Context, limit int) (MonitorLeaseRecoveryResult, error) {
 	var result MonitorLeaseRecoveryResult
 	if s == nil || s.pool == nil {
@@ -6,6 +6,7 @@ import (
 	"regexp"
 	"strings"
 	"testing"
+	"time"

 	"github.com/geo-platform/tenant-api/internal/shared/response"
 	"github.com/geo-platform/tenant-api/internal/tenant/repository"
@@ -53,6 +54,71 @@ func TestDesktopTaskRecoverySelectQueryLeaseExpiryFiltersExpiredLeases(t *testin
 	}
 }

+func TestStaleQueuedPublishTasksCandidateSQLUsesExecutableQueueAge(t *testing.T) {
+	t.Parallel()
+
+	query := normalizeSQLForDesktopTaskTest(staleQueuedPublishTasksCandidateSQL())
+	for _, fragment := range []string{
+		"FROM desktop_tasks AS t",
+		"JOIN desktop_publish_jobs AS j",
+		"t.kind = 'publish'",
+		"t.status = 'queued'",
+		"GREATEST( COALESCE(t.enqueued_at, t.created_at), COALESCE(j.scheduled_at, t.created_at) ) < $1",
+		"LIMIT $2",
+		"FOR UPDATE OF t SKIP LOCKED",
+	} {
+		if !strings.Contains(query, fragment) {
+			t.Fatalf("stale queued publish query missing %q: %s", fragment, query)
+		}
+	}
+}
+
+func TestPublishQueueTimeoutErrorPayloadIncludesConfig(t *testing.T) {
+	t.Parallel()
+
+	payload, err := publishQueueTimeoutErrorPayload(3 * 24 * time.Hour)
+	if err != nil {
+		t.Fatalf("publishQueueTimeoutErrorPayload() error = %v", err)
+	}
+	text := string(payload)
+	for _, fragment := range []string{
+		`"code":"publish_queue_timeout"`,
+		`发布任务在等待通道中超过 3 天未被桌面端领取`,
+		`请确认对应桌面端在线、账号登录正常且任务通道在线后，再重新发布。`,
+		`"source":"publish_queue_timeout"`,
+		`"queued_timeout_seconds":259200`,
+	} {
+		if !strings.Contains(text, fragment) {
+			t.Fatalf("queue timeout payload missing %q: %s", fragment, text)
+		}
+	}
+}
+
+func TestPublishQueueTimeoutDisplayDurationUsesReadableChinese(t *testing.T) {
+	t.Parallel()
+
+	cases := []struct {
+		name    string
+		timeout time.Duration
+		want    string
+	}{
+		{name: "days", timeout: 7 * 24 * time.Hour, want: "7 天"},
+		{name: "hours", timeout: 36 * time.Hour, want: "36 小时"},
+		{name: "hours and minutes", timeout: 90 * time.Minute, want: "1 小时 30 分钟"},
+		{name: "minutes", timeout: 45 * time.Minute, want: "45 分钟"},
+	}
+
+	for _, tc := range cases {
+		tc := tc
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			if got := publishQueueTimeoutDisplayDuration(tc.timeout); got != tc.want {
+				t.Fatalf("publishQueueTimeoutDisplayDuration(%s) = %q, want %q", tc.timeout, got, tc.want)
+			}
+		})
+	}
+}
+
 func TestReconcileRetryPublishesTaskAvailableForAllDesktopTaskKinds(t *testing.T) {
 	t.Parallel()