feat(轮询优先队列): AUG26-016 卡轮询优先队列、人工入队与读侧接口，归档并同步主 Spec 与证据矩阵

新增 000228 成对迁移 tb_polling_priority_item：卡、任务类型、状态、触发类型、来源订单/套餐使用记录、
触发次数与来源集合、尝试次数、失败原因、人工原因与操作者、店铺快照与各时间列；以活动项部分唯一索引
uq_polling_priority_item_active（仅 deleted_at IS NULL AND status IN ('pending','processing') 占键位）
表达「同卡同任务类型至多一条活动项」，另有状态/时间索引与全列注释；down 守卫在存在活动项或未终态行时
拒绝回滚并给出中文原因。

新增优先轮询请求可靠事件 polling.priority.requested（载荷版本 v1、事件键前缀 prio:）与消费者：只在原
业务事务内追加、幂等键稳定；消费者按卡 × 纳入任务类型（realname/carddata/card_status/package）逐条
建项并在提交后下发执行提示，重复投递只合并触发次数、来源集合与最近触发时间，不新建行也不重复调用。
触发点为四类自动场景 purchase_activated / renewal_activated（按同载体更早套餐使用记录判定）/
queue_activated / addon_activated 与「无有效套餐」no_valid_package（仅在普通套餐轮询来源且存在待生效
套餐使用记录时追加；事件通道显式拒绝 manual_trigger）；入队对象恒为卡，绑定设备资产在触发事务内冻结
在用卡快照逐卡建项，不使用设备当前卡槽口径。

轮询共享基类新增认领接缝：四个 Handler（realname/carddata/card_status/package）在并发信号量之后、调用
上游之前探测活动项——待执行条件认领、执行中且 90 秒租约未到期则跳过并延后、无活动项时行为与既有完全
等价；超租约允许相邻执行接管，尝试次数只在真正发起执行后累加，未达上限（3）回到活动态按既有间隔重排，
达上限或业务校验类失败进入失败终态并保留可安全展示原因；执行前校验卡自身与绑定设备的轮询开关。未引入
通用卡级锁与 Redis 活动标记，分片队列的出队、入队与移除路径未改动。

提示通道按任务类型独立键（polling:priority:{taskType}），与既有手动触发队列分离；调度器在同一周期内先
排空优先提示、再排空手动触发队列，提示排空不受分片背压跳过影响；未新建调度设施或异步任务类型。

新增人工优先入队与只读查询三条路由 POST /api/admin/polling-priority-items、
GET /api/admin/polling-priority-items、GET /api/admin/polling-priority-items/:id：人工入队复用既有轮询
权限判定（抽取为同包共享函数），原因必填，不受每日 500 次上限与 24 小时去重约束，重复抑制由活动项合并
承担；读侧按店铺快照下推数据范围，越权与不存在不可区分，不提供优先级分级、有效期或人工重触发入口。
新增 7 个审计动作（enqueue/claim/fail/retry/complete/dequeue/manual_denied）与资源
polling_priority_item，并按（操作者类型，来源）注册，人工侧与 Worker 侧均通过来源校验。

同步 OpenAPI 文档装配三处与路由注册；归档 Change 至
openspec/changes/archive/2026-09-17-add-priority-polling-queue/ 并同步主 Spec（新增
priority-polling-queue、polling-operations 追加单次执行互斥 Requirement 与三条路由索引）与上下文健康
证据（requirement-evidence 150 行、入口矩阵 http 403 / async 56）。

本机验证：junhong_cmp_test 与隔离 Redis DB 15，未连生产、未启动 Worker/API、未调用运营商上游；迁移
up/down/up 与 down 守卫实测（含 dirty=true 记账口径与 force 恢复），A–F 批 94 PASS、接缝 63 PASS、
提示通道 12 PASS、清理零残留 20 PASS。成功路径 Complete、真并发互斥、尝试上限第 3 次判定、HTTP 层权限
矩阵、通道阈值持锁复机边界与三类生效触发点生产集成留待测试部署验证（见
docs/verification/add-priority-polling-queue-verification.md 第 4 节）。自动化测试按项目决策为 N/A，
未新增 *_test.go。

internal/task/auto_purchase.go

