feat: 完善履约路由、订单服务与自动核销

2026-11-05 14:59:58 +08:00
parent 517d763f17
commit 1df88cc639
54 changed files with 3166 additions and 318 deletions
@@ -11,6 +11,7 @@ export const APP_CONFIG_KEYS = {
  workerProductNameSuffix: 'worker_product_name_suffix',
  workerRiskControl: 'worker_risk_control',
  workerMockShops: 'worker_mock_shops',
+  voucherConsumePolicy: 'voucher_consume_policy',
  kuaishouIndustrySource: 'kuaishou_industry_source',
  affiliateDash: 'affiliate_dash',
  pddSource: 'pdd_source',
@@ -0,0 +1,108 @@
+-- 129_fulfillment_retry_semantics.sql
+--
+-- 履约载体重试语义：
+--   1. 同一订单项/单元只限制一个“活跃”任务，历史 closed/failed 任务保留；
+--   2. 载体补建使用独立水位，避免每次重复扫描最早的 200 条；
+--   3. 回填历史内部 SKU 的大小写/全半角/空白规范化结果，并合并规范后重复渠道。
+
+-- 允许同一订单项/单元在旧任务关闭后创建新的履约尝试。
+ALTER TABLE fulfillment_tasks
+  DROP CONSTRAINT IF EXISTS fulfillment_tasks_order_item_id_unit_index_key;
+
+DROP INDEX IF EXISTS fulfillment_tasks_order_item_id_unit_index_key;
+
+CREATE UNIQUE INDEX IF NOT EXISTS uq_fulfillment_tasks_active_unit
+  ON fulfillment_tasks (order_item_id, unit_index)
+  WHERE task_status NOT IN (
+    'closed', 'expired', 'failed', 'manual_review', 'retry_pending', 'cancelled'
+  );
+
+COMMENT ON INDEX uq_fulfillment_tasks_active_unit IS
+  '同一订单项/单元只允许一个活跃履约载体；关闭、失败、待重试和人工复核任务保留为历史记录';
+
+ALTER TABLE work_orders
+  DROP CONSTRAINT IF EXISTS idx_work_orders_order_item_unique;
+
+DROP INDEX IF EXISTS idx_work_orders_order_item_unique;
+
+CREATE UNIQUE INDEX IF NOT EXISTS uq_work_orders_active_order_item
+  ON work_orders (order_item_id)
+  WHERE order_item_id IS NOT NULL AND status <> 'cancelled';
+
+COMMENT ON INDEX uq_work_orders_active_order_item IS
+  '同一订单项只允许一个未取消工单；取消工单保留为历史并可重新建单';
+
+INSERT INTO finance_watermarks (pipeline, last_item_id, last_success_at)
+VALUES ('carrier_reconcile', 0, NULL)
+ON CONFLICT (pipeline) DO NOTHING;
+
+COMMENT ON COLUMN finance_watermarks.pipeline IS
+  '同步管道：channel_assign / carrier_reconcile / carrier_dispatch / product_normalize / cost_snapshot / cost_snapshot_hot / sales_facts / ledger_post';
+
+-- 规范化前先收敛跨渠道主渠道。同一规范 SKU 只能有一个 enabled 主渠道；
+-- 优先保留带载体参数的行，再按优先级、id 稳定选择。
+WITH ranked_primary AS (
+  SELECT
+    id,
+    ROW_NUMBER() OVER (
+      PARTITION BY canonicalize_internal_sku(internal_sku)
+      ORDER BY
+        CASE WHEN config_json = '{}'::jsonb THEN 1 ELSE 0 END,
+        priority,
+        id
+    ) AS candidate_rank
+  FROM internal_sku_channels
+  WHERE enabled = TRUE AND is_fallback = FALSE
+)
+UPDATE internal_sku_channels channel
+SET is_fallback = TRUE,
+    priority = GREATEST(priority, 90),
+    updated_at = NOW()
+FROM ranked_primary
+WHERE channel.id = ranked_primary.id
+  AND ranked_primary.candidate_rank > 1;
+
+-- 同一规范 SKU + 渠道只保留一行，优先保留启用、带载体参数的行。
+WITH ranked AS (
+  SELECT
+    internal_sku,
+    channel,
+    id,
+    canonicalize_internal_sku(internal_sku) AS canonical_sku,
+    ROW_NUMBER() OVER (
+      PARTITION BY canonicalize_internal_sku(internal_sku), channel
+      ORDER BY
+        enabled DESC,
+        CASE WHEN config_json = '{}'::jsonb THEN 1 ELSE 0 END,
+        is_fallback,
+        priority,
+        id
+    ) AS candidate_rank
+  FROM internal_sku_channels
+)
+DELETE FROM internal_sku_channels channel
+USING ranked
+WHERE channel.id = ranked.id
+  AND ranked.candidate_rank > 1;
+
+UPDATE internal_sku_channels
+SET internal_sku = canonicalize_internal_sku(internal_sku),
+    updated_at = NOW()
+WHERE internal_sku <> canonicalize_internal_sku(internal_sku);
+
+UPDATE manual_fulfillment_configs
+SET sku_key = canonicalize_internal_sku(sku_key),
+    updated_at = NOW()
+WHERE sku_key <> canonicalize_internal_sku(sku_key);
+
+UPDATE manual_sku_registry
+SET sku_key = canonicalize_internal_sku(sku_key),
+    updated_at = NOW()
+WHERE sku_key <> canonicalize_internal_sku(sku_key);
+
+UPDATE internal_sku_mappings
+SET match_value = canonicalize_internal_sku(match_value),
+    internal_sku = canonicalize_internal_sku(internal_sku),
+    updated_at = NOW()
+WHERE match_value <> canonicalize_internal_sku(match_value)
+   OR internal_sku <> canonicalize_internal_sku(internal_sku);
@@ -0,0 +1,50 @@
+-- Persistent delayed voucher consumption jobs.
+--
+-- Business events only enqueue a job. A scheduler worker performs the external
+-- platform call later, so a restart cannot lose a delayed consumption.
+
+CREATE TABLE IF NOT EXISTS voucher_consume_jobs (
+  id BIGSERIAL PRIMARY KEY,
+  provider TEXT NOT NULL
+    CHECK (provider IN ('kuaishou_industry', 'douyin_mashida')),
+  channel TEXT NOT NULL
+    CHECK (channel IN ('affiliate_dash', 'work_order', 'manual', 'third_party_voucher')),
+  trigger_event TEXT NOT NULL,
+  target_key TEXT NOT NULL,
+  platform_order_id TEXT NOT NULL DEFAULT '',
+  order_id BIGINT REFERENCES orders(id) ON DELETE SET NULL,
+  order_item_id BIGINT REFERENCES order_items(id) ON DELETE SET NULL,
+  task_id BIGINT REFERENCES fulfillment_tasks(id) ON DELETE SET NULL,
+  work_order_id BIGINT REFERENCES work_orders(id) ON DELETE SET NULL,
+  due_at TIMESTAMPTZ NOT NULL,
+  status TEXT NOT NULL DEFAULT 'pending'
+    CHECK (status IN ('pending', 'processing', 'succeeded', 'failed', 'cancelled')),
+  attempt_count INTEGER NOT NULL DEFAULT 0,
+  max_attempts INTEGER NOT NULL DEFAULT 8,
+  locked_until TIMESTAMPTZ,
+  last_error TEXT NOT NULL DEFAULT '',
+  result_json JSONB NOT NULL DEFAULT '{}'::jsonb,
+  created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+  updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+  finished_at TIMESTAMPTZ
+);
+
+-- Keep one active/successful target per provider. Failed jobs may be replaced
+-- by a later business event, while succeeded jobs suppress duplicate events.
+CREATE UNIQUE INDEX IF NOT EXISTS uq_voucher_consume_jobs_active_target
+  ON voucher_consume_jobs(provider, target_key)
+  WHERE status IN ('pending', 'processing', 'succeeded');
+
+CREATE INDEX IF NOT EXISTS idx_voucher_consume_jobs_due
+  ON voucher_consume_jobs(status, due_at, id);
+
+CREATE INDEX IF NOT EXISTS idx_voucher_consume_jobs_order
+  ON voucher_consume_jobs(order_id, order_item_id, id DESC);
+
+CREATE INDEX IF NOT EXISTS idx_voucher_consume_jobs_updated
+  ON voucher_consume_jobs(updated_at DESC, id DESC);
+
+COMMENT ON TABLE voucher_consume_jobs IS
+  'Delayed voucher verification jobs; events enqueue, scheduler executes';
+COMMENT ON COLUMN voucher_consume_jobs.target_key IS
+  'Stable dedupe target: order-item when available, otherwise task/order';
@@ -28,6 +28,7 @@ export type OrderServiceProfileRow = {
 export type OrderServicePublishedProfileRow = {
  profile_id: number
  category_id: number
+  source_category_id: number
  version_id: number
  version_no: number
  config_json: Record<string, unknown>
@@ -221,12 +222,29 @@ export async function backfillPendingOrderServiceSnapshotsForCategory(
 ): Promise<number> {
  const result = await client.query(
    `
+      WITH affected_categories AS (
+        SELECT $1::bigint AS category_id
+        UNION
+        SELECT child.id
+        FROM work_categories child
+        WHERE child.parent_id = $1
+          AND NOT EXISTS (
+            SELECT 1
+            FROM order_service_profiles child_profile
+            JOIN order_service_profile_versions child_version
+              ON child_version.id = child_profile.current_version_id
+            WHERE child_profile.category_id = child.id
+              AND child_profile.status <> 'archived'
+          )
+      )
      UPDATE work_orders wo
      SET service_profile_id = $2,
          service_profile_version_id = $3,
          service_profile_snapshot_json = jsonb_build_object(
            'profileId', p.id,
-            'categoryId', p.category_id,
+            'categoryId', wo.category_id,
+            'sourceCategoryId', p.category_id,
+            'inherited', wo.category_id <> p.category_id,
            'versionId', v.id,
            'versionNo', v.version_no,
            'config', v.config_json
@@ -240,7 +258,7 @@ export async function backfillPendingOrderServiceSnapshotsForCategory(
      WHERE p.id = $2
        AND p.category_id = $1
        AND v.id = $3
-        AND wo.category_id = p.category_id
+        AND wo.category_id IN (SELECT category_id FROM affected_categories)
        AND wo.status = 'pending_material'
        AND wo.assigned_worker_id IS NULL
        AND NOT EXISTS (
@@ -295,6 +313,7 @@ export async function getPublishedOrderServiceProfileByCategory(
      SELECT
        p.id AS profile_id,
        p.category_id,
+        p.category_id AS source_category_id,
        v.id AS version_id,
        v.version_no,
        v.config_json
@@ -311,6 +330,57 @@ export async function getPublishedOrderServiceProfileByCategory(
    ? {
        profile_id: Number(row.profile_id),
        category_id: Number(row.category_id),
+        source_category_id: Number(row.source_category_id),
+        version_id: Number(row.version_id),
+        version_no: Number(row.version_no),
+        config_json: row.config_json,
+      }
+    : null
+}
+
+/**
+ * 分类配置解析：先从当前二级分类取已发布版本，没有再逐级继承父分类。
+ * 返回值中的 source_category_id 表示最终命中的配置所属分类。
+ */
+export async function getPublishedOrderServiceProfileByCategoryWithInheritance(
+  categoryId: number | string,
+  client?: Pick<PoolClient, 'query'>,
+): Promise<OrderServicePublishedProfileRow | null> {
+  const db = client || { query }
+  const result = await db.query<OrderServicePublishedProfileRow>(
+    `
+      WITH RECURSIVE category_chain AS (
+        SELECT id, parent_id, 0 AS depth
+        FROM work_categories
+        WHERE id = $1
+        UNION ALL
+        SELECT parent.id, parent.parent_id, child.depth + 1
+        FROM work_categories parent
+        JOIN category_chain child ON parent.id = child.parent_id
+        WHERE child.depth < 8
+      )
+      SELECT
+        p.id AS profile_id,
+        p.category_id,
+        chain.id AS source_category_id,
+        v.id AS version_id,
+        v.version_no,
+        v.config_json
+      FROM category_chain chain
+      JOIN order_service_profiles p ON p.category_id = chain.id
+      JOIN order_service_profile_versions v ON v.id = p.current_version_id
+      WHERE p.status <> 'archived'
+      ORDER BY chain.depth ASC
+      LIMIT 1
+    `,
+    [Number(categoryId)],
+  )
+  const row = result.rows[0]
+  return row
+    ? {
+        profile_id: Number(row.profile_id),
+        category_id: Number(row.category_id),
+        source_category_id: Number(row.source_category_id),
        version_id: Number(row.version_id),
        version_no: Number(row.version_no),
        config_json: row.config_json,
@@ -298,7 +298,10 @@ export async function getWorkOrderByOrderItemId(
  orderItemId: number | string,
 ): Promise<WorkOrderRow | null> {
  const result = await query<WorkOrderRow>(
-    `${WORK_ORDER_SELECT} WHERE wo.order_item_id = $1 LIMIT 1`,
+    `${WORK_ORDER_SELECT}
+     WHERE wo.order_item_id = $1
+     ORDER BY (wo.status <> 'cancelled') DESC, wo.id DESC
+     LIMIT 1`,
    [Number(orderItemId)],
  )
  return result.rows[0] || null
@@ -17,6 +17,7 @@ import {
  forceDeleteChannelRules,
  listChannelRulesPage,
  listPendingChannelGroups,
+  retryPendingCarrierGroup,
  switchChannelForOrderItems,
  unlockChannel,
  upsertChannelRule,
@@ -37,6 +38,15 @@ import {
  getFinancePipelineStatus,
  isFinancePipelineName,
 } from '../../services/scheduler/finance-pipeline-status.js'
+import {
+  getVoucherConsumePolicy,
+  saveVoucherConsumePolicy,
+} from '../../services/fulfillment/voucher-consume-policy-service.js'
+import {
+  listVoucherConsumeJobs,
+  retryVoucherConsumeJob,
+} from '../../services/fulfillment/voucher-consume-queue-service.js'
+import { createHttpError } from '../../utils/http.js'
 import { bodyOf } from './request-helpers.js'
 import { createJsonHandler, requireAdminPermission, requireAdminRoles } from './session.js'

@@ -44,6 +54,78 @@ const router = Router()

 router.use('/fulfillment', requireAdminRoles(['admin', 'operator', 'finance', 'support']))

+router.get(
+  '/fulfillment/voucher-consume-policy',
+  requireAdminPermission('finance.view'),
+  createJsonHandler(() => getVoucherConsumePolicy(), {
+    successMessage: 'ok',
+    errorMessage: '读取核销策略失败',
+    scope: '[admin/fulfillment/voucher-consume-policy]',
+  }),
+)
+
+router.post(
+  '/fulfillment/voucher-consume-policy',
+  requireAdminPermission('finance.config'),
+  createJsonHandler((req) => saveVoucherConsumePolicy(bodyOf(req)), {
+    successMessage: '核销策略已保存',
+    errorMessage: '保存核销策略失败',
+    scope: '[admin/fulfillment/voucher-consume-policy]',
+    audit: (_req, data) => ({
+      action: 'fulfillment.save_voucher_consume_policy',
+      targetType: 'voucher_consume_policy',
+      targetId: 'default',
+      data: data as Record<string, unknown>,
+    }),
+  }),
+)
+
+router.get(
+  '/fulfillment/voucher-consume-jobs',
+  requireAdminPermission('finance.view'),
+  createJsonHandler(
+    (req) =>
+      listVoucherConsumeJobs({
+        status: String(req.query.status || ''),
+        page: Number(req.query.page),
+        pageSize: Number(req.query.pageSize),
+      }),
+    {
+      successMessage: 'ok',
+      errorMessage: '读取核销任务失败',
+      scope: '[admin/fulfillment/voucher-consume-jobs]',
+    },
+  ),
+)
+
+router.post(
+  '/fulfillment/voucher-consume-jobs/:id/retry',
+  requireAdminPermission('finance.config'),
+  createJsonHandler(
+    async (req) => {
+      const job = await retryVoucherConsumeJob(req.params.id)
+      if (!job) {
+        throw createHttpError('该核销任务当前不可重试，或已有更新的任务在队列中', {
+          statusCode: 409,
+          errorCode: 'voucher_consume_job_not_retryable',
+        })
+      }
+      return job
+    },
+    {
+      successMessage: '核销任务已重新排队',
+      errorMessage: '重试核销任务失败',
+      scope: '[admin/fulfillment/voucher-consume-jobs/retry]',
+      audit: (req, data) => ({
+        action: 'fulfillment.retry_voucher_consume_job',
+        targetType: 'voucher_consume_job',
+        targetId: String(req.params.id || ''),
+        data: data as Record<string, unknown>,
+      }),
+    },
+  ),
+)
+
 router.get(
  '/fulfillment/pending',
  requireAdminPermission('finance.view'),
@@ -110,6 +192,28 @@ router.post(
  ),
 )

+router.post(
+  '/fulfillment/pending/retry-carriers',
+  requireAdminPermission('finance.config'),
+  createJsonHandler(
+    (req) =>
+      retryPendingCarrierGroup({
+        internalSku: String(bodyOf(req).internalSku || ''),
+      }),
+    {
+      successMessage: '履约载体重试完成',
+      errorMessage: '重试履约载体失败',
+      scope: '[admin/fulfillment/pending/retry-carriers]',
+      audit: (req, data) => ({
+        action: 'fulfillment.retry_pending_carriers',
+        targetType: 'internal_sku',
+        targetId: String(bodyOf(req).internalSku || ''),
+        data: data as Record<string, unknown>,
+      }),
+    },
+  ),
+)
+
 router.get(
  '/fulfillment/internal-sku-mappings',
  requireAdminPermission('finance.view'),
@@ -319,7 +319,7 @@ const CANDIDATE_SQL = `
    wo.reward_amount
  FROM order_items oi
  JOIN orders o ON o.id = oi.order_id
-  LEFT JOIN work_orders wo ON wo.order_item_id = oi.id
+  LEFT JOIN work_orders wo ON wo.order_item_id = oi.id AND wo.status <> 'cancelled'
  LEFT JOIN work_product_rules wpr ON wpr.id = wo.product_rule_id
  WHERE oi.fulfillment_status = 'assigned'
    AND oi.fulfillment_channel IN (
@@ -325,7 +325,7 @@ export async function normalizeOrderItemProducts(
          cfg.canonical_product_id AS manual_config_product_id
        FROM order_items oi
        JOIN orders o ON o.id = oi.order_id
-        LEFT JOIN work_orders wo ON wo.order_item_id = oi.id
+        LEFT JOIN work_orders wo ON wo.order_item_id = oi.id AND wo.status <> 'cancelled'
        LEFT JOIN work_product_rules wpr ON wpr.id = wo.product_rule_id
        LEFT JOIN manual_fulfillment_configs cfg ON cfg.id = oi.fulfillment_config_id
        WHERE oi.fulfillment_status = 'assigned'
@@ -479,7 +479,7 @@ export async function normalizeProductsForItems(itemIds: number[]): Promise<{
        cfg.canonical_product_id AS manual_config_product_id
      FROM order_items oi
      JOIN orders o ON o.id = oi.order_id
-      LEFT JOIN work_orders wo ON wo.order_item_id = oi.id
+      LEFT JOIN work_orders wo ON wo.order_item_id = oi.id AND wo.status <> 'cancelled'
      LEFT JOIN work_product_rules wpr ON wpr.id = wo.product_rule_id
      LEFT JOIN manual_fulfillment_configs cfg ON cfg.id = oi.fulfillment_config_id
      WHERE oi.fulfillment_status = 'assigned'
@@ -728,7 +728,7 @@ async function loadTemplateProductBootstrapData(range: { fromDate: string; toDat
          COUNT(DISTINCT oi.id)::int AS item_count
        FROM order_items oi
        JOIN orders o ON o.id = oi.order_id
-        JOIN work_orders wo ON wo.order_item_id = oi.id
+        JOIN work_orders wo ON wo.order_item_id = oi.id AND wo.status <> 'cancelled'
        JOIN work_product_rules r ON r.id = wo.product_rule_id AND r.enabled = TRUE
        WHERE wo.product_rule_id IS NOT NULL
          AND (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date
@@ -33,7 +33,7 @@ function resolveWorkOrderRange(payload: Record<string, unknown>) {
 export async function backfillAdminWorkOrderProductSkus(payload: Record<string, unknown> = {}) {
  const now = nowIso()
  const range = resolveWorkOrderRange(payload)
-  // order_item_id 有唯一索引，这里可以直接 count，避免 DISTINCT ON 被
+  // 未取消工单对 order_item_id 有唯一索引，这里可以直接 count，避免 DISTINCT ON 被
  // 三个子查询反复物化后撞 15 秒语句超时。
  const result = await query<WorkOrderSkuBackfillRow>(
    `
@@ -51,6 +51,7 @@ export async function backfillAdminWorkOrderProductSkus(payload: Record<string,
      JOIN order_items oi ON oi.id = wo.order_item_id
      JOIN orders o ON o.id = oi.order_id
      WHERE wo.order_item_id IS NOT NULL
+        AND wo.status <> 'cancelled'
        AND (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date
          BETWEEN $1::date AND $2::date
    `,
@@ -73,6 +74,7 @@ export async function backfillAdminWorkOrderProductSkus(payload: Record<string,
          JOIN order_items oi ON oi.id = wo.order_item_id
          JOIN orders o ON o.id = oi.order_id
          WHERE wo.order_item_id IS NOT NULL
+            AND wo.status <> 'cancelled'
            AND (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date
              BETWEEN $1::date AND $2::date
            AND wo.order_item_id > $3
@@ -125,7 +127,10 @@ export async function backfillAdminWorkOrderProductSkus(payload: Record<string,
          AND btrim(COALESCE(oi.normalized_sku_name, '')) <> ''
          AND (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date
            BETWEEN $2::date AND $3::date
-          AND EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id)
+          AND EXISTS (
+            SELECT 1 FROM work_orders wo
+            WHERE wo.order_item_id = oi.id AND wo.status <> 'cancelled'
+          )
        GROUP BY 1
      ),
      missing AS (
@@ -452,6 +457,7 @@ export async function syncAdminWorkOrderTemplateCostVersions(ruleId?: number) {
      FROM work_orders wo
      JOIN rule_version rv ON rv.work_product_rule_id = wo.product_rule_id
      WHERE cs.order_item_id = wo.order_item_id
+        AND wo.status <> 'cancelled'
        AND cs.cost_item_id = rv.cost_item_id
        AND cs.superseded_at IS NULL AND reversed_at IS NULL
        AND cs.cost_version_id <> rv.cost_version_id
@@ -54,6 +54,53 @@ async function hasAffiliateTask(orderItemId: number): Promise<boolean> {
  return result.rows[0]?.exists === true
 }

+const BLOCKED_INACTIVE_CARRIER_STATUSES = new Set(['closed', 'cancelled', 'manual_review'])
+
+async function findBlockingInactiveCarrier(input: {
+  orderItemId: number
+  channel: FulfillmentChannel
+}): Promise<{ status: string; reason: string; message: string } | null> {
+  if (input.channel !== 'affiliate_dash' && input.channel !== 'work_order') return null
+
+  const result =
+    input.channel === 'affiliate_dash'
+      ? await query<{ status: string }>(
+          `SELECT task_status AS status
+           FROM fulfillment_tasks
+           WHERE order_item_id = $1 AND executor_key = 'affiliate_dash'
+           ORDER BY id DESC
+           LIMIT 1`,
+          [input.orderItemId],
+        )
+      : await query<{ status: string }>(
+          `SELECT status
+           FROM work_orders
+           WHERE order_item_id = $1
+           ORDER BY id DESC
+           LIMIT 1`,
+          [input.orderItemId],
+        )
+
+  const status = String(result.rows[0]?.status || '')
+  if (!BLOCKED_INACTIVE_CARRIER_STATUSES.has(status)) return null
+
+  if (status === 'manual_review') {
+    return {
+      status,
+      reason: 'carrier_task_manual_review',
+      message: '已有履约载体进入人工复核，需先处理后再重试',
+    }
+  }
+  return {
+    status,
+    reason: 'carrier_task_closed',
+    message:
+      status === 'cancelled'
+        ? '已有履约载体已取消，需由运营确认后重新创建'
+        : '已有履约载体已关闭，需由运营确认后重新创建',
+  }
+}
+
 function toResolvedItem(item: OrderItemRow): ResolvedFulfillmentItem {
  return {
    ...item,
@@ -79,8 +126,16 @@ export async function ensureCarrierForOrderItem(input: {
  order: OrderRow
  item: OrderItemRow
  channel: FulfillmentChannel
+  retryInactiveCarrier?: boolean
+  allowInactiveTaskRetry?: boolean
 }): Promise<EnsureCarrierResult> {
-  const { order, item, channel } = input
+  const {
+    order,
+    item,
+    channel,
+    retryInactiveCarrier = false,
+    allowInactiveTaskRetry = false,
+  } = input
  const existingCarrier = findConflictingCarrier(
    await loadActiveCarrierEvidenceForItem(Number(item.id)),
    channel,
@@ -131,6 +186,18 @@ export async function ensureCarrierForOrderItem(input: {
    if (await hasWorkOrder(Number(item.id))) {
      return { ready: true, created: false, reason: 'work_order_exists', message: '' }
    }
+    const inactiveCarrier = await findBlockingInactiveCarrier({
+      orderItemId: Number(item.id),
+      channel,
+    })
+    if (inactiveCarrier && !retryInactiveCarrier) {
+      return {
+        ready: false,
+        created: false,
+        reason: inactiveCarrier.reason,
+        message: inactiveCarrier.message,
+      }
+    }
    const internalSku = canonicalizeInternalSku(
      resolveInternalSkuKey({
        manualSkuKey: item.manual_sku_key,
@@ -179,6 +246,7 @@ export async function ensureCarrierForOrderItem(input: {
      autoOnly: true,
      skipChannelAssign: true,
      preferredWorkRuleKey,
+      retryCancelledWorkOrder: retryInactiveCarrier,
    })
    const ready = (await hasWorkOrder(Number(item.id))) || result.created.length > 0
    if (ready) {
@@ -198,6 +266,18 @@ export async function ensureCarrierForOrderItem(input: {
  if (await hasAffiliateTask(Number(item.id))) {
    return { ready: true, created: false, reason: 'affiliate_task_exists', message: '' }
  }
+  const inactiveCarrier = await findBlockingInactiveCarrier({
+    orderItemId: Number(item.id),
+    channel,
+  })
+  if (inactiveCarrier && !retryInactiveCarrier) {
+    return {
+      ready: false,
+      created: false,
+      reason: inactiveCarrier.reason,
+      message: inactiveCarrier.message,
+    }
+  }

  const resolved = await applyIntakeChannelDecision(
    toResolvedItem(item),
@@ -232,14 +312,18 @@ export async function ensureCarrierForOrderItem(input: {
  )
  const before = await hasAffiliateTask(Number(item.id))
  if (!before) {
-    await syncDeliveryTasksForOrder(order, [
-      {
-        ...item,
-        sku_code: String(resolved.skuCode || item.sku_code),
-        spec_json: resolved.spec || {},
-        item_snapshot_json: snapshot,
-      } as OrderItemRow,
-    ])
+    await syncDeliveryTasksForOrder(
+      order,
+      [
+        {
+          ...item,
+          sku_code: String(resolved.skuCode || item.sku_code),
+          spec_json: resolved.spec || {},
+          item_snapshot_json: snapshot,
+        } as OrderItemRow,
+      ],
+      { allowInactiveTaskRetry: retryInactiveCarrier || allowInactiveTaskRetry },
+    )
  }
  const ready = await hasAffiliateTask(Number(item.id))
  return {
@@ -49,11 +49,21 @@ async function loadCandidateItems(input: {
  limit: number
  orderItemIds?: number[]
  assignedMissingCarrier?: boolean
+  afterId?: number
+  includeLocked?: boolean
 }): Promise<CarrierCandidate[]> {
  const ids = (input.orderItemIds || [])
    .map(Number)
    .filter((value) => Number.isInteger(value) && value > 0)
-  const idFilter = ids.length > 0 ? 'AND oi.id = ANY($2::bigint[])' : ''
+  const params: unknown[] = [input.limit]
+  let idFilter = ''
+  if (ids.length > 0) {
+    params.push(ids)
+    idFilter = `AND oi.id = ANY($${params.length}::bigint[])`
+  } else if (Number(input.afterId || 0) > 0) {
+    params.push(Math.max(0, Math.trunc(Number(input.afterId))))
+    idFilter = `AND oi.id > $${params.length}`
+  }
  const statusFilter = input.assignedMissingCarrier
    ? `oi.fulfillment_status = 'assigned'
       AND oi.fulfillment_channel IN ('affiliate_dash', 'work_order')
@@ -66,10 +76,9 @@ async function loadCandidateItems(input: {
         WHERE ft.order_item_id = oi.id AND ft.executor_key = 'affiliate_dash'
           AND ft.task_status NOT IN (${INACTIVE_CARRIER_TASK_STATUSES.map((status) => `'${status}'`).join(', ')})
       )`
-    : `oi.fulfillment_status = 'pending' AND oi.fulfillment_locked = FALSE`
+    : `oi.fulfillment_status = 'pending'
+       ${input.includeLocked ? '' : 'AND oi.fulfillment_locked = FALSE'}`

-  const params: unknown[] = [input.limit]
-  if (ids.length > 0) params.push(ids)
  const result = await query<CarrierCandidate>(
    `
      SELECT
@@ -119,6 +128,7 @@ async function markPendingWithoutCarrier(input: {
  internalSku: string
  attempts: Array<{ channel: FulfillmentChannel; reason: string; message: string }>
 }): Promise<void> {
+  const carrierIssue = input.attempts[0]?.reason || 'carrier_attempt_failed'
  await query(
    `
      UPDATE order_items
@@ -131,6 +141,8 @@ async function markPendingWithoutCarrier(input: {
      input.itemId,
      JSON.stringify({
        internalSku: input.internalSku,
+        carrierIssue,
+        carrierMessage: input.attempts[0]?.message || '',
        attempts: input.attempts,
      }),
    ],
@@ -191,6 +203,9 @@ export async function reconcilePendingCarriers(
    orderItemIds?: number[]
    includeAssignedMissingCarrier?: boolean
    allowFallback?: boolean
+    afterId?: number
+    includeLocked?: boolean
+    retryInactiveCarrier?: boolean
  } = {},
 ): Promise<{
  scanned: number
@@ -199,11 +214,14 @@ export async function reconcilePendingCarriers(
  fallbacksSelected: number
  failed: number
  remaining: number
+  nextCursor: number
 }> {
  const limit = Math.min(500, Math.max(1, Math.trunc(Number(input.limit) || 200)))
  const rows = await loadCandidateItems({
    limit,
    assignedMissingCarrier: input.includeAssignedMissingCarrier === true,
+    ...(input.afterId != null ? { afterId: input.afterId } : {}),
+    ...(input.includeLocked ? { includeLocked: true } : {}),
    ...(input.orderItemIds ? { orderItemIds: input.orderItemIds } : {}),
  })

@@ -273,6 +291,8 @@ export async function reconcilePendingCarriers(
            order,
            item,
            channel: attempt.channel,
+            retryInactiveCarrier: input.retryInactiveCarrier === true,
+            allowInactiveTaskRetry: true,
          })
          if (!carrier.ready) {
            failures.push({
@@ -364,5 +384,10 @@ export async function reconcilePendingCarriers(
    fallbacksSelected,
    failed,
    remaining: Number(remainingResult.rows[0]?.count || 0),
+    nextCursor: input.orderItemIds?.length
+      ? 0
+      : rows.length < limit
+        ? 0
+        : Number(rows[rows.length - 1]?.id || 0),
  }
 }
@@ -30,6 +30,7 @@ import {
  type InternalSkuChannelRow,
 } from './internal-sku-channel-service.js'
 import { ensureCarrierForOrderItem } from './carrier-creation-service.js'
+import { reconcilePendingCarriers } from './carrier-reconciliation-service.js'
 import { TASK_STATUS } from '../../domain/task-status.js'
 import { canonicalizeInternalSku, containsCjk } from './internal-sku-resolver.js'

@@ -116,12 +117,23 @@ function resolvePendingReason(item: ChannelCandidateItem): {
      detail: { internalSku: item.internal_sku },
    }
  }
+  const primary = plan.primary
+  const primaryConfig = primary.config_json || {}
+  const carrierIssue =
+    primary.channel === 'manual' && !Number(primaryConfig.manualConfigId || 0)
+      ? 'manual_config_missing'
+      : primary.channel === 'work_order' && !String(primaryConfig.workRuleKey || '').trim()
+        ? 'work_rule_missing'
+        : primary.channel === 'affiliate_dash' && !String(primaryConfig.affiliateSku || '').trim()
+          ? 'affiliate_mapping_missing'
+          : 'carrier_not_attempted'
  return {
    reason: 'no_carrier',
    detail: {
      internalSku: item.internal_sku,
-      channel: plan.primary.channel,
+      channel: primary.channel,
      fallbacks: plan.fallbacks.map((channel) => channel.channel),
+      carrierIssue,
    },
  }
 }
@@ -234,7 +246,13 @@ export type AssignChannelsResult = {

 /** 批量判定：默认只扫 pending 且未锁定的订单项。 */
 export async function assignPendingOrderItemChannels(
-  options: { batchSize?: number; maxBatches?: number; afterId?: number } = {},
+  options: {
+    batchSize?: number
+    maxBatches?: number
+    afterId?: number
+    orderItemIds?: number[]
+    includeLocked?: boolean
+  } = {},
 ): Promise<AssignChannelsResult> {
  const batchSize = Math.min(5000, Math.max(1, Math.trunc(options.batchSize || 500)))
  const maxBatches = Math.min(200, Math.max(1, Math.trunc(options.maxBatches || 20)))
@@ -257,8 +275,20 @@ export async function assignPendingOrderItemChannels(
    pendingReasons: {},
  }
  const affectedDates = new Set<string>()
+  const requestedIds = [
+    ...new Set(
+      (options.orderItemIds || []).map(Number).filter((id) => Number.isInteger(id) && id > 0),
+    ),
+  ]
+  const params: unknown[] = [cursor, batchSize]
+  let requestedFilter = ''
+  if (requestedIds.length > 0) {
+    params.push(requestedIds)
+    requestedFilter = `AND oi.id = ANY($${params.length}::bigint[])`
+  }

  for (let batch = 0; batch < maxBatches; batch += 1) {
+    params[0] = cursor
    const rows = await query<ChannelCandidateItem>(
      `
        SELECT
@@ -288,12 +318,13 @@ export async function assignPendingOrderItemChannels(
        FROM order_items oi
        JOIN orders o ON o.id = oi.order_id
        WHERE oi.fulfillment_status = 'pending'
-          AND oi.fulfillment_locked = FALSE
+          ${options.includeLocked ? '' : 'AND oi.fulfillment_locked = FALSE'}
          AND oi.id > $1
+          ${requestedFilter}
        ORDER BY oi.id
        LIMIT $2
      `,
-      [cursor, batchSize],
+      params,
    )

    if (rows.rows.length === 0) {
@@ -353,7 +384,13 @@ export async function assignPendingOrderItemChannels(
        await query(
          `UPDATE order_items
           SET fulfillment_pending_reason = $2,
-               fulfillment_pending_detail = $3,
+               fulfillment_pending_detail = CASE
+                 WHEN $2 = 'no_carrier'
+                   AND COALESCE(fulfillment_pending_detail->>'carrierIssue', '')
+                       NOT IN ('', 'carrier_not_attempted')
+                 THEN fulfillment_pending_detail
+                 ELSE $3::jsonb
+               END,
               updated_at = NOW()
           WHERE id = $1 AND fulfillment_status = 'pending'`,
          [item.id, pending.reason, JSON.stringify(pending.detail)],
@@ -445,7 +482,13 @@ export async function assignOrderItemsAfterCarrierReady(input: {

    let carrier
    try {
-      carrier = await ensureCarrierForOrderItem({ order, item, channel: input.channel })
+      carrier = await ensureCarrierForOrderItem({
+        order,
+        item,
+        channel: input.channel,
+        retryInactiveCarrier: true,
+        allowInactiveTaskRetry: true,
+      })
    } catch (error) {
      carrier = {
        ready: false,
@@ -498,6 +541,7 @@ export async function assignOrderItemsAfterCarrierReady(input: {
              channel: input.channel,
              reason: carrier.reason,
              message: carrier.message,
+              carrierIssue: carrier.reason,
            }),
          ],
        )
@@ -905,6 +949,9 @@ export type PendingGroupRow = {
  shops: Array<{ id: string; name: string }>
  sku_codes: string[]
  pending_reasons: string[]
+  pending_subreasons: string[]
+  configured_channels: string[]
+  locked_item_count: number
  item_count: number
  revenue_fen: number
  first_seen_at: string
@@ -963,14 +1010,32 @@ export async function listPendingChannelGroups(input: {
        array_agg(DISTINCT o.shop_id) AS shop_ids,
        array_agg(DISTINCT oi.sku_code) AS sku_codes,
        array_remove(array_agg(DISTINCT oi.fulfillment_pending_reason), NULL) AS pending_reasons,
+        array_remove(
+          array_agg(
+            DISTINCT COALESCE(
+              oi.fulfillment_pending_detail->>'carrierIssue',
+              oi.fulfillment_pending_detail->>'reason'
+            )
+          ),
+          NULL
+        ) AS pending_subreasons,
+        COALESCE(cfg.channels, ARRAY[]::text[]) AS configured_channels,
+        COUNT(*) FILTER (WHERE oi.fulfillment_locked)::int AS locked_item_count,
        COUNT(*)::int AS item_count,
        COALESCE(SUM(oi.revenue_amount_fen), 0)::bigint AS revenue_fen,
        MIN(oi.created_at) AS first_seen_at,
        MAX(oi.created_at) AS last_seen_at
      FROM order_items oi
      JOIN orders o ON o.id = oi.order_id
+      LEFT JOIN LATERAL (
+        SELECT array_agg(DISTINCT channel.channel ORDER BY channel.channel) AS channels
+        FROM internal_sku_channels channel
+        WHERE channel.enabled = TRUE
+          AND canonicalize_internal_sku(channel.internal_sku) =
+              ${INTERNAL_SKU_SQL}
+      ) cfg ON TRUE
      WHERE ${where}
-      GROUP BY ${INTERNAL_SKU_SQL}
+      GROUP BY ${INTERNAL_SKU_SQL}, cfg.channels
      ORDER BY item_count DESC, internal_sku
      LIMIT $${params.length - 1} OFFSET $${params.length}
    `,
@@ -1034,6 +1099,102 @@ export async function listPendingChannelGroups(input: {
  }
 }

+export type RetryPendingCarrierGroupResult = {
+  scanned: number
+  workOrdersCreated: number
+  affiliateTasksCreated: number
+  assigned: number
+  failed: number
+  remaining: number
+}
+
+/** 运营确认后重试整组载体：允许重开已关闭的历史尝试，并包含运营锁定的订单项。 */
+export async function retryPendingCarrierGroup(input: {
+  internalSku: string
+}): Promise<RetryPendingCarrierGroupResult> {
+  const internalSku = canonicalizeInternalSku(input.internalSku)
+  if (!internalSku || !containsCjk(internalSku)) {
+    throw createHttpError('必须使用包含中文的内部 SKU', {
+      statusCode: 400,
+      errorCode: 'internal_sku_required',
+    })
+  }
+
+  const pending = await query<{ id: number }>(
+    `SELECT oi.id
+     FROM order_items oi
+     WHERE oi.fulfillment_status = 'pending'
+       AND ${INTERNAL_SKU_SQL} = $1
+     ORDER BY oi.id`,
+    [internalSku],
+  )
+  const ids = pending.rows.map((row) => Number(row.id))
+  let workOrdersCreated = 0
+  let affiliateTasksCreated = 0
+  let failed = 0
+  const assignedItemIds: number[] = []
+
+  for (let offset = 0; offset < ids.length; offset += 500) {
+    const batchIds = ids.slice(offset, offset + 500)
+    const carrier = await reconcilePendingCarriers({
+      orderItemIds: batchIds,
+      limit: batchIds.length,
+      includeLocked: true,
+      retryInactiveCarrier: true,
+    })
+    workOrdersCreated += carrier.workOrdersCreated
+    affiliateTasksCreated += carrier.affiliateTasksCreated
+    failed += carrier.failed
+    const applied = await assignPendingOrderItemChannels({
+      orderItemIds: batchIds,
+      batchSize: batchIds.length,
+      maxBatches: 1,
+      afterId: 0,
+      includeLocked: true,
+    })
+    assignedItemIds.push(...applied.assignedItemIds)
+  }
+
+  await query(
+    `UPDATE order_items oi
+     SET fulfillment_config_id = cfg.id,
+         canonical_product_id = COALESCE(oi.canonical_product_id, cfg.canonical_product_id),
+         updated_at = NOW()
+     FROM manual_fulfillment_configs cfg
+     WHERE oi.id = ANY($1::bigint[])
+       AND oi.fulfillment_status = 'assigned'
+       AND oi.fulfillment_channel = 'manual'
+       AND cfg.enabled = TRUE
+       AND canonicalize_internal_sku(cfg.sku_key) =
+           resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code)`,
+    [ids],
+  )
+  if (assignedItemIds.length > 0) {
+    await normalizeProductsForItems(assignedItemIds).catch(() => undefined)
+    const cost = await syncCostSnapshotsForItems(assignedItemIds).catch(() => null)
+    const dates = [
+      ...new Set([...(await orderItemBasisDates(assignedItemIds)), ...(cost?.affectedDates || [])]),
+    ]
+    await refreshFactsForDates(dates).catch(() => undefined)
+  }
+  const final = await query<{ assigned: number; remaining: number }>(
+    `SELECT
+       count(*) FILTER (WHERE fulfillment_status = 'assigned')::int AS assigned,
+       count(*) FILTER (WHERE fulfillment_status = 'pending')::int AS remaining
+     FROM order_items
+     WHERE id = ANY($1::bigint[])`,
+    [ids],
+  )
+  return {
+    scanned: ids.length,
+    workOrdersCreated,
+    affiliateTasksCreated,
+    assigned: Number(final.rows[0]?.assigned || 0),
+    failed,
+    remaining: Number(final.rows[0]?.remaining || 0),
+  }
+}
+
 export type DeletePendingChannelGroupResult = {
  archivedItems: number
  preservedOrders: number
@@ -30,6 +30,10 @@ export const INACTIVE_CARRIER_TASK_STATUSES = [
  'cancelled',
 ] as const

+export function isInactiveCarrierTaskStatus(value: unknown): boolean {
+  return (INACTIVE_CARRIER_TASK_STATUSES as readonly string[]).includes(String(value || ''))
+}
+
 export const FULFILLMENT_SOURCES = [
  'auto_rule',
  'auto_mapping',
@@ -1,6 +1,8 @@
 import { buildOrderClaimUrl } from '../../claim/claim-service.js'
 import { TASK_STATUS } from '../../../domain/task-status.js'
+import { logWarn } from '../../../utils/logger.js'
 import { prepareAffiliateDashTask } from '../affiliate-dash/index.js'
+import { scheduleVoucherConsumeForTask } from '../voucher-consume-queue-service.js'
 import {
  FULFILLMENT_EXECUTOR_KEYS,
  type FulfillmentDeliveryLink,
@@ -32,7 +34,18 @@ async function preparePaidTask(
        })) || task
    }

-    return await prepareAffiliateDashTask(workingTask)
+    const prepared = await prepareAffiliateDashTask(workingTask)
+    const claimToken = String(prepared?.primary_claim_token || prepared?.claim_token || '').trim()
+    if (prepared && prepared.task_status !== TASK_STATUS.MANUAL_REVIEW && claimToken) {
+      await scheduleVoucherConsumeForTask(prepared, 'link_generated').catch((error) => {
+        logWarn('[affiliate-dash-executor]', '领取链接生成后安排自动核销失败', {
+          taskId: prepared.id,
+          platformOrderId: prepared.platform_order_id,
+          error: error instanceof Error ? error.message : String(error),
+        })
+      })
+    }
+    return prepared
  } catch (error) {
    const message = error instanceof Error ? error.message : 'affiliate-dash 订单创建失败'
    const updatedTask = await deps.updateTask(task.id, {
@@ -1,6 +1,6 @@
 import { logWarn } from '../../../utils/logger.js'
 import { TASK_STATUS } from '../../../domain/task-status.js'
-import { autoConsumeKuaishouIndustryVouchersForTask } from '../../platforms/kuaishou-industry/voucher-auto-consume-service.js'
+import { scheduleVoucherConsumeForTask } from '../voucher-consume-queue-service.js'
 import {
  FULFILLMENT_EXECUTOR_KEYS,
  type FulfillmentExecutor,
@@ -32,10 +32,8 @@ async function preparePaidTask(
  })

  const effectiveTask = updatedTask || task
-  await autoConsumeKuaishouIndustryVouchersForTask(effectiveTask, {
-    source: 'manual_dispatch_prepared',
-  }).catch((error) => {
-    logWarn('[manual-executor]', '人工派单准备后自动推进电子凭证核销失败', {
+  await scheduleVoucherConsumeForTask(effectiveTask, 'task_created').catch((error) => {
+    logWarn('[manual-executor]', '人工派单准备后安排自动核销失败', {
      taskId: effectiveTask.id,
      platformOrderId: effectiveTask.platform_order_id,
      error: error instanceof Error ? error.message : String(error),
@@ -27,6 +27,7 @@ test(
      listChannelRules,
      listChannelRulesPage,
      listPendingChannelGroups,
+      retryPendingCarrierGroup,
      switchChannelForOrderItems,
      upsertChannelRule,
    } = await import('./channel-assign-service.js')
@@ -117,15 +118,18 @@ test(
        internal_sku: string | null
        platform_only: string | null
        cjk_extension: string | null
+        case_variant: string | null
      }>(
        `SELECT
           resolve_internal_sku('platform-code', ' 套装－浪漫 天命 ', 'platform-code') AS internal_sku,
           resolve_internal_sku('platform-code', 'Platform Title', 'platform-code') AS platform_only,
-           resolve_internal_sku('', '𠀀', '') AS cjk_extension`,
+           resolve_internal_sku('', '𠀀', '') AS cjk_extension,
+           resolve_internal_sku('', '黑色ROCK长袖T恤', '') AS case_variant`,
      )
      assert.equal(canonicalResolution.rows[0]?.internal_sku, '套装-浪漫天命')
      assert.equal(canonicalResolution.rows[0]?.platform_only, null)
      assert.equal(canonicalResolution.rows[0]?.cjk_extension, '𠀀')
+      assert.equal(canonicalResolution.rows[0]?.case_variant, '黑色rock长袖t恤')

      await query(
        `INSERT INTO internal_sku_channels (internal_sku, channel, priority, is_fallback, enabled, source)
@@ -331,6 +335,12 @@ test(
         )`,
        [retryWorkOrderRuleId, retryWorkOrderSku],
      )
+      await query(
+        `INSERT INTO internal_sku_channels (
+           internal_sku, channel, priority, is_fallback, enabled, config_json, source
+         ) VALUES ($1, 'work_order', 10, FALSE, TRUE, $2::jsonb, 'integration_test')`,
+        [retryWorkOrderSku, JSON.stringify({ workRuleKey: retryWorkOrderRuleKey })],
+      )
      const retriedWorkOrders = await retryPendingWorkOrderRouting({
        internalSku: retryWorkOrderSku,
      })
@@ -353,6 +363,34 @@ test(
      assert.equal(retriedItem.rows[0]?.fulfillment_status, 'assigned')
      assert.equal(retriedItem.rows[0]?.fulfillment_channel, 'work_order')
      assert.equal(retriedItem.rows[0]?.work_rule_key, retryWorkOrderRuleKey)
+      await query(
+        `UPDATE work_orders SET status = 'cancelled', updated_at = NOW()
+         WHERE order_item_id = $1`,
+        [retryWorkOrderItemId],
+      )
+      await query(
+        `UPDATE order_items
+         SET fulfillment_status = 'pending', fulfillment_channel = NULL,
+             fulfillment_source = 'auto_mapping', fulfillment_pending_reason = 'no_carrier',
+             fulfillment_pending_detail = $2::jsonb, updated_at = NOW()
+         WHERE id = $1`,
+        [
+          retryWorkOrderItemId,
+          JSON.stringify({ carrierIssue: 'carrier_task_closed', channel: 'work_order' }),
+        ],
+      )
+      const retriedCarrier = await retryPendingCarrierGroup({
+        internalSku: retryWorkOrderSku,
+      })
+      assert.equal(retriedCarrier.scanned, 1)
+      assert.equal(retriedCarrier.assigned, 1)
+      assert.equal(retriedCarrier.remaining, 0)
+      const activeWorkOrders = await query<{ count: number }>(
+        `SELECT count(*)::int AS count FROM work_orders
+         WHERE order_item_id = $1 AND status <> 'cancelled'`,
+        [retryWorkOrderItemId],
+      )
+      assert.equal(activeWorkOrders.rows[0]?.count, 1)

      const crossShopA = `it-routing-a-${suffix}`
      const crossShopB = `it-routing-b-${suffix}`
@@ -555,6 +593,57 @@ test(
        Number(manualFallbackConfig.rows[0]!.id),
      )

+      const retryTaskItemId = await createPendingOrderItem(autoManualSku)
+      const retryTaskOrder = await query<{
+        order_id: number
+        provider: string
+        platform: string
+        shop_id: string
+        shop_name: string
+        platform_order_id: string
+      }>(
+        `SELECT oi.order_id, o.provider, o.platform, o.shop_id, o.shop_name, o.platform_order_id
+         FROM order_items oi
+         JOIN orders o ON o.id = oi.order_id
+         WHERE oi.id = $1`,
+        [retryTaskItemId],
+      )
+      const retryTaskOrderRow = retryTaskOrder.rows[0]!
+      const insertRetryTask = (taskNo: string, status: string) =>
+        query(
+          `INSERT INTO fulfillment_tasks (
+             order_id, order_item_id, unit_index, task_no, provider, platform,
+             shop_id, shop_name, platform_order_id, profile_id, executor_key,
+             task_status, created_at, updated_at
+           ) VALUES ($1, $2, 1, $3, $4, $5, $6, $7, $8, $9, 'affiliate_dash', $10, NOW(), NOW())`,
+          [
+            retryTaskOrderRow.order_id,
+            retryTaskItemId,
+            taskNo,
+            retryTaskOrderRow.provider,
+            retryTaskOrderRow.platform,
+            retryTaskOrderRow.shop_id,
+            retryTaskOrderRow.shop_name,
+            retryTaskOrderRow.platform_order_id,
+            fallbackProfileId,
+            status,
+          ],
+        )
+      await insertRetryTask(`IT-AFF-CLOSED-${suffix}`, 'closed')
+      await insertRetryTask(`IT-AFF-ACTIVE-${suffix}`, 'paid')
+      await assert.rejects(
+        () => insertRetryTask(`IT-AFF-DUPLICATE-${suffix}`, 'claimed'),
+        /duplicate key value violates unique constraint/,
+      )
+      const retryTaskRows = await query<{ task_status: string }>(
+        `SELECT task_status FROM fulfillment_tasks WHERE order_item_id = $1 ORDER BY id`,
+        [retryTaskItemId],
+      )
+      assert.deepEqual(
+        retryTaskRows.rows.map((row) => row.task_status),
+        ['closed', 'paid'],
+      )
+
      await ensureVoucherRuleForSku(voucherSku)
      const voucherChannel = await query<{ channel: string; source: string }>(
        `SELECT channel, source FROM internal_sku_channels
@@ -52,4 +52,8 @@ test('内部 SKU：没有中文时返回空，平台码和英文码不能作为
 test('内部 SKU 规范化：NFKC + 去全部空白 + 小写', () => {
  assert.equal(canonicalizeInternalSku(' 套装－浪漫天命　'), '套装-浪漫天命')
  assert.equal(canonicalizeInternalSku(' M416-仓鼠 灰灰 '), 'm416-仓鼠灰灰')
+  assert.equal(
+    canonicalizeInternalSku('黑色ROCK长袖T恤'),
+    canonicalizeInternalSku('黑色rock长袖t恤'),
+  )
 })
@@ -37,6 +37,14 @@ export async function retryPendingWorkOrderRouting(
    `SELECT oi.*
     FROM order_items oi
     WHERE oi.fulfillment_status = 'pending'
+       AND EXISTS (
+         SELECT 1
+         FROM internal_sku_channels channel
+         WHERE channel.enabled = TRUE
+           AND channel.channel = 'work_order'
+           AND canonicalize_internal_sku(channel.internal_sku) =
+               resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code)
+       )
       ${skuClause}
     ORDER BY oi.id
     LIMIT $1`,
@@ -71,6 +79,7 @@ export async function retryPendingWorkOrderRouting(
      source: 'admin_pending_retry',
      autoOnly: true,
      skipChannelAssign: true,
+      retryCancelledWorkOrder: true,
    })
    createdWorkOrders += result.createdCount
    for (const skipped of result.skipped) {
@@ -0,0 +1,85 @@
+import assert from 'node:assert/strict'
+import test from 'node:test'
+
+import {
+  createDefaultVoucherConsumePolicy,
+  normalizeVoucherConsumePolicy,
+  resolveVoucherConsumePolicy,
+} from './voucher-consume-policy-service.js'
+
+test('默认核销策略：自动发货 5 秒，工单发布或指派后 10 秒，人工任务 10 秒', () => {
+  const policy = createDefaultVoucherConsumePolicy()
+
+  assert.deepEqual(policy.platforms.kuaishou_industry.channels.affiliate_dash, {
+    enabled: true,
+    mode: 'after_link_generated',
+    delaySeconds: 5,
+  })
+  assert.deepEqual(policy.platforms.douyin_mashida.channels.work_order, {
+    enabled: true,
+    mode: 'after_work_order_published_or_assigned',
+    delaySeconds: 10,
+  })
+  assert.deepEqual(policy.platforms.kuaishou_industry.channels.manual, {
+    enabled: true,
+    mode: 'after_task_created',
+    delaySeconds: 10,
+  })
+})
+
+test('工单默认策略同时匹配发布和指派，且不匹配链接生成', () => {
+  const policy = createDefaultVoucherConsumePolicy()
+
+  assert.equal(
+    resolveVoucherConsumePolicy({
+      provider: 'kuaishou_industry',
+      channel: 'work_order',
+      event: 'work_order_published',
+      policy,
+    }).enabled,
+    true,
+  )
+  assert.equal(
+    resolveVoucherConsumePolicy({
+      provider: 'douyin_mashida',
+      channel: 'work_order',
+      event: 'work_order_assigned',
+      policy,
+    }).enabled,
+    true,
+  )
+  assert.equal(
+    resolveVoucherConsumePolicy({
+      provider: 'douyin_mashida',
+      channel: 'work_order',
+      event: 'link_generated',
+      policy,
+    }).enabled,
+    false,
+  )
+})
+
+test('策略归一化会丢弃非法模式并限制延迟范围', () => {
+  const policy = normalizeVoucherConsumePolicy({
+    enabled: true,
+    platforms: {
+      kuaishou_industry: {
+        enabled: true,
+        channels: {
+          affiliate_dash: {
+            enabled: true,
+            mode: 'after_work_order_assigned',
+            delaySeconds: 99999,
+          },
+        },
+      },
+    },
+  })
+
+  assert.deepEqual(policy.platforms.kuaishou_industry.channels.affiliate_dash, {
+    enabled: true,
+    mode: 'after_link_generated',
+    delaySeconds: 3600,
+  })
+  assert.equal(policy.platforms.douyin_mashida.enabled, true)
+})
@@ -0,0 +1,222 @@
+import { APP_CONFIG_KEYS } from '../../config/app-config-keys.js'
+import type { JsonObject } from '../../types/json.js'
+import { readAppConfigEntry, saveAppConfigEntry } from '../config/app-config-store.js'
+
+export const VOUCHER_CONSUME_PROVIDERS = ['kuaishou_industry', 'douyin_mashida'] as const
+export type VoucherConsumeProvider = (typeof VOUCHER_CONSUME_PROVIDERS)[number]
+
+export const VOUCHER_CONSUME_CHANNELS = [
+  'affiliate_dash',
+  'work_order',
+  'manual',
+  'third_party_voucher',
+] as const
+export type VoucherConsumeChannel = (typeof VOUCHER_CONSUME_CHANNELS)[number]
+
+export const VOUCHER_CONSUME_EVENTS = [
+  'link_generated',
+  'task_created',
+  'work_order_published',
+  'work_order_assigned',
+] as const
+export type VoucherConsumeEvent = (typeof VOUCHER_CONSUME_EVENTS)[number]
+
+export const VOUCHER_CONSUME_MODES = [
+  'manual',
+  'after_link_generated',
+  'after_task_created',
+  'after_work_order_published',
+  'after_work_order_assigned',
+  'after_work_order_published_or_assigned',
+] as const
+export type VoucherConsumeMode = (typeof VOUCHER_CONSUME_MODES)[number]
+
+export type VoucherChannelConsumePolicy = {
+  enabled: boolean
+  mode: VoucherConsumeMode
+  delaySeconds: number
+}
+
+export type VoucherPlatformConsumePolicy = {
+  enabled: boolean
+  channels: Record<VoucherConsumeChannel, VoucherChannelConsumePolicy>
+}
+
+export type VoucherConsumePolicy = {
+  enabled: boolean
+  platforms: Record<VoucherConsumeProvider, VoucherPlatformConsumePolicy>
+}
+
+export function createDefaultVoucherConsumePolicy(): VoucherConsumePolicy {
+  return {
+    enabled: true,
+    platforms: {
+      kuaishou_industry: createDefaultPlatformPolicy(),
+      douyin_mashida: createDefaultPlatformPolicy(),
+    },
+  }
+}
+
+export function getVoucherConsumePolicy(): VoucherConsumePolicy {
+  return readAppConfigEntry({
+    configKey: APP_CONFIG_KEYS.voucherConsumePolicy,
+    fallback: createDefaultVoucherConsumePolicy,
+    normalize: normalizeVoucherConsumePolicy,
+  })
+}
+
+export async function saveVoucherConsumePolicy(rawValue: unknown): Promise<VoucherConsumePolicy> {
+  return saveAppConfigEntry({
+    configKey: APP_CONFIG_KEYS.voucherConsumePolicy,
+    value: rawValue,
+    normalize: normalizeVoucherConsumePolicy,
+  })
+}
+
+export function normalizeVoucherConsumePolicy(rawValue: unknown): VoucherConsumePolicy {
+  const defaults = createDefaultVoucherConsumePolicy()
+  const source = isPlainObject(rawValue) ? rawValue : {}
+  const rawPlatforms = isPlainObject(source.platforms) ? source.platforms : {}
+
+  return {
+    enabled: normalizeBoolean(source.enabled, defaults.enabled),
+    platforms: {
+      kuaishou_industry: normalizePlatformPolicy(
+        rawPlatforms.kuaishou_industry,
+        defaults.platforms.kuaishou_industry,
+      ),
+      douyin_mashida: normalizePlatformPolicy(
+        rawPlatforms.douyin_mashida,
+        defaults.platforms.douyin_mashida,
+      ),
+    },
+  }
+}
+
+export function resolveVoucherConsumePolicy(input: {
+  provider: VoucherConsumeProvider
+  channel: VoucherConsumeChannel
+  event: VoucherConsumeEvent
+  policy?: VoucherConsumePolicy
+}): { enabled: boolean; delaySeconds: number; mode: VoucherConsumeMode } {
+  const policy = input.policy || getVoucherConsumePolicy()
+  const platform = policy.platforms[input.provider]
+  const channel = platform.channels[input.channel]
+
+  if (!policy.enabled || !platform.enabled || !channel.enabled || channel.mode === 'manual') {
+    return { enabled: false, delaySeconds: 0, mode: channel.mode }
+  }
+
+  return {
+    enabled: modeMatchesEvent(channel.mode, input.event),
+    delaySeconds: channel.delaySeconds,
+    mode: channel.mode,
+  }
+}
+
+export function isVoucherConsumeProvider(value: unknown): value is VoucherConsumeProvider {
+  return (VOUCHER_CONSUME_PROVIDERS as readonly string[]).includes(String(value || '').trim())
+}
+
+function createDefaultPlatformPolicy(): VoucherPlatformConsumePolicy {
+  return {
+    enabled: true,
+    channels: {
+      affiliate_dash: {
+        enabled: true,
+        mode: 'after_link_generated',
+        delaySeconds: 5,
+      },
+      work_order: {
+        enabled: true,
+        mode: 'after_work_order_published_or_assigned',
+        delaySeconds: 10,
+      },
+      manual: {
+        enabled: true,
+        mode: 'after_task_created',
+        delaySeconds: 10,
+      },
+      third_party_voucher: {
+        enabled: false,
+        mode: 'manual',
+        delaySeconds: 0,
+      },
+    },
+  }
+}
+
+function normalizePlatformPolicy(
+  rawValue: unknown,
+  fallback: VoucherPlatformConsumePolicy,
+): VoucherPlatformConsumePolicy {
+  const source = isPlainObject(rawValue) ? rawValue : {}
+  const rawChannels = isPlainObject(source.channels) ? source.channels : {}
+
+  return {
+    enabled: normalizeBoolean(source.enabled, fallback.enabled),
+    channels: {
+      affiliate_dash: normalizeChannelPolicy(
+        rawChannels.affiliate_dash,
+        fallback.channels.affiliate_dash,
+        ['manual', 'after_link_generated', 'after_task_created'],
+      ),
+      work_order: normalizeChannelPolicy(rawChannels.work_order, fallback.channels.work_order, [
+        'manual',
+        'after_task_created',
+        'after_work_order_published',
+        'after_work_order_assigned',
+        'after_work_order_published_or_assigned',
+      ]),
+      manual: normalizeChannelPolicy(rawChannels.manual, fallback.channels.manual, [
+        'manual',
+        'after_task_created',
+      ]),
+      third_party_voucher: normalizeChannelPolicy(
+        rawChannels.third_party_voucher,
+        fallback.channels.third_party_voucher,
+        ['manual'],
+      ),
+    },
+  }
+}
+
+function normalizeChannelPolicy(
+  rawValue: unknown,
+  fallback: VoucherChannelConsumePolicy,
+  allowedModes: readonly VoucherConsumeMode[],
+): VoucherChannelConsumePolicy {
+  const source = isPlainObject(rawValue) ? rawValue : {}
+  const mode = String(source.mode || '').trim() as VoucherConsumeMode
+  const fallbackMode = allowedModes.includes(fallback.mode) ? fallback.mode : 'manual'
+
+  return {
+    enabled: normalizeBoolean(source.enabled, fallback.enabled),
+    mode: allowedModes.includes(mode) ? mode : fallbackMode,
+    delaySeconds: normalizeInteger(source.delaySeconds, fallback.delaySeconds, 0, 3600),
+  }
+}
+
+function modeMatchesEvent(mode: VoucherConsumeMode, event: VoucherConsumeEvent): boolean {
+  if (mode === 'manual') return false
+  if (mode === 'after_link_generated') return event === 'link_generated'
+  if (mode === 'after_task_created') return event === 'task_created'
+  if (mode === 'after_work_order_published') return event === 'work_order_published'
+  if (mode === 'after_work_order_assigned') return event === 'work_order_assigned'
+  return event === 'work_order_published' || event === 'work_order_assigned'
+}
+
+function normalizeBoolean(value: unknown, fallback: boolean): boolean {
+  if (value === undefined || value === null || value === '') return fallback
+  return value !== false && String(value) !== 'false'
+}
+
+function normalizeInteger(value: unknown, fallback: number, min: number, max: number): number {
+  const parsed = Number(value)
+  if (!Number.isFinite(parsed)) return fallback
+  return Math.min(max, Math.max(min, Math.trunc(parsed)))
+}
+
+function isPlainObject(value: unknown): value is JsonObject {
+  return Object.prototype.toString.call(value) === '[object Object]'
+}
@@ -0,0 +1,777 @@
+import { query } from '../../db/client.js'
+import { getOrderById } from '../../repositories/order-repo.js'
+import { getOrderItemById, listOrderItemsByOrderId } from '../../repositories/order-item-repo.js'
+import { getTaskById, listTasksByOrderId } from '../../repositories/task-repo.js'
+import { listKuaishouIndustryVouchersByOid } from '../../repositories/kuaishou-industry-voucher-repo.js'
+import {
+  findRedeemCertificateByOrder,
+  listRedeemCertificatesByLocalOrder,
+  type RedeemCertificateRow,
+} from '../../repositories/redeem-certificate-repo.js'
+import type { WorkOrderRow } from '../../repositories/worker-platform/index.js'
+import {
+  DOUYIN_MASHIDA_PROVIDER,
+  isDouyinMashidaProvider,
+} from '../platforms/douyin-mashida/order-mapper.js'
+import {
+  consumeKuaishouIndustryVoucher,
+  isKuaishouIndustryVoucherSendCallbackSuccess,
+  normalizeVoucherStatus,
+} from '../platforms/kuaishou-industry/voucher-service.js'
+import { attachKuaishouIndustryVoucherToTask } from '../platforms/kuaishou-industry/voucher-binding-service.js'
+import { verifyCertificate } from '../redeem/certificate-service.js'
+import { logIntegration, logWarn } from '../../utils/logger.js'
+import {
+  getVoucherConsumePolicy,
+  resolveVoucherConsumePolicy,
+  type VoucherConsumeChannel,
+  type VoucherConsumeEvent,
+  type VoucherConsumeProvider,
+} from './voucher-consume-policy-service.js'
+import type {
+  KuaishouIndustryVoucherRow,
+  OrderItemRow,
+  OrderRow,
+  TaskRow,
+} from '../../types/repository/rows.js'
+import { OPEN_91_PLATFORM, OPEN_91_PROVIDER } from '../open-91/config.js'
+import { FULFILLMENT_EXECUTOR_KEYS } from './executors/types.js'
+
+export type VoucherConsumeScheduleResult = {
+  ok: boolean
+  voucherCount: number
+  consumedCount: number
+  alreadyConsumedCount: number
+  failedCount: number
+  voucherCodes: string[]
+  errorMessage: string
+  queuedCount: number
+  jobIds: number[]
+  dueAt: string
+  status: 'queued' | 'already_scheduled' | 'skipped' | 'failed'
+}
+
+export type VoucherConsumeJobRow = {
+  id: number
+  provider: VoucherConsumeProvider
+  channel: VoucherConsumeChannel
+  trigger_event: VoucherConsumeEvent
+  target_key: string
+  platform_order_id: string
+  order_id: number | null
+  order_item_id: number | null
+  task_id: number | null
+  work_order_id: number | null
+  due_at: string
+  status: 'pending' | 'processing' | 'succeeded' | 'failed' | 'cancelled'
+  attempt_count: number
+  max_attempts: number
+  locked_until: string | null
+  last_error: string
+  result_json: string | Record<string, unknown>
+  created_at: string
+  updated_at: string
+  finished_at: string | null
+}
+
+type VoucherConsumeScheduleTarget = {
+  provider: VoucherConsumeProvider
+  channel: VoucherConsumeChannel
+  event: VoucherConsumeEvent
+  targetKey: string
+  platformOrderId: string
+  orderId: number | null
+  orderItemId: number | null
+  taskId: number | null
+  workOrderId: number | null
+  voucherCodes: string[]
+  voucherCount: number
+}
+
+const RETRY_DELAYS_SECONDS = [10, 30, 60, 300, 600, 900, 1800]
+const PROCESSING_LOCK_SECONDS = 120
+
+export async function scheduleVoucherConsumeForTask(
+  task: TaskRow,
+  event: VoucherConsumeEvent,
+): Promise<VoucherConsumeScheduleResult> {
+  const order = await getOrderById(task.order_id)
+  if (!order) {
+    return buildScheduleResult('skipped', { errorMessage: '订单不存在，无法安排核销' })
+  }
+
+  const provider = resolveVoucherProvider(order)
+  if (!provider) {
+    return buildScheduleResult('skipped')
+  }
+
+  const orderItem = await getOrderItemById(task.order_item_id)
+  const channel = resolveTaskVoucherChannel(task, orderItem)
+  if (!channel) {
+    return buildScheduleResult('skipped')
+  }
+
+  return scheduleVoucherConsumeTarget({
+    provider,
+    channel,
+    event,
+    targetKey: buildVoucherConsumeTargetKey(task.order_item_id, `task:${task.id}`),
+    platformOrderId: String(order.platform_order_id || task.platform_order_id || '').trim(),
+    orderId: Number(order.id),
+    orderItemId: Number(task.order_item_id || 0) || null,
+    taskId: Number(task.id),
+    workOrderId: null,
+  })
+}
+
+export async function scheduleVoucherConsumeForWorkOrder(
+  workOrder: WorkOrderRow,
+  event: Extract<VoucherConsumeEvent, 'work_order_published' | 'work_order_assigned'>,
+): Promise<VoucherConsumeScheduleResult> {
+  const order = workOrder.order_id ? await getOrderById(workOrder.order_id) : null
+  if (!order) {
+    return buildScheduleResult('skipped', { errorMessage: '订单不存在，无法安排核销' })
+  }
+
+  const provider = resolveVoucherProvider(order)
+  if (!provider) {
+    return buildScheduleResult('skipped')
+  }
+
+  return scheduleVoucherConsumeTarget({
+    provider,
+    channel: 'work_order',
+    event,
+    targetKey: buildVoucherConsumeTargetKey(workOrder.order_item_id, `work_order:${workOrder.id}`),
+    platformOrderId: String(order.platform_order_id || workOrder.platform_order_id || '').trim(),
+    orderId: Number(order.id),
+    orderItemId: Number(workOrder.order_item_id || 0) || null,
+    // work_orders.task_id is vestigial (migration 091); never use it as a join key.
+    taskId: null,
+    workOrderId: Number(workOrder.id),
+  })
+}
+
+async function scheduleVoucherConsumeTarget(
+  target: Omit<VoucherConsumeScheduleTarget, 'voucherCodes' | 'voucherCount'>,
+): Promise<VoucherConsumeScheduleResult> {
+  const policy = resolveVoucherConsumePolicy({
+    provider: target.provider,
+    channel: target.channel,
+    event: target.event,
+    policy: getVoucherConsumePolicy(),
+  })
+  if (!policy.enabled) {
+    return buildScheduleResult('skipped')
+  }
+
+  const candidates = await listVoucherCandidatesForTarget(target)
+  const dueAt = new Date(Date.now() + policy.delaySeconds * 1000).toISOString()
+  const inserted = await query<VoucherConsumeJobRow>(
+    `
+      INSERT INTO voucher_consume_jobs (
+        provider,
+        channel,
+        trigger_event,
+        target_key,
+        platform_order_id,
+        order_id,
+        order_item_id,
+        task_id,
+        work_order_id,
+        due_at,
+        status,
+        attempt_count,
+        max_attempts,
+        updated_at
+      ) VALUES (
+        $1, $2, $3, $4, $5, $6, $7, $8, $9, $10,
+        'pending', 0, 8, NOW()
+      )
+      ON CONFLICT (provider, target_key)
+        WHERE status IN ('pending', 'processing', 'succeeded')
+      DO NOTHING
+      RETURNING *
+    `,
+    [
+      target.provider,
+      target.channel,
+      target.event,
+      target.targetKey,
+      target.platformOrderId,
+      target.orderId,
+      target.orderItemId,
+      target.taskId,
+      target.workOrderId,
+      dueAt,
+    ],
+  )
+
+  const job = inserted.rows[0]
+  if (!job) {
+    return buildScheduleResult('already_scheduled', {
+      voucherCount: candidates.voucherCount,
+      voucherCodes: candidates.voucherCodes,
+    })
+  }
+
+  logIntegration('[voucher-consume/queue]', '核销任务已安排', {
+    jobId: job.id,
+    provider: target.provider,
+    channel: target.channel,
+    event: target.event,
+    dueAt,
+    orderId: target.orderId,
+    orderItemId: target.orderItemId,
+    workOrderId: target.workOrderId,
+    voucherCount: candidates.voucherCount,
+  })
+
+  return buildScheduleResult('queued', {
+    queuedCount: 1,
+    jobIds: [Number(job.id)],
+    dueAt,
+    voucherCount: candidates.voucherCount,
+    voucherCodes: candidates.voucherCodes,
+  })
+}
+
+export async function runVoucherConsumeDispatchJob(
+  jobConfig: Record<string, unknown> = {},
+): Promise<Record<string, unknown>> {
+  const batchSize = clampInteger(jobConfig.batchSize, 20, 1, 200)
+  const jobs = await claimDueVoucherConsumeJobs(batchSize)
+  let okCount = 0
+  let failedCount = 0
+  let retryCount = 0
+  let cancelledCount = 0
+
+  for (const job of jobs) {
+    try {
+      const policy = resolveVoucherConsumePolicy({
+        provider: job.provider,
+        channel: job.channel,
+        event: job.trigger_event,
+        policy: getVoucherConsumePolicy(),
+      })
+      if (!policy.enabled) {
+        await finishVoucherConsumeJob(job.id, 'cancelled', {
+          reason: 'policy_disabled_or_changed',
+        })
+        cancelledCount += 1
+        continue
+      }
+
+      const result = await executeVoucherConsumeJob(job)
+      await finishVoucherConsumeJob(job.id, 'succeeded', result)
+      okCount += 1
+    } catch (error) {
+      const retry = Number(job.attempt_count || 0) < Number(job.max_attempts || 8)
+      await failVoucherConsumeJob(job, error, retry)
+      if (retry) {
+        retryCount += 1
+      } else {
+        failedCount += 1
+      }
+      logWarn('[voucher-consume/dispatch]', '核销任务执行失败', {
+        jobId: job.id,
+        provider: job.provider,
+        channel: job.channel,
+        event: job.trigger_event,
+        attemptCount: job.attempt_count,
+        willRetry: retry,
+        error: error instanceof Error ? error.message : String(error),
+      })
+    }
+  }
+
+  return {
+    status: failedCount > 0 ? 'warning' : 'ok',
+    message:
+      `核销任务：领取 ${jobs.length}，成功 ${okCount}，重试 ${retryCount}，` +
+      `失败 ${failedCount}，取消 ${cancelledCount}`,
+    checkedCount: jobs.length,
+    okCount,
+    failedCount,
+    retryCount,
+    cancelledCount,
+  }
+}
+
+export async function listVoucherConsumeJobs(
+  input: {
+    status?: string
+    page?: number
+    pageSize?: number
+  } = {},
+) {
+  const page = clampInteger(input.page, 1, 1, Number.MAX_SAFE_INTEGER)
+  const pageSize = clampInteger(input.pageSize, 20, 1, 100)
+  const status = String(input.status || '').trim()
+  const params: unknown[] = []
+  const where: string[] = []
+
+  if (status) {
+    params.push(status)
+    where.push(`status = $${params.length}`)
+  }
+
+  const whereSql = where.length > 0 ? `WHERE ${where.join(' AND ')}` : ''
+  const countResult = await query<{ total: string | number }>(
+    `SELECT COUNT(*) AS total FROM voucher_consume_jobs ${whereSql}`,
+    params,
+  )
+  const listParams = [...params, pageSize, (page - 1) * pageSize]
+  const result = await query<VoucherConsumeJobRow>(
+    `
+      SELECT *
+      FROM voucher_consume_jobs
+      ${whereSql}
+      ORDER BY updated_at DESC, id DESC
+      LIMIT $${params.length + 1}
+      OFFSET $${params.length + 2}
+    `,
+    listParams,
+  )
+
+  return {
+    items: result.rows,
+    total: Number(countResult.rows[0]?.total || 0) || 0,
+    page,
+    pageSize,
+  }
+}
+
+export async function retryVoucherConsumeJob(jobId: unknown): Promise<VoucherConsumeJobRow | null> {
+  const id = Number(jobId)
+  if (!Number.isInteger(id) || id <= 0) {
+    return null
+  }
+
+  const result = await query<VoucherConsumeJobRow>(
+    `
+      UPDATE voucher_consume_jobs job
+      SET
+        status = 'pending',
+        attempt_count = 0,
+        locked_until = NULL,
+        last_error = '',
+        result_json = '{}'::jsonb,
+        due_at = NOW(),
+        updated_at = NOW(),
+        finished_at = NULL
+      WHERE job.id = $1
+        AND job.status IN ('failed', 'cancelled')
+        AND NOT EXISTS (
+          SELECT 1
+          FROM voucher_consume_jobs active
+          WHERE active.provider = job.provider
+            AND active.target_key = job.target_key
+            AND active.id <> job.id
+            AND active.status IN ('pending', 'processing', 'succeeded')
+        )
+      RETURNING job.*
+    `,
+    [id],
+  )
+  return result.rows[0] || null
+}
+
+async function executeVoucherConsumeJob(job: VoucherConsumeJobRow) {
+  if (job.provider === 'kuaishou_industry') {
+    return executeKuaishouConsumeJob(job)
+  }
+  if (job.provider === DOUYIN_MASHIDA_PROVIDER) {
+    return executeDouyinConsumeJob(job)
+  }
+  throw new Error(`不支持的核销平台：${job.provider}`)
+}
+
+async function executeKuaishouConsumeJob(job: VoucherConsumeJobRow) {
+  const vouchers = await resolveKuaishouVouchersForJob(job)
+  if (vouchers.length === 0) {
+    throw new Error('未找到待核销的快手电子凭证')
+  }
+
+  const task = job.task_id ? await getTaskById(job.task_id) : null
+  let consumedCount = 0
+  let alreadyConsumedCount = 0
+  let skippedCount = 0
+
+  for (const voucher of vouchers) {
+    const status = normalizeVoucherStatus(voucher.status)
+    if (status === 'CONSUMED') {
+      alreadyConsumedCount += 1
+      continue
+    }
+    if (status === 'DESTROYED') {
+      skippedCount += 1
+      continue
+    }
+    if (!isKuaishouIndustryVoucherSendCallbackSuccess(voucher)) {
+      throw new Error(`电子凭证 ${voucher.voucher_code} 尚未完成发码回调`)
+    }
+
+    const result = await consumeKuaishouIndustryVoucher(voucher, {
+      source: `delayed_${job.trigger_event}`,
+      consumeTime: Date.now(),
+      ...(task ? { task } : {}),
+    })
+    if (!result.ok || !result.voucher) {
+      throw new Error(result.errorMessage || `电子凭证 ${voucher.voucher_code} 核销失败`)
+    }
+
+    consumedCount += 1
+    if (task) {
+      await attachKuaishouIndustryVoucherToTask(task, result.voucher, {
+        source: `delayed_${job.trigger_event}`,
+      })
+    }
+  }
+
+  return {
+    provider: job.provider,
+    voucherCount: vouchers.length,
+    consumedCount,
+    alreadyConsumedCount,
+    skippedCount,
+  }
+}
+
+async function executeDouyinConsumeJob(job: VoucherConsumeJobRow) {
+  const certificates = await resolveDouyinCertificatesForJob(job)
+  if (certificates.length === 0) {
+    throw new Error('未找到待核销的抖音券')
+  }
+
+  let consumedCount = 0
+  let alreadyConsumedCount = 0
+  let skippedCount = 0
+
+  for (const certificate of certificates) {
+    if (certificate.status === 'verified') {
+      alreadyConsumedCount += 1
+      continue
+    }
+    if (certificate.status === 'invalid') {
+      skippedCount += 1
+      continue
+    }
+
+    await verifyCertificate({ certNo: certificate.cert_no })
+    consumedCount += 1
+  }
+
+  return {
+    provider: job.provider,
+    voucherCount: certificates.length,
+    consumedCount,
+    alreadyConsumedCount,
+    skippedCount,
+  }
+}
+
+async function resolveKuaishouVouchersForJob(
+  job: VoucherConsumeJobRow,
+): Promise<KuaishouIndustryVoucherRow[]> {
+  const oid = String(job.platform_order_id || '').trim()
+  if (!oid) return []
+  const vouchers = await listKuaishouIndustryVouchersByOid(oid)
+  if (vouchers.length === 0) return []
+
+  if (job.task_id) {
+    const task = await getTaskById(job.task_id)
+    if (task) {
+      return vouchers.filter(
+        (voucher) =>
+          Number(voucher.task_id || 0) === Number(task.id) ||
+          Number(voucher.unit_index || 0) === Number(task.unit_index || 0),
+      )
+    }
+  }
+
+  const orderId = Number(job.order_id || 0)
+  const orderItemId = Number(job.order_item_id || 0)
+  if (!orderId) return vouchers
+
+  const orderItems = await listOrderItemsByOrderId(orderId)
+  if (orderItems.length <= 1 || !orderItemId) {
+    return vouchers
+  }
+
+  const tasks = await listTasksByOrderId(orderId)
+  const itemTaskIds = new Set(
+    tasks
+      .filter((task) => Number(task.order_item_id) === orderItemId)
+      .map((task) => Number(task.id)),
+  )
+  const byTask = vouchers.filter((voucher) => itemTaskIds.has(Number(voucher.task_id || 0)))
+  if (byTask.length > 0) {
+    return byTask
+  }
+
+  const range = resolveOrderItemUnitRange(orderItems, orderItemId)
+  if (!range) return []
+  return vouchers.filter((voucher) => {
+    const unitIndex = Number(voucher.unit_index || 0)
+    return unitIndex >= range.start && unitIndex <= range.end
+  })
+}
+
+async function resolveDouyinCertificatesForJob(
+  job: VoucherConsumeJobRow,
+): Promise<RedeemCertificateRow[]> {
+  const orderId = Number(job.order_id || 0)
+  const orderItemId = Number(job.order_item_id || 0)
+  const certificates = orderId ? await listRedeemCertificatesByLocalOrder(orderId) : []
+  const filtered = orderItemId
+    ? certificates.filter(
+        (certificate) =>
+          !certificate.order_item_id || Number(certificate.order_item_id) === orderItemId,
+      )
+    : certificates
+  if (filtered.length > 0) {
+    return filtered
+  }
+
+  const receiptOrderId = String(job.platform_order_id || '').trim()
+  if (!receiptOrderId) return []
+  const fallback = await findRedeemCertificateByOrder({
+    provider: DOUYIN_MASHIDA_PROVIDER,
+    orderId: receiptOrderId,
+  })
+  return fallback ? [fallback] : []
+}
+
+async function listVoucherCandidatesForTarget(
+  target: Pick<
+    VoucherConsumeScheduleTarget,
+    'provider' | 'platformOrderId' | 'orderId' | 'orderItemId' | 'taskId'
+  >,
+): Promise<{ voucherCount: number; voucherCodes: string[] }> {
+  if (target.provider === 'kuaishou_industry') {
+    const vouchers = target.platformOrderId
+      ? await listKuaishouIndustryVouchersByOid(target.platformOrderId)
+      : []
+    return {
+      voucherCount: vouchers.length,
+      voucherCodes: vouchers.map((voucher) => String(voucher.voucher_code || '')).filter(Boolean),
+    }
+  }
+
+  const certificates = await resolveDouyinCertificatesForJob({
+    ...buildSyntheticJob(target),
+  })
+  return {
+    voucherCount: certificates.length,
+    voucherCodes: certificates.map((certificate) => certificate.cert_no),
+  }
+}
+
+function buildSyntheticJob(
+  target: Pick<
+    VoucherConsumeScheduleTarget,
+    'provider' | 'platformOrderId' | 'orderId' | 'orderItemId' | 'taskId'
+  >,
+): VoucherConsumeJobRow {
+  return {
+    id: 0,
+    provider: target.provider,
+    channel: 'affiliate_dash',
+    trigger_event: 'link_generated',
+    target_key: '',
+    platform_order_id: target.platformOrderId,
+    order_id: target.orderId,
+    order_item_id: target.orderItemId,
+    task_id: target.taskId,
+    work_order_id: null,
+    due_at: '',
+    status: 'pending',
+    attempt_count: 0,
+    max_attempts: 8,
+    locked_until: null,
+    last_error: '',
+    result_json: {},
+    created_at: '',
+    updated_at: '',
+    finished_at: null,
+  }
+}
+
+async function claimDueVoucherConsumeJobs(limit: number): Promise<VoucherConsumeJobRow[]> {
+  const result = await query<VoucherConsumeJobRow>(
+    `
+      WITH due AS (
+        SELECT id
+        FROM voucher_consume_jobs
+        WHERE attempt_count < max_attempts
+          AND (
+            (status = 'pending' AND due_at <= NOW())
+            OR (status = 'processing' AND locked_until < NOW())
+          )
+        ORDER BY due_at ASC, id ASC
+        LIMIT $1
+        FOR UPDATE SKIP LOCKED
+      )
+      UPDATE voucher_consume_jobs job
+      SET
+        status = 'processing',
+        attempt_count = job.attempt_count + 1,
+        locked_until = NOW() + ($2 * INTERVAL '1 second'),
+        updated_at = NOW()
+      FROM due
+      WHERE job.id = due.id
+      RETURNING job.*
+    `,
+    [limit, PROCESSING_LOCK_SECONDS],
+  )
+  return result.rows
+}
+
+async function finishVoucherConsumeJob(
+  jobId: number,
+  status: 'succeeded' | 'cancelled',
+  result: Record<string, unknown>,
+) {
+  await query(
+    `
+      UPDATE voucher_consume_jobs
+      SET
+        status = $1,
+        locked_until = NULL,
+        last_error = '',
+        result_json = $2::jsonb,
+        updated_at = NOW(),
+        finished_at = NOW()
+      WHERE id = $3
+    `,
+    [status, JSON.stringify(result), jobId],
+  )
+}
+
+async function failVoucherConsumeJob(job: VoucherConsumeJobRow, error: unknown, retry: boolean) {
+  const message = error instanceof Error ? error.message : String(error)
+  if (!retry) {
+    await query(
+      `
+        UPDATE voucher_consume_jobs
+        SET
+          status = 'failed',
+          locked_until = NULL,
+          last_error = $1,
+          result_json = '{}'::jsonb,
+          updated_at = NOW(),
+          finished_at = NOW()
+        WHERE id = $2
+      `,
+      [message, job.id],
+    )
+    return
+  }
+
+  const attemptIndex = Math.max(0, Number(job.attempt_count || 1) - 1)
+  const retryDelaySeconds =
+    RETRY_DELAYS_SECONDS[Math.min(attemptIndex, RETRY_DELAYS_SECONDS.length - 1)] || 60
+  await query(
+    `
+      UPDATE voucher_consume_jobs
+      SET
+        status = 'pending',
+        due_at = NOW() + ($1 * INTERVAL '1 second'),
+        locked_until = NULL,
+        last_error = $2,
+        updated_at = NOW()
+      WHERE id = $3
+    `,
+    [retryDelaySeconds, message, job.id],
+  )
+}
+
+function resolveVoucherProvider(order: OrderRow): VoucherConsumeProvider | null {
+  if (isDouyinMashidaProvider(order.provider)) {
+    return 'douyin_mashida'
+  }
+  const provider = String(order.provider || '').trim()
+  const platform = String(order.platform || '').trim()
+  if (
+    [OPEN_91_PROVIDER, 'open_91', '91'].includes(provider) ||
+    [OPEN_91_PLATFORM, 'kuaishou'].includes(platform)
+  ) {
+    return 'kuaishou_industry'
+  }
+  return null
+}
+
+function resolveTaskVoucherChannel(
+  task: Pick<TaskRow, 'executor_key'>,
+  orderItem: OrderItemRow | null,
+): VoucherConsumeChannel | null {
+  const executorKey = String(task.executor_key || '').trim()
+  if (executorKey === FULFILLMENT_EXECUTOR_KEYS.AFFILIATE_DASH) return 'affiliate_dash'
+  if (
+    executorKey === FULFILLMENT_EXECUTOR_KEYS.MANUAL_DISPATCH ||
+    executorKey === FULFILLMENT_EXECUTOR_KEYS.KUAISHOU_CLOUD ||
+    executorKey === FULFILLMENT_EXECUTOR_KEYS.KUAISHOU_FEIFEI ||
+    executorKey === FULFILLMENT_EXECUTOR_KEYS.KUAISHOU_INDUSTRY
+  ) {
+    return 'manual'
+  }
+
+  const channel = String(orderItem?.fulfillment_channel || '').trim()
+  if (
+    channel === 'affiliate_dash' ||
+    channel === 'work_order' ||
+    channel === 'manual' ||
+    channel === 'third_party_voucher'
+  ) {
+    return channel
+  }
+  return null
+}
+
+function buildVoucherConsumeTargetKey(orderItemId: unknown, fallback: string): string {
+  const normalizedItemId = Number(orderItemId || 0)
+  return normalizedItemId > 0 ? `order_item:${normalizedItemId}` : fallback
+}
+
+function resolveOrderItemUnitRange(
+  orderItems: OrderItemRow[],
+  orderItemId: number,
+): { start: number; end: number } | null {
+  let cursor = 0
+  for (const item of orderItems) {
+    const quantity = Math.max(1, Number(item.quantity || 1))
+    const start = cursor + 1
+    const end = cursor + quantity
+    if (Number(item.id) === orderItemId) {
+      return { start, end }
+    }
+    cursor = end
+  }
+  return null
+}
+
+function buildScheduleResult(
+  status: VoucherConsumeScheduleResult['status'],
+  input: Partial<VoucherConsumeScheduleResult> = {},
+): VoucherConsumeScheduleResult {
+  return {
+    ok: status !== 'failed',
+    voucherCount: Number(input.voucherCount || 0),
+    consumedCount: 0,
+    alreadyConsumedCount: 0,
+    failedCount: 0,
+    voucherCodes: input.voucherCodes || [],
+    errorMessage: String(input.errorMessage || ''),
+    queuedCount: Number(input.queuedCount || 0),
+    jobIds: input.jobIds || [],
+    dueAt: String(input.dueAt || ''),
+    status,
+  }
+}
+
+function clampInteger(value: unknown, fallback: number, min: number, max: number): number {
+  const parsed = Number(value)
+  if (!Number.isInteger(parsed)) return fallback
+  return Math.min(max, Math.max(min, parsed))
+}
@@ -1,14 +1,10 @@
 /**
- * 91 卡券交付前核销：
- * 在查询接口即将返回 orderStatus=20 + cards 前，同步核销行业电子凭证，
- * 把「核销时间」提前到链接交付时刻，降低「拿链即退」误退窗口。
- *
- * 语义：核销 = 领取链接已交付（售出闭环），≠ 游戏内道具已到账。
+ * 91 卡券交付时安排延迟核销。
 *
+ * 查询接口仍按任务 ready 状态返回 cards，但平台核销不再同步阻塞交付；
+ * 这里投递 link_generated 事件，后续由持久化定时任务按配置延迟执行。
 */
-import { attachKuaishouIndustryVoucherToTask } from '../platforms/kuaishou-industry/voucher-binding-service.js'
-import { consumeKuaishouIndustryVoucher } from '../platforms/kuaishou-industry/voucher-service.js'
-import { listKuaishouIndustryVouchersByOid } from '../../repositories/kuaishou-industry-voucher-repo.js'
+import { scheduleVoucherConsumeForTask } from '../fulfillment/voucher-consume-queue-service.js'
 import { logIntegration, logWarn } from '../../utils/logger.js'
 import type { KuaishouIndustryVoucherRow, OrderRow, TaskRow } from '../../types/repository/rows.js'

@@ -55,18 +51,7 @@ export function resolveIndustryVoucherForTask(
  return null
 }

-function isVoucherConsumed(voucher: Pick<KuaishouIndustryVoucherRow, 'status'>): boolean {
-  return (
-    String(voucher.status || '')
-      .trim()
-      .toUpperCase() === 'CONSUMED'
-  )
-}
-
-/**
- * 对即将交付给 91 的 ready 任务同步核销行业凭证（幂等）。
- * 失败时调用方应返回 orderStatus=10，避免「未核销却已发链」。
- */
+/** 对即将交付给 91 的 ready 任务投递延迟核销任务。 */
 export async function consumeIndustryVouchersBeforeOpen91Deliver(options: {
  order: OrderRow
  tasks: TaskRow[]
@@ -95,59 +80,29 @@ export async function consumeIndustryVouchersBeforeOpen91Deliver(options: {
  }

  const oid = String(options.order.platform_order_id || '').trim()
-  const vouchers = oid ? await listKuaishouIndustryVouchersByOid(oid) : []
-  const consumedTaskIds: number[] = []
  const alreadyConsumedTaskIds: number[] = []
  const failed: Array<{ taskId: number; errorMessage: string }> = []

  for (const task of readyTasks) {
    const taskId = Number(task.id || 0)
-    const voucher = resolveIndustryVoucherForTask(task, vouchers)
-
-    if (!voucher) {
+    try {
+      await scheduleVoucherConsumeForTask(task, 'link_generated')
+    } catch (error) {
      failed.push({
        taskId,
-        errorMessage: '交付前核销失败：未找到对应电子凭证',
+        errorMessage: error instanceof Error ? error.message : '安排自动核销失败',
      })
-      continue
-    }
-
-    const alreadyConsumed = isVoucherConsumed(voucher)
-    const result = await consumeKuaishouIndustryVoucher(voucher, {
-      source,
-      task,
-      token: String(voucher.token || '').trim(),
-      consumeTime: Date.now(),
-    })
-
-    if (!result.ok || !result.voucher) {
-      failed.push({
-        taskId,
-        errorMessage: result.errorMessage || '电子凭证核销回调失败',
-      })
-      continue
-    }
-
-    await attachKuaishouIndustryVoucherToTask(task, result.voucher, {
-      source,
-    })
-
-    if (alreadyConsumed) {
-      alreadyConsumedTaskIds.push(taskId)
-    } else {
-      consumedTaskIds.push(taskId)
    }
  }

  if (failed.length > 0) {
-    const errorMessage = failed[0]?.errorMessage || '电子凭证核销失败，暂缓交付领取链接'
-    logWarn('[open-91/query]', '91交付前核销失败，暂不返回 cards', {
+    const errorMessage = failed[0]?.errorMessage || '安排自动核销失败'
+    logWarn('[open-91/query]', '91交付时安排自动核销失败', {
      requestId: options.requestId || '',
      orderId: options.order.id,
      orderNo: oid,
      failedTaskIds: failed.map((item) => item.taskId),
      failed,
-      consumedTaskIds,
      alreadyConsumedTaskIds,
      errorMessage,
    })
@@ -156,24 +111,24 @@ export async function consumeIndustryVouchersBeforeOpen91Deliver(options: {
      ok: false,
      errorMessage,
      failedTaskIds: failed.map((item) => item.taskId),
-      consumedTaskIds,
+      consumedTaskIds: [],
      alreadyConsumedTaskIds,
    }
  }

-  logIntegration('[open-91/query]', '91交付前核销完成', {
+  logIntegration('[open-91/query]', '91交付时核销任务已投递', {
    requestId: options.requestId || '',
    orderId: options.order.id,
    orderNo: oid,
    readyTaskCount: readyTasks.length,
-    consumedTaskIds,
+    taskCount: readyTasks.length,
    alreadyConsumedTaskIds,
    source,
  })

  return {
    ok: true,
-    consumedTaskIds,
+    consumedTaskIds: [],
    alreadyConsumedTaskIds,
  }
 }
@@ -197,6 +197,88 @@ test('syncDeliveryTasksForOrder 已有部分任务时按订单项补建缺失任
  assert.deepEqual(taskEvents, [{ taskId: 202, eventType: 'fulfillment_task_planned' }])
 })

+test('syncDeliveryTasksForOrder 关闭任务不阻塞新的履约尝试', async () => {
+  const closedTask = stubCreatedTask(205, {
+    orderId: 5,
+    orderItemId: 25,
+    unitIndex: 1,
+    executorKey: 'affiliate_dash',
+    taskStatus: 'closed',
+  })
+  const createdInputs: CreatedTaskInput[] = []
+
+  await syncDeliveryTasksForOrderWithDeps(
+    stubOrder({ id: 5, platform_order_id: 'retry-closed-order' }),
+    [
+      stubOrderItem({
+        id: 25,
+        order_id: 5,
+        sku_code: 'SKU-retry',
+        item_snapshot_json: buildAffiliateSnapshot('AFF-RETRY', '商品 RETRY'),
+      }),
+    ],
+    {
+      loadActiveCarrierEvidence: async () => new Map(),
+      listTasksByOrderId: async () => [closedTask],
+      getFulfillmentProfileByKey: async () => buildAffiliateProfile(),
+      createTask: async (input) => {
+        createdInputs.push(input as CreatedTaskInput)
+        return stubCreatedTask(206, input as CreatedTaskInput)
+      },
+      createTaskEvent: async () => null,
+      logInfo: () => null,
+      logWarn: () => null,
+      allowInactiveTaskRetry: true,
+      nowIso: () => '2026-05-29T09:05:46.658Z',
+      randomId: () => 'DT-retry-closed',
+    },
+  )
+
+  assert.deepEqual(
+    createdInputs.map((input) => [input.orderItemId, input.unitIndex]),
+    [[25, 1]],
+  )
+})
+
+test('syncDeliveryTasksForOrder 默认不会因历史关闭任务自动补建', async () => {
+  const closedTask = stubCreatedTask(208, {
+    orderId: 6,
+    orderItemId: 26,
+    unitIndex: 1,
+    executorKey: 'affiliate_dash',
+    taskStatus: 'closed',
+  })
+  const createdInputs: CreatedTaskInput[] = []
+
+  await syncDeliveryTasksForOrderWithDeps(
+    stubOrder({ id: 6, platform_order_id: 'closed-order-no-retry' }),
+    [
+      stubOrderItem({
+        id: 26,
+        order_id: 6,
+        sku_code: 'SKU-closed-no-retry',
+        item_snapshot_json: buildAffiliateSnapshot('AFF-CLOSED', '商品 CLOSED'),
+      }),
+    ],
+    {
+      loadActiveCarrierEvidence: async () => new Map(),
+      listTasksByOrderId: async () => [closedTask],
+      getFulfillmentProfileByKey: async () => buildAffiliateProfile(),
+      createTask: async (input) => {
+        createdInputs.push(input as CreatedTaskInput)
+        return stubCreatedTask(209, input as CreatedTaskInput)
+      },
+      createTaskEvent: async () => null,
+      logInfo: () => null,
+      logWarn: () => null,
+      nowIso: () => '2026-05-29T09:05:46.658Z',
+      randomId: () => 'DT-closed-no-retry',
+    },
+  )
+
+  assert.equal(createdInputs.length, 0)
+})
+
 test('syncDeliveryTasksForOrder 数量增加时只补建缺失单元', async () => {
  const existingTask = stubCreatedTask(301, {
    orderId: 4,
@@ -22,7 +22,10 @@ import {
  findConflictingCarrier,
  loadActiveCarrierEvidence,
 } from '../fulfillment/carrier-evidence-service.js'
-import type { FulfillmentChannel } from '../fulfillment/channel-types.js'
+import {
+  isInactiveCarrierTaskStatus,
+  type FulfillmentChannel,
+} from '../fulfillment/channel-types.js'

 type ClaimTokenLike = {
  token: string
@@ -52,13 +55,15 @@ type DeliveryTaskDeps = {
  createTaskEvent?: typeof createTaskEvent
  logInfo?: typeof logInfo
  logWarn?: typeof logWarn
+  allowInactiveTaskRetry?: boolean
 }

 export async function syncDeliveryTasksForOrder(
  order: OrderRow,
  orderItems: OrderItemRow[],
+  options: { allowInactiveTaskRetry?: boolean } = {},
 ): Promise<TaskRow[]> {
-  return syncDeliveryTasksForOrderWithDeps(order, orderItems)
+  return syncDeliveryTasksForOrderWithDeps(order, orderItems, options)
 }

 export async function syncDeliveryTasksForOrderWithDeps(
@@ -79,6 +84,7 @@ export async function syncDeliveryTasksForOrderWithDeps(
    createTaskEvent: appendTaskEvent = createTaskEvent,
    logInfo: writeInfoLog = logInfo,
    logWarn: writeWarnLog = logWarn,
+    allowInactiveTaskRetry = false,
  } = deps

  const runtimeDeps: FulfillmentPrepareDeps = {
@@ -90,7 +96,12 @@ export async function syncDeliveryTasksForOrderWithDeps(

  const existingTasks = await listTasks(order.id)
  const tasks: DeliveryTaskRow[] = [...existingTasks]
-  const existingTaskKeys = new Set(existingTasks.map(buildTaskUnitKey))
+  const existingTaskKeys = new Set(
+    (allowInactiveTaskRetry
+      ? existingTasks.filter((task) => !isInactiveCarrierTaskStatus(task.task_status))
+      : existingTasks
+    ).map(buildTaskUnitKey),
+  )
  const carrierEvidence = await loadCarrierEvidence(orderItems.map((item) => Number(item.id)))

  writeInfoLog('[delivery-task-service]', '开始核对订单履约任务完整性', {
@@ -205,6 +216,9 @@ export async function syncDeliveryTasksForOrderWithDeps(
        task =
          concurrentTasks.find(
            (candidate) =>
+              (allowInactiveTaskRetry
+                ? !isInactiveCarrierTaskStatus(candidate.task_status)
+                : true) &&
              buildTaskUnitKey(candidate) === buildOrderItemUnitKey(item.id, unitIndex),
          ) || null
        if (!task) {
@@ -30,7 +30,6 @@ import {
  resolveKuaishouIndustryVoucherValidity,
 } from './voucher-service.js'
 import { bindKuaishouIndustryVouchersToOrderTasks } from './voucher-binding-service.js'
-import { autoConsumeKuaishouIndustryVouchersForOrder } from './voucher-auto-consume-service.js'
 import { parseAmountToFen } from '../../../utils/money.js'
 import {
  publishAdminOrderRealtimeChange,
@@ -518,32 +517,6 @@ async function syncOpen91OrderAfterSendCallbackSuccess(
      )
    }
  }
-
-  try {
-    const autoConsumeResult = await autoConsumeKuaishouIndustryVouchersForOrder(normalizedOid, {
-      source: `${source}_auto_consume`,
-    })
-    if (autoConsumeResult.consumedCount > 0 || autoConsumeResult.failedCount > 0) {
-      logIntegration('[kuaishou-industry/send-code]', '发码回调成功后自动推进电子凭证核销完成', {
-        source,
-        oid: normalizedOid,
-        orderId: order.id,
-        consumedCount: autoConsumeResult.consumedCount,
-        alreadyConsumedCount: autoConsumeResult.alreadyConsumedCount,
-        skippedCount: autoConsumeResult.skippedCount,
-        failedCount: autoConsumeResult.failedCount,
-        skippedReasons: autoConsumeResult.skippedReasons,
-        errors: autoConsumeResult.errors,
-      })
-    }
-  } catch (consumeError) {
-    logWarn('[kuaishou-industry/send-code]', '发码回调成功后自动推进电子凭证核销异常', {
-      source,
-      oid: normalizedOid,
-      orderId: order.id,
-      error: consumeError instanceof Error ? consumeError.message : String(consumeError),
-    })
-  }
 }

 function isOpen91OrderAwaitingFulfillmentConfig(error: unknown) {
@@ -45,6 +45,7 @@ test('normalizeScheduledJobsConfig ensures default jobs exist', () => {
  assert.equal(ids.includes('cost-finance-sync'), false)
  assert.ok(ids.includes('channel-assign'))
  assert.ok(ids.includes('ledger-post'))
+  assert.ok(ids.includes('voucher-consume-dispatch'))
 })

 test('旧的 cost_finance_sync 任务已彻底移除，不再出现在配置里', () => {
@@ -19,6 +19,7 @@ const COST_SNAPSHOT_JOB_ID = 'cost-snapshot'
 const SALES_FACTS_JOB_ID = 'sales-facts'
 const LEDGER_POST_JOB_ID = 'ledger-post'
 const ORDER_SERVICE_AUTO_PUBLISH_JOB_ID = 'order-service-auto-publish'
+const VOUCHER_CONSUME_DISPATCH_JOB_ID = 'voucher-consume-dispatch'

 export function getScheduledJobsFilePath() {
  return SCHEDULED_JOBS_FILE_PATH
@@ -93,6 +94,7 @@ export function normalizeScheduledJobsConfig(rawValue: unknown) {
  const hasOrderServiceAutoPublish = jobs.some(
    (item) => item.id === ORDER_SERVICE_AUTO_PUBLISH_JOB_ID,
  )
+  const hasVoucherConsumeDispatch = jobs.some((item) => item.id === VOUCHER_CONSUME_DISPATCH_JOB_ID)

  if (!hasWorkOrderTimeout) {
    jobs.push(createDefaultWorkOrderTimeoutJob())
@@ -116,6 +118,7 @@ export function normalizeScheduledJobsConfig(rawValue: unknown) {
  if (!hasSalesFacts) jobs.push(createDefaultSalesFactsJob())
  if (!hasLedgerPost) jobs.push(createDefaultLedgerPostJob())
  if (!hasOrderServiceAutoPublish) jobs.push(createDefaultOrderServiceAutoPublishJob())
+  if (!hasVoucherConsumeDispatch) jobs.push(createDefaultVoucherConsumeDispatchJob())

  return {
    enabled: typeof source.enabled === 'boolean' ? source.enabled : true,
@@ -166,6 +169,9 @@ function normalizeScheduledJob(rawValue: unknown) {
  if (type === 'order_service_auto_publish') {
    return normalizeOrderServiceAutoPublishJob(rawValue)
  }
+  if (type === 'voucher_consume_dispatch') {
+    return normalizeVoucherConsumeDispatchJob(rawValue)
+  }

  return null
 }
@@ -237,6 +243,7 @@ function createDefaultScheduledJobsConfig() {
      createDefaultSalesFactsJob(),
      createDefaultLedgerPostJob(),
      createDefaultOrderServiceAutoPublishJob(),
+      createDefaultVoucherConsumeDispatchJob(),
    ],
  }
 }
@@ -322,6 +329,19 @@ function normalizeOrderServiceAutoPublishJob(rawValue: JsonObject) {
  }
 }

+function normalizeVoucherConsumeDispatchJob(rawValue: JsonObject) {
+  const config = isPlainObject(rawValue.config) ? rawValue.config : {}
+  return {
+    id: VOUCHER_CONSUME_DISPATCH_JOB_ID,
+    type: 'voucher_consume_dispatch',
+    enabled: rawValue.enabled !== false,
+    intervalSeconds: normalizeRangeInteger(rawValue.intervalSeconds, 5, 1, 60),
+    config: {
+      batchSize: normalizeRangeInteger(config.batchSize, 20, 1, 200),
+    },
+  }
+}
+
 function normalizeLedgerPostJob(rawValue: JsonObject) {
  const config = isPlainObject(rawValue.config) ? rawValue.config : {}
  return {
@@ -385,6 +405,16 @@ function createDefaultOrderServiceAutoPublishJob() {
  }
 }

+function createDefaultVoucherConsumeDispatchJob() {
+  return {
+    id: VOUCHER_CONSUME_DISPATCH_JOB_ID,
+    type: 'voucher_consume_dispatch',
+    enabled: true,
+    intervalSeconds: 5,
+    config: { batchSize: 20 },
+  }
+}
+
 function createDefaultSalesFactsJob() {
  return {
    id: SALES_FACTS_JOB_ID,
@@ -86,7 +86,15 @@ export async function runChannelAssignJob(job: JsonObject = {}) {
 export async function runCarrierDispatchJob(job: JsonObject = {}) {
  const config = jobConfig(job)
  const limit = clampInt(config.limit, 200, 1, 2000)
-  const reconciled = await reconcilePendingCarriers({ limit: Math.min(limit, 200) })
+  const reconcileCursor =
+    config.carrierReconcileAfterId != null
+      ? Math.max(0, Math.trunc(Number(config.carrierReconcileAfterId)))
+      : await readWatermarkItemId('carrier_reconcile')
+  const reconciled = await reconcilePendingCarriers({
+    limit: Math.min(limit, 200),
+    afterId: reconcileCursor,
+  })
+  await touchWatermark('carrier_reconcile', { lastItemId: reconciled.nextCursor })
  const repairedAssigned = await reconcilePendingCarriers({
    limit: Math.min(limit, 200),
    includeAssignedMissingCarrier: true,
@@ -335,7 +343,8 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) {
    message:
      `人工载体挂载 ${attached}，缺配置 ${missingManual.rows.length}，` +
      `缺履约载体 ${missingCarrier.rows.length}，多载体冲突 ${conflictingCarriers.rows.length}，异常收口 ${resolved}，` +
-      `自动转人工 ${fallback.switched}，链路补载体 ${reconciled.workOrdersCreated + reconciled.affiliateTasksCreated + repairedAssigned.workOrdersCreated + repairedAssigned.affiliateTasksCreated}，` +
+      `自动转人工 ${fallback.switched}，本轮补载体扫描 ${reconciled.scanned}、创建 ${reconciled.workOrdersCreated + reconciled.affiliateTasksCreated}、失败 ${reconciled.failed}，` +
+      `已分配补载体 ${repairedAssigned.workOrdersCreated + repairedAssigned.affiliateTasksCreated}，` +
      `兜底 ${reconciled.fallbacksSelected + repairedAssigned.fallbacksSelected}`,
    attached,
    missingManual: missingManual.rows.length,
@@ -541,6 +550,7 @@ const PIPELINE_RUNNERS: Record<
  (job?: JsonObject) => Promise<Record<string, unknown>>
 > = {
  channel_assign: runChannelAssignJob,
+  carrier_reconcile: runCarrierDispatchJob,
  carrier_dispatch: runCarrierDispatchJob,
  product_normalize: runProductNormalizeJob,
  cost_snapshot: runCostSnapshotJob,
@@ -14,6 +14,7 @@ import type { WatermarkPipeline } from '../watermark-store.js'

 export const FINANCE_PIPELINES = [
  { name: 'channel_assign', label: '渠道判定', jobId: 'channel-assign' },
+  { name: 'carrier_reconcile', label: '载体补建', jobId: 'carrier-dispatch' },
  { name: 'carrier_dispatch', label: '履约载体', jobId: 'carrier-dispatch' },
  { name: 'product_normalize', label: '商品归一', jobId: 'product-normalize' },
  { name: 'cost_snapshot', label: '成本快照', jobId: 'cost-snapshot' },
@@ -15,6 +15,7 @@ import {
  runProductNormalizeJob,
  runSalesFactsJob,
 } from './finance-pipeline-jobs.js'
+import { runVoucherConsumeDispatchJob } from '../fulfillment/voucher-consume-queue-service.js'

 const timers = new Map<string, NodeJS.Timeout>()
 const jobStates = new Map<string, JsonObject>()
@@ -88,7 +89,7 @@ function rescheduleJob(job: JsonObject) {
  if (!jobId) return
  const latestJob = getScheduledJobById(jobId)
  if (!latestJob || latestJob.enabled !== true) return
-  const intervalMs = Math.max(10, Number(latestJob.intervalSeconds || 20)) * 1000
+  const intervalMs = Math.max(1, Number(latestJob.intervalSeconds || 20)) * 1000
  scheduleJob(latestJob, intervalMs)
 }

@@ -234,6 +235,11 @@ function dispatchJob(job: JsonObject) {
  if (job.type === 'order_service_auto_publish') {
    return runOrderServiceAutoPublishJobs(job)
  }
+  if (job.type === 'voucher_consume_dispatch') {
+    return runVoucherConsumeDispatchJob(
+      job.config && typeof job.config === 'object' ? (job.config as Record<string, unknown>) : {},
+    )
+  }

  throw createHttpError(`不支持的定时任务类型：${job.type}`, {
    statusCode: 400,
@@ -9,6 +9,7 @@ import { nowIso } from '../utils/time.js'

 export type WatermarkPipeline =
  | 'channel_assign'
+  | 'carrier_reconcile'
  | 'carrier_dispatch'
  | 'product_normalize'
  | 'cost_snapshot'
@@ -152,6 +152,8 @@ export async function createAdminMockWorkOrder(payload: JsonObject = {}) {
      ? JSON.stringify({
          profileId: serviceProfile.profileId,
          categoryId: serviceProfile.categoryId,
+          sourceCategoryId: serviceProfile.sourceCategoryId,
+          inherited: serviceProfile.inherited,
          versionId: serviceProfile.versionId,
          versionNo: serviceProfile.versionNo,
          config: serviceProfile.config,
@@ -13,7 +13,10 @@ import {
  publishWorkerWalletRealtimeChange,
  publishWorkOrderRealtimeChange,
 } from '../realtime/realtime-event-service.js'
-import { consumeIndustryVouchersBeforeWorkOrderAssign } from './publish-voucher-service.js'
+import {
+  scheduleVoucherConsumeForWorkOrder,
+  type VoucherConsumeScheduleResult,
+} from '../fulfillment/voucher-consume-queue-service.js'
 import {
  mapWorkOrderAdmin,
  normalizeOptionalId,
@@ -65,8 +68,7 @@ export async function assignAdminWorkOrderToWorker(
    })
  }

-  // 已有拼单份额的订单不能再按整单指派；放在核销电子凭证之前，
-  // 避免“凭证已核销但指派失败”的副作用。
+  // 已有拼单份额的订单不能再按整单指派。
  const activeShares = (await listWorkOrderShares(workOrder.id)).filter(
    (share) => share.status !== 'cancelled',
  )
@@ -77,26 +79,6 @@ export async function assignAdminWorkOrderToWorker(
    })
  }

-  const voucherConsume = await consumeIndustryVouchersBeforeWorkOrderAssign(workOrder)
-  if (!voucherConsume.ok) {
-    const failedAt = nowIso()
-    await createWorkOrderEvent({
-      workOrderId: workOrder.id,
-      actorType: 'admin',
-      actorId: actorName,
-      eventType: 'assign_voucher_consume_failed',
-      fromStatus: workOrder.status,
-      toStatus: workOrder.status,
-      payloadJson: JSON.stringify({ workerId: worker.id, voucherConsume }),
-      now: failedAt,
-    })
-    throw createHttpError(voucherConsume.errorMessage || '电子凭证核销失败，订单未指派', {
-      statusCode: 409,
-      errorCode: 'work_order_assign_voucher_consume_failed',
-      context: voucherConsume,
-    })
-  }
-
  const now = nowIso()
  const { order: updated, failureReason } = await assignWorkOrderToWorker({
    workOrderId: workOrder.id,
@@ -108,38 +90,70 @@ export async function assignAdminWorkOrderToWorker(
    actorName,
  })
  if (failureReason === 'work_order_not_assignable') {
-    throw createHttpError(
-      resolveAssignFailureMessage(voucherConsume.voucherCount, '订单状态已变化，无法指派'),
-      { statusCode: 409, errorCode: 'work_order_not_assignable' },
-    )
+    throw createHttpError('订单状态已变化，无法指派', {
+      statusCode: 409,
+      errorCode: 'work_order_not_assignable',
+    })
  }
  if (failureReason === 'worker_active_order_limit') {
-    throw createHttpError(
-      resolveAssignFailureMessage(voucherConsume.voucherCount, '该打手已达到最大同时接单量'),
-      { statusCode: 409, errorCode: 'worker_active_order_limit' },
-    )
+    throw createHttpError('该打手已达到最大同时接单量', {
+      statusCode: 409,
+      errorCode: 'worker_active_order_limit',
+    })
  }
  if (failureReason === 'worker_deposit_insufficient') {
-    throw createHttpError(
-      resolveAssignFailureMessage(voucherConsume.voucherCount, '该打手余额不足，无法冻结所需押金'),
-      { statusCode: 409, errorCode: 'worker_deposit_insufficient' },
-    )
+    throw createHttpError('该打手余额不足，无法冻结所需押金', {
+      statusCode: 409,
+      errorCode: 'worker_deposit_insufficient',
+    })
  }
  if (failureReason === 'work_order_has_shares') {
-    throw createHttpError(
-      resolveAssignFailureMessage(
-        voucherConsume.voucherCount,
-        '该订单已有拼单份额，不能按整单指派；请使用拼单管理',
-      ),
-      { statusCode: 409, errorCode: 'work_order_has_shares' },
-    )
+    throw createHttpError('该订单已有拼单份额，不能按整单指派；请使用拼单管理', {
+      statusCode: 409,
+      errorCode: 'work_order_has_shares',
+    })
  }
  if (!updated) {
-    throw createHttpError(
-      resolveAssignFailureMessage(voucherConsume.voucherCount, '指派失败，订单状态可能已变化'),
-      { statusCode: 409, errorCode: 'work_order_assign_conflict' },
-    )
+    throw createHttpError('指派失败，订单状态可能已变化', {
+      statusCode: 409,
+      errorCode: 'work_order_assign_conflict',
+    })
  }
+
+  let voucherConsume: VoucherConsumeScheduleResult
+  try {
+    voucherConsume = await scheduleVoucherConsumeForWorkOrder(updated, 'work_order_assigned')
+  } catch (error) {
+    voucherConsume = {
+      ok: false,
+      voucherCount: 0,
+      consumedCount: 0,
+      alreadyConsumedCount: 0,
+      failedCount: 1,
+      voucherCodes: [],
+      errorMessage: error instanceof Error ? error.message : '安排自动核销失败',
+      queuedCount: 0,
+      jobIds: [],
+      dueAt: '',
+      status: 'failed',
+    }
+  }
+  await createWorkOrderEvent({
+    workOrderId: workOrder.id,
+    actorType: 'admin',
+    actorId: actorName,
+    eventType:
+      voucherConsume.status === 'queued'
+        ? 'assign_voucher_consume_scheduled'
+        : voucherConsume.status === 'failed'
+          ? 'assign_voucher_consume_schedule_failed'
+          : 'assign_voucher_consume_schedule_skipped',
+    fromStatus: workOrder.status,
+    toStatus: updated.status,
+    payloadJson: JSON.stringify({ workerId: worker.id, voucherConsume }),
+    now,
+  }).catch(() => undefined)
+
  publishWorkOrderRealtimeChange({
    workOrderId: Number(workOrder.id),
    workerIds: [Number(worker.id)],
@@ -149,10 +163,6 @@ export async function assignAdminWorkOrderToWorker(
  return { order: mapWorkOrderAdmin(updated), voucherConsume }
 }

-function resolveAssignFailureMessage(voucherCount: number, fallback: string) {
-  return voucherCount > 0 ? `电子凭证已核销，但${fallback}，请重试指派` : fallback
-}
-
 function resolveAssignedDeadlineAt(timeoutMinutes: number, now: string): string | null {
  if (timeoutMinutes <= 0) return null
  return new Date(new Date(now).getTime() + timeoutMinutes * 60 * 1000).toISOString()
@@ -22,9 +22,9 @@ import { resolveAdminNotificationEntity } from '../admin/admin-notification-serv
 import { enqueueCustomerOrderNotification } from '../notification/customer-order-notification-service.js'
 import { publishWorkOrderRealtimeChange } from '../realtime/realtime-event-service.js'
 import {
-  consumeIndustryVouchersBeforeWorkOrderPublish,
-  type WorkOrderVoucherConsumeResult,
-} from './publish-voucher-service.js'
+  scheduleVoucherConsumeForWorkOrder,
+  type VoucherConsumeScheduleResult,
+} from '../fulfillment/voucher-consume-queue-service.js'
 import {
  mapWorkOrderAdmin,
  mapWorkOrderShare,
@@ -347,7 +347,7 @@ export async function reopenAdminCancelledWorkOrder(workOrderId: number | string
 export async function publishAdminWorkOrder(
  workOrderId: number | string,
  actorName = '',
-  options: { requireVoucherConsume?: boolean } = {},
+  _options: { requireVoucherConsume?: boolean } = {},
 ) {
  const workOrder = await getRequiredWorkOrder(workOrderId)
  if (workOrder.status !== WORK_ORDER_STATUS.UNASSIGNED) {
@@ -364,39 +364,6 @@ export async function publishAdminWorkOrder(
  }

  const now = nowIso()
-  let voucherConsume: WorkOrderVoucherConsumeResult
-  try {
-    voucherConsume = await consumeIndustryVouchersBeforeWorkOrderPublish(workOrder)
-  } catch (error) {
-    voucherConsume = {
-      ok: false,
-      voucherCount: 0,
-      consumedCount: 0,
-      alreadyConsumedCount: 0,
-      failedCount: 1,
-      voucherCodes: [],
-      errorMessage: error instanceof Error ? error.message : '电子凭证核销失败',
-    }
-  }
-  if (!voucherConsume.ok) {
-    await createWorkOrderEvent({
-      workOrderId: workOrder.id,
-      actorType: 'system',
-      actorId: 'kuaishou_send_code',
-      eventType: 'publish_voucher_consume_failed',
-      fromStatus: workOrder.status,
-      toStatus: workOrder.status,
-      payloadJson: JSON.stringify(voucherConsume),
-      now: nowIso(),
-    })
-    if (options.requireVoucherConsume === true) {
-      throw createHttpError(voucherConsume.errorMessage || '电子凭证核销失败，订单未发布', {
-        statusCode: 409,
-        errorCode: 'work_order_publish_voucher_consume_failed',
-        context: { voucherConsume },
-      })
-    }
-  }

  const published = await publishUnassignedWorkOrder({
    workOrderId: workOrder.id,
@@ -411,6 +378,39 @@ export async function publishAdminWorkOrder(
  }

  const currentOrder = await getRequiredWorkOrder(workOrder.id)
+  let voucherConsume: VoucherConsumeScheduleResult
+  try {
+    voucherConsume = await scheduleVoucherConsumeForWorkOrder(currentOrder, 'work_order_published')
+  } catch (error) {
+    voucherConsume = {
+      ok: false,
+      voucherCount: 0,
+      consumedCount: 0,
+      alreadyConsumedCount: 0,
+      failedCount: 1,
+      voucherCodes: [],
+      errorMessage: error instanceof Error ? error.message : '安排自动核销失败',
+      queuedCount: 0,
+      jobIds: [],
+      dueAt: '',
+      status: 'failed',
+    }
+  }
+  await createWorkOrderEvent({
+    workOrderId: currentOrder.id,
+    actorType: 'system',
+    actorId: 'voucher_consume_policy',
+    eventType:
+      voucherConsume.status === 'queued'
+        ? 'publish_voucher_consume_scheduled'
+        : voucherConsume.status === 'failed'
+          ? 'publish_voucher_consume_schedule_failed'
+          : 'publish_voucher_consume_schedule_skipped',
+    fromStatus: currentOrder.status,
+    toStatus: currentOrder.status,
+    payloadJson: JSON.stringify(voucherConsume),
+    now: nowIso(),
+  }).catch(() => undefined)
  publishWorkOrderRealtimeChange({ workOrderId: Number(workOrder.id), hallChanged: true })
  await enqueueCustomerOrderNotification({
    orderId: currentOrder.order_id,
@@ -47,9 +47,9 @@ import { assignAdminWorkOrderToWorker } from './admin-work-order-assignment-serv
 import { unassignAdminWorkOrder } from './admin-work-order-cancellation-service.js'
 import { mapWorkOrderAdmin } from './mappers.js'
 import {
-  consumeIndustryVouchersBeforeWorkOrderPublish,
-  type WorkOrderVoucherConsumeResult,
-} from './publish-voucher-service.js'
+  scheduleVoucherConsumeForWorkOrder,
+  type VoucherConsumeScheduleResult,
+} from '../fulfillment/voucher-consume-queue-service.js'
 import {
  type ConsignmentMatchOption,
  type ConsignmentMatchResult,
@@ -383,7 +383,7 @@ export async function fulfillWorkOrderWithConsignment(input: {
 }): Promise<{
  order: ReturnType<typeof mapWorkOrderAdmin>
  inventory: ReturnType<typeof mapConsignmentInventory>
-  voucherConsume: WorkOrderVoucherConsumeResult
+  voucherConsume: VoucherConsumeScheduleResult
 }> {
  const workOrder = await getWorkOrderById(input.workOrderId)
  if (!workOrder) {
@@ -576,13 +576,10 @@ export async function fulfillWorkOrderWithConsignment(input: {
    return { updatedOrder: wo, updatedInventory: inv }
  })

-  // 顺带核销快手电子凭证。
-  // 核销要走快手外部接口、无法和上面的数据库事务合并回滚，因此这里只能做两件事：
-  //   1) 失败时不吞错：写事件留痕，并把 voucherConsume.ok=false 交给前端提示运营人工核销；
-  //   2) 绝不因为"凭证没核销"就谎报成功（此前前端无条件弹成功提示，属于假成功）。
-  let voucherConsume: WorkOrderVoucherConsumeResult
+  // 行内发货完成后安排延迟核销，外部平台调用不阻塞当前履约事务。
+  let voucherConsume: VoucherConsumeScheduleResult
  try {
-    voucherConsume = await consumeIndustryVouchersBeforeWorkOrderPublish(updatedOrder)
+    voucherConsume = await scheduleVoucherConsumeForWorkOrder(updatedOrder, 'work_order_assigned')
  } catch (error) {
    voucherConsume = {
      ok: false,
@@ -591,7 +588,11 @@ export async function fulfillWorkOrderWithConsignment(input: {
      alreadyConsumedCount: 0,
      failedCount: 1,
      voucherCodes: [],
-      errorMessage: error instanceof Error ? error.message : '电子凭证核销失败',
+      errorMessage: error instanceof Error ? error.message : '安排自动核销失败',
+      queuedCount: 0,
+      jobIds: [],
+      dueAt: '',
+      status: 'failed',
    }
  }
  if (!voucherConsume.ok) {
@@ -599,8 +600,13 @@ export async function fulfillWorkOrderWithConsignment(input: {
      await createWorkOrderEvent({
        workOrderId: Number(updatedOrder.id),
        actorType: 'system',
-        actorId: 'kuaishou_send_code',
-        eventType: 'consignment_fulfill_voucher_consume_failed',
+        actorId: 'voucher_consume_policy',
+        eventType:
+          voucherConsume.status === 'queued'
+            ? 'consignment_fulfill_voucher_consume_scheduled'
+            : voucherConsume.status === 'failed'
+              ? 'consignment_fulfill_voucher_consume_schedule_failed'
+              : 'consignment_fulfill_voucher_consume_schedule_skipped',
        fromStatus: String(updatedOrder.status),
        toStatus: String(updatedOrder.status),
        payloadJson: JSON.stringify(voucherConsume),
@@ -668,7 +674,7 @@ export async function assignWorkOrderWithConsignmentWorker(input: {
 }): Promise<{
  order: ReturnType<typeof mapWorkOrderAdmin>
  inventory: ReturnType<typeof mapConsignmentInventory>
-  voucherConsume: WorkOrderVoucherConsumeResult
+  voucherConsume: VoucherConsumeScheduleResult
 }> {
  const workOrder = await getWorkOrderById(input.workOrderId)
  if (!workOrder) {
@@ -5,7 +5,10 @@ import {
  isOrderServiceMaterialRequired,
  normalizeProfileConfig,
 } from './order-service-profile-service.js'
-import { backfillPendingOrderServiceSnapshotsForCategory } from '../../repositories/worker-platform/order-service-profile-repo.js'
+import {
+  backfillPendingOrderServiceSnapshotsForCategory,
+  getPublishedOrderServiceProfileByCategoryWithInheritance,
+} from '../../repositories/worker-platform/order-service-profile-repo.js'

 function config(patch: Record<string, unknown> = {}) {
  return normalizeProfileConfig({
@@ -181,4 +184,40 @@ test('发布配置：待提交且无有效拼单份额的工单可更新到最
  assert.equal(count, 2)
  assert.doesNotMatch(capturedSql, /service_profile_snapshot_json IS NULL/)
  assert.match(capturedSql, /share\.status <> 'cancelled'/)
+  assert.match(capturedSql, /WITH affected_categories/)
+  assert.match(capturedSql, /child_profile\.status <> 'archived'/)
+  assert.match(capturedSql, /wo\.category_id IN/)
+  assert.match(capturedSql, /'sourceCategoryId', p\.category_id/)
+  assert.match(capturedSql, /'inherited', wo\.category_id <> p\.category_id/)
+})
+
+test('订单服务配置：二级分类优先，未发布时沿分类链继承父级', async () => {
+  let capturedSql = ''
+  const client = {
+    query: async (sql: string) => {
+      capturedSql = sql
+      return {
+        rows: [
+          {
+            profile_id: 9,
+            category_id: 3,
+            source_category_id: 3,
+            version_id: 19,
+            version_no: 4,
+            config_json: { fields: [] },
+          },
+        ],
+      }
+    },
+  } as unknown as Parameters<typeof getPublishedOrderServiceProfileByCategoryWithInheritance>[1]
+
+  const resolved = await getPublishedOrderServiceProfileByCategoryWithInheritance(81, client)
+
+  assert.equal(resolved?.profile_id, 9)
+  assert.equal(resolved?.category_id, 3)
+  assert.equal(resolved?.source_category_id, 3)
+  assert.equal(resolved?.version_no, 4)
+  assert.match(capturedSql, /WITH RECURSIVE category_chain/)
+  assert.match(capturedSql, /ORDER BY chain\.depth ASC/)
+  assert.match(capturedSql, /p\.status <> 'archived'/)
 })
@@ -10,7 +10,7 @@ import {
  backfillPendingOrderServiceSnapshotsForCategory,
  createOrderServiceProfileVersion,
  getOrderServiceProfileVersion,
-  getPublishedOrderServiceProfileByCategory,
+  getPublishedOrderServiceProfileByCategoryWithInheritance,
  listOrderServiceProfiles,
  setOrderServiceProfileCurrentVersion,
  upsertOrderServiceProfile,
@@ -83,6 +83,8 @@ export type OrderServiceProfileConfig = {
 export type OrderServiceProfileSnapshot = {
  profileId: number
  categoryId: number
+  sourceCategoryId: number
+  inherited: boolean
  versionId: number
  versionNo: number
  config: OrderServiceProfileConfig
@@ -307,11 +309,14 @@ export async function getPublishedOrderServiceProfileSnapshotByCategory(
  categoryId: number | string,
  client?: Pick<PoolClient, 'query'>,
 ): Promise<OrderServiceProfileSnapshot | null> {
-  const row = await getPublishedOrderServiceProfileByCategory(categoryId, client)
+  const row = await getPublishedOrderServiceProfileByCategoryWithInheritance(categoryId, client)
  if (!row) return null
+  const targetCategoryId = Number(categoryId)
  return {
    profileId: Number(row.profile_id),
-    categoryId: Number(row.category_id),
+    categoryId: targetCategoryId,
+    sourceCategoryId: Number(row.source_category_id),
+    inherited: Number(row.source_category_id) !== targetCategoryId,
    versionId: Number(row.version_id),
    versionNo: Number(row.version_no),
    config: normalizeProfileConfig(row.config_json),
@@ -329,6 +334,8 @@ export async function getOrderServiceProfileSnapshotByVersion(input: {
  return {
    profileId: Number(input.profileId),
    categoryId: Number(input.categoryId),
+    sourceCategoryId: Number(input.categoryId),
+    inherited: false,
    versionId: Number(version.id),
    versionNo: Number(version.version_no),
    config: normalizeProfileConfig(version.config_json),
@@ -560,6 +560,13 @@ async function resolveProfileConfig(
    })
    if (snapshotFromVersion) return snapshotFromVersion.config
  }
+  if (order.category_id) {
+    const published = await getPublishedOrderServiceProfileSnapshotByCategory(
+      Number(order.category_id),
+      client,
+    )
+    if (published) return published.config
+  }
  const legacy = safeParseJson(order.requirement_json)
  if (Array.isArray(legacy.fields)) {
    return {
@@ -8,7 +8,7 @@ import { query } from '../../db/client.js'
 * - 订单不存在（91 未到/缺失）：用快手数据兜底建单 + 建工单 + 凭证回填
 *
 * 幂等依据：orders 唯一键 provider+platform+shop_id+platform_order_id
- * （由 91 侧查重逻辑保证同 oid 单条记录）、work_orders.order_item_id 唯一索引、
+ * （由 91 侧查重逻辑保证同 oid 单条记录）、work_orders 未取消工单对 order_item_id 的唯一索引、
 * listWorkOrdersByPlatformOrderId 查重。
 *
 * 本服务只写 orders / order_items / work_orders，不触碰
@@ -125,6 +125,7 @@ export async function syncWorkerOrdersForSourceOrder(
    sourceMetadata?: JsonObject
    skipChannelAssign?: boolean
    preferredWorkRuleKey?: string
+    retryCancelledWorkOrder?: boolean
  } = {},
 ) {
  const [rules, ruleMappings] = await Promise.all([
@@ -252,7 +253,10 @@ export async function syncWorkerOrdersForSourceOrder(

    // 已有工单：跳过匹配日志，避免 PDD 定时同步对旧单反复刷「命中」日志。
    const existingWorkOrder = await getWorkOrderByOrderItemId(item.id)
-    if (existingWorkOrder) {
+    if (
+      existingWorkOrder &&
+      !(options.retryCancelledWorkOrder === true && existingWorkOrder.status === 'cancelled')
+    ) {
      skipped.push({ orderItemId: Number(item.id), reason: 'already_exists' })
      decisions.push(
        buildWorkerConsumerDecision(item, {
@@ -425,6 +429,8 @@ export async function syncWorkerOrdersForSourceOrder(
        ? JSON.stringify({
            profileId: serviceProfile.profileId,
            categoryId: serviceProfile.categoryId,
+            sourceCategoryId: serviceProfile.sourceCategoryId,
+            inherited: serviceProfile.inherited,
            versionId: serviceProfile.versionId,
            versionNo: serviceProfile.versionNo,
            config: serviceProfile.config,
@@ -2554,14 +2554,15 @@ test('履约与指派两条路径都要过服务端规格复检（不只依赖
  assert.ok(src.includes('consignment_match_note_required'))
 })

-test('内部履约的电子凭证核销失败要写事件留痕（不能静默假成功）', async () => {
+test('内部履约要投递延迟核销任务并写事件留痕', async () => {
  const readFile = await import('node:fs/promises')
  const src = await readFile.readFile(
    new URL('./consignment-fulfillment-service.ts', import.meta.url),
    'utf8',
  )

-  assert.ok(src.includes('consignment_fulfill_voucher_consume_failed'))
+  assert.ok(src.includes('consignment_fulfill_voucher_consume_scheduled'))
+  assert.ok(src.includes('scheduleVoucherConsumeForWorkOrder'))
  assert.ok(src.includes('if (!voucherConsume.ok)'))
 })