fix(finance): 重写履约路由迁移与统一渠道配置

重写 120/121 迁移，统一内部 SKU、渠道 id、config_json 和旧规则转换。后台渠道配置改读统一渠道表，卡券与人工配置不再写旧规则。增加安全渠道切换，并排除已关闭、失败或过期的历史任务载体。按生产 119 基线重跑迁移、完整履约财务链路和严格预检。
2026-11-02 15:29:13 +08:00
parent 2f87445ed9
commit 6bf28489e0
31 changed files with 958 additions and 464 deletions
@@ -1,10 +1,22 @@
 -- 120_fulfillment_routing_invariants.sql
 --
-- 履约路由唯一口径与数据清理：
--   1. 订单项内部 SKU 由一个不可变 SQL 函数统一解析；
--   2. 人工配置自动成为 internal_sku_channels 的 manual 能力；
--   3. manual 与其他渠道并存时统一降为兜底，避免同优先级被静默抢占；
--   4. 可确定的旧 sku_code 规则迁移为内部 SKU 渠道并停用旧规则。
+-- 生产迁移 119 之后的履约路由统一迁移。
+--
+-- 目标：
+--   1. 订单项内部 SKU 只有一个 SQL 解析口径；
+--   2. internal_sku_channels 成为唯一渠道配置源；
+--   3. 人工配置、旧 sku_code 规则和渠道适配参数全部回填到统一表；
+--   4. 原有 sku_code 规则只做一次性迁移，迁移后可确定的规则全部停用。
+
+-- 稳定 id 让后台渠道配置可以通过统一 API 增删改，而不必继续暴露旧规则表。
+ALTER TABLE internal_sku_channels
+  ADD COLUMN IF NOT EXISTS id BIGSERIAL,
+  ADD COLUMN IF NOT EXISTS note TEXT NOT NULL DEFAULT '',
+  ADD COLUMN IF NOT EXISTS created_by_admin_user_id BIGINT
+    REFERENCES admin_users(id) ON DELETE SET NULL;
+
+CREATE UNIQUE INDEX IF NOT EXISTS uq_internal_sku_channels_id
+  ON internal_sku_channels (id);

 CREATE OR REPLACE FUNCTION resolve_internal_sku(
  p_manual_sku_key TEXT,
@@ -33,7 +45,7 @@ $$;
 COMMENT ON FUNCTION resolve_internal_sku(TEXT, TEXT, TEXT) IS
  '订单项内部 SKU 唯一 SQL 口径：中文 manual_sku_key/sku_name 优先，再按历史生成列顺序回退';

-- 人工配置本身就是一个 manual 渠道声明。存在其他渠道时默认作为 90 优先级兜底。
+-- 人工配置就是 manual 渠道声明。存在其他渠道时先按 90 优先级兜底。
 INSERT INTO internal_sku_channels (
  internal_sku, channel, priority, is_fallback, enabled,
  canonical_product_id, config_json, source, created_at, updated_at
@@ -42,15 +54,13 @@ SELECT
  BTRIM(cfg.sku_key),
  'manual',
  CASE WHEN EXISTS (
-    SELECT 1
-    FROM internal_sku_channels c
+    SELECT 1 FROM internal_sku_channels c
    WHERE BTRIM(c.internal_sku) = BTRIM(cfg.sku_key)
      AND c.enabled = TRUE
      AND c.channel <> 'manual'
  ) THEN 90 ELSE 10 END,
  CASE WHEN EXISTS (
-    SELECT 1
-    FROM internal_sku_channels c
+    SELECT 1 FROM internal_sku_channels c
    WHERE BTRIM(c.internal_sku) = BTRIM(cfg.sku_key)
      AND c.enabled = TRUE
      AND c.channel <> 'manual'
@@ -69,20 +79,241 @@ SELECT
 FROM manual_fulfillment_configs cfg
 WHERE BTRIM(COALESCE(cfg.sku_key, '')) <> ''
 ON CONFLICT (internal_sku, channel) DO UPDATE SET
-  priority = CASE
-    WHEN internal_sku_channels.source = 'operator' THEN internal_sku_channels.priority
-    ELSE EXCLUDED.priority
-  END,
-  is_fallback = CASE
-    WHEN internal_sku_channels.source = 'operator' THEN internal_sku_channels.is_fallback
-    ELSE EXCLUDED.is_fallback
-  END,
+  priority = CASE WHEN internal_sku_channels.source = 'operator'
+    THEN internal_sku_channels.priority ELSE EXCLUDED.priority END,
+  is_fallback = CASE WHEN internal_sku_channels.source = 'operator'
+    THEN internal_sku_channels.is_fallback ELSE EXCLUDED.is_fallback END,
  enabled = EXCLUDED.enabled,
  canonical_product_id = COALESCE(EXCLUDED.canonical_product_id, internal_sku_channels.canonical_product_id),
  config_json = internal_sku_channels.config_json || EXCLUDED.config_json,
  updated_at = NOW();

-- 只要同一内部 SKU 还有自动发货/接单/卡券能力，人工渠道就不能作为并列主渠道。
+-- 旧 sku_code 规则能稳定解析出的内部 SKU，统一补到入口映射表。
+WITH legacy_candidates AS (
+  SELECT
+    rule.provider,
+    rule.shop_id,
+    rule.match_key AS sku_code,
+    BTRIM(product.canonical_name) AS internal_sku,
+    10 AS priority
+  FROM sku_channel_rules rule
+  JOIN product_catalog product
+    ON product.canonical_sku = CONCAT('aff:', rule.match_key)
+   AND product.status = 'active'
+  WHERE rule.match_key_type = 'sku_code'
+    AND BTRIM(COALESCE(product.canonical_name, '')) <> ''
+
+  UNION ALL
+
+  SELECT
+    rule.provider,
+    rule.shop_id,
+    rule.match_key AS sku_code,
+    matched.internal_sku,
+    20 AS priority
+  FROM sku_channel_rules rule
+  CROSS JOIN LATERAL (
+    SELECT
+      MIN(resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code)) AS internal_sku,
+      COUNT(DISTINCT resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code)) AS internal_count
+    FROM order_items oi
+    JOIN orders source_order ON source_order.id = oi.order_id
+    WHERE oi.sku_code = rule.match_key
+      AND (rule.provider = '' OR source_order.provider = rule.provider)
+      AND (rule.shop_id = '' OR source_order.shop_id = rule.shop_id)
+  ) matched
+  WHERE rule.match_key_type = 'sku_code'
+    AND matched.internal_count = 1
+    AND BTRIM(COALESCE(matched.internal_sku, '')) <> ''
+),
+ranked_legacy_candidates AS (
+  SELECT *,
+         ROW_NUMBER() OVER (
+           PARTITION BY provider, shop_id, sku_code
+           ORDER BY priority, internal_sku
+         ) AS candidate_rank
+  FROM legacy_candidates
+)
+INSERT INTO internal_sku_mappings (
+  provider, shop_id, match_type, match_value, internal_sku,
+  priority, enabled, source, note, created_at, updated_at
+)
+SELECT
+  provider, shop_id, 'sku_code', sku_code, internal_sku,
+  priority, TRUE, 'legacy-rule', '由旧 sku_code 渠道规则迁移', NOW(), NOW()
+FROM ranked_legacy_candidates
+WHERE candidate_rank = 1
+ON CONFLICT (provider, shop_id, match_type, match_value) DO UPDATE SET
+  internal_sku = EXCLUDED.internal_sku,
+  priority = LEAST(internal_sku_mappings.priority, EXCLUDED.priority),
+  enabled = TRUE,
+  source = EXCLUDED.source,
+  note = EXCLUDED.note,
+  updated_at = NOW();
+
+-- 可确定的旧 sku_code 渠道转换为 internal_sku_channels。
+WITH resolved_rules AS (
+  SELECT
+    rule.id,
+    rule.provider,
+    rule.shop_id,
+    rule.match_key,
+    rule.channel,
+    rule.priority,
+    COALESCE(
+      (
+        SELECT BTRIM(product.canonical_name)
+        FROM product_catalog product
+        WHERE product.canonical_sku = CONCAT('aff:', rule.match_key)
+          AND product.status = 'active'
+        ORDER BY product.id
+        LIMIT 1
+      ),
+      matched.internal_sku
+    ) AS internal_sku,
+    matched.canonical_product_id
+  FROM sku_channel_rules rule
+  LEFT JOIN LATERAL (
+    SELECT
+      MIN(resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code)) AS internal_sku,
+      COUNT(DISTINCT resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code)) AS internal_count,
+      MIN(oi.canonical_product_id) AS canonical_product_id
+    FROM order_items oi
+    JOIN orders source_order ON source_order.id = oi.order_id
+    WHERE oi.sku_code = rule.match_key
+      AND (rule.provider = '' OR source_order.provider = rule.provider)
+      AND (rule.shop_id = '' OR source_order.shop_id = rule.shop_id)
+  ) matched ON matched.internal_count = 1
+  WHERE rule.match_key_type = 'sku_code'
+    AND rule.enabled = TRUE
+)
+INSERT INTO internal_sku_channels (
+  internal_sku, channel, priority, is_fallback, enabled,
+  canonical_product_id, config_json, source, created_at, updated_at
+)
+SELECT
+  BTRIM(internal_sku),
+  channel,
+  MIN(priority),
+  FALSE,
+  TRUE,
+  MIN(canonical_product_id),
+  jsonb_build_object(
+    'legacySkuCode', MIN(match_key),
+    'legacyProvider', MIN(provider),
+    'legacyShopId', MIN(shop_id)
+  ),
+  'legacy-rule',
+  NOW(),
+  NOW()
+FROM resolved_rules
+WHERE BTRIM(COALESCE(internal_sku, '')) <> ''
+GROUP BY BTRIM(internal_sku), channel
+ON CONFLICT (internal_sku, channel) DO UPDATE SET
+  priority = LEAST(internal_sku_channels.priority, EXCLUDED.priority),
+  enabled = TRUE,
+  canonical_product_id = COALESCE(EXCLUDED.canonical_product_id, internal_sku_channels.canonical_product_id),
+  config_json = internal_sku_channels.config_json || EXCLUDED.config_json,
+  updated_at = NOW();
+
+-- 已转成内部 SKU 的旧规则停用；真正无法解析的规则保留 enabled，交给严格预检阻断。
+WITH resolved_rule_ids AS (
+  SELECT rule.id
+  FROM sku_channel_rules rule
+  LEFT JOIN LATERAL (
+    SELECT
+      MIN(resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code)) AS internal_sku,
+      COUNT(DISTINCT resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code)) AS internal_count
+    FROM order_items oi
+    JOIN orders source_order ON source_order.id = oi.order_id
+    WHERE oi.sku_code = rule.match_key
+      AND (rule.provider = '' OR source_order.provider = rule.provider)
+      AND (rule.shop_id = '' OR source_order.shop_id = rule.shop_id)
+  ) matched ON matched.internal_count = 1
+  WHERE rule.match_key_type = 'sku_code'
+    AND rule.enabled = TRUE
+    AND (
+      EXISTS (
+        SELECT 1 FROM product_catalog product
+        WHERE product.canonical_sku = CONCAT('aff:', rule.match_key)
+          AND product.status = 'active'
+      )
+      OR BTRIM(COALESCE(matched.internal_sku, '')) <> ''
+    )
+)
+UPDATE sku_channel_rules rule
+SET enabled = FALSE,
+    note = CONCAT(COALESCE(rule.note, ''), ' [已迁移 internal_sku]'),
+    updated_at = NOW()
+FROM resolved_rule_ids
+WHERE rule.id = resolved_rule_ids.id;
+
+-- 旧 internal_sku 规则已完整复制到统一表，不再参与运行时判定。
+UPDATE sku_channel_rules
+SET enabled = FALSE,
+    note = CONCAT(COALESCE(note, ''), ' [internal_sku_channels 已成为真源]'),
+    updated_at = NOW()
+WHERE match_key_type = 'internal_sku' AND enabled = TRUE;
+
+-- 回填适配参数，避免统一表只有“渠道名”而实际还要猜 affiliate SKU / 工单模板。
+WITH affiliate_config AS (
+  SELECT
+    BTRIM(product.canonical_name) AS internal_sku,
+    MIN(SUBSTRING(product.canonical_sku FROM 5)) AS affiliate_sku,
+    MIN(product.id) AS canonical_product_id
+  FROM product_catalog product
+  WHERE product.status = 'active' AND product.canonical_sku LIKE 'aff:%'
+  GROUP BY BTRIM(product.canonical_name)
+)
+UPDATE internal_sku_channels channel
+SET config_json = channel.config_json || jsonb_build_object(
+      'affiliateSku', affiliate_config.affiliate_sku
+    ),
+    canonical_product_id = COALESCE(channel.canonical_product_id, affiliate_config.canonical_product_id),
+    updated_at = NOW()
+FROM affiliate_config
+WHERE channel.channel = 'affiliate_dash'
+  AND BTRIM(channel.internal_sku) = affiliate_config.internal_sku;
+
+WITH work_order_config AS (
+  SELECT
+    BTRIM(product.canonical_name) AS internal_sku,
+    MIN(SUBSTRING(product.canonical_sku FROM 5)) AS work_rule_key,
+    MIN(product.id) AS canonical_product_id
+  FROM product_catalog product
+  WHERE product.status = 'active' AND product.canonical_sku LIKE 'tpl:%'
+  GROUP BY BTRIM(product.canonical_name)
+)
+UPDATE internal_sku_channels channel
+SET config_json = channel.config_json || jsonb_build_object(
+      'workRuleKey', work_order_config.work_rule_key
+    ),
+    canonical_product_id = COALESCE(channel.canonical_product_id, work_order_config.canonical_product_id),
+    updated_at = NOW()
+FROM work_order_config
+WHERE channel.channel = 'work_order'
+  AND BTRIM(channel.internal_sku) = work_order_config.internal_sku;
+
+WITH voucher_config AS (
+  SELECT
+    BTRIM(product.canonical_name) AS internal_sku,
+    MIN(SUBSTRING(product.canonical_sku FROM 5)) AS voucher_sku,
+    MIN(product.id) AS canonical_product_id
+  FROM product_catalog product
+  WHERE product.status = 'active' AND product.canonical_sku LIKE 'tpv:%'
+  GROUP BY BTRIM(product.canonical_name)
+)
+UPDATE internal_sku_channels channel
+SET config_json = channel.config_json || jsonb_build_object(
+      'voucherSku', voucher_config.voucher_sku
+    ),
+    canonical_product_id = COALESCE(channel.canonical_product_id, voucher_config.canonical_product_id),
+    updated_at = NOW()
+FROM voucher_config
+WHERE channel.channel = 'third_party_voucher'
+  AND BTRIM(channel.internal_sku) = voucher_config.internal_sku;
+
+-- 只要同一内部 SKU 还有非人工渠道，人工统一降为显式兜底。
 UPDATE internal_sku_channels manual_channel
 SET priority = GREATEST(manual_channel.priority, 90),
    is_fallback = TRUE,