@@ -14,6 +14,7 @@ import (
 	"gorm.io/gorm/clause"
 
 	cardObservationApp "github.com/break/junhong_cmp_fiber/internal/application/cardobservation"
+	priorityapp "github.com/break/junhong_cmp_fiber/internal/application/prioritypolling"
 	packagedomain "github.com/break/junhong_cmp_fiber/internal/domain/package"
 	"github.com/break/junhong_cmp_fiber/internal/infrastructure/audit"
 	"github.com/break/junhong_cmp_fiber/internal/infrastructure/commissiondelivery"
@@ -51,6 +52,7 @@ type AutoPurchaseHandler struct {
 	asynqClient                *asynq.Client // 用于事务提交成功后触发佣金计算任务
 	logger                     *zap.Logger
 	observationSeriesEvents    cardObservationApp.SeriesEventWriter
+	priorityEvents             priorityapp.PriorityEventWriter
 	auditWriter                *audit.Writer
 }
 
@@ -67,6 +69,7 @@ func NewAutoPurchaseHandler(
 	asynqClient *asynq.Client,
 	logger *zap.Logger,
 	observationSeriesEvents cardObservationApp.SeriesEventWriter,
+	priorityEvents priorityapp.PriorityEventWriter,
 	auditWriter *audit.Writer,
 ) *AutoPurchaseHandler {
 	if orderStore == nil {
@@ -103,6 +106,7 @@ func NewAutoPurchaseHandler(
 		asynqClient:                asynqClient,
 		logger:                     logger,
 		observationSeriesEvents:    observationSeriesEvents,
+		priorityEvents:             priorityEvents,
 		auditWriter:                auditWriter,
 	}
 }
@@ -712,10 +716,22 @@ func (h *AutoPurchaseHandler) activateMainPackage(
 		return err
 	}
 
-	return tx.Model(usage).Updates(map[string]any{
+	if err = tx.Model(usage).Updates(map[string]any{
 		"status":                      usage.Status,
 		"pending_realname_activation": usage.PendingRealnameActivation,
-	}).Error
+	}).Error; err != nil {
+		return err
+	}
+
+	// 仅在生效状态分支追加优先轮询请求：待生效（含等待实名激活）不属于触发场景。
+	if status == constants.PackageUsageStatusActive {
+		triggerType, classifyErr := packagepkg.ResolveActivationTriggerType(ctx, tx, carrierType, carrierID, usage.ID)
+		if classifyErr != nil {
+			return classifyErr
+		}
+		return packagepkg.AppendActivatedPriorityRequested(ctx, tx, h.priorityEvents, usage, carrierType, carrierID, triggerType, activatedAt)
+	}
+	return nil
 }
 
 // isCarrierRealnamed 查询自动购包载体是否已满足实名激活条件。
@@ -822,7 +838,13 @@ func (h *AutoPurchaseHandler) activateAddonPackage(
 		usage.DeviceID = carrierID
 	}
 
-	return tx.Create(usage).Error
+	if err := tx.Create(usage).Error; err != nil {
+		return err
+	}
+
+	// 加油包立即生效：以生效状态提交后追加优先轮询请求。
+	return packagepkg.AppendActivatedPriorityRequested(ctx, tx, h.priorityEvents, usage, carrierType, carrierID,
+		constants.PollingPriorityTriggerAddonActivated, now)
 }
 
 func (h *AutoPurchaseHandler) resolvePackageTerms(ctx context.Context, tx *gorm.DB, pkg *model.Package, sellerShopID *uint) (packagedomain.TermsSnapshot, error) {

internal/task/polling_base.go

@@ -6,7 +6,9 @@ import (
 
 	"github.com/redis/go-redis/v9"
 	"go.uber.org/zap"
+	"gorm.io/gorm"
 
+	"github.com/break/junhong_cmp_fiber/internal/infrastructure/audit"
 	"github.com/break/junhong_cmp_fiber/internal/infrastructure/cardtrafficlock"
 	"github.com/break/junhong_cmp_fiber/internal/polling"
 	"github.com/break/junhong_cmp_fiber/internal/store/postgres"
@@ -54,7 +56,7 @@ func pollingFallbackContext() (context.Context, context.CancelFunc) {
 }
 
 // PollingBase 轮询共享基类
-// 封装并发控制、卡缓存、重入队、配置间隔查询等公共方法，所有 Handler 共享
+// 封装并发控制、卡缓存、重入队、配置间隔查询与优先轮询项认领等公共方法，所有 Handler 共享
 type PollingBase struct {
 	redis               *redis.Client
 	queueMgr            *polling.PollingQueueManager
@@ -64,6 +66,10 @@ type PollingBase struct {
 	verboseLog          bool
 	totalMaxConcurrency int
 	trafficLock         *cardtrafficlock.Lock
+	db                  *gorm.DB
+	priorityStore       *postgres.PollingPriorityItemStore
+	priorityAudit       *audit.Writer
+	scopeChecker        *polling.PollingLifecycleService
 }
 
 // NewPollingBase 创建轮询共享基类
@@ -72,6 +78,10 @@ func NewPollingBase(
 	queueMgr *polling.PollingQueueManager,
 	configMgr *polling.PollingConfigManager,
 	iotCardStore *postgres.IotCardStore,
+	scopeChecker *polling.PollingLifecycleService,
+	priorityStore *postgres.PollingPriorityItemStore,
+	priorityAudit *audit.Writer,
+	db *gorm.DB,
 	logger *zap.Logger,
 	verboseLog bool,
 	totalMaxConcurrency int,
@@ -85,6 +95,10 @@ func NewPollingBase(
 		verboseLog:          verboseLog,
 		totalMaxConcurrency: totalMaxConcurrency,
 		trafficLock:         cardtrafficlock.New(redisClient),
+		db:                  db,
+		priorityStore:       priorityStore,
+		priorityAudit:       priorityAudit,
+		scopeChecker:        scopeChecker,
 	}
 }
 

internal/task/polling_carddata_handler.go

@@ -49,29 +49,53 @@ func (h *PollingCarddataHandler) Handle(ctx context.Context, task *asynq.Task) e
 	}
 	defer h.base.releaseCardTrafficSyncLock(ctx, cardID, lockToken)
 
+	// 优先轮询互斥接缝：放在流量同步锁获取成功之后，避免认领后拿不到锁导致优先项空占。
+	priority, skip, priorityErr := h.base.beginPollingPriority(ctx, cardID, constants.TaskTypePollingCarddata)
+	if priorityErr != nil {
+		return h.failAndRequeue(ctx, cardID, startedAt, "查询活动优先轮询项失败", priorityErr)
+	}
+	if skip {
+		h.base.logger.Debug("卡轮询优先项已被其它执行持有，本轮跳过并延后",
+			zap.Uint("card_id", cardID), zap.String("task_type", constants.TaskTypePollingCarddata))
+		return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingCarddata)
+	}
+
 	h.base.invalidateCardCache(ctx, cardID)
 	card, err := h.base.iotCardStore.GetByID(ctx, cardID)
 	if err != nil {
 		if isNotFound(err) {
+			priority.FailFinal(ctx, "卡不存在")
 			return nil
 		}
+		// 本地读卡失败属可恢复失败：先按真实执行累加尝试次数，再按上限收敛，避免加急事实被静默丢弃。
+		priority.MarkExecution(ctx)
+		priority.FailRetryable(ctx, "获取卡信息失败")
 		return h.failAndRequeue(ctx, cardID, startedAt, "获取卡信息失败", err)
 	}
-	if h.gateway == nil {
+	// 执行前轮询范围校验：优先通道不得重启已停止轮询的卡。
+	if !h.base.priorityInPollingScope(ctx, priority, card) {
 		return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingCarddata)
 	}
+	if h.gateway == nil {
+		priority.FailFinal(ctx, "流量查询能力未配置")
+		return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingCarddata)
+	}
+	priority.MarkExecution(ctx)
 	attempt := newGatewayAttempt(cardID, constants.IntegrationOperationGatewayTraffic, constants.CardObservationSceneTrafficPolling)
 	result, err := h.gateway.QueryFlow(ctx, &gateway.FlowQueryReq{CardNo: card.ICCID})
 	if err != nil {
 		_ = recordGatewayAttempt(ctx, h.integration, attempt, constants.IntegrationResultFailed, false, "查询流量失败")
+		priority.FailRetryable(ctx, "查询流量失败")
 		return h.failAndRequeue(ctx, cardID, startedAt, "查询流量失败", err)
 	}
 	if strings.TrimSpace(result.ICCID) == "" {
 		_ = recordGatewayAttempt(ctx, h.integration, attempt, constants.IntegrationResultInvalidPayload, false, "流量查询响应缺少 ICCID")
+		priority.FailRetryable(ctx, "流量查询响应缺少 ICCID")
 		return h.failAndRequeue(ctx, cardID, startedAt, "流量查询响应缺少 ICCID", nil)
 	}
 	ctx = withPollingWorkerAuditContext(ctx, constants.TaskTypePollingCarddata, "卡流量轮询任务", attempt.IntegrationID)
 	if h.observation == nil || h.carrier == nil {
+		priority.FailFinal(ctx, "卡流量观测能力未配置")
 		return h.failAndRequeue(ctx, cardID, startedAt, "卡流量观测能力未配置", nil)
 	}
 	now := time.Now()