@@ -98,128 +329,17 @@ WHERE manual_channel.channel = 'manual'
      AND other_channel.channel <> 'manual'
  );

-- 自动发货旧 sku_code 规则能通过 aff:<sku> 商品稳定定位内部 SKU，直接迁移。
-INSERT INTO internal_sku_channels (
-  internal_sku, channel, priority, is_fallback, enabled,
-  canonical_product_id, config_json, source, created_at, updated_at
-)
-SELECT
-  BTRIM(product.canonical_name),
-  'affiliate_dash',
-  MIN(rule.priority),
-  FALSE,
-  TRUE,
-  MIN(product.id),
-  jsonb_build_object('legacySkuCode', MIN(rule.match_key)),
-  'legacy-rule',
-  NOW(),
-  NOW()
-FROM sku_channel_rules rule
-JOIN product_catalog product
-  ON product.canonical_sku = CONCAT('aff:', rule.match_key)
- AND product.status = 'active'
-WHERE rule.match_key_type = 'sku_code'
-  AND rule.channel = 'affiliate_dash'
-  AND rule.enabled = TRUE
-  AND BTRIM(COALESCE(product.canonical_name, '')) <> ''
-GROUP BY BTRIM(product.canonical_name)
-ON CONFLICT (internal_sku, channel) DO UPDATE SET
-  priority = LEAST(internal_sku_channels.priority, EXCLUDED.priority),
-  enabled = TRUE,
-  canonical_product_id = COALESCE(EXCLUDED.canonical_product_id, internal_sku_channels.canonical_product_id),
-  config_json = internal_sku_channels.config_json || EXCLUDED.config_json,
-  updated_at = NOW();
-
-UPDATE sku_channel_rules rule
-SET enabled = FALSE,
-    note = CONCAT(COALESCE(rule.note, ''), ' [已迁移 internal_sku]'),
-    updated_at = NOW()
-WHERE rule.match_key_type = 'sku_code'
-  AND rule.channel = 'affiliate_dash'
-  AND rule.enabled = TRUE
-  AND EXISTS (
-    SELECT 1
-    FROM product_catalog product
-    WHERE product.canonical_sku = CONCAT('aff:', rule.match_key)
-      AND product.status = 'active'
-      AND BTRIM(COALESCE(product.canonical_name, '')) <> ''
-  );
-
-- 卡券旧规则按历史订单中唯一可解析的内部 SKU 迁移；歧义数据保持原规则不动。
-WITH resolved_legacy AS (
-  SELECT
-    rule.id,
-    matched.internal_sku,
-    MIN(rule.priority) AS priority
-  FROM sku_channel_rules rule
-  CROSS JOIN LATERAL (
-    SELECT
-      MIN(resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code)) AS internal_sku,
-      COUNT(DISTINCT resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code)) AS internal_count
-    FROM order_items oi
-    JOIN orders source_order ON source_order.id = oi.order_id
-    WHERE oi.sku_code = rule.match_key
-      AND (rule.provider = '' OR source_order.provider = rule.provider)
-      AND (rule.shop_id = '' OR source_order.shop_id = rule.shop_id)
-  ) matched
-  WHERE rule.match_key_type = 'sku_code'
-    AND rule.channel = 'third_party_voucher'
-    AND rule.enabled = TRUE
-    AND matched.internal_count = 1
-    AND BTRIM(COALESCE(matched.internal_sku, '')) <> ''
-  GROUP BY rule.id, matched.internal_sku
-)
-INSERT INTO internal_sku_channels (
-  internal_sku, channel, priority, is_fallback, enabled,
-  config_json, source, created_at, updated_at
-)
-SELECT
-  internal_sku,
-  'third_party_voucher',
-  priority,
-  FALSE,
-  TRUE,
-  '{}'::jsonb,
-  'legacy-rule',
-  NOW(),
-  NOW()
-FROM resolved_legacy
-ON CONFLICT (internal_sku, channel) DO UPDATE SET
-  priority = LEAST(internal_sku_channels.priority, EXCLUDED.priority),
-  enabled = TRUE,
-  updated_at = NOW();
-
-WITH resolved_legacy AS (
-  SELECT rule.id
-  FROM sku_channel_rules rule
-  CROSS JOIN LATERAL (
-    SELECT
-      MIN(resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code)) AS internal_sku,
-      COUNT(DISTINCT resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code)) AS internal_count
-    FROM order_items oi
-    JOIN orders source_order ON source_order.id = oi.order_id
-    WHERE oi.sku_code = rule.match_key
-      AND (rule.provider = '' OR source_order.provider = rule.provider)
-      AND (rule.shop_id = '' OR source_order.shop_id = rule.shop_id)
-  ) matched
-  WHERE rule.match_key_type = 'sku_code'
-    AND rule.channel = 'third_party_voucher'
-    AND rule.enabled = TRUE
-    AND matched.internal_count = 1
-    AND BTRIM(COALESCE(matched.internal_sku, '')) <> ''
-)
-UPDATE sku_channel_rules rule
-SET enabled = FALSE,
-    note = CONCAT(COALESCE(rule.note, ''), ' [已迁移 internal_sku]'),
-    updated_at = NOW()
-FROM resolved_legacy
-WHERE rule.id = resolved_legacy.id;
-
 ALTER TABLE internal_sku_channels
  DROP CONSTRAINT IF EXISTS chk_internal_sku_channels_priority_nonnegative;
 ALTER TABLE internal_sku_channels
  ADD CONSTRAINT chk_internal_sku_channels_priority_nonnegative CHECK (priority >= 0);