@@ -84,6 +108,7 @@ func (h *PollingCarddataHandler) Handle(ctx context.Context, task *asynq.Task) e
 	})
 	if err != nil {
 		_ = recordGatewayAttempt(ctx, h.integration, attempt, constants.IntegrationResultUnknown, false, "应用流量观测失败")
+		priority.FailRetryable(ctx, "应用流量观测失败")
 		return h.failAndRequeue(ctx, cardID, startedAt, "应用流量观测失败", err)
 	}
 	changed := decision.IncrementMB > 0 || decision.CrossMonth
@@ -100,6 +125,7 @@ func (h *PollingCarddataHandler) Handle(ctx context.Context, task *asynq.Task) e
 		metric = "polling.observation.persisted"
 	}
 	h.base.logger.Info("流量轮询观测完成", zap.Uint("card_id", cardID), zap.Bool("changed", changed), zap.String("metric", metric))
+	priority.Complete(ctx)
 	h.base.updateStats(ctx, constants.TaskTypePollingCarddata, true, time.Since(startedAt))
 	return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingCarddata)
 }

internal/task/polling_cardstatus_handler.go

@@ -49,28 +49,53 @@ func (h *PollingCardStatusHandler) Handle(ctx context.Context, task *asynq.Task)
 		return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingCardStatus)
 	}
 	defer h.base.releaseConcurrency(ctx, constants.TaskTypePollingCardStatus)
+
+	// 优先轮询互斥接缝：存在活动优先项时本轮即优先执行或被跳过；无活动项时行为与既有实现一致。
+	priority, skip, priorityErr := h.base.beginPollingPriority(ctx, cardID, constants.TaskTypePollingCardStatus)
+	if priorityErr != nil {
+		return h.failAndRequeue(ctx, cardID, startedAt, "查询活动优先轮询项失败", priorityErr)
+	}
+	if skip {
+		h.base.logger.Debug("卡轮询优先项已被其它执行持有，本轮跳过并延后",
+			zap.Uint("card_id", cardID), zap.String("task_type", constants.TaskTypePollingCardStatus))
+		return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingCardStatus)
+	}
+
 	card, err := h.base.getCardWithCache(ctx, cardID)
 	if err != nil {
 		if isNotFound(err) {
+			priority.FailFinal(ctx, "卡不存在")
 			return nil
 		}
+		// 本地读卡失败属可恢复失败：先按真实执行累加尝试次数，再按上限收敛，避免加急事实被静默丢弃。
+		priority.MarkExecution(ctx)
+		priority.FailRetryable(ctx, "获取卡信息失败")
 		return h.failAndRequeue(ctx, cardID, startedAt, "获取卡信息失败", err)
 	}
-	if h.gateway == nil {
+	// 执行前轮询范围校验：优先通道不得重启已停止轮询的卡。
+	if !h.base.priorityInPollingScope(ctx, priority, card) {
 		return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingCardStatus)
 	}
+	if h.gateway == nil {
+		priority.FailFinal(ctx, "卡状态查询能力未配置")
+		return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingCardStatus)
+	}
+	priority.MarkExecution(ctx)
 	attempt := newGatewayAttempt(cardID, constants.IntegrationOperationGatewayNetwork, constants.CardObservationSceneNetworkPolling)
 	result, err := h.gateway.QueryCardStatus(ctx, &gateway.CardStatusReq{CardNo: card.ICCID})
 	if err != nil {
 		_ = recordGatewayAttempt(ctx, h.integration, attempt, constants.IntegrationResultFailed, false, "查询卡状态失败")
+		priority.FailRetryable(ctx, "查询卡状态失败")
 		return h.failAndRequeue(ctx, cardID, startedAt, "查询卡状态失败", err)
 	}
 	if strings.TrimSpace(result.ICCID) == "" {
 		_ = recordGatewayAttempt(ctx, h.integration, attempt, constants.IntegrationResultInvalidPayload, false, "卡状态查询响应缺少 ICCID")
+		priority.FailRetryable(ctx, "卡状态查询响应缺少 ICCID")
 		return h.failAndRequeue(ctx, cardID, startedAt, "卡状态查询响应缺少 ICCID", nil)
 	}
 	ctx = withPollingWorkerAuditContext(ctx, constants.TaskTypePollingCardStatus, "卡网络状态轮询任务", attempt.IntegrationID)
 	if h.observation == nil {
+		priority.FailFinal(ctx, "卡网络观测能力未配置")
 		return h.failAndRequeue(ctx, cardID, startedAt, "卡网络观测能力未配置", nil)
 	}
 	now := time.Now()
@@ -83,6 +108,7 @@ func (h *PollingCardStatusHandler) Handle(ctx context.Context, task *asynq.Task)
 	})
 	if err != nil {
 		_ = recordGatewayAttempt(ctx, h.integration, attempt, constants.IntegrationResultUnknown, false, "应用网络状态观测失败")
+		priority.FailRetryable(ctx, "应用网络状态观测失败")
 		return h.failAndRequeue(ctx, cardID, startedAt, "应用网络状态观测失败", err)
 	}
 	changed := decision.StatusChanged || decision.StopReasonChanged || decision.StopPolling
@@ -100,6 +126,7 @@ func (h *PollingCardStatusHandler) Handle(ctx context.Context, task *asynq.Task)
 		metric = "polling.observation.persisted"
 	}
 	h.base.logger.Info("卡状态轮询观测完成", zap.Uint("card_id", cardID), zap.Bool("changed", changed), zap.String("metric", metric))
+	priority.Complete(ctx)
 	h.base.updateStats(ctx, constants.TaskTypePollingCardStatus, true, time.Since(startedAt))
 	if decision.StopPolling {
 		h.base.logger.Info("独立卡命中风险状态，已关闭轮询", zap.Uint("card_id", cardID), zap.String("gateway_extend", decision.GatewayExtend))

internal/task/polling_package_handler.go

@@ -7,6 +7,7 @@ import (
 	"github.com/hibiken/asynq"
 	"go.uber.org/zap"
 
+	priorityapp "github.com/break/junhong_cmp_fiber/internal/application/prioritypolling"
 	iot_card_svc "github.com/break/junhong_cmp_fiber/internal/service/iot_card"
 	"github.com/break/junhong_cmp_fiber/internal/store/postgres"
 	"github.com/break/junhong_cmp_fiber/pkg/constants"
@@ -52,39 +53,76 @@ func (h *PollingPackageHandler) Handle(ctx context.Context, t *asynq.Task) error
 	}
 	defer h.base.releaseConcurrency(ctx, constants.TaskTypePollingPackage)
 
-	if _, cacheErr := h.base.getCardWithCache(ctx, cardID); cacheErr != nil {
-		if isNotFound(cacheErr) {
-			return nil
-		}
-		h.base.logger.Error("获取卡信息失败", zap.Uint("card_id", cardID), zap.Error(cacheErr))
+	// 优先轮询互斥接缝：存在活动优先项时本轮即优先执行或被跳过；无活动项时行为与既有实现一致。
+	priority, skip, priorityErr := h.base.beginPollingPriority(ctx, cardID, constants.TaskTypePollingPackage)
+	if priorityErr != nil {
+		h.base.logger.Error("查询活动优先轮询项失败", zap.Uint("card_id", cardID), zap.Error(priorityErr))
 		h.base.updateStats(ctx, constants.TaskTypePollingPackage, false, time.Since(startTime))
 		return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingPackage)
 	}
-
-	if h.stopResumeSvc != nil {
-		freshCard, loadErr := h.iotCardStore.GetByID(ctx, cardID)
-		if loadErr == nil {
-			h.base.logger.Info("套餐检查：执行停复机评估",
-				zap.Uint("card_id", cardID),
-				zap.Int("network_status", freshCard.NetworkStatus),
-				zap.String("stop_reason", freshCard.StopReason))
-			if h.base.verboseLog {
-				h.base.logger.Info("套餐检查详情",
-					zap.Uint("card_id", cardID),
-					zap.String("iccid", freshCard.ICCID),
-					zap.Int("network_status", freshCard.NetworkStatus),
-					zap.String("stop_reason", freshCard.StopReason))
-			}
-			if evalErr := h.stopResumeSvc.EvaluateAndAct(ctx, freshCard); evalErr != nil {
-				h.base.logger.Warn("套餐检查后停复机评估失败",
-					zap.Uint("card_id", cardID), zap.Error(evalErr))
-			}
-		} else {
-			h.base.logger.Warn("套餐检查：加载卡信息失败",
-				zap.Uint("card_id", cardID), zap.Error(loadErr))
-		}
+	if skip {
+		h.base.logger.Debug("卡轮询优先项已被其它执行持有，本轮跳过并延后",
+			zap.Uint("card_id", cardID), zap.String("task_type", constants.TaskTypePollingPackage))
+		return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingPackage)
 	}
 
+	card, cacheErr := h.base.getCardWithCache(ctx, cardID)
+	if cacheErr != nil {
+		if isNotFound(cacheErr) {
+			priority.FailFinal(ctx, "卡不存在")
+			return nil
+		}
+		h.base.logger.Error("获取卡信息失败", zap.Uint("card_id", cardID), zap.Error(cacheErr))
+		// 本地读卡失败属可恢复失败：与其余三个轮询 handler 统一按尝试上限收敛，避免加急事实被静默丢弃。
+		priority.MarkExecution(ctx)
+		priority.FailRetryable(ctx, "获取卡信息失败")
+		h.base.updateStats(ctx, constants.TaskTypePollingPackage, false, time.Since(startTime))
+		return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingPackage)
+	}
+	// 执行前轮询范围校验：优先通道不得重启已停止轮询的卡。
+	if !h.base.priorityInPollingScope(ctx, priority, card) {
+		return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingPackage)
+	}
+
+	if h.stopResumeSvc == nil {
+		priority.FailFinal(ctx, "套餐评估能力未配置")
+		h.base.updateStats(ctx, constants.TaskTypePollingPackage, true, time.Since(startTime))
+		return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingPackage)
+	}
+
+	// 停复机评估就是本任务的上游执行动作：先累加尝试次数再发起。
+	priority.MarkExecution(ctx)
+	freshCard, loadErr := h.iotCardStore.GetByID(ctx, cardID)
+	if loadErr != nil {
+		h.base.logger.Warn("套餐检查：加载卡信息失败",
+			zap.Uint("card_id", cardID), zap.Error(loadErr))
+		priority.FailRetryable(ctx, "获取卡信息失败")
+		h.base.updateStats(ctx, constants.TaskTypePollingPackage, true, time.Since(startTime))
+		return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingPackage)
+	}
+	h.base.logger.Info("套餐检查：执行停复机评估",
+		zap.Uint("card_id", cardID),
+		zap.Int("network_status", freshCard.NetworkStatus),
+		zap.String("stop_reason", freshCard.StopReason))
+	if h.base.verboseLog {
+		h.base.logger.Info("套餐检查详情",
+			zap.Uint("card_id", cardID),
+			zap.String("iccid", freshCard.ICCID),
+			zap.Int("network_status", freshCard.NetworkStatus),
+			zap.String("stop_reason", freshCard.StopReason))
+	}
+	// 标记本次评估来自普通套餐轮询：无有效套餐场景据此才允许追加优先轮询请求，
+	// 其余调用链（观测消费者、手动实名、保护期轮询等）不打标即不写。
+	evaluationCtx := priorityapp.WithPollingPriorityTrigger(ctx)
+	if evalErr := h.stopResumeSvc.EvaluateAndAct(evaluationCtx, freshCard); evalErr != nil {
+		h.base.logger.Warn("套餐检查后停复机评估失败",
+			zap.Uint("card_id", cardID), zap.Error(evalErr))
+		priority.FailRetryable(ctx, "停复机评估失败")
+		h.base.updateStats(ctx, constants.TaskTypePollingPackage, true, time.Since(startTime))
+		return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingPackage)
+	}
+
+	priority.Complete(ctx)
 	h.base.updateStats(ctx, constants.TaskTypePollingPackage, true, time.Since(startTime))
 	return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingPackage)
 }