+ALTER TABLE internal_sku_channels
+  DROP CONSTRAINT IF EXISTS chk_internal_sku_channels_config_object;
+ALTER TABLE internal_sku_channels
+  ADD CONSTRAINT chk_internal_sku_channels_config_object
+  CHECK (jsonb_typeof(config_json) = 'object');
+
 CREATE INDEX IF NOT EXISTS idx_internal_sku_channels_fallback
  ON internal_sku_channels (internal_sku, priority, channel)
  WHERE enabled = TRUE AND is_fallback = TRUE;
@@ -23,3 +23,9 @@ END $$;
 CREATE UNIQUE INDEX IF NOT EXISTS uq_internal_sku_channels_single_primary
  ON internal_sku_channels (internal_sku)
  WHERE enabled = TRUE AND is_fallback = FALSE;
+
+ALTER TABLE internal_sku_channels
+  DROP CONSTRAINT IF EXISTS chk_internal_sku_channels_nonempty_trimmed;
+ALTER TABLE internal_sku_channels
+  ADD CONSTRAINT chk_internal_sku_channels_nonempty_trimmed
+  CHECK (internal_sku <> '' AND internal_sku = BTRIM(internal_sku));
@@ -374,8 +374,8 @@ router.post(
    errorMessage: '补齐卡券渠道规则失败',
    scope: '[admin/cost-center/voucher-products/ensure-rules]',
    audit: (_req, data) => ({
-      action: 'voucher_channel_rules_ensured',
-      targetType: 'sku_channel_rules',
+      action: 'voucher_internal_channels_ensured',
+      targetType: 'internal_sku_channels',
      targetId: '',
      data: data as Record<string, unknown>,
    }),
@@ -14,6 +14,7 @@ import {
  deleteChannelRule,
  listChannelRules,
  listPendingChannelGroups,
+  switchChannelForOrderItems,
  unlockChannel,
  upsertChannelRule,
 } from '../../services/fulfillment/channel-assign-service.js'
@@ -143,6 +144,35 @@ router.post(
  ),
 )