internal/task/polling_priority_claim.go

@@ -0,0 +1,324 @@
+package task
+
+import (
+	"context"
+	"strconv"
+	"time"
+
+	"go.uber.org/zap"
+	"gorm.io/gorm"
+
+	"github.com/break/junhong_cmp_fiber/internal/infrastructure/audit"
+	"github.com/break/junhong_cmp_fiber/internal/model"
+	"github.com/break/junhong_cmp_fiber/pkg/auditfailure"
+	"github.com/break/junhong_cmp_fiber/pkg/constants"
+)
+
+// pollingPriorityTaskNames 是纳入优先轮询的既有任务类型的中文名，用于日志与审计展示。
+var pollingPriorityTaskNames = map[string]string{
+	constants.TaskTypePollingRealname:   "实名状态轮询任务",
+	constants.TaskTypePollingCarddata:   "卡流量轮询任务",
+	constants.TaskTypePollingPackage:    "套餐状态轮询任务",
+	constants.TaskTypePollingCardStatus: "卡网络状态轮询任务",
+}
+
+// pollingPriorityRound 承载本轮轮询执行是否同时是一次优先执行。
+//
+// 无活动优先项时接缝返回 nil，其所有方法都是无操作，因此调用方在「无优先项」路径上
+// 完全保持既有行为（入队、出队、移除与延后一行不改）。领取成功时必须恰好调用一次结果收口方法
+// （Complete / FailRetryable / FailFinal），以保证不出现「领了优先项却没执行也没落终态或退回活动态」的路径。
+//
+// 上下文约定（照 PollingBase 既有约定）：本类型的写库与审计一律忽略调用方 ctx，
+// 内部改用 pollingFallbackContext 的独立短超时上下文。Asynq 任务 ctx 超时或取消后，
+// 失败事实、终态与审计仍必须尽量落下，否则优先项会停在执行中只能等租约接管。
+type pollingPriorityRound struct {
+	base     *PollingBase
+	item     *model.PollingPriorityItem
+	taskType string
+	taskName string
+	// attempts 与库内 attempt_count 保持同步，用于判定是否达到尝试上限。
+	attempts int
+}
+
+// beginPollingPriority 轮询共享基类的优先项认领接缝。
+//
+// 位置约定：必须在取并发信号量之后、调用上游之前（carddata 还需在流量同步锁获取成功之后）。
+// 返回语义：
+//   - round 非 nil：本轮领取成功，本轮执行同时就是优先执行；
+//   - skip 为 true：存在活动优先项但本轮必须跳过并延后（已被其它执行持有，或超租约接管未命中），
+//     调用方必须释放并发、按既有间隔 requeueCard 延后、零上游调用、不写失败审计；
+//   - err 非 nil：活动项探测失败，调用方按既有失败分支处理；
+//   - round 为 nil 且 skip 为 false：无活动优先项，调用方后续行为与既有实现完全一致。
+func (b *PollingBase) beginPollingPriority(ctx context.Context, cardID uint, taskType string) (*pollingPriorityRound, bool, error) {
+	if b == nil || b.priorityStore == nil {
+		return nil, false, nil
+	}
+	item, err := b.priorityStore.FindActive(ctx, cardID, taskType)
+	if err != nil {
+		return nil, false, err
+	}
+	if item == nil {
+		return nil, false, nil
+	}
+
+	now := time.Now()
+	switch item.Status {
+	case constants.PollingPriorityStatusPending:
+		// 待执行项必须由到达的执行认领并执行，否则提示通道产生的执行会被自己跳过、优先项永远无人执行。
+		claimed, claimErr := b.priorityStore.ClaimPending(ctx, cardID, taskType, now)
+		if claimErr != nil {
+			return nil, false, claimErr
+		}
+		if claimed == nil {
+			return nil, true, nil
+		}
+		return b.newPriorityRound(ctx, claimed, taskType, now), false, nil
+	case constants.PollingPriorityStatusProcessing:
+		// 执行中且认领租约未到期：不发第二次上游调用。
+		if item.ClaimedAt != nil && now.Before(item.ClaimedAt.Add(time.Duration(constants.PollingPriorityClaimLease)*time.Second)) {
+			return nil, true, nil
+		}
+		// 超过租约或认领时间缺失视为崩溃遗留：允许相邻执行接管，该卡该任务类型不被永久跳过。
+		takenOver, takeoverErr := b.priorityStore.TakeOverExpiredClaim(ctx, cardID, taskType, now, constants.PollingPriorityClaimLease)
+		if takeoverErr != nil {
+			return nil, false, takeoverErr
+		}
+		if takenOver == nil {
+			return nil, true, nil
+		}
+		return b.newPriorityRound(ctx, takenOver, taskType, now), false, nil
+	default:
+		// 终态行不属于活动集合，FindActive 不会返回；防御性按跳过处理。
+		return nil, true, nil
+	}
+}
+
+// newPriorityRound 创建领取成功后的执行轮次，并记录领取事实。
+func (b *PollingBase) newPriorityRound(ctx context.Context, item *model.PollingPriorityItem, taskType string, now time.Time) *pollingPriorityRound {
+	round := &pollingPriorityRound{
+		base:     b,
+		item:     item,
+		taskType: taskType,
+		taskName: pollingPriorityTaskName(taskType),
+		attempts: item.AttemptCount,
+	}
+	before := round.stateData()
+	item.Status = constants.PollingPriorityStatusProcessing
+	item.ClaimedAt = &now
+	round.writeAudit(constants.AuditActionPollingPriorityClaimed, "卡轮询优先项已领取", constants.AuditResultSuccess, "", "", before)
+	return round
+}
+
+// priorityInPollingScope 执行前校验该卡是否仍在轮询范围内：卡自身与绑定设备的轮询开关均为启用。
+//
+// 优先通道不得重启已停止轮询的卡；既有人工触发入口的范围语义不受本方法影响。
+// 无活动优先项（round 为 nil）时恒为真，普通轮询保持既有行为；不在范围内时按业务校验类失败
+// 终结优先项并返回 false，调用方据此跳过上游调用。
+func (b *PollingBase) priorityInPollingScope(_ context.Context, round *pollingPriorityRound, card *model.IotCard) bool {
+	if round == nil {
+		return true
+	}
+	if card == nil {
+		round.FailFinal(nil, "卡不存在")
+		return false
+	}
+	scopeCtx, cancel := pollingFallbackContext()
+	inScope := b.scopeChecker == nil || b.scopeChecker.ShouldEnqueue(scopeCtx, card)
+	cancel()
+	if inScope {
+		return true
+	}
+	round.FailFinal(nil, "卡已不在轮询范围内")
+	return false
+}
+
+// MarkExecution 标记即将真正发起上游调用，并在调用之前累加尝试次数。
+// 崩溃遗留的执行中项被接管后据此按尝试上限收敛。
+func (r *pollingPriorityRound) MarkExecution(_ context.Context) {
+	if r == nil {
+		return
+	}
+	ctx, cancel := pollingFallbackContext()
+	defer cancel()
+	hit, err := r.base.priorityStore.IncrementAttempt(ctx, r.item.ID)
+	if err != nil {
+		r.base.logger.Warn("累加优先轮询项尝试次数失败",
+			zap.Uint("priority_item_id", r.item.ID), zap.Error(err))
+	}
+	if !hit {
+		r.base.logger.Warn("优先轮询项已不处于执行中，尝试次数未累加",
+			zap.Uint("priority_item_id", r.item.ID))
+	}
+	r.attempts = r.item.AttemptCount + 1
+	r.item.AttemptCount = r.attempts
+}
+
+// Complete 记录优先项执行成功并出队，该卡继续按普通轮询运行。
+func (r *pollingPriorityRound) Complete(_ context.Context) {
+	if r == nil {
+		return
+	}
+	ctx, cancel := pollingFallbackContext()
+	defer cancel()
+	before := r.stateData()
+	hit, err := r.base.priorityStore.MarkCompleted(ctx, r.item.ID)
+	if err != nil {
+		r.base.logger.Warn("更新优先轮询项完成状态失败",
+			zap.Uint("priority_item_id", r.item.ID), zap.Error(err))
+	}
+	if !hit {
+		r.base.logger.Warn("优先轮询项已完成或不处于执行中，跳过完成收口",
+			zap.Uint("priority_item_id", r.item.ID))
+		return
+	}
+	r.item.Status = constants.PollingPriorityStatusCompleted
+	r.item.Result = constants.PollingPriorityResultSuccess
+	r.item.FailureReason = ""
+	r.writeAudit(constants.AuditActionPollingPriorityCompleted, "卡轮询优先项执行完成并出队", constants.AuditResultSuccess, "", "", before)
+}
+
+// FailRetryable 记录一次可恢复失败（上游超时、失败或响应无效）。
+// 未达尝试上限时退回活动态并按既有轮询间隔由调用方重排；达到上限时记录失败终态并出队。
+func (r *pollingPriorityRound) FailRetryable(_ context.Context, reason string) {
+	if r == nil {
+		return
+	}
+	ctx, cancel := pollingFallbackContext()
+	defer cancel()
+	before := r.stateData()
+	if r.attempts < constants.PollingPriorityMaxAttempts {
+		hit, err := r.base.priorityStore.ReturnToPending(ctx, r.item.ID, reason)
+		if err != nil {
+			r.base.logger.Warn("退回优先轮询项活动状态失败",
+				zap.Uint("priority_item_id", r.item.ID), zap.Error(err))
+		}
+		if !hit {
+			r.base.logger.Warn("优先轮询项不处于执行中，跳过退回活动态收口",
+				zap.Uint("priority_item_id", r.item.ID))
+			return
+		}
+		r.item.Status = constants.PollingPriorityStatusPending
+		r.item.Result = constants.PollingPriorityResultFailed
+		r.item.FailureReason = reason
+		r.item.ClaimedAt = nil
+		r.writeAudit(constants.AuditActionPollingPriorityFailed, "卡轮询优先项执行失败", constants.AuditResultFailed, "", reason, before)
+		r.writeAudit(constants.AuditActionPollingPriorityRetried, "卡轮询优先项未达尝试上限，退回活动态重新排期", constants.AuditResultFailed, "", reason, before)
+		return
+	}
+
+	hit, err := r.base.priorityStore.MarkFailed(ctx, r.item.ID, reason)
+	if err != nil {
+		r.base.logger.Warn("更新优先轮询项失败状态失败",
+			zap.Uint("priority_item_id", r.item.ID), zap.Error(err))
+	}
+	if !hit {
+		r.base.logger.Warn("优先轮询项不处于执行中，跳过失败出队收口",
+			zap.Uint("priority_item_id", r.item.ID))
+		return
+	}
+	r.item.Status = constants.PollingPriorityStatusFailed
+	r.item.Result = constants.PollingPriorityResultFailed
+	r.item.FailureReason = reason
+	r.writeAudit(constants.AuditActionPollingPriorityFailed, "卡轮询优先项执行失败", constants.AuditResultFailed, "", reason, before)
+	r.writeAudit(constants.AuditActionPollingPriorityDequeued, "卡轮询优先项达到尝试上限，失败出队", constants.AuditResultFailed, "", reason, before)
+}
+
+// FailFinal 记录业务校验类失败（卡不存在、不在轮询范围、能力未配置等）：不重试，直接失败出队。
+func (r *pollingPriorityRound) FailFinal(_ context.Context, reason string) {
+	if r == nil {
+		return
+	}
+	ctx, cancel := pollingFallbackContext()
+	defer cancel()
+	before := r.stateData()
+	hit, err := r.base.priorityStore.MarkFailed(ctx, r.item.ID, reason)
+	if err != nil {
+		r.base.logger.Warn("更新优先轮询项失败状态失败",
+			zap.Uint("priority_item_id", r.item.ID), zap.Error(err))
+	}
+	if !hit {
+		r.base.logger.Warn("优先轮询项不处于执行中，跳过失败出队收口",
+			zap.Uint("priority_item_id", r.item.ID))
+		return
+	}
+	r.item.Status = constants.PollingPriorityStatusFailed
+	r.item.Result = constants.PollingPriorityResultFailed
+	r.item.FailureReason = reason
+	r.writeAudit(constants.AuditActionPollingPriorityFailed, "卡轮询优先项执行业务校验失败", constants.AuditResultFailed, "", reason, before)
+	r.writeAudit(constants.AuditActionPollingPriorityDequeued, "卡轮询优先项失败出队且不重试", constants.AuditResultFailed, "", reason, before)
+}
+
+// stateData 返回当前优先项状态的前后值快照，覆盖触发来源、操作来源、时间与执行结果。
+func (r *pollingPriorityRound) stateData() map[string]any {
+	return map[string]any{
+		"status":            r.item.Status,
+		"result":            r.item.Result,
+		"trigger_type":      r.item.TriggerType,
+		"trigger_types":     r.item.TriggerTypes,
+		"trigger_count":     r.item.TriggerCount,
+		"last_triggered_at": r.item.LastTriggeredAt,
+		"attempt_count":     r.item.AttemptCount,
+		"claimed_at":        r.item.ClaimedAt,
+		"failure_reason":    r.item.FailureReason,
+		"source":            constants.AuditSourceWorker,
+		"occurred_at":       time.Now(),
+	}
+}
+
+// writeAudit 以独立短事务写入一条优先项生命周期审计。
+// 审计写入失败不阻断轮询执行，按既有约定登记为失败审计二次写入故障。
+func (r *pollingPriorityRound) writeAudit(
+	actionCode, summary, result, errorCode, errorSummary string,
+	before map[string]any,
+) {
+	base := r.base
+	if base == nil || base.priorityAudit == nil || base.db == nil || r.item == nil || r.item.ID == 0 {
+		return
+	}
+	ctx, cancel := pollingFallbackContext()
+	defer cancel()
+	input := audit.TaskInput{
+		ActionCode: actionCode, Summary: summary,
+		TaskID: r.item.ID, TaskNo: pollingPriorityResourceKey(r.item), DisplayName: "卡轮询优先项",
+		Actor: audit.ActorInput{
+			Kind: constants.AuditActorSystemTask, ID: r.taskType, Name: r.taskName,
+		},
+		Source:        constants.AuditSourceWorker,
+		ScopeType:     constants.AuditScopePlatform,
+		Result:        result,
+		ErrorCode:     errorCode,
+		ErrorSummary:  errorSummary,
+		CorrelationID: r.taskType,
+		IdentitySnapshot: map[string]any{
+			"id": r.item.ID, "card_id": r.item.CardID, "task_type": r.item.TaskType,
+			"status": r.item.Status, "trigger_type": r.item.TriggerType,
+			"trigger_count": r.item.TriggerCount, "manual_operator_id": r.item.ManualOperatorID,
+		},
+		BeforeData: before,
+		AfterData:  r.stateData(),
+	}
+	err := base.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
+		return base.priorityAudit.WriteTask(ctx, tx, input)
+	})
+	if err != nil {
+		auditfailure.RecordSecondaryWriteFailure(actionCode, input.TaskNo, "", r.taskType, errorCode, err)
+		base.logger.Error("优先轮询项审计写入失败",
+			zap.String("action", actionCode), zap.Uint("priority_item_id", r.item.ID), zap.Error(err))
+	}
+}
+
+// pollingPriorityTaskName 返回优先轮询任务类型的中文名，未知类型回退为完整任务类型。
+func pollingPriorityTaskName(taskType string) string {
+	if name, ok := pollingPriorityTaskNames[taskType]; ok {
+		return name
+	}
+	return taskType
+}
+
+// pollingPriorityResourceKey 返回优先项在审计里的稳定资源键。
+func pollingPriorityResourceKey(item *model.PollingPriorityItem) string {
+	if item == nil {
+		return ""
+	}
+	return strconv.FormatUint(uint64(item.ID), 10)
+}

internal/task/polling_realname_handler.go

@@ -41,28 +41,52 @@ func (h *PollingRealnameHandler) Handle(ctx context.Context, task *asynq.Task) e
 	}
 	defer h.base.releaseConcurrency(ctx, constants.TaskTypePollingRealname)
 
+	// 优先轮询互斥接缝：存在活动优先项时本轮即优先执行或被跳过；无活动项时行为与既有实现一致。
+	priority, skip, priorityErr := h.base.beginPollingPriority(ctx, cardID, constants.TaskTypePollingRealname)
+	if priorityErr != nil {
+		return h.failAndRequeue(ctx, cardID, startedAt, "查询活动优先轮询项失败", priorityErr)
+	}
+	if skip {
+		h.base.logger.Debug("卡轮询优先项已被其它执行持有，本轮跳过并延后",
+			zap.Uint("card_id", cardID), zap.String("task_type", constants.TaskTypePollingRealname))
+		return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingRealname)
+	}
+
 	card, err := h.base.getCardWithCache(ctx, cardID)
 	if err != nil {
 		if isNotFound(err) {
+			priority.FailFinal(ctx, "卡不存在")
 			return nil
 		}
+		// 本地读卡失败属可恢复失败：先按真实执行累加尝试次数，再按上限收敛，避免加急事实被静默丢弃。
+		priority.MarkExecution(ctx)
+		priority.FailRetryable(ctx, "获取卡信息失败")
 		return h.failAndRequeue(ctx, cardID, startedAt, "获取卡信息失败", err)
 	}