+router.post(
+  '/fulfillment/switch',
+  requireAdminPermission('finance.config'),
+  createJsonHandler(
+    (req) => {
+      const body = bodyOf(req)
+      return switchChannelForOrderItems({
+        orderItemIds: Array.isArray(body.orderItemIds)
+          ? body.orderItemIds.map((value: unknown) => Number(value))
+          : [Number(body.orderItemId)],
+        channel: String(body.channel || ''),
+        adminUserId: req.adminSession?.userId ?? null,
+        note: String(body.note || ''),
+      })
+    },
+    {
+      successMessage: '履约渠道已切换',
+      errorMessage: '切换履约渠道失败',
+      scope: '[admin/fulfillment/switch]',
+      audit: (req, data) => ({
+        action: 'fulfillment.switch_channel',
+        targetType: 'order_item',
+        targetId: String(bodyOf(req).orderItemId || ''),
+        data: { payload: bodyOf(req), result: data as Record<string, unknown> },
+      }),
+    },
+  ),
+)
+
 router.get(
  '/fulfillment/rules',
  requireAdminPermission('finance.view'),
@@ -170,9 +200,7 @@ router.post(
      const body = bodyOf(req)
      return upsertChannelRule({
        ...(body.id ? { id: Number(body.id) } : {}),
-        provider: String(body.provider || ''),
-        shopId: String(body.shopId || ''),
-        matchKeyType: String(body.matchKeyType || 'sku_code'),
+        matchKeyType: String(body.matchKeyType || 'internal_sku'),
        matchKey: String(body.matchKey || ''),
        channel: String(body.channel || ''),
        priority: Number(body.priority),
@@ -188,7 +216,7 @@ router.post(
      scope: '[admin/fulfillment/rules]',
      audit: (req, data) => ({
        action: 'fulfillment.upsert_rule',
-        targetType: 'sku_channel_rule',
+        targetType: 'internal_sku_channel',
        targetId: String((data as { id?: number })?.id || ''),
        data: { payload: bodyOf(req) },
      }),
@@ -178,6 +178,7 @@ export async function backfillAdminAffiliateDashSkus() {
          ) AS sku_code
        FROM fulfillment_tasks ft
        WHERE ft.executor_key = 'affiliate_dash'
+          AND ft.task_status NOT IN ('closed','expired','failed','manual_review','retry_pending','cancelled')
          AND ft.created_at >= NOW() - INTERVAL '30 days'
          AND btrim(COALESCE(ft.context_json->>'skuCode', '')) <> ''
        ORDER BY ft.order_item_id, ft.created_at DESC, ft.id DESC
@@ -202,6 +203,7 @@ export async function backfillAdminAffiliateDashSkus() {
          SELECT COUNT(DISTINCT ft.order_item_id)::int
          FROM fulfillment_tasks ft
          WHERE ft.executor_key = 'affiliate_dash'
+            AND ft.task_status NOT IN ('closed','expired','failed','manual_review','retry_pending','cancelled')
            AND ft.created_at >= NOW() - INTERVAL '30 days'
        ) AS affiliate_item_count,
        (SELECT COUNT(*)::int FROM target) AS matched_count,
@@ -245,6 +247,7 @@ export async function backfillAdminAffiliateDashSkus() {
          SELECT 1 FROM fulfillment_tasks ft
          WHERE ft.order_item_id = oi.id
            AND ft.executor_key = 'affiliate_dash'
+            AND ft.task_status NOT IN ('closed','expired','failed','manual_review','retry_pending','cancelled')
            AND ft.created_at >= NOW() - INTERVAL '30 days'
        )
    `,
@@ -408,7 +408,6 @@ export async function deleteManualFulfillmentConfig(skuKey: string) {
    `DELETE FROM manual_fulfillment_configs WHERE sku_key = $1 RETURNING id`,
    [key],
  )
-  await query(`DELETE FROM sku_channel_rules WHERE source = 'manual' AND match_key = $1`, [key])
  await upsertInternalSkuChannel({
    internalSku: key,
    channel: 'manual',
@@ -722,9 +721,7 @@ function requiredRow<T>(row: T | undefined, message: string): T {
 }

 /**
- * 人工配置 → 渠道规则（内部 SKU 精确匹配）。
- *
- * 幂等：按唯一键 (provider, shop, match_key_type, match_key, channel) upsert。
+ * 人工配置 → internal_sku_channels（内部 SKU 精确匹配）。
 */
 async function syncManualChannelRule(config: {
  id: number
@@ -757,29 +754,4 @@ async function syncManualChannelRule(config: {
    configJson: { manualConfigId: config.id, note: String(config.note || '').slice(0, 200) },
    replacePriority: true,
  })
-
-  await query(
-    `
-      INSERT INTO sku_channel_rules (
-        provider, shop_id, match_key_type, match_key, channel, priority,
-        source, source_ref_id, enabled, note, created_at, updated_at
-      )
-      VALUES ('', '', 'internal_sku', $1, 'manual', $5, 'manual', $2, $3, $4, NOW(), NOW())
-      ON CONFLICT (provider, shop_id, match_key_type, match_key, channel)
-      DO UPDATE SET
-        enabled = EXCLUDED.enabled,
-        priority = EXCLUDED.priority,
-        source = 'manual',
-        source_ref_id = EXCLUDED.source_ref_id,
-        note = EXCLUDED.note,
-        updated_at = NOW()
-    `,
-    [
-      skuKey,
-      config.id,
-      config.enabled !== false,
-      String(config.note || '').slice(0, 200),
-      hasOtherChannel ? 90 : 10,
-    ],
-  )
 }
@@ -198,23 +198,10 @@ async function ensureVoucherProductForItem(item: ItemRow): Promise<number | null
    priority: 10,
    isFallback: false,
    canonicalProductId: productId,
+    configJson: { voucherSku: internalSku },
    source: 'auto',
  }).catch(() => undefined)

-  // 同一内部中文 SKU 跨平台码统一分流到卡券渠道。
-  await query(
-    `
-      INSERT INTO sku_channel_rules (
-        provider, shop_id, match_key_type, match_key, channel, priority,
-        source, enabled, note, created_at, updated_at
-      )
-      VALUES ('', '', 'internal_sku', $1, 'third_party_voucher', 10, 'voucher', TRUE,
-              '卡券商品自动建商品后生成', NOW(), NOW())
-      ON CONFLICT (provider, shop_id, match_key_type, match_key, channel) DO NOTHING
-    `,
-    [internalSku],
-  )
-
  return productId
 }

@@ -31,6 +31,7 @@ export async function runAffiliateFallback(
    internal_sku: string
    sku_name: string
    config_id: number | null
+    config_product_id: number | null
    basis_date: string
  }>(
    `
@@ -42,6 +43,7 @@ export async function runAffiliateFallback(
        resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code) AS internal_sku,
        oi.sku_name,
        cfg.id AS config_id,
+        cfg.canonical_product_id AS config_product_id,
        (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date::text AS basis_date
      FROM order_items oi
      JOIN orders o ON o.id = oi.order_id
@@ -95,6 +97,7 @@ export async function runAffiliateFallback(
            channel_switch_reason = 'affiliate_failed',
            channel_switched_at = NOW(),
            fulfillment_config_id = COALESCE(fulfillment_config_id, $2),
+            canonical_product_id = $3,
            updated_at = NOW()
          WHERE id = $1
            AND fulfillment_status = 'assigned'
@@ -102,7 +105,7 @@ export async function runAffiliateFallback(
            AND fulfillment_locked = FALSE
          RETURNING id
        `,
-        [row.id, row.config_id],
+        [row.id, row.config_id, row.config_product_id],
      )
      if ((result.rowCount || 0) === 0) return false

@@ -2,9 +2,10 @@ import { query } from '../../db/client.js'
 import type { OrderItemRow, OrderRow } from '../../types/repository/rows.js'
 import { syncDeliveryTasksForOrder } from '../order/delivery-task-service.js'
 import { syncWorkerOrdersForSourceOrder } from '../worker-platform/sync-worker-orders-service.js'
-import type { FulfillmentChannel } from './channel-types.js'
+import { INACTIVE_CARRIER_TASK_STATUSES, type FulfillmentChannel } from './channel-types.js'
 import { applyIntakeChannelDecision } from './intake-channel-decision-service.js'
 import type { ResolvedFulfillmentItem } from './product-resolution-service.js'
+import { resolveInternalSkuKey } from './internal-sku-resolver.js'

 function parseObject(value: unknown): Record<string, unknown> {
  if (value && typeof value === 'object' && !Array.isArray(value)) {
@@ -29,7 +30,9 @@ export type EnsureCarrierResult = {

 async function hasWorkOrder(orderItemId: number): Promise<boolean> {
  const result = await query<{ exists: boolean }>(
-    `SELECT EXISTS (SELECT 1 FROM work_orders WHERE order_item_id = $1) AS exists`,
+    `SELECT EXISTS (
+       SELECT 1 FROM work_orders WHERE order_item_id = $1 AND status <> 'cancelled'
+     ) AS exists`,
    [orderItemId],
  )
  return result.rows[0]?.exists === true
@@ -40,8 +43,9 @@ async function hasAffiliateTask(orderItemId: number): Promise<boolean> {
    `SELECT EXISTS (
       SELECT 1 FROM fulfillment_tasks
       WHERE order_item_id = $1 AND executor_key = 'affiliate_dash'
+         AND task_status <> ALL($2::text[])
     ) AS exists`,
-    [orderItemId],
+    [orderItemId, INACTIVE_CARRIER_TASK_STATUSES],
  )
  return result.rows[0]?.exists === true
 }
@@ -110,6 +114,23 @@ export async function ensureCarrierForOrderItem(input: {
    if (await hasWorkOrder(Number(item.id))) {
      return { ready: true, created: false, reason: 'work_order_exists', message: '' }
    }
+    const internalSku = resolveInternalSkuKey({
+      manualSkuKey: item.manual_sku_key,
+      skuCode: item.sku_code,
+      skuName: item.sku_name,
+      spec: parseObject(item.spec_json),
+      snapshot: parseObject(item.item_snapshot_json),
+    })
+    const channelConfig = await query<{ config_json: Record<string, unknown> }>(
+      `SELECT config_json
+       FROM internal_sku_channels
+       WHERE btrim(internal_sku) = $1 AND channel = 'work_order' AND enabled = TRUE
+       LIMIT 1`,
+      [internalSku],
+    )
+    const preferredWorkRuleKey = String(
+      channelConfig.rows[0]?.config_json?.workRuleKey || '',
+    ).trim()
    const snapshot = parseObject(item.item_snapshot_json)
    const classification = parseObject(snapshot.classification)
    const route = parseObject(snapshot.fulfillmentRoute)
@@ -130,6 +151,7 @@ export async function ensureCarrierForOrderItem(input: {
      source: 'carrier_creation',
      autoOnly: true,
      skipChannelAssign: true,
+      ...(preferredWorkRuleKey ? { preferredWorkRuleKey } : {}),
    })
    const ready = (await hasWorkOrder(Number(item.id))) || result.created.length > 0
    if (ready) {
@@ -2,7 +2,11 @@ import { query } from '../../db/client.js'
 import { getOrderById } from '../../repositories/order-repo.js'
 import type { OrderItemRow } from '../../types/repository/rows.js'
 import { logWarn } from '../../utils/logger.js'
-import { FULFILLMENT_CHANNELS, type FulfillmentChannel } from './channel-types.js'
+import {
+  FULFILLMENT_CHANNELS,
+  INACTIVE_CARRIER_TASK_STATUSES,
+  type FulfillmentChannel,
+} from './channel-types.js'
 import { ensureCarrierForOrderItem } from './carrier-creation-service.js'
 import {
  planInternalSkuChannels,
@@ -21,6 +25,7 @@ function normalizeChannels(value: unknown): InternalSkuChannelRow[] {
  return value
    .filter((row): row is Record<string, unknown> => Boolean(row) && typeof row === 'object')
    .map((row) => ({
+      id: Number(row.id || 0),
      internal_sku: String(row.internal_sku || '').trim(),
      channel: String(row.channel || '') as FulfillmentChannel,
      priority: Number(row.priority || 100),
@@ -33,6 +38,9 @@ function normalizeChannels(value: unknown): InternalSkuChannelRow[] {
          ? (row.config_json as Record<string, unknown>)
          : {},
      source: String(row.source || ''),
+      note: String(row.note || ''),
+      created_by_admin_user_id:
+        row.created_by_admin_user_id == null ? null : Number(row.created_by_admin_user_id),
    }))
    .filter((row) => FULFILLMENT_CHANNELS.includes(row.channel))
 }
@@ -49,10 +57,14 @@ async function loadCandidateItems(input: {
  const statusFilter = input.assignedMissingCarrier
    ? `oi.fulfillment_status = 'assigned'
       AND oi.fulfillment_channel IN ('affiliate_dash', 'work_order')
-       AND NOT EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id)
+       AND NOT EXISTS (
+         SELECT 1 FROM work_orders wo
+         WHERE wo.order_item_id = oi.id AND wo.status <> 'cancelled'
+       )
       AND NOT EXISTS (
         SELECT 1 FROM fulfillment_tasks ft
         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`

@@ -70,6 +82,7 @@ async function loadCandidateItems(input: {
        SELECT COALESCE(
          jsonb_agg(
            jsonb_build_object(
+              'id', channel.id,
              'internal_sku', channel.internal_sku,
              'channel', channel.channel,
              'priority', channel.priority,
@@ -78,6 +91,8 @@ async function loadCandidateItems(input: {
              'canonical_product_id', channel.canonical_product_id,
              'config_json', channel.config_json,
              'source', channel.source
+              ,'note', channel.note
+              ,'created_by_admin_user_id', channel.created_by_admin_user_id
            )
            ORDER BY channel.priority, channel.channel
          ),
@@ -138,6 +153,8 @@ async function selectSafeFallback(input: {
          fulfillment_reason = $3,
          channel_switch_reason = $3,
          channel_switched_at = NOW(),
+          canonical_product_id = NULL,
+          fulfillment_config_id = NULL,
          fulfillment_pending_reason = NULL,
          fulfillment_pending_detail = NULL,
          updated_at = NOW()
@@ -328,10 +345,14 @@ export async function reconcilePendingCarriers(
      FROM order_items oi
      WHERE oi.fulfillment_status IN ('pending', 'assigned')
        AND oi.fulfillment_channel IN ('affiliate_dash', 'work_order')
-        AND NOT EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id)
+        AND NOT EXISTS (
+          SELECT 1 FROM work_orders wo
+          WHERE wo.order_item_id = oi.id AND wo.status <> 'cancelled'
+        )
        AND NOT EXISTS (
          SELECT 1 FROM fulfillment_tasks ft
          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(', ')})
        )
    `,
  )
@@ -29,6 +29,7 @@ function buildItem(patch: Partial<ChannelCandidateItem> = {}): ChannelCandidateI

 function buildChannel(patch: Partial<InternalSkuChannelRow> = {}): InternalSkuChannelRow {
  return {
+    id: 1,
    internal_sku: '套装-测试',
    channel: 'manual',
    priority: 10,
@@ -37,6 +38,8 @@ function buildChannel(patch: Partial<InternalSkuChannelRow> = {}): InternalSkuCh
    canonical_product_id: null,
    config_json: {},
    source: 'test',
+    note: '',
+    created_by_admin_user_id: null,
    ...patch,
  }
 }
@@ -1,5 +1,8 @@
-import { query } from '../../db/client.js'
+import { query, withTransaction } from '../../db/client.js'
+import { closeClaimTokensByTaskId } from '../../repositories/claim-token-repo.js'
 import { getOrderById } from '../../repositories/order-repo.js'
+import { createTaskEvent } from '../../repositories/task-event-repo.js'
+import { createWorkOrderEvent } from '../../repositories/worker-platform/work-order-event-repo.js'
 import type { OrderItemRow } from '../../types/repository/rows.js'
 import { createHttpError } from '../../utils/http.js'
 import { nowIso } from '../../utils/time.js'
@@ -10,10 +13,10 @@ import { normalizeProductsForItems } from '../cost-center/product-normalize-serv
 import { orderItemBasisDates, refreshFactsForDates } from '../finance/refresh-facts.js'
 import {
  FULFILLMENT_CHANNELS,
+  INACTIVE_CARRIER_TASK_STATUSES,
  MATCH_KEY_TYPES,
  RULE_SOURCES,
  isFulfillmentChannel,
-  isMatchKeyType,
  isRuleSource,
  type FulfillmentChannel,
  type FulfillmentSource,
@@ -21,13 +24,14 @@ import {
  type RuleSource,
 } from './channel-types.js'
 import {
-  disableInternalSkuChannel,
+  deleteInternalSkuChannelById,
  planInternalSkuChannels,
  setInternalSkuPrimaryChannel,
  upsertInternalSkuChannel,
  type InternalSkuChannelRow,
 } from './internal-sku-channel-service.js'
 import { ensureCarrierForOrderItem } from './carrier-creation-service.js'
+import { TASK_STATUS } from '../../domain/task-status.js'

 /**
 * 管道 A：渠道判定。
@@ -45,6 +49,7 @@ import { ensureCarrierForOrderItem } from './carrier-creation-service.js'

 /** 数据库函数 `resolve_internal_sku` 是订单项内部 SKU 的唯一 SQL 口径。 */
 const INTERNAL_SKU_SQL = `resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code)`
+const ACTIVE_CARRIER_TASK_SQL = `task.task_status NOT IN (${INACTIVE_CARRIER_TASK_STATUSES.map((status) => `'${status}'`).join(', ')})`

 export type ChannelRule = {
  id: number
@@ -304,15 +309,17 @@ export async function assignPendingOrderItemChannels(
          (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date::text
            AS basis_date,
          EXISTS (
-            SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id
+            SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id AND wo.status <> 'cancelled'
          ) AS has_work_order,
          EXISTS (
-            SELECT 1 FROM fulfillment_tasks ft
-            WHERE ft.order_item_id = oi.id AND ft.executor_key = 'affiliate_dash'
+            SELECT 1 FROM fulfillment_tasks task
+            WHERE task.order_item_id = oi.id AND task.executor_key = 'affiliate_dash'
+              AND ${ACTIVE_CARRIER_TASK_SQL}
          ) AS has_affiliate_task,
          EXISTS (
-            SELECT 1 FROM fulfillment_tasks ft
-            WHERE ft.order_item_id = oi.id AND ft.executor_key = 'manual_dispatch'
+            SELECT 1 FROM fulfillment_tasks task
+            WHERE task.order_item_id = oi.id AND task.executor_key = 'manual_dispatch'
+              AND ${ACTIVE_CARRIER_TASK_SQL}
          ) AS has_manual_task
        FROM order_items oi
        JOIN orders o ON o.id = oi.order_id
@@ -577,6 +584,205 @@ async function assignOrderItemsAfterCarrierReady(input: {
  return { updated: Number(final.rows[0]?.count || assignedIds.length), carrierFailed }
 }

+const WORK_ORDER_SWITCH_BLOCKED = new Set([
+  'open',
+  'in_progress',
+  'pending_acceptance',
+  'problem',
+  'accepted',
+])
+const TASK_SWITCH_BLOCKED = new Set<string>([
+  TASK_STATUS.REDEEMING,
+  TASK_STATUS.DISPATCHED_PENDING_RETURN,
+  TASK_STATUS.COMPLETED,
+  TASK_STATUS.REDEEMED,
+])
+
+/**
+ * 显式切换渠道：先取消尚未实际履约的旧载体，再创建新载体。
+ * 已经成功或进入实际交付阶段的订单不允许自动切换。
+ */
+export async function switchChannelForOrderItems(input: {
+  orderItemIds: number[]
+  channel: string
+  adminUserId: number | null
+  note?: string
+}): Promise<{
+  switched: number
+  failed: number
+  errors: Array<{ orderItemId: number; reason: string }>
+}> {
+  if (!isFulfillmentChannel(input.channel)) {
+    throw createHttpError('渠道无效', { statusCode: 400, errorCode: 'invalid_channel' })
+  }
+  const ids = [
+    ...new Set(input.orderItemIds.map(Number).filter((id) => Number.isInteger(id) && id > 0)),
+  ]
+  if (ids.length === 0) {
+    throw createHttpError('请选择订单项', { statusCode: 400, errorCode: 'order_item_required' })
+  }
+
+  const items = (
+    await query<OrderItemRow>(`SELECT * FROM order_items WHERE id = ANY($1::bigint[])`, [ids])
+  ).rows
+  const orderIds = [...new Set(items.map((item) => Number(item.order_id)))]
+  const orders = new Map(
+    (await Promise.all(orderIds.map((orderId) => getOrderById(orderId))))
+      .filter((order): order is NonNullable<typeof order> => Boolean(order))
+      .map((order) => [Number(order.id), order]),
+  )
+  const readyIds: number[] = []
+  const errors: Array<{ orderItemId: number; reason: string }> = []
+
+  for (const item of items) {
+    const order = orders.get(Number(item.order_id))
+    if (!order) {
+      errors.push({ orderItemId: Number(item.id), reason: 'order_not_found' })
+      continue
+    }
+    if (item.fulfillment_status === 'archived') {
+      errors.push({ orderItemId: Number(item.id), reason: 'archived' })
+      continue
+    }
+    if (item.fulfillment_channel === input.channel) {
+      readyIds.push(Number(item.id))
+      continue
+    }
+
+    try {
+      const workOrders = await query<{ id: number; status: string }>(
+        `SELECT id, status FROM work_orders
+         WHERE order_item_id = $1 AND status <> 'cancelled'`,
+        [item.id],
+      )
+      const blockedWorkOrder = workOrders.rows.find((row) =>
+        WORK_ORDER_SWITCH_BLOCKED.has(String(row.status)),
+      )
+      if (blockedWorkOrder) {
+        throw createHttpError('旧接单工单已进入履约阶段，不能自动切换渠道', {
+          statusCode: 409,
+          errorCode: 'channel_switch_work_order_active',
+        })
+      }
+      const tasks = await query<{ id: number; task_status: string }>(
+        `SELECT id, task_status FROM fulfillment_tasks WHERE order_item_id = $1`,
+        [item.id],
+      )
+      const blockedTask = tasks.rows.find((row) => TASK_SWITCH_BLOCKED.has(String(row.task_status)))
+      if (blockedTask) {
+        throw createHttpError('旧履约任务已进入实际交付阶段，不能自动切换渠道', {
+          statusCode: 409,
+          errorCode: 'channel_switch_task_active',
+        })
+      }
+
+      const now = nowIso()
+      await withTransaction(async (client) => {
+        const locked = await client.query<OrderItemRow>(
+          `SELECT * FROM order_items WHERE id = $1 FOR UPDATE`,
+          [item.id],
+        )
+        if (!locked.rows[0] || locked.rows[0].fulfillment_status === 'archived') {
+          throw createHttpError('订单项状态已变化', {
+            statusCode: 409,
+            errorCode: 'channel_switch_order_item_changed',
+          })
+        }
+        for (const workOrder of workOrders.rows) {
+          await client.query(
+            `UPDATE work_orders
+             SET status = 'cancelled', assigned_worker_id = NULL, assigned_at = NULL,
+                 deadline_at = NULL, published_at = NULL, updated_at = $2
+             WHERE id = $1 AND status <> 'cancelled'`,
+            [workOrder.id, now],
+          )
+        }
+        for (const task of tasks.rows) {
+          await client.query(
+            `UPDATE fulfillment_tasks
+             SET task_status = $2, delivery_status = 'closed', result_code = 'closed_by_channel_switch',
+                 result_message = $3, user_action_status = 'closed', claim_expires_at = $4,
+                 updated_at = $4
+             WHERE id = $1`,
+            [task.id, TASK_STATUS.CLOSED, '履约渠道已切换，原任务关闭', now],
+          )
+        }
+        await client.query(
+          `UPDATE cost_snapshots
+           SET superseded_at = $2
+           WHERE order_item_id = $1 AND superseded_at IS NULL AND reversed_at IS NULL`,
+          [item.id, now],
+        )
+        await client.query(
+          `UPDATE order_items
+           SET original_channel = COALESCE(original_channel, fulfillment_channel),
+               fulfillment_channel = NULL,
+               fulfillment_status = 'pending',
+               fulfillment_source = 'operator_assigned',
+               fulfillment_locked = FALSE,
+               fulfillment_reason = $2,
+               channel_switch_reason = 'operator',
+               channel_switched_at = $3,
+               canonical_product_id = NULL,
+               fulfillment_config_id = NULL,
+               fulfillment_pending_reason = NULL,
+               fulfillment_pending_detail = NULL,
+               updated_at = $3
+           WHERE id = $1`,
+          [item.id, String(input.note || '运营切换履约渠道').slice(0, 200), now],
+        )
+      })
+
+      for (const workOrder of workOrders.rows) {
+        await createWorkOrderEvent({
+          workOrderId: Number(workOrder.id),
+          actorType: 'admin',
+          actorId: String(input.adminUserId || ''),
+          eventType: 'cancelled_by_channel_switch',
+          fromStatus: workOrder.status,
+          toStatus: 'cancelled',
+          payloadJson: JSON.stringify({ channel: input.channel, note: input.note || '' }),
+          now,
+        })
+      }
+      for (const task of tasks.rows) {
+        await closeClaimTokensByTaskId(Number(task.id), now)
+        await createTaskEvent(
+          Number(task.id),
+          'task_closed_by_channel_switch',
+          { channel: input.channel, note: input.note || '', previousStatus: task.task_status },
+          now,
+        )
+      }
+      readyIds.push(Number(item.id))
+    } catch (error) {
+      errors.push({
+        orderItemId: Number(item.id),
+        reason: error instanceof Error ? error.message : String(error),
+      })
+    }
+  }
+
+  const assigned = await assignOrderItemsAfterCarrierReady({
+    orderItemIds: readyIds,
+    channel: input.channel,
+    source: 'operator_assigned',
+    lock: true,
+    note: input.note,
+  })
+  await query(
+    `UPDATE order_items
+     SET fulfillment_decided_by = $2, fulfillment_decided_at = NOW()
+     WHERE id = ANY($1::bigint[])`,
+    [readyIds, input.adminUserId],
+  )
+  return {
+    switched: assigned.updated,
+    failed: errors.length + assigned.carrierFailed,
+    errors,
+  }
+}
+
 /** 运营指定渠道：先保证真实载体，成功后才写入并锁定。 */
 export async function assignChannelToOrderItems(input: {
  orderItemIds: number[]
@@ -700,8 +906,8 @@ export async function assignChannelByInternalSku(input: {
  internalSku: string
  channel: string
  adminUserId: number | null
-  note?: string
-  createRule?: boolean
+  note?: string | undefined
+  createRule?: boolean | undefined
 }): Promise<{ updated: number; ruleCreated: boolean; carrierFailed: number }> {
  if (!isFulfillmentChannel(input.channel)) {
    throw createHttpError('渠道无效', { statusCode: 400, errorCode: 'invalid_channel' })
@@ -713,6 +919,12 @@ export async function assignChannelByInternalSku(input: {
      errorCode: 'internal_sku_required',
    })
  }
+  if (input.createRule === false) {
+    throw createHttpError('内部 SKU 渠道配置必须持久化，不能只改当前订单', {
+      statusCode: 400,
+      errorCode: 'persistent_channel_rule_required',
+    })
+  }

  const idsResult = await query<{ id: number }>(
    `
@@ -743,24 +955,11 @@ export async function assignChannelByInternalSku(input: {
    canonicalProductId: product.rows[0]?.id == null ? null : Number(product.rows[0].id),
    source: 'operator',
    configJson: { note: String(input.note || '').slice(0, 200) },
+    note: input.note,
+    adminUserId: input.adminUserId,
  })

-  let ruleCreated = false
-  if (input.createRule !== false) {
-    const rule = await upsertChannelRule({
-      provider: input.provider,
-      shopId: input.shopId,
-      matchKeyType: 'internal_sku',
-      matchKey: internalSku,
-      channel: input.channel,
-      priority: 10,
-      source: 'operator',
-      enabled: true,
-      note: `运营在待分类队列指定：${String(input.note || '').slice(0, 120)}`,
-      adminUserId: input.adminUserId,
-    })
-    ruleCreated = rule.id > 0
-  }
+  const ruleCreated = true

  const ids = idsResult.rows.map((row) => Number(row.id))
  const applied = await assignOrderItemsAfterCarrierReady({
@@ -787,8 +986,8 @@ export async function assignChannelBySku(input: {
  skuCode: string
  channel: string
  adminUserId: number | null
-  note?: string
-  createRule?: boolean
+  note?: string | undefined
+  createRule?: boolean | undefined
 }): Promise<{ updated: number; ruleCreated: boolean; carrierFailed: number }> {
  if (!isFulfillmentChannel(input.channel)) {
    throw createHttpError('渠道无效', { statusCode: 400, errorCode: 'invalid_channel' })
@@ -813,48 +1012,24 @@ export async function assignChannelBySku(input: {
  const internalSkus = [
    ...new Set(idsResult.rows.map((row) => String(row.internal_sku || '').trim()).filter(Boolean)),
  ]
-  if (internalSkus.length === 1) {
-    await setInternalSkuPrimaryChannel({
-      internalSku: internalSkus[0]!,
-      channel: input.channel,
-      source: 'operator',
-      configJson: { note: String(input.note || '').slice(0, 200) },
+  if (internalSkus.length === 0) {
+    return { updated: 0, ruleCreated: false, carrierFailed: 0 }
+  }
+  if (internalSkus.length > 1) {
+    throw createHttpError('同一平台 SKU 对应多个内部 SKU，请按内部 SKU 指定渠道', {
+      statusCode: 409,
+      errorCode: 'sku_code_ambiguous_internal_sku',
    })
  }
-
-  // 默认同时落一条渠道规则：否则同 SKU 的下一单还是会进待分类。
-  let ruleCreated = false
-  if (input.createRule !== false) {
-    const rule = await upsertChannelRule({
-      provider: input.provider,
-      shopId: input.shopId,
-      matchKeyType: 'sku_code',
-      matchKey: input.skuCode,
-      channel: input.channel,
-      priority: 10,
-      source: 'operator',
-      enabled: true,
-      note: `运营在待分类队列指定：${String(input.note || '').slice(0, 120)}`,
-      adminUserId: input.adminUserId,
-    })
-    ruleCreated = rule.id > 0
-  }
-
-  const ids = idsResult.rows.map((row) => Number(row.id))
-  const applied = await assignOrderItemsAfterCarrierReady({
-    orderItemIds: ids,
+  return assignChannelByInternalSku({
+    provider: input.provider,
+    shopId: input.shopId,
+    internalSku: internalSkus[0]!,
    channel: input.channel,
-    source: 'operator_assigned',
-    lock: true,
+    adminUserId: input.adminUserId,
    note: input.note,
+    createRule: input.createRule,
  })
-  await query(
-    `UPDATE order_items
-     SET fulfillment_decided_by = $2, fulfillment_decided_at = NOW()
-     WHERE id = ANY($1::bigint[])`,
-    [ids, input.adminUserId],
-  )
-  return { ...applied, ruleCreated }
 }

 export async function listChannelRules(
@@ -877,19 +1052,26 @@ export async function listChannelRules(
  const rows = await query<ChannelRule>(
    `
      SELECT id, provider, shop_id, match_key_type, match_key, channel, priority, enabled
-      FROM sku_channel_rules
+      FROM (
+        SELECT id, ''::text AS provider, ''::text AS shop_id,
+               'internal_sku'::text AS match_key_type, internal_sku AS match_key,
+               channel, priority, enabled
+        FROM internal_sku_channels
+      ) internal_rules
      ${where}
      ORDER BY priority, channel, match_key
    `,
    params,
  )
-  return rows.rows
+  return rows.rows.map((row) => ({
+    ...row,
+    id: Number(row.id),
+    priority: Number(row.priority),
+  }))
 }

 export async function upsertChannelRule(input: {
  id?: number
-  provider?: string
-  shopId?: string
  matchKeyType: string
  matchKey: string
  channel: string
@@ -899,8 +1081,8 @@ export async function upsertChannelRule(input: {
  note?: string
  adminUserId: number | null
 }): Promise<{ id: number }> {
-  if (!isMatchKeyType(input.matchKeyType)) {
-    throw createHttpError('匹配键类型无效', {
+  if (input.matchKeyType !== 'internal_sku') {
+    throw createHttpError('渠道配置只接受内部 SKU；旧 sku_code 规则仅用于一次性迁移', {
      statusCode: 400,
      errorCode: 'invalid_match_key_type',
    })
@@ -914,48 +1096,19 @@ export async function upsertChannelRule(input: {
    throw createHttpError('匹配键不能为空', { statusCode: 400, errorCode: 'match_key_required' })
  }

-  const result = await query<{ id: number }>(
-    `
-      INSERT INTO sku_channel_rules (
-        provider, shop_id, match_key_type, match_key, channel, priority,
-        source, enabled, note, created_by_admin_user_id, created_at, updated_at
-      )
-      VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, NOW(), NOW())
-      ON CONFLICT (provider, shop_id, match_key_type, match_key, channel)
-      DO UPDATE SET
-        priority = EXCLUDED.priority,
-        enabled = EXCLUDED.enabled,
-        note = EXCLUDED.note,
-        updated_at = NOW()
-      RETURNING id
-    `,
-    [
-      String(input.provider || '').trim(),
-      String(input.shopId || '').trim(),
-      input.matchKeyType,
-      matchKey,
-      input.channel,
-      Math.trunc(Number(input.priority) || 100),
-      source,
-      input.enabled !== false,
-      String(input.note || '').slice(0, 200),
-      input.adminUserId,
-    ],
-  )
-
-  if (input.matchKeyType === 'internal_sku') {
-    await upsertInternalSkuChannel({
-      internalSku: matchKey,
-      channel: input.channel,
-      priority: Math.trunc(Number(input.priority) || 100),
-      isFallback: Math.trunc(Number(input.priority) || 100) >= 90,
-      enabled: input.enabled !== false,
-      source: source === 'operator' ? 'operator' : `rule:${source}`,
-      replacePriority: true,
-    })
-  }
-
-  return { id: result.rows[0]!.id }
+  const priority = Math.trunc(Number(input.priority) || 100)
+  const id = await upsertInternalSkuChannel({
+    internalSku: matchKey,
+    channel: input.channel,
+    priority,
+    isFallback: priority >= 90,
+    enabled: input.enabled !== false,
+    source: source === 'operator' ? 'operator' : `rule:${source}`,
+    note: String(input.note || '').slice(0, 300),
+    adminUserId: input.adminUserId,
+    replacePriority: true,
+  })
+  return { id }
 }

 export async function deleteChannelRule(ruleId: number): Promise<{ deleted: boolean }> {
@@ -963,25 +1116,7 @@ export async function deleteChannelRule(ruleId: number): Promise<{ deleted: bool
  if (id <= 0) {
    throw createHttpError('规则 id 无效', { statusCode: 400, errorCode: 'invalid_rule_id' })
  }
-  const result = await query<{
-    id: number
-    match_key_type: string
-    match_key: string
-    channel: string
-  }>(
-    `DELETE FROM sku_channel_rules
-     WHERE id = $1
-     RETURNING id, match_key_type, match_key, channel`,
-    [id],
-  )
-  const deleted = result.rows[0]
-  if (deleted?.match_key_type === 'internal_sku' && isFulfillmentChannel(deleted.channel)) {
-    await disableInternalSkuChannel({
-      internalSku: deleted.match_key,
-      channel: deleted.channel,
-    })
-  }
-  return { deleted: Boolean(deleted) }
+  return { deleted: await deleteInternalSkuChannelById(id) }
 }

 async function loadEnabledRules(): Promise<ChannelRule[]> {
@@ -992,7 +1127,11 @@ async function loadEnabledRules(): Promise<ChannelRule[]> {
      WHERE enabled = TRUE
    `,
  )
-  return rows.rows
+  return rows.rows.map((row) => ({
+    ...row,
+    id: Number(row.id),
+    priority: Number(row.priority),
+  }))
 }

 async function loadInternalSkuChannelsForKeys(
@@ -1004,8 +1143,8 @@ async function loadInternalSkuChannelsForKeys(

  const rows = await query<InternalSkuChannelRow>(
    `
-      SELECT internal_sku, channel, priority, is_fallback, enabled,
-             canonical_product_id, config_json, source
+      SELECT id, internal_sku, channel, priority, is_fallback, enabled,
+             canonical_product_id, config_json, source, note, created_by_admin_user_id
      FROM internal_sku_channels
      WHERE enabled = TRUE AND btrim(internal_sku) = ANY($1::text[])
      ORDER BY internal_sku, priority, channel
@@ -1029,19 +1168,23 @@ export const CHANNEL_VALUES = FULFILLMENT_CHANNELS
 export async function ensureVoucherRuleForSku(skuCode: string): Promise<boolean> {
  const key = String(skuCode || '').trim()
  if (!key) return false
-  const result = await query<{ id: number }>(
-    `
-      INSERT INTO sku_channel_rules (
-        provider, shop_id, match_key_type, match_key, channel, priority,
-        source, enabled, note, created_at, updated_at
-      )
-      VALUES ('', '', 'sku_code', $1, 'third_party_voucher', 70, 'voucher', TRUE, '卡券商品自动生成', NOW(), NOW())
-      ON CONFLICT (provider, shop_id, match_key_type, match_key, channel) DO NOTHING
-      RETURNING id
-    `,
-    [key],
+  const product = await query<{ id: number }>(
+    `SELECT id FROM product_catalog
+     WHERE canonical_sku = $1 AND status = 'active'
+     LIMIT 1`,
+    [`tpv:${key}`],
  )
-  return (result.rowCount || 0) > 0
+  await upsertInternalSkuChannel({
+    internalSku: key,
+    channel: 'third_party_voucher',
+    priority: 10,
+    isFallback: false,
+    enabled: true,
+    canonicalProductId: Number(product.rows[0]?.id || 0) || null,
+    configJson: { voucherSku: key },
+    source: 'voucher',
+  })
+  return true
 }

 /**
@@ -1054,28 +1197,38 @@ export async function ensureVoucherChannelRules(): Promise<{
  created: number
  productCount: number
 }> {
+  await query(
+    `UPDATE internal_sku_channels channel
+     SET is_fallback = TRUE,
+         priority = GREATEST(priority, 90),
+         updated_at = NOW()
+     FROM product_catalog product
+     WHERE product.status = 'active'
+       AND product.canonical_sku LIKE 'tpv:%'
+       AND BTRIM(channel.internal_sku) = BTRIM(product.canonical_name)
+       AND channel.channel <> 'third_party_voucher'
+       AND channel.enabled = TRUE
+       AND channel.is_fallback = FALSE`,
+  )
  const created = await query<{ id: number }>(
    `
-      INSERT INTO sku_channel_rules (
-        provider, shop_id, match_key_type, match_key, channel, priority,
-        source, enabled, note, created_at, updated_at
+      INSERT INTO internal_sku_channels (
+        internal_sku, channel, priority, is_fallback, enabled,
+        canonical_product_id, config_json, source, created_at, updated_at
      )
      SELECT
-        '', '', 'sku_code', substring(p.canonical_sku from 5), 'third_party_voucher', 70,
-        'voucher', TRUE, '卡券商品自动生成', NOW(), NOW()
+        BTRIM(p.canonical_name), 'third_party_voucher', 10, FALSE, TRUE,
+        p.id, jsonb_build_object('voucherSku', substring(p.canonical_sku from 5)),
+        'voucher', NOW(), NOW()
      FROM product_catalog p
      WHERE p.status = 'active'
        AND p.canonical_sku LIKE 'tpv:%'
-        AND btrim(substring(p.canonical_sku from 5)) <> ''
-        AND NOT EXISTS (
-          SELECT 1 FROM sku_channel_rules r
-          WHERE r.provider = ''
-            AND r.shop_id = ''
-            AND r.match_key_type = 'sku_code'
-            AND r.match_key = substring(p.canonical_sku from 5)
-            AND r.channel = 'third_party_voucher'
-        )
-      ON CONFLICT DO NOTHING
+        AND BTRIM(COALESCE(p.canonical_name, '')) <> ''
+      ON CONFLICT (internal_sku, channel) DO UPDATE SET
+        enabled = TRUE,
+        canonical_product_id = COALESCE(EXCLUDED.canonical_product_id, internal_sku_channels.canonical_product_id),
+        config_json = internal_sku_channels.config_json || EXCLUDED.config_json,
+        updated_at = NOW()
      RETURNING id
    `,
  )
@@ -20,6 +20,16 @@ export const FULFILLMENT_STATUSES = ['pending', 'assigned', 'archived'] as const

 export type FulfillmentStatus = (typeof FULFILLMENT_STATUSES)[number]

+/** 这些任务状态只保留审计价值，不再代表当前履约载体。 */
+export const INACTIVE_CARRIER_TASK_STATUSES = [
+  'closed',
+  'expired',
+  'failed',
+  'manual_review',
+  'retry_pending',
+  'cancelled',
+] as const
+
 export const FULFILLMENT_SOURCES = [
  'auto_rule',
  'auto_mapping',
@@ -13,19 +13,27 @@ import test from 'node:test'
 const TEST_DATABASE_URL = String(process.env.FULFILLMENT_ROUTING_TEST_DATABASE_URL || '').trim()

 test(
-  '运营指定渠道：先保证载体，再写 assigned；缺载体保持 pending',
+  '运营指定渠道：先保证载体；支持安全切换；缺载体保持 pending',
  { skip: !TEST_DATABASE_URL ? '未设置 FULFILLMENT_ROUTING_TEST_DATABASE_URL' : false },
  async () => {
    process.env.DATABASE_URL = TEST_DATABASE_URL
    const { closeDb, query } = await import('../../db/client.js')
-    const { assignChannelByInternalSku } = await import('./channel-assign-service.js')
+    const {
+      assignChannelByInternalSku,
+      deleteChannelRule,
+      ensureVoucherRuleForSku,
+      listChannelRules,
+      switchChannelForOrderItems,
+      upsertChannelRule,
+    } = await import('./channel-assign-service.js')
    const suffix = `${Date.now()}_${Math.floor(Math.random() * 1_000_000)}`
    const shopId = `it-routing-${suffix}`
    const manualSku = `集成测试人工-${suffix}`
    const missingAffiliateSku = `集成测试自动-${suffix}`
+    const voucherSku = `集成测试卡券-${suffix}`
    const orderIds: number[] = []
    const orderItemIds: number[] = []
-    const internalSkus = [manualSku, missingAffiliateSku]
+    const internalSkus = [manualSku, missingAffiliateSku, voucherSku]

    async function createPendingOrderItem(sku: string) {
      const order = await query<{ id: number }>(
@@ -84,6 +92,33 @@ test(
      assert.equal(manualItem.rows[0]?.fulfillment_channel, 'manual')
      assert.ok(Number(manualItem.rows[0]?.fulfillment_config_id) > 0)

+      const switched = await switchChannelForOrderItems({
+        orderItemIds: [manualItemId],
+        channel: 'third_party_voucher',
+        adminUserId: null,
+        note: '集成测试切换',
+      })
+      assert.equal(switched.switched, 1)
+      assert.equal(switched.failed, 0)
+      const switchedItem = await query<{
+        fulfillment_status: string
+        fulfillment_channel: string
+        original_channel: string | null
+        canonical_product_id: number | null
+        canonical_sku: string | null
+      }>(
+        `SELECT oi.fulfillment_status, oi.fulfillment_channel, oi.original_channel,
+                oi.canonical_product_id, product.canonical_sku
+         FROM order_items oi
+         LEFT JOIN product_catalog product ON product.id = oi.canonical_product_id
+         WHERE oi.id = $1`,
+        [manualItemId],
+      )
+      assert.equal(switchedItem.rows[0]?.fulfillment_status, 'assigned')
+      assert.equal(switchedItem.rows[0]?.fulfillment_channel, 'third_party_voucher')
+      assert.equal(switchedItem.rows[0]?.original_channel, 'manual')
+      assert.ok(switchedItem.rows[0]?.canonical_sku?.startsWith('tpv:'))
+
      const missingAffiliateItemId = await createPendingOrderItem(missingAffiliateSku)
      const affiliateFailed = await assignChannelByInternalSku({
        provider: '91kaquan',
@@ -107,6 +142,34 @@ test(
      assert.equal(pendingItem.rows[0]?.fulfillment_status, 'pending')
      assert.equal(pendingItem.rows[0]?.fulfillment_channel, null)
      assert.equal(pendingItem.rows[0]?.fulfillment_pending_reason, 'no_carrier')
+
+      await ensureVoucherRuleForSku(voucherSku)
+      const voucherChannel = await query<{ channel: string; source: string }>(
+        `SELECT channel, source FROM internal_sku_channels
+         WHERE internal_sku = $1 ORDER BY channel`,
+        [voucherSku],
+      )
+      assert.deepEqual(voucherChannel.rows, [{ channel: 'third_party_voucher', source: 'voucher' }])
+      const legacyVoucherRules = await query<{ count: number }>(
+        `SELECT count(*)::int AS count FROM sku_channel_rules
+         WHERE match_key = $1`,
+        [voucherSku],
+      )
+      assert.equal(legacyVoucherRules.rows[0]?.count, 0)
+      const savedRule = await upsertChannelRule({
+        matchKeyType: 'internal_sku',
+        matchKey: voucherSku,
+        channel: 'third_party_voucher',
+        priority: 10,
+        enabled: true,
+        note: '集成测试统一渠道',
+        adminUserId: null,
+      })
+      const listedRules = await listChannelRules({ channel: 'third_party_voucher', enabled: true })
+      assert.ok(
+        listedRules.some((rule) => rule.id === savedRule.id && rule.match_key === voucherSku),
+      )
+      assert.equal((await deleteChannelRule(savedRule.id)).deleted, true)
    } finally {
      if (orderItemIds.length > 0) {
        await query(`DELETE FROM cost_exception_items WHERE order_item_id = ANY($1::bigint[])`, [
@@ -78,10 +78,14 @@ export async function applyIntakeChannelDecision(

  const canOverrideDefaultManual = !selected || isDefaultManualFallback
  const existingAffiliateSku = String(parseObject(snapshot.affiliateDash).sku || '').trim()
+  const configuredAffiliateSku = String(primary.config_json.affiliateSku || '').trim()

  const desiredChannel = options.preferredChannel || primary.channel
  if (desiredChannel === 'affiliate_dash' || selected === 'affiliate_dash') {
-    const dashSku = existingAffiliateSku || matchAffiliateDashSkuByInternalSku(internalSku)
+    const dashSku =
+      existingAffiliateSku ||
+      configuredAffiliateSku ||
+      matchAffiliateDashSkuByInternalSku(internalSku)
    if (dashSku) {
      if (canOverrideDefaultManual || selected === 'affiliate_dash') {
        nextSnapshot.fulfillmentRoute = {
@@ -8,6 +8,7 @@ import {

 function channel(patch: Partial<InternalSkuChannelRow>): InternalSkuChannelRow {
  return {
+    id: 1,
    internal_sku: '内部SKU',
    channel: 'manual',
    priority: 10,
@@ -16,6 +17,8 @@ function channel(patch: Partial<InternalSkuChannelRow>): InternalSkuChannelRow {
    canonical_product_id: null,
    config_json: {},
    source: 'test',
+    note: '',
+    created_by_admin_user_id: null,
    ...patch,
  }
 }
@@ -3,6 +3,7 @@ import type { FulfillmentChannel } from './channel-types.js'
 import { compareFulfillmentChannel } from './internal-sku-resolver.js'

 export type InternalSkuChannelRow = {
+  id: number
  internal_sku: string
  channel: FulfillmentChannel
  priority: number
@@ -11,6 +12,8 @@ export type InternalSkuChannelRow = {
  canonical_product_id: number | null
  config_json: Record<string, unknown>
  source: string
+  note: string
+  created_by_admin_user_id: number | null
 }

 export type InternalSkuChannelPlan = {
@@ -27,8 +30,8 @@ export async function listInternalSkuChannels(
  if (!key) return []
  const result = await query<InternalSkuChannelRow>(
    `
-      SELECT internal_sku, channel, priority, is_fallback, enabled, canonical_product_id,
-             config_json, source
+      SELECT id, internal_sku, channel, priority, is_fallback, enabled, canonical_product_id,
+             config_json, source, note, created_by_admin_user_id
      FROM internal_sku_channels
      WHERE btrim(internal_sku) = $1 AND enabled = TRUE
      ORDER BY
@@ -89,11 +92,13 @@ export async function upsertInternalSkuChannel(input: {
  canonicalProductId?: number | null
  configJson?: Record<string, unknown> | undefined
  source?: string
+  note?: string | undefined
+  adminUserId?: number | null | undefined
  replacePriority?: boolean
-}): Promise<void> {
+}): Promise<number> {
  const internalSku = String(input.internalSku || '').trim()
-  if (!internalSku) return
-  await withTransaction(async (client) => {
+  if (!internalSku) return 0
+  return withTransaction(async (client) => {
    if (input.enabled !== false && input.isFallback !== true) {
      await client.query(
        `UPDATE internal_sku_channels
@@ -107,15 +112,16 @@ export async function upsertInternalSkuChannel(input: {
        [internalSku, input.channel],
      )
    }
-    await client.query(
+    const result = await client.query<{ id: number }>(
      `
      INSERT INTO internal_sku_channels (
        internal_sku, channel, priority, is_fallback, enabled,
-        canonical_product_id, config_json, source, created_at, updated_at
-      ) VALUES ($1, $2, $3, $4, $5, $6, $7::jsonb, $8, NOW(), NOW())
+        canonical_product_id, config_json, source, note, created_by_admin_user_id,
+        created_at, updated_at
+      ) VALUES ($1, $2, $3, $4, $5, $6, $7::jsonb, $8, $9, $10, NOW(), NOW())
      ON CONFLICT (internal_sku, channel) DO UPDATE SET
        priority = CASE
-          WHEN $9::boolean THEN EXCLUDED.priority
+          WHEN $11::boolean THEN EXCLUDED.priority
          ELSE LEAST(internal_sku_channels.priority, EXCLUDED.priority)
        END,
        is_fallback = EXCLUDED.is_fallback,
@@ -123,7 +129,10 @@ export async function upsertInternalSkuChannel(input: {
        canonical_product_id = COALESCE(EXCLUDED.canonical_product_id, internal_sku_channels.canonical_product_id),
        config_json = internal_sku_channels.config_json || EXCLUDED.config_json,
        source = EXCLUDED.source,
+        note = CASE WHEN EXCLUDED.note <> '' THEN EXCLUDED.note ELSE internal_sku_channels.note END,
+        created_by_admin_user_id = COALESCE(EXCLUDED.created_by_admin_user_id, internal_sku_channels.created_by_admin_user_id),
        updated_at = NOW()
+      RETURNING id
      `,
      [
        internalSku,
@@ -134,9 +143,12 @@ export async function upsertInternalSkuChannel(input: {
        input.canonicalProductId ?? null,
        JSON.stringify(input.configJson || {}),
        String(input.source || 'auto').trim() || 'auto',
+        String(input.note || '').slice(0, 300),
+        input.adminUserId ?? null,
        input.replacePriority === true,
      ],
    )
+    return Number(result.rows[0]?.id || 0)
  })
 }

@@ -149,22 +161,12 @@ export async function setInternalSkuPrimaryChannel(input: {
  canonicalProductId?: number | null
  source?: string
  configJson?: Record<string, unknown> | undefined
+  note?: string | undefined
+  adminUserId?: number | null
 }): Promise<void> {
  const internalSku = String(input.internalSku || '').trim()
  if (!internalSku) return

-  await query(
-    `
-      UPDATE internal_sku_channels
-      SET priority = GREATEST(priority, 90),
-          is_fallback = TRUE,
-          updated_at = NOW()
-      WHERE btrim(internal_sku) = $1
-        AND channel <> $2
-        AND enabled = TRUE
-    `,
-    [internalSku, input.channel],
-  )
  await upsertInternalSkuChannel({
    internalSku,
    channel: input.channel,
@@ -174,6 +176,8 @@ export async function setInternalSkuPrimaryChannel(input: {
    canonicalProductId: input.canonicalProductId ?? null,
    configJson: input.configJson,
    source: input.source || 'operator',
+    note: input.note,
+    adminUserId: input.adminUserId,
    replacePriority: true,
  })
 }
@@ -192,3 +196,10 @@ export async function disableInternalSkuChannel(input: {
  )
  return result.rowCount || 0
 }
+
+export async function deleteInternalSkuChannelById(channelId: number): Promise<boolean> {
+  const id = Math.trunc(Number(channelId) || 0)
+  if (id <= 0) return false
+  const result = await query(`DELETE FROM internal_sku_channels WHERE id = $1`, [id])
+  return (result.rowCount || 0) > 0
+}
@@ -213,7 +213,7 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) {
    `
      SELECT oi.id, oi.fulfillment_channel AS channel, oi.sku_code
      FROM order_items oi
-      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'
      WHERE oi.fulfillment_status = 'assigned'
        AND oi.fulfillment_channel IN ('affiliate_dash', 'work_order')
        AND wo.id IS NULL
@@ -221,6 +221,7 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) {
          SELECT 1 FROM fulfillment_tasks ft
          WHERE ft.order_item_id = oi.id
            AND ft.executor_key = 'affiliate_dash'
+            AND ft.task_status NOT IN ('closed', 'expired', 'failed', 'manual_review', 'retry_pending', 'cancelled')
        )
        AND oi.id > $2
      ORDER BY oi.id
@@ -261,16 +262,18 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) {
      FROM order_items oi
      WHERE oi.fulfillment_status = 'assigned'
        AND (
-          EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id)
+          EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id AND wo.status <> 'cancelled')
        )::int + (
          EXISTS (
            SELECT 1 FROM fulfillment_tasks ft
            WHERE ft.order_item_id = oi.id AND ft.executor_key = 'affiliate_dash'
+              AND ft.task_status NOT IN ('closed', 'expired', 'failed', 'manual_review', 'retry_pending', 'cancelled')
          )
        )::int + (
          EXISTS (
            SELECT 1 FROM fulfillment_tasks ft
            WHERE ft.order_item_id = oi.id AND ft.executor_key = 'manual_dispatch'
+              AND ft.task_status NOT IN ('closed', 'expired', 'failed', 'manual_review', 'retry_pending', 'cancelled')
          )
        )::int > 1
      ORDER BY oi.id
@@ -362,11 +365,12 @@ async function resolveCarrierDispatchExceptions(): Promise<number> {
          AND ce.status <> 'resolved'
          AND (
            (oi.fulfillment_channel = 'work_order'
-              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'))
            OR (oi.fulfillment_channel = 'affiliate_dash' AND EXISTS (
              SELECT 1 FROM fulfillment_tasks ft
              WHERE ft.order_item_id = oi.id
                AND ft.executor_key = 'affiliate_dash'
+                AND ft.task_status NOT IN ('closed', 'expired', 'failed', 'manual_review', 'retry_pending', 'cancelled')
            ))
          )
        LIMIT 500
@@ -55,6 +55,7 @@ export async function syncWorkerOrdersForSourceOrder(
    autoOnly?: boolean
    sourceMetadata?: JsonObject
    skipChannelAssign?: boolean
+    preferredWorkRuleKey?: string
  } = {},
 ) {
  const [rules, mappings] = await Promise.all([
@@ -80,9 +81,36 @@ export async function syncWorkerOrdersForSourceOrder(
      continue
    }

-    const match = resolveMatchingProductRuleDecision(order, item, rules, mappings, {
-      legacyFallbackEnabled: productMatchConfig.legacyFallbackEnabled,
-    })
+    const preferredRuleKey = String(options.preferredWorkRuleKey || '').trim()
+    const preferredRule = preferredRuleKey
+      ? rules.find((rule) => String(rule.rule_key || '').trim() === preferredRuleKey) || null
+      : null
+    const match =
+      preferredRuleKey && !preferredRule
+        ? {
+            rule: null,
+            mappingId: null,
+            reason: 'unmatched' as const,
+            score: 0,
+            candidates: [],
+            diagnosis: [
+              {
+                dimension: 'internal_sku_channel',
+                reason: 'configured_work_rule_missing',
+              },
+            ],
+          }
+        : preferredRule
+          ? {
+              rule: preferredRule,
+              mappingId: null,
+              reason: 'matched' as const,
+              score: 1,
+              candidates: [{ ruleKey: preferredRuleKey, score: 1, mappingId: null }],
+            }
+          : resolveMatchingProductRuleDecision(order, item, rules, mappings, {
+              legacyFallbackEnabled: productMatchConfig.legacyFallbackEnabled,
+            })
    const matchContext = resolveKuaishouWorkProductMatchContext(item)
    const snapshot = safeParseJson(item.item_snapshot_json)
    const sellerId =