-	if card.CardCategory == constants.CardCategoryIndustry || h.gateway == nil {
+	// 执行前轮询范围校验：优先通道不得重启已停止轮询的卡。
+	if !h.base.priorityInPollingScope(ctx, priority, card) {
 		return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingRealname)
 	}
+	if card.CardCategory == constants.CardCategoryIndustry || h.gateway == nil {
+		priority.FailFinal(ctx, "卡类型不适用实名轮询或实名查询能力未配置")
+		return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingRealname)
+	}
+	priority.MarkExecution(ctx)
 	attempt := newGatewayAttempt(cardID, constants.IntegrationOperationGatewayRealname, constants.CardObservationSceneRealnamePolling)
 	result, err := h.gateway.QueryRealnameStatus(ctx, &gateway.CardStatusReq{CardNo: card.ICCID})
 	if err != nil {
 		_ = recordGatewayAttempt(ctx, h.integration, attempt, constants.IntegrationResultFailed, false, "查询实名状态失败")
+		priority.FailRetryable(ctx, "查询实名状态失败")
 		return h.failAndRequeue(ctx, cardID, startedAt, "查询实名状态失败", err)
 	}
 	if strings.TrimSpace(result.ICCID) == "" {
 		_ = recordGatewayAttempt(ctx, h.integration, attempt, constants.IntegrationResultInvalidPayload, false, "实名查询响应缺少 ICCID")
+		priority.FailRetryable(ctx, "实名查询响应缺少 ICCID")
 		return h.failAndRequeue(ctx, cardID, startedAt, "实名查询响应缺少 ICCID", nil)
 	}
 	ctx = withPollingWorkerAuditContext(ctx, constants.TaskTypePollingRealname, "实名状态轮询任务", attempt.IntegrationID)
 	if h.observation == nil {
+		priority.FailFinal(ctx, "卡实名观测能力未配置")
 		return h.failAndRequeue(ctx, cardID, startedAt, "卡实名观测能力未配置", nil)
 	}
 	now := time.Now()
@@ -75,6 +99,7 @@ func (h *PollingRealnameHandler) Handle(ctx context.Context, task *asynq.Task) e
 	})
 	if err != nil {
 		_ = recordGatewayAttempt(ctx, h.integration, attempt, constants.IntegrationResultUnknown, false, "应用实名观测失败")
+		priority.FailRetryable(ctx, "应用实名观测失败")
 		return h.failAndRequeue(ctx, cardID, startedAt, "应用实名观测失败", err)
 	}
 	changed := decision.StatusChanged
@@ -91,6 +116,7 @@ func (h *PollingRealnameHandler) Handle(ctx context.Context, task *asynq.Task) e
 		metric = "polling.observation.persisted"
 	}
 	h.base.logger.Info("实名轮询观测完成", zap.Uint("card_id", cardID), zap.Bool("changed", changed), zap.String("metric", metric))
+	priority.Complete(ctx)
 	h.base.updateStats(ctx, constants.TaskTypePollingRealname, true, time.Since(startedAt))
 	return h.base.requeueCard(ctx, cardID, constants.TaskTypePollingRealname)
 }