Merge branch 'main' into dev_match

2026-11-06 17:39:08 +08:00
parent 3ee679ad4d 91e0bc8c9a
commit fde35624cd
59 changed files with 3319 additions and 956 deletions
@@ -0,0 +1,82 @@
+-- 核销任务直接保存目标券码，供后台“最近核销任务”展示和检索。
+-- 任务创建时会先解析候选券码；执行时再刷新一次，兼容券码晚于任务到达的场景。
+
+CREATE EXTENSION IF NOT EXISTS pg_trgm;
+
+ALTER TABLE voucher_consume_jobs
+  ADD COLUMN IF NOT EXISTS voucher_codes TEXT[] NOT NULL DEFAULT ARRAY[]::TEXT[];
+
+COMMENT ON COLUMN voucher_consume_jobs.voucher_codes IS
+  '核销任务关联的券码；快手电子凭证保存 voucher_code，抖音码速达保存我方与平台侧券码';
+
+-- array_to_string 在 PostgreSQL 中被标记为 STABLE，不能直接用于表达式索引；
+-- text[] 的拼接是本项目的确定性展示口径，用 IMMUTABLE 包装支持核销码模糊检索。
+CREATE OR REPLACE FUNCTION fulfillment_voucher_codes_search_text(codes TEXT[])
+RETURNS TEXT
+LANGUAGE sql
+IMMUTABLE
+PARALLEL SAFE
+RETURN ARRAY_TO_STRING(codes, ' ');
+
+COMMENT ON FUNCTION fulfillment_voucher_codes_search_text(TEXT[]) IS
+  '核销任务券码模糊搜索文本（用于 GIN trigram 表达式索引）';
+
+CREATE INDEX IF NOT EXISTS idx_voucher_consume_jobs_voucher_codes
+  ON voucher_consume_jobs
+  USING GIN ((fulfillment_voucher_codes_search_text(voucher_codes)) gin_trgm_ops);
+
+-- 回填已有任务。快手优先按任务/单元定位；没有任务时退回订单号维度。
+UPDATE voucher_consume_jobs job
+SET voucher_codes = COALESCE((
+  SELECT ARRAY_AGG(DISTINCT voucher.voucher_code ORDER BY voucher.voucher_code)
+  FROM kuaishou_industry_vouchers voucher
+  WHERE job.provider = 'kuaishou_industry'
+    AND voucher.oid = job.platform_order_id
+    AND BTRIM(voucher.voucher_code) <> ''
+    AND (
+      job.task_id IS NULL
+      OR voucher.task_id = job.task_id
+      OR voucher.unit_index = (
+        SELECT task.unit_index FROM fulfillment_tasks task WHERE task.id = job.task_id
+      )
+    )
+), ARRAY[]::TEXT[])
+WHERE job.provider = 'kuaishou_industry'
+  AND CARDINALITY(job.voucher_codes) = 0;
+
+-- 抖音码速达可能发生平台侧换码，因此两个券码都保留并可搜索。
+UPDATE voucher_consume_jobs job
+SET voucher_codes = COALESCE((
+  SELECT ARRAY_AGG(DISTINCT code ORDER BY code)
+  FROM (
+    SELECT certificate.cert_no AS code
+    FROM redeem_certificates certificate
+    WHERE certificate.provider = 'douyin_mashida'
+      AND (
+        certificate.local_order_id = job.order_id
+        OR (job.order_id IS NULL AND certificate.order_id = job.platform_order_id)
+      )
+      AND BTRIM(certificate.cert_no) <> ''
+      AND (
+        job.order_item_id IS NULL
+        OR certificate.order_item_id IS NULL
+        OR certificate.order_item_id = job.order_item_id
+      )
+    UNION
+    SELECT certificate.platform_cert_no AS code
+    FROM redeem_certificates certificate
+    WHERE certificate.provider = 'douyin_mashida'
+      AND (
+        certificate.local_order_id = job.order_id
+        OR (job.order_id IS NULL AND certificate.order_id = job.platform_order_id)
+      )
+      AND BTRIM(certificate.platform_cert_no) <> ''
+      AND (
+        job.order_item_id IS NULL
+        OR certificate.order_item_id IS NULL
+        OR certificate.order_item_id = job.order_item_id
+      )
+  ) codes
+), ARRAY[]::TEXT[])
+WHERE job.provider = 'douyin_mashida'
+  AND CARDINALITY(job.voucher_codes) = 0;
@@ -0,0 +1,173 @@
+-- 143_repair_third_party_voucher_products.sql
+--
+-- 第三方卡券订单项必须关联 tpv:<内部 SKU>。历史商品归一在缺少 tpv 商品时
+-- 会回退到同名 man 商品，导致卡券订单使用人工商品维度而无法生成卡券采购成本。
+-- 本迁移只修正商品归属、渠道规则和能力，不删除任何商品、订单或历史快照。
+-- 部署后由 cost_snapshot / sales_facts 管道补建成本快照并重算受影响日期。
+
+DROP TABLE IF EXISTS pg_temp.tmp_repair_voucher_items;
+CREATE TEMP TABLE tmp_repair_voucher_items ON COMMIT DROP AS
+SELECT
+  oi.id AS order_item_id,
+  oi.order_id,
+  resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code) AS internal_sku,
+  (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
+LEFT JOIN product_catalog current_product ON current_product.id = oi.canonical_product_id
+WHERE oi.fulfillment_channel = 'third_party_voucher'
+  AND BTRIM(resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code)) <> ''
+  AND char_length(
+    'tpv:' || resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code)
+  ) <= 64
+  AND (
+    current_product.id IS NULL
+    OR current_product.canonical_sku IS DISTINCT FROM
+      'tpv:' || resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code)
+  );
+
+CREATE UNIQUE INDEX ON tmp_repair_voucher_items (order_item_id);
+
+INSERT INTO product_catalog (
+  canonical_sku, canonical_name, category, unit_name, status,
+  attributes_json, created_at, updated_at
+)
+SELECT DISTINCT
+  'tpv:' || item.internal_sku,
+  item.internal_sku,
+  '卡券',
+  '张',
+  'active',
+  '{"source":"third_party_voucher","auto_created":true,"repaired":true}'::jsonb,
+  NOW(),
+  NOW()
+FROM tmp_repair_voucher_items item
+ON CONFLICT (canonical_sku) DO UPDATE SET
+  status = 'active',
+  canonical_name = CASE
+    WHEN BTRIM(product_catalog.canonical_name) = '' THEN EXCLUDED.canonical_name
+    ELSE product_catalog.canonical_name
+  END,
+  attributes_json = product_catalog.attributes_json || EXCLUDED.attributes_json,
+  updated_at = EXCLUDED.updated_at;
+
+UPDATE order_items oi
+SET canonical_product_id = product.id,
+    updated_at = NOW()
+FROM tmp_repair_voucher_items item
+JOIN product_catalog product
+  ON product.canonical_sku = 'tpv:' || item.internal_sku
+ AND product.status = 'active'
+WHERE oi.id = item.order_item_id;
+
+UPDATE cost_snapshots snapshot
+SET canonical_product_id = product.id
+FROM tmp_repair_voucher_items item
+JOIN product_catalog product
+  ON product.canonical_sku = 'tpv:' || item.internal_sku
+WHERE snapshot.order_item_id = item.order_item_id
+  AND snapshot.superseded_at IS NULL
+  AND snapshot.reversed_at IS NULL;
+
+INSERT INTO product_channel_capabilities (
+  product_id, channel, priority, is_fallback, enabled, source, created_at, updated_at
+)
+SELECT DISTINCT
+  product.id,
+  'third_party_voucher',
+  10,
+  FALSE,
+  TRUE,
+  'voucher',
+  NOW(),
+  NOW()
+FROM tmp_repair_voucher_items item
+JOIN product_catalog product
+  ON product.canonical_sku = 'tpv:' || item.internal_sku
+ON CONFLICT (product_id, channel) DO UPDATE SET
+  priority = 10,
+  is_fallback = FALSE,
+  enabled = TRUE,
+  source = 'voucher',
+  updated_at = EXCLUDED.updated_at;
+
+-- 卡券没有外部履约载体，规则命中即完成渠道判定；同 SKU 其他能力保留为兜底。
+UPDATE internal_sku_channels channel
+SET is_fallback = TRUE,
+    priority = GREATEST(channel.priority, 90),
+    updated_at = NOW()
+FROM (
+  SELECT DISTINCT internal_sku
+  FROM tmp_repair_voucher_items
+) repaired
+WHERE canonicalize_internal_sku(channel.internal_sku) = repaired.internal_sku
+  AND channel.channel <> 'third_party_voucher'
+  AND channel.enabled = TRUE
+  AND channel.is_fallback = FALSE;
+
+INSERT INTO internal_sku_channels (
+  internal_sku, channel, priority, is_fallback, enabled,
+  canonical_product_id, config_json, source, created_at, updated_at
+)
+SELECT DISTINCT
+  item.internal_sku,
+  'third_party_voucher',
+  10,
+  FALSE,
+  TRUE,
+  product.id,
+  jsonb_build_object('voucherSku', item.internal_sku, 'repaired', TRUE),
+  'voucher',
+  NOW(),
+  NOW()
+FROM tmp_repair_voucher_items item
+JOIN product_catalog product
+  ON product.canonical_sku = 'tpv:' || item.internal_sku
+ON CONFLICT (internal_sku, channel) DO UPDATE SET
+  priority = 10,
+  is_fallback = FALSE,
+  enabled = TRUE,
+  canonical_product_id = EXCLUDED.canonical_product_id,
+  config_json = internal_sku_channels.config_json || EXCLUDED.config_json,
+  source = EXCLUDED.source,
+  updated_at = EXCLUDED.updated_at;
+
+-- 修复商品归属后，商品缺失和旧成本缺失异常由 cost_snapshot 管道重新计算并自动收口。
+WITH removed_links AS (
+  DELETE FROM cost_exception_items
+  WHERE order_item_id IN (
+    SELECT order_item_id FROM tmp_repair_voucher_items
+  )
+  RETURNING exception_id
+),
+affected_heads AS (
+  SELECT DISTINCT exception_id FROM removed_links
+)
+UPDATE cost_exceptions exception
+SET status = 'resolved',
+    resolved_at = NOW(),
+    item_count = 0,
+    last_seen_at = NOW()
+WHERE exception.id IN (SELECT exception_id FROM affected_heads)
+  AND NOT EXISTS (
+    SELECT 1
+    FROM cost_exception_items item
+    WHERE item.exception_id = exception.id
+  );
+
+UPDATE cost_exceptions exception
+SET status = 'resolved',
+    resolved_at = NOW(),
+    item_count = 0,
+    last_seen_at = NOW()
+WHERE exception.order_item_id IN (
+    SELECT order_item_id FROM tmp_repair_voucher_items
+  )
+  AND NOT EXISTS (
+    SELECT 1
+    FROM cost_exception_items item
+    WHERE item.exception_id = exception.id
+  );
+
+COMMENT ON COLUMN product_catalog.attributes_json IS
+  '标准商品扩展信息；第三方卡券修复记录 repaired=true，商品归属以 tpv:<内部 SKU> 为唯一口径';
@@ -0,0 +1,120 @@
+-- 144_fulfillment_shadow_configuration_sync.sql
+--
+-- 统一履约视图会读取人工配置和 internal_sku_channels 的派生状态。
+-- 这两张配置表变化时也必须刷新影子模型，否则影子表会保留旧载体状态和旧异常。
+
+CREATE OR REPLACE FUNCTION sync_fulfillment_shadow_for_manual_config()
+RETURNS TRIGGER
+LANGUAGE plpgsql
+AS $$
+DECLARE
+  old_key TEXT;
+  new_key TEXT;
+BEGIN
+  IF TG_OP <> 'INSERT' THEN
+    old_key := canonicalize_internal_sku(OLD.sku_key);
+  END IF;
+  IF TG_OP <> 'DELETE' THEN
+    new_key := canonicalize_internal_sku(NEW.sku_key);
+  END IF;
+
+  PERFORM sync_fulfillment_shadow(item.id)
+  FROM order_items item
+  WHERE item.fulfillment_status IN ('pending', 'assigned')
+    AND canonicalize_internal_sku(
+      resolve_internal_sku(item.manual_sku_key, item.sku_name, item.sku_code)
+    ) IN (old_key, new_key);
+
+  RETURN COALESCE(NEW, OLD);
+END;
+$$;
+
+DROP TRIGGER IF EXISTS trg_manual_fulfillment_config_shadow ON manual_fulfillment_configs;
+CREATE TRIGGER trg_manual_fulfillment_config_shadow
+AFTER INSERT OR UPDATE OR DELETE ON manual_fulfillment_configs
+FOR EACH ROW
+EXECUTE FUNCTION sync_fulfillment_shadow_for_manual_config();
+
+CREATE OR REPLACE FUNCTION sync_fulfillment_shadow_for_internal_channel()
+RETURNS TRIGGER
+LANGUAGE plpgsql
+AS $$
+DECLARE
+  old_key TEXT;
+  new_key TEXT;
+BEGIN
+  IF TG_OP <> 'INSERT' THEN
+    old_key := canonicalize_internal_sku(OLD.internal_sku);
+  END IF;
+  IF TG_OP <> 'DELETE' THEN
+    new_key := canonicalize_internal_sku(NEW.internal_sku);
+  END IF;
+
+  PERFORM sync_fulfillment_shadow(item.id)
+  FROM order_items item
+  WHERE item.fulfillment_status IN ('pending', 'assigned')
+    AND canonicalize_internal_sku(
+      resolve_internal_sku(item.manual_sku_key, item.sku_name, item.sku_code)
+    ) IN (old_key, new_key);
+
+  RETURN COALESCE(NEW, OLD);
+END;
+$$;
+
+DROP TRIGGER IF EXISTS trg_internal_sku_channel_shadow ON internal_sku_channels;
+CREATE TRIGGER trg_internal_sku_channel_shadow
+AFTER INSERT OR UPDATE OR DELETE ON internal_sku_channels
+FOR EACH ROW
+EXECUTE FUNCTION sync_fulfillment_shadow_for_internal_channel();
+
+CREATE OR REPLACE FUNCTION assert_manual_config_product_source()
+RETURNS TRIGGER
+LANGUAGE plpgsql
+AS $$
+BEGIN
+  IF NEW.canonical_product_id IS NOT NULL
+    AND NOT EXISTS (
+      SELECT 1
+      FROM product_catalog product
+      WHERE product.id = NEW.canonical_product_id
+        AND product.status = 'active'
+        AND product.canonical_sku LIKE 'man:%'
+    )
+  THEN
+    RAISE EXCEPTION
+      'manual fulfillment config product % must be an active man: product',
+      NEW.canonical_product_id
+      USING ERRCODE = '23514';
+  END IF;
+  RETURN NEW;
+END;
+$$;
+
+DROP TRIGGER IF EXISTS trg_manual_config_product_source ON manual_fulfillment_configs;
+CREATE TRIGGER trg_manual_config_product_source
+BEFORE INSERT OR UPDATE ON manual_fulfillment_configs
+FOR EACH ROW
+EXECUTE FUNCTION assert_manual_config_product_source();
+
+-- 已启用的配置仍可能指向后来被归档的人工商品。这里恢复其 man: 商品，
+-- 避免迁移后归一继续把后续订单挂到不可用商品上。
+UPDATE product_catalog product
+SET status = 'active',
+    updated_at = NOW()
+FROM manual_fulfillment_configs config
+WHERE config.canonical_product_id = product.id
+  AND config.enabled = TRUE
+  AND product.canonical_sku LIKE 'man:%'
+  AND product.status <> 'active';
+
+-- 修正迁移 141 后由配置变化造成的存量陈旧影子记录。
+SELECT sync_fulfillment_shadow(item.id)
+FROM order_items item
+WHERE item.fulfillment_status IN ('pending', 'assigned');
+
+COMMENT ON FUNCTION sync_fulfillment_shadow_for_manual_config() IS
+  '人工配置变更后重算同内部 SKU 的履约影子';
+COMMENT ON FUNCTION sync_fulfillment_shadow_for_internal_channel() IS
+  '统一渠道规则变更后重算同内部 SKU 的履约影子';
+COMMENT ON FUNCTION assert_manual_config_product_source() IS
+  '人工履约配置只能关联有效的 man: 标准商品';
@@ -0,0 +1,38 @@
+-- 145_fulfillment_current_attempt_latest.sql
+--
+-- 统一读模型按 task.id DESC 取当前尝试。为保持与 is_current 指针一致，
+-- 从现在开始禁止把较旧任务重新标记为 current；新尝试仍由业务先退休旧指针。
+
+CREATE OR REPLACE FUNCTION assert_fulfillment_current_is_latest()
+RETURNS TRIGGER
+LANGUAGE plpgsql
+AS $$
+BEGIN
+  IF NEW.is_current = TRUE
+    AND EXISTS (
+      SELECT 1
+      FROM fulfillment_tasks newer
+      WHERE newer.order_item_id = NEW.order_item_id
+        AND newer.unit_index = NEW.unit_index
+        AND newer.id > COALESCE(NEW.id, 0)
+    )
+  THEN
+    RAISE EXCEPTION
+      'fulfillment current attempt must be the latest attempt for order_item % unit %',
+      NEW.order_item_id,
+      NEW.unit_index
+      USING ERRCODE = '23514';
+  END IF;
+  RETURN NEW;
+END;
+$$;
+
+DROP TRIGGER IF EXISTS trg_fulfillment_current_is_latest ON fulfillment_tasks;
+CREATE TRIGGER trg_fulfillment_current_is_latest
+BEFORE INSERT OR UPDATE OF is_current, order_item_id, unit_index
+ON fulfillment_tasks
+FOR EACH ROW
+EXECUTE FUNCTION assert_fulfillment_current_is_latest();
+
+COMMENT ON FUNCTION assert_fulfillment_current_is_latest() IS
+  '当前履约尝试必须是同订单项单元中 id 最大的尝试，保证统一视图 id DESC 与 is_current 一致';
@@ -1,5 +1,6 @@
 import test from 'node:test'
 import assert from 'node:assert/strict'
+import { readFile } from 'node:fs/promises'

 import {
  preserveAssignedOrderItemState,
@@ -202,3 +203,19 @@ test('已分配订单重推写库时不会用平台新 SKU 覆盖冻结值', asy
  assert.equal(updateParams[0]?.[0], 'lucky_coin_x2')
  assert.equal(updateParams[0]?.[1], '幸运币2个')
 })
+
+test('所有替换订单项入口都在同一事务内分摊订单收入', async () => {
+  const source = await readFile(new URL('./order-item-repo.ts', import.meta.url), 'utf8')
+  const replaceSource = source.slice(
+    source.indexOf('export async function replaceOrderItems('),
+    source.indexOf('export async function applyOrderItemRevenueAllocationWithExecutor('),
+  )
+
+  assert.ok(replaceSource.includes('withTransaction'))
+  assert.ok(replaceSource.includes('applyOrderItemRevenueAllocationWithExecutor'))
+  assert.ok(replaceSource.includes('SELECT total_amount FROM orders'))
+  assert.ok(
+    replaceSource.indexOf('replaceOrderItemsWithExecutor') <
+      replaceSource.indexOf('applyOrderItemRevenueAllocationWithExecutor'),
+  )
+})
@@ -42,9 +42,22 @@ export async function replaceOrderItems(
  orderId: number | string,
  items: OrderItemReplaceInput[],
 ): Promise<OrderItemRow[]> {
-  return withTransaction((client: PoolClient) =>
-    replaceOrderItemsWithExecutor(client.query.bind(client), orderId, items),
-  )
+  return withTransaction(async (client: PoolClient) => {
+    const executor = client.query.bind(client)
+    await replaceOrderItemsWithExecutor(executor, orderId, items)
+    const order = await client.query<{ total_amount: string | number }>(
+      `SELECT total_amount FROM orders WHERE id = $1`,
+      [orderId],
+    )
+    if (order.rows[0]) {
+      await applyOrderItemRevenueAllocationWithExecutor(
+        executor,
+        orderId,
+        Number(order.rows[0].total_amount) || 0,
+      )
+    }
+    return listOrderItemsByOrderIdWithExecutor(executor, orderId)
+  })
 }

 export async function applyOrderItemRevenueAllocationWithExecutor(
@@ -1,16 +1,12 @@
 /**
 * 成本中心 · 商品与映射路由。
 *
- * 覆盖：接单模板初始化、自动发货/接单平台商品同步与回填、人工商品归一、
+ * 覆盖：接单模板初始化、自动发货/接单平台商品同步与回填、人工商品自动投影、
 * 标准商品 CRUD、渠道映射 CRUD、未归一队列、归一数量回填。
 */
 import { Router } from 'express'

-import { ensureVoucherChannelRules } from '../../services/fulfillment/channel-assign-service.js'
-
 import {
-  archiveHistoricalManualProducts,
-  archiveSupersededManualProducts,
  backfillAdminNormalizedQuantities,
  backfillAdminProductMappings,
  createAdminChannelMapping,
@@ -27,6 +23,7 @@ import {
  updateAdminChannelMapping,
  updateAdminProduct,
 } from '../../services/cost-center/product-catalog-service.js'
+import { upsertAdminVoucherProduct } from '../../services/cost-center/voucher-product-service.js'
 import {
  importAdminAffiliateDashPricesBaseline,
  setAdminAffiliateProductPrice,
@@ -164,38 +161,6 @@ router.post(
  }),
 )

-router.post(
-  '/cost-center/products/archive-historical-manual',
-  requireAdminPermission('finance.config'),
-  createJsonHandler((req) => archiveHistoricalManualProducts(bodyOf(req)), {
-    successMessage: '历史人工商品已归档',
-    errorMessage: '归档历史人工商品失败',
-    scope: '[admin/cost-center/products/archive-historical-manual]',
-    audit: (_req, data) => ({
-      action: 'cost_historical_manual_products_archived',
-      targetType: 'product_catalog',
-      targetId: '',
-      data: data as Record<string, unknown>,
-    }),
-  }),
-)
-
-router.post(
-  '/cost-center/products/archive-superseded-manual',
-  requireAdminPermission('finance.config'),
-  createJsonHandler((req) => archiveSupersededManualProducts(bodyOf(req)), {
-    successMessage: '重复人工归一商品已归档',
-    errorMessage: '归档重复人工归一商品失败',
-    scope: '[admin/cost-center/products/archive-superseded-manual]',
-    audit: (_req, data) => ({
-      action: 'cost_superseded_manual_products_archived',
-      targetType: 'product_catalog',
-      targetId: '',
-      data: data as Record<string, unknown>,
-    }),
-  }),
-)
-
 router.patch(
  '/cost-center/products/:productId',
  requireAdminPermission('finance.config'),
@@ -402,16 +367,16 @@ router.post(
 )

 router.post(
-  '/cost-center/voucher-products/ensure-rules',
+  '/cost-center/voucher-products',
  requireAdminPermission('finance.config'),
-  createJsonHandler(() => ensureVoucherChannelRules(), {
-    successMessage: '卡券渠道规则已补齐',
-    errorMessage: '补齐卡券渠道规则失败',
-    scope: '[admin/cost-center/voucher-products/ensure-rules]',
+  createJsonHandler((req) => upsertAdminVoucherProduct(bodyOf(req), sessionOf(req)), {
+    successMessage: '卡券 SKU 已保存',
+    errorMessage: '保存卡券 SKU 失败',
+    scope: '[admin/cost-center/voucher-products]',
    audit: (_req, data) => ({
-      action: 'voucher_internal_channels_ensured',
-      targetType: 'internal_sku_channels',
-      targetId: '',
+      action: 'voucher_product_upserted',
+      targetType: 'product_catalog',
+      targetId: String((data as { product?: { id?: number } })?.product?.id || ''),
      data: data as Record<string, unknown>,
    }),
  }),
@@ -13,6 +13,10 @@ import {
  confirmManualFulfillment,
  listManualFulfillmentPending,
 } from '../../services/finance/manual-fulfillment-service.js'
+import {
+  listOnAccountBills,
+  reconcileOnAccountBill,
+} from '../../services/finance/on-account-bill-service.js'
 import {
  createJsonHandler,
  requireAdminPermission,
@@ -152,4 +156,44 @@ router.post(
  }),
 )

+router.get(
+  '/finance/on-account-bills',
+  requireAdminRoles(['admin', 'operator', 'support', 'finance']),
+  createJsonHandler((req) => listOnAccountBills(req.query as Record<string, unknown>), {
+    successMessage: 'ok',
+    errorMessage: '读取统一结算账单失败',
+    scope: '[admin/finance/on-account-bills]',
+  }),
+)
+
+router.post(
+  '/finance/on-account-bills/:billId/reconcile',
+  requireAdminPermission('finance.reconcile'),
+  createJsonHandler(
+    (req) => reconcileOnAccountBill(req.params.billId, bodyOf(req), sessionOf(req)),
+    {
+      successMessage: '统一结算账单已对账',
+      errorMessage: '统一结算账单对账失败',
+      scope: '[admin/finance/on-account-bills/:billId/reconcile]',
+      audit: (req, data) => {
+        const result = data as {
+          bill?: { id?: number; billNo?: string }
+          amountFen?: number
+          snapshotId?: number
+        }
+        return {
+          action: 'on_account_bill_reconciled',
+          targetType: 'on_account_bill',
+          targetId: String(result.bill?.id || req.params.billId || ''),
+          data: {
+            billNo: result.bill?.billNo || '',
+            amountFen: result.amountFen || 0,
+            costSnapshotId: result.snapshotId || null,
+          },
+        }
+      },
+    },
+  ),
+)
+
 export default router
@@ -9,6 +9,7 @@ import {
  listAdminKuaishouIndustryRefunds,
  listAdminKuaishouIndustryShops,
  listAdminKuaishouIndustryVouchers,
+  reissueAndConsumeAdminKuaishouIndustryVoucher,
  resendAdminKuaishouIndustryVoucherCode,
  reverseAdminKuaishouIndustryVoucher,
 } from '../../services/admin/kuaishou-industry-admin-service.js'
@@ -21,6 +22,7 @@ import type {
  AdminKuaishouIndustryVoucherCheckAvailableRouteBody,
  AdminKuaishouIndustryVoucherConsumeRouteBody,
  AdminKuaishouIndustryVoucherDestroyRouteBody,
+  AdminKuaishouIndustryVoucherReissueConsumeRouteBody,
  AdminKuaishouIndustryVoucherResendRouteBody,
  AdminKuaishouIndustryVoucherReverseRouteBody,
 } from '../../types/admin/route-inputs.js'
@@ -204,6 +206,24 @@ router.post(
  ),
 )

+router.post(
+  '/kuaishou-industry/vouchers/reissue-consume',
+  requireAdminRoles(['admin', 'operator']),
+  createJsonHandler(
+    (req) =>
+      reissueAndConsumeAdminKuaishouIndustryVoucher(
+        req.body as AdminKuaishouIndustryVoucherReissueConsumeRouteBody,
+      ),
+    {
+      successMessage: '历史券补发并核销已执行',
+      errorMessage: '历史券补发并核销失败',
+      scope: '[admin/kuaishou-industry/vouchers/reissue-consume]',
+      audit: (req, data) =>
+        buildKuaishouIndustryAudit('kuaishou_industry_voucher_reissue_consume', req.body, data),
+    },
+  ),
+)
+
 router.post(
  '/kuaishou-industry/vouchers/destroy',
  requireAdminRoles(['admin', 'operator']),
@@ -238,6 +258,13 @@ function buildKuaishouIndustryAudit(action: string, body: unknown, data: unknown
      success: Boolean(result.success),
      httpStatus: Number(result.httpStatus || 0) || 0,
      error: String(result.error || ''),
+      stage: String(result.stage || ''),
+      reason: String(result.reason || ''),
+      validMinutes: Number(result.validMinutes || 0) || 0,
+      originalValidEndTime: Number(result.originalValidEndTime || 0) || 0,
+      validEndTime: Number(result.validEndTime || 0) || 0,
+      sendSuccess: Boolean(result.sendSuccess),
+      consumeSuccess: Boolean(result.consumeSuccess),
    },
  }
 }
@@ -27,6 +27,7 @@ import { resendKuaishouIndustryVoucherSendCallback } from '../platforms/kuaishou
 import {
  consumeKuaishouIndustryVoucher,
  destroyKuaishouIndustryVoucher,
+  isKuaishouIndustryVoucherSendCallbackSuccess,
  resolveKuaishouIndustryEticketType,
 } from '../platforms/kuaishou-industry/voucher-service.js'
 import {
@@ -39,6 +40,8 @@ import { attachKuaishouIndustryVoucherToTask } from '../platforms/kuaishou-indus
 import type { KuaishouIndustryOpenApiCallResult } from '../platforms/kuaishou-industry/openapi-client.js'
 import type { KuaishouIndustryVoucherRow, TaskRow } from '../../types/repository/rows.js'

+const historicalReissueInFlight = new Set<string>()
+
 export async function listAdminKuaishouIndustryVouchers(
  input: KuaishouIndustryVoucherAdminListQuery = {},
 ) {
@@ -201,11 +204,204 @@ export async function resendAdminKuaishouIndustryVoucherCode(input: JsonObject =
  return {
    success: result.success,
    response: 'response' in result ? result.response || null : null,
-    error: result.error || '',
+    error:
+      result.error ||
+      resolveKuaishouOpenApiErrorMessage('response' in result ? result.response : null),
    voucher: result.voucher ? mapAdminKuaishouIndustryVoucher(result.voucher) : null,
  }
 }

+/**
+ * 历史发码失败券的受控补偿入口。
+ *
+ * 只允许未使用券进入；先把本地有效期临时改到未来，再复用原券码发码。
+ * 发码成功后立即核销，避免为了补回调把历史券重新暴露给买家。
+ */
+export async function reissueAndConsumeAdminKuaishouIndustryVoucher(input: JsonObject = {}) {
+  const voucher = await resolveRequiredVoucher(input)
+  const voucherCode = String(voucher.voucher_code || '').trim()
+  const accountStatus = String(voucher.status || '')
+    .trim()
+    .toUpperCase()
+  const nowMs = Date.now()
+  const originalValidEndTime = Number(voucher.valid_end_time || 0) || 0
+
+  if (accountStatus !== 'UNUSED') {
+    throw createHttpError('仅未使用电子凭证可以补发并核销', {
+      statusCode: 409,
+      errorCode: 'admin_kuaishou_industry_reissue_voucher_not_unused',
+    })
+  }
+
+  if (isKuaishouIndustryVoucherSendCallbackSuccess(voucher) && originalValidEndTime > nowMs) {
+    throw createHttpError('该电子凭证已发码成功且仍在有效期，请直接执行手动核销', {
+      statusCode: 409,
+      errorCode: 'admin_kuaishou_industry_reissue_voucher_already_active',
+    })
+  }
+
+  if (historicalReissueInFlight.has(voucherCode)) {
+    throw createHttpError('该电子凭证正在补发，请勿重复提交', {
+      statusCode: 409,
+      errorCode: 'admin_kuaishou_industry_reissue_in_progress',
+    })
+  }
+
+  const validity = resolveKuaishouIndustryHistoricalReissueWindow(input.validMinutes, nowMs)
+  const reason = normalizeKuaishouIndustryHistoricalReissueReason(input.reason)
+  const originalValidStartTime = Number(voucher.valid_start_time || 0) || 0
+  const validStartTime =
+    originalValidStartTime > 0 ? Math.min(originalValidStartTime, nowMs - 1_000) : nowMs - 1_000
+
+  historicalReissueInFlight.add(voucherCode)
+  try {
+    const prepared = await updateKuaishouIndustryVoucherByCode(voucherCode, {
+      validStartTime,
+      validEndTime: validity.validEndTime,
+      sendCallbackStatus: 'pending',
+      sendCallbackLastError: '',
+      sendCallbackResponseJson: {},
+      sendCallbackSentAt: null,
+      updatedAt: new Date().toISOString(),
+    })
+    if (!prepared) {
+      throw createHttpError('准备历史券补发时未找到电子凭证', {
+        statusCode: 404,
+        errorCode: 'admin_kuaishou_industry_reissue_voucher_not_found',
+      })
+    }
+
+    const sendResult = await resendAdminKuaishouIndustryVoucherCode({
+      voucherCode,
+      oid: voucher.oid,
+      orderId: voucher.oid,
+    })
+    if (!sendResult.success) {
+      return {
+        success: false,
+        stage: 'send',
+        reason,
+        validMinutes: validity.validMinutes,
+        originalValidEndTime,
+        validEndTime: validity.validEndTime,
+        sendSuccess: false,
+        consumeSuccess: false,
+        response: sendResult.response,
+        error: sendResult.error || '电子凭证发码回调失败',
+        voucher: sendResult.voucher,
+        task: null,
+      }
+    }
+
+    const sentVoucher = await findKuaishouIndustryVoucherByCode(voucherCode, voucher.oid)
+    if (!sentVoucher || !isKuaishouIndustryVoucherSendCallbackSuccess(sentVoucher)) {
+      return {
+        success: false,
+        stage: 'send',
+        reason,
+        validMinutes: validity.validMinutes,
+        originalValidEndTime,
+        validEndTime: validity.validEndTime,
+        sendSuccess: false,
+        consumeSuccess: false,
+        response: sendResult.response,
+        error: '电子凭证发码回调返回成功，但本地状态未更新',
+        voucher: sentVoucher ? mapAdminKuaishouIndustryVoucher(sentVoucher) : null,
+        task: null,
+      }
+    }
+
+    const task = sentVoucher.task_id ? await getTaskById(sentVoucher.task_id) : null
+    const consumeType = String(input.consumeType || '').trim()
+    const consumeInput: Parameters<typeof consumeKuaishouIndustryVoucher>[1] = {
+      source: 'admin_historical_reissue_consume',
+      token: String(input.token || sentVoucher.token || '').trim(),
+      consumeTime: Date.now(),
+      ...(consumeType ? { consumeType } : {}),
+      ...(input.serialNum ? { serialNum: String(input.serialNum).trim() } : {}),
+      ...(input.storeName ? { storeName: String(input.storeName).trim() } : {}),
+      ...(input.storeAddress ? { storeAddress: String(input.storeAddress).trim() } : {}),
+      ...(input.expressCode ? { expressCode: String(input.expressCode).trim() } : {}),
+      ...(input.expressNo ? { expressNo: String(input.expressNo).trim() } : {}),
+    }
+    if (task) {
+      consumeInput.task = task
+    }
+
+    const consumeResult = await consumeKuaishouIndustryVoucher(sentVoucher, consumeInput)
+    if (!consumeResult.ok || !consumeResult.voucher) {
+      return {
+        success: false,
+        stage: 'consume',
+        reason,
+        validMinutes: validity.validMinutes,
+        originalValidEndTime,
+        validEndTime: validity.validEndTime,
+        sendSuccess: true,
+        consumeSuccess: false,
+        response: sendResult.response,
+        error: consumeResult.errorMessage || '电子凭证发码成功，但核销失败',
+        voucher: consumeResult.voucher
+          ? mapAdminKuaishouIndustryVoucher(consumeResult.voucher)
+          : mapAdminKuaishouIndustryVoucher(sentVoucher),
+        task: task ? mapTaskReference(task) : null,
+      }
+    }
+
+    return {
+      success: true,
+      stage: 'completed',
+      reason,
+      validMinutes: validity.validMinutes,
+      originalValidEndTime,
+      validEndTime: validity.validEndTime,
+      sendSuccess: true,
+      consumeSuccess: true,
+      response: sendResult.response,
+      error: '',
+      voucher: mapAdminKuaishouIndustryVoucher(consumeResult.voucher),
+      task: task ? mapTaskReference(task) : null,
+    }
+  } finally {
+    historicalReissueInFlight.delete(voucherCode)
+  }
+}
+
+export function resolveKuaishouIndustryHistoricalReissueWindow(
+  rawMinutes: unknown,
+  nowMs = Date.now(),
+) {
+  const validMinutes = Number(rawMinutes)
+  if (!Number.isInteger(validMinutes) || validMinutes < 1 || validMinutes > 60) {
+    throw createHttpError('临时有效期必须是 1 到 60 分钟的整数', {
+      statusCode: 400,
+      errorCode: 'admin_kuaishou_industry_reissue_valid_minutes_invalid',
+    })
+  }
+
+  return {
+    validMinutes,
+    validEndTime: nowMs + validMinutes * 60_000,
+  }
+}
+
+export function normalizeKuaishouIndustryHistoricalReissueReason(rawReason: unknown) {
+  const reason = String(rawReason || '').trim()
+  if (!reason) {
+    throw createHttpError('补发原因不能为空', {
+      statusCode: 400,
+      errorCode: 'admin_kuaishou_industry_reissue_reason_required',
+    })
+  }
+  if (reason.length > 200) {
+    throw createHttpError('补发原因不能超过 200 个字符', {
+      statusCode: 400,
+      errorCode: 'admin_kuaishou_industry_reissue_reason_too_long',
+    })
+  }
+  return reason
+}
+
 export async function destroyAdminKuaishouIndustryVoucherByCode(input: JsonObject = {}) {
  const voucher = await resolveRequiredVoucher(input)
  const checkPayload = buildVoucherOpenApiPayload(input, voucher)
@@ -485,6 +681,14 @@ function mapOpenApiResult(result: KuaishouIndustryOpenApiCallResult) {
  }
 }

+function resolveKuaishouOpenApiErrorMessage(response: unknown) {
+  if (!response || typeof response !== 'object' || Array.isArray(response)) {
+    return ''
+  }
+  const current = response as Record<string, unknown>
+  return String(current.error_msg || current.sub_msg || current.message || current.msg || '').trim()
+}
+
 function sanitizeOpenApiRequest(request: JsonObject | undefined): JsonObject | null {
  if (!request) {
    return null
@@ -0,0 +1,76 @@
+import assert from 'node:assert/strict'
+import { readFile } from 'node:fs/promises'
+import test from 'node:test'
+
+import {
+  normalizeKuaishouIndustryHistoricalReissueReason,
+  resolveKuaishouIndustryHistoricalReissueWindow,
+} from './kuaishou-industry-admin-service.js'
+
+test('历史券补发窗口只允许 1 到 60 分钟', () => {
+  const nowMs = 1_800_000_000_000
+
+  assert.deepEqual(resolveKuaishouIndustryHistoricalReissueWindow(10, nowMs), {
+    validMinutes: 10,
+    validEndTime: nowMs + 10 * 60_000,
+  })
+  assert.throws(() => resolveKuaishouIndustryHistoricalReissueWindow(0, nowMs), /1 到 60 分钟/)
+  assert.throws(() => resolveKuaishouIndustryHistoricalReissueWindow(61, nowMs), /1 到 60 分钟/)
+  assert.throws(() => resolveKuaishouIndustryHistoricalReissueWindow('1.5', nowMs), /1 到 60 分钟/)
+})
+
+test('历史券补发原因必填且限制长度', () => {
+  assert.equal(
+    normalizeKuaishouIndustryHistoricalReissueReason('  历史发码失败补发  '),
+    '历史发码失败补发',
+  )
+  assert.throws(() => normalizeKuaishouIndustryHistoricalReissueReason('  '), /不能为空/)
+  assert.throws(
+    () => normalizeKuaishouIndustryHistoricalReissueReason('x'.repeat(201)),
+    /不能超过 200/,
+  )
+})
+
+test('受控补发必须先重置有效期，再发码，最后立即核销', async () => {
+  const source = await readFile(
+    new URL('./kuaishou-industry-admin-service.ts', import.meta.url),
+    'utf8',
+  )
+  const start = source.indexOf(
+    'export async function reissueAndConsumeAdminKuaishouIndustryVoucher',
+  )
+  assert.ok(start > 0, '未找到历史券补发并核销服务')
+  const body = source.slice(
+    start,
+    source.indexOf('\nexport function resolveKuaishouIndustryHistoricalReissueWindow', start),
+  )
+
+  const prepareIndex = body.indexOf('updateKuaishouIndustryVoucherByCode')
+  const sendIndex = body.indexOf('resendAdminKuaishouIndustryVoucherCode')
+  const consumeIndex = body.indexOf('consumeKuaishouIndustryVoucher(sentVoucher')
+
+  assert.ok(prepareIndex > 0, '必须先更新本地临时有效期')
+  assert.ok(sendIndex > prepareIndex, '必须先重置有效期再重发发码')
+  assert.ok(consumeIndex > sendIndex, '必须等发码成功后再执行核销')
+  assert.ok(body.includes("source: 'admin_historical_reissue_consume'"), '核销来源必须可审计')
+  assert.ok(body.includes("sendCallbackStatus: 'pending'"), '重发前必须重置发码状态')
+  assert.ok(body.includes("stage: 'send'"), '发码失败必须返回明确阶段')
+  assert.ok(body.includes("stage: 'consume'"), '核销失败必须返回明确阶段')
+  assert.ok(body.includes("stage: 'completed'"), '成功必须返回完成阶段')
+})
+
+test('补发并核销路由只允许管理员和运营执行', async () => {
+  const source = await readFile(
+    new URL('../../routes/admin/kuaishou-industry.ts', import.meta.url),
+    'utf8',
+  )
+  const routeIndex = source.indexOf("'/kuaishou-industry/vouchers/reissue-consume'")
+  assert.ok(routeIndex > 0, '未注册历史券补发并核销路由')
+  const routeBody = source.slice(routeIndex, source.indexOf('\nrouter.post(', routeIndex + 1))
+
+  assert.ok(routeBody.includes("requireAdminRoles(['admin', 'operator'])"), '必须限制管理员和运营')
+  assert.ok(
+    routeBody.includes("'kuaishou_industry_voucher_reissue_consume'"),
+    '必须写入独立审计动作',
+  )
+})
@@ -34,28 +34,12 @@ export function productSourceOf(canonicalSku: string): string {
 }

 /** 商品来源筛选的 SQL 片段：按 canonical_sku 前缀过滤。 */
-export function productSourceFilterSql(
-  paramIndex: number,
-  column = 'p.canonical_sku',
-  nameColumn = 'p.canonical_name',
-): string {
+export function productSourceFilterSql(paramIndex: number, column = 'p.canonical_sku'): string {
  return `(
    ($${paramIndex} = 'affiliate_dash' AND ${column} LIKE 'aff:%')
    OR ($${paramIndex} = 'work_order' AND (${column} LIKE 'tpl:%' OR ${column} LIKE 'wo:%'))
    OR ($${paramIndex} = 'third_party_voucher' AND ${column} LIKE 'tpv:%')
-    OR (
-      $${paramIndex} = 'manual'
-      AND ${column} LIKE 'man:%'
-      AND NOT EXISTS (
-        SELECT 1
-        FROM internal_sku_channels live_channel
-        WHERE live_channel.enabled = TRUE
-          AND live_channel.is_fallback = FALSE
-          AND live_channel.channel <> 'manual'
-          AND canonicalize_internal_sku(live_channel.internal_sku)
-              = canonicalize_internal_sku(${nameColumn})
-      )
-    )
+    OR ($${paramIndex} = 'manual' AND ${column} LIKE 'man:%')
  )`
 }

@@ -6,6 +6,7 @@
 *   2. 成本项统计按 cost_model 分组，不再读 cost_items.carrier_type。
 */
 import { query } from '../../db/client.js'
+import { productSourceFilterSql } from './cost-center-mappers.js'
 import { PROVIDERS, productSourceOf, toNumber } from './cost-center-shared.js'

 export type CostCenterScope = 'affiliate_dash' | 'work_order' | 'manual'
@@ -37,25 +38,7 @@ export async function getAdminCostCenterOverview(params: { scope?: string } = {}
        SELECT p.id
        FROM product_catalog p
        WHERE p.status = 'active'
-          AND (
-            $1::text IS NULL
-            OR ($1 = 'affiliate_dash' AND p.canonical_sku LIKE 'aff:%')
-            OR ($1 = 'work_order' AND (p.canonical_sku LIKE 'tpl:%' OR p.canonical_sku LIKE 'wo:%'))
-            OR ($1 = 'third_party_voucher' AND p.canonical_sku LIKE 'tpv:%')
-            OR (
-              $1 = 'manual'
-              AND p.canonical_sku LIKE 'man:%'
-              AND NOT EXISTS (
-                SELECT 1
-                FROM internal_sku_channels live_channel
-                WHERE live_channel.enabled = TRUE
-                  AND live_channel.is_fallback = FALSE
-                  AND live_channel.channel <> 'manual'
-                  AND canonicalize_internal_sku(live_channel.internal_sku)
-                      = canonicalize_internal_sku(p.canonical_name)
-              )
-            )
-          )
+          AND ($1::text IS NULL OR ${productSourceFilterSql(1)})
      ),
      scoped_cost_models AS (
        SELECT CASE $1
@@ -8,7 +8,7 @@ import {
  resolveMappingNameKey,
 } from './cost-center-shared.js'
 import { calculateCostTotal, validateCostVersionScope } from './cost-resolution-service.js'
-import { resolveMappingProductId } from './product-catalog-service.js'
+import { createAdminProduct, resolveMappingProductId } from './product-catalog-service.js'
 import { productSourceFilterSql } from './cost-center-mappers.js'
 import {
  buildTemplateMappingPlan,
@@ -310,14 +310,11 @@ test('第三方卡券删除：无历史引用硬删除，有历史引用自动
  assert.ok(routeSource.includes("'/cost-center/voucher-products/:productId'"))
 })

-test('人工客服筛选排除已有其他主渠道的旧 man 商品', () => {
+test('人工客服筛选保留所有可维护的 man 商品和显式兜底配置', () => {
  const sql = productSourceFilterSql(1)
  assert.ok(sql.includes(`$1 = 'manual'`))
-  assert.ok(sql.includes("live_channel.channel <> 'manual'"))
-  assert.ok(
-    sql.includes('canonicalize_internal_sku(live_channel.internal_sku)') &&
-      sql.includes('canonicalize_internal_sku(p.canonical_name)'),
-  )
+  assert.ok(sql.includes("LIKE 'man:%'"))
+  assert.ok(!sql.includes('live_channel'))
 })

 test('成本中心总览异常统计必须走异常表关联，不能全表扫描订单项子查询', async () => {
@@ -328,6 +325,7 @@ test('成本中心总览异常统计必须走异常表关联，不能全表扫
  )
  assert.ok(source.includes('JOIN order_items oi ON oi.id = ce.order_item_id'))
  assert.ok(!source.includes('ce.order_item_id IN ('))
+  assert.ok(source.includes('productSourceFilterSql(1)'))
 })

 test('tpv: 商品来源归为第三方卡券', async () => {
@@ -336,3 +334,29 @@ test('tpv: 商品来源归为第三方卡券', async () => {
  assert.equal(productSourceOf('tpv:模拟卡券'), 'third_party_voucher')
  assert.equal(productSourceOf('man:模拟人工'), 'manual')
 })
+
+test('人工标准商品禁止绕过人工 SKU 配置手工创建', async () => {
+  await assert.rejects(
+    createAdminProduct(
+      {
+        canonicalSku: 'man:破片手榴弹-兔兔白日梦',
+        canonicalName: '破片手榴弹-兔兔白日梦',
+      },
+      {} as never,
+    ),
+    /人工 SKU 配置自动创建/,
+  )
+})
+
+test('第三方卡券商品禁止绕过卡券 SKU 配置手工创建', async () => {
+  await assert.rejects(
+    createAdminProduct(
+      {
+        canonicalSku: 'tpv:11月改名卡',
+        canonicalName: '11月改名卡',
+      },
+      {} as never,
+    ),
+    /卡券 SKU 配置自动创建/,
+  )
+})
@@ -1,3 +1,5 @@
+import type { PoolClient } from 'pg'
+
 import { query } from '../../db/client.js'
 import { nowIso } from '../../utils/time.js'

@@ -19,6 +21,7 @@ export const COST_EXCEPTION_REASONS = [
  'cost_missing',
  'manual_config_missing',
  'manual_cost_missing',
+  'on_account_cost_pending',
  'carrier_missing',
 ] as const

@@ -29,14 +32,16 @@ export type UpsertExceptionInput = {
  scopeKey: string
  orderItemId: number
  detail?: Record<string, unknown>
+  client?: Pick<PoolClient, 'query'>
 }

 /** 幂等上浮一条聚合问题：已存在则累加明细并刷新 last_seen。 */
 export async function upsertAggregatedException(input: UpsertExceptionInput): Promise<void> {
  const now = nowIso()
  const detail = JSON.stringify(input.detail || {})
+  const run = input.client ? input.client.query.bind(input.client) : query

-  const result = await query<{ id: number }>(
+  const result = await run<{ id: number }>(
    `
      INSERT INTO cost_exceptions (
        subject_type, subject_id, order_item_id, reason, status, detail_json,
@@ -55,7 +60,7 @@ export async function upsertAggregatedException(input: UpsertExceptionInput): Pr
  const exceptionId = result.rows[0]?.id
  if (!exceptionId) return

-  await query(
+  await run(
    `
      INSERT INTO cost_exception_items (exception_id, order_item_id, created_at)
      VALUES ($1, $2, $3)
@@ -64,7 +69,7 @@ export async function upsertAggregatedException(input: UpsertExceptionInput): Pr
    [exceptionId, input.orderItemId, now],
  )

-  await query(
+  await run(
    `
      UPDATE cost_exceptions ce
      SET item_count = (
@@ -1,84 +0,0 @@
-/** 旧人工归一商品清理集成测试（真实 PostgreSQL）。 */
-import assert from 'node:assert/strict'
-import test from 'node:test'
-
-const TEST_DATABASE_URL = String(
-  process.env.LEGACY_MANUAL_PRODUCT_CLEANUP_TEST_DATABASE_URL || '',
-).trim()
-
-test(
-  '已被其他主渠道接管的历史 man 商品只归档并修正兜底指向',
-  { skip: !TEST_DATABASE_URL ? '未设置 LEGACY_MANUAL_PRODUCT_CLEANUP_TEST_DATABASE_URL' : false },
-  async () => {
-    process.env.DATABASE_URL = TEST_DATABASE_URL
-    const { closeDb, query } = await import('../../db/client.js')
-    const { archiveSupersededManualProducts } = await import('./product-catalog-service.js')
-    const suffix = `${Date.now()}_${Math.floor(Math.random() * 1_000_000)}`
-    const internalSku = `清理测试人工商品-${suffix}`
-    const manSku = `man:${internalSku}`
-    const liveSku = `aff:cleanup-${suffix}`
-    let manProductId = 0
-    let liveProductId = 0
-
-    try {
-      const inserted = await query<{ id: number; canonical_sku: string }>(
-        `
-          INSERT INTO product_catalog (
-            canonical_sku, canonical_name, category, unit_name, status,
-            attributes_json, created_at, updated_at
-          ) VALUES
-            ($1, $3, '人工客服', '件', 'active', '{"source":"manual_normalization"}'::jsonb, NOW(), NOW()),
-            ($2, $3, '自动发货', '件', 'active', '{}'::jsonb, NOW(), NOW())
-          RETURNING id, canonical_sku
-        `,
-        [manSku, liveSku, internalSku],
-      )
-      manProductId = Number(inserted.rows.find((row) => row.canonical_sku === manSku)?.id)
-      liveProductId = Number(inserted.rows.find((row) => row.canonical_sku === liveSku)?.id)
-      assert.ok(manProductId > 0)
-      assert.ok(liveProductId > 0)
-
-      await query(
-        `
-          INSERT INTO internal_sku_channels (
-            internal_sku, channel, priority, is_fallback, enabled,
-            canonical_product_id, config_json, source, created_at, updated_at
-          ) VALUES
-            ($1, 'affiliate_dash', 10, FALSE, TRUE, $3, '{}'::jsonb, 'test', NOW(), NOW()),
-            ($1, 'manual', 90, TRUE, TRUE, $2, '{}'::jsonb, 'test', NOW(), NOW())
-        `,
-        [internalSku, manProductId, liveProductId],
-      )
-
-      const preview = await archiveSupersededManualProducts()
-      assert.equal(preview.candidateCount, 1)
-      assert.ok(preview.sample.some((row) => Number(row.id) === manProductId))
-
-      const result = await archiveSupersededManualProducts({ execute: true })
-      assert.equal(result.archivedCount, 1)
-      assert.equal(result.repointedChannelCount, 1)
-
-      const products = await query<{ canonical_sku: string; status: string }>(
-        'SELECT canonical_sku, status FROM product_catalog WHERE id = ANY($1::bigint[])',
-        [[manProductId, liveProductId]],
-      )
-      assert.equal(products.rows.find((row) => row.canonical_sku === manSku)?.status, 'archived')
-      assert.equal(products.rows.find((row) => row.canonical_sku === liveSku)?.status, 'active')
-      const fallback = await query<{ canonical_product_id: number }>(
-        `SELECT canonical_product_id
-         FROM internal_sku_channels
-         WHERE internal_sku = $1 AND channel = 'manual'`,
-        [internalSku],
-      )
-      assert.equal(Number(fallback.rows[0]?.canonical_product_id), liveProductId)
-    } finally {
-      if (manProductId > 0 || liveProductId > 0) {
-        await query('DELETE FROM internal_sku_channels WHERE internal_sku = $1', [internalSku])
-        await query('DELETE FROM product_catalog WHERE id = ANY($1::bigint[])', [
-          [manProductId, liveProductId],
-        ])
-      }
-      await closeDb()
-    }
-  },
-)
@@ -1,10 +1,15 @@
+import type { PoolClient } from 'pg'
+
 import { query, withTransaction } from '../../db/client.js'
 import { resolveExceptionsForItemsByReason } from './cost-exception-service.js'
 import type { JsonObject } from '../../types/json.js'
 import { createHttpError } from '../../utils/http.js'
 import { formatShanghaiDate, shiftDate } from '../../utils/finance-date-range.js'
 import { nowIso } from '../../utils/time.js'
-import { upsertInternalSkuChannel } from '../fulfillment/internal-sku-channel-service.js'
+import {
+  upsertInternalSkuChannel,
+  upsertInternalSkuChannelWithClient,
+} from '../fulfillment/internal-sku-channel-service.js'
 import { reconcilePendingCarriers } from '../fulfillment/carrier-reconciliation-service.js'
 import {
  attachFulfillmentConfigurations,
@@ -463,6 +468,26 @@ export async function saveManualFulfillmentConfig(payload: JsonObject) {
    const existingId = Number(existing.rows[0]?.id || 0)
    let resolvedProductId = canonicalProductId

+    if (resolvedProductId) {
+      const product = await client.query<{ canonical_sku: string; status: string }>(
+        `SELECT canonical_sku, status
+         FROM product_catalog
+         WHERE id = $1
+         LIMIT 1`,
+        [resolvedProductId],
+      )
+      if (
+        !product.rows[0] ||
+        product.rows[0].status !== 'active' ||
+        !String(product.rows[0].canonical_sku || '').startsWith('man:')
+      ) {
+        throw createHttpError('人工履约配置只能关联有效的 man: 标准商品', {
+          statusCode: 400,
+          errorCode: 'manual_config_product_source_invalid',
+        })
+      }
+    }
+
    // 人工配置本身就是该 SKU 的标准商品声明。没有显式关联时自动创建/复用
    // man:<内部 SKU>，避免运营再去「标准商品」手工建一次。
    if (!resolvedProductId) {
@@ -543,14 +568,14 @@ export async function saveManualFulfillmentConfig(payload: JsonObject) {
            now,
          ],
        )
-    return requiredRow(result.rows[0], '保存人工履约配置失败')
+    const saved = requiredRow(result.rows[0], '保存人工履约配置失败')
+    await syncRegistryConfigId(saved.sku_key, saved.id, client)
+    await upsertManualRegistryConfig(saved, client)
+    // 标准商品、人工配置和渠道规则必须一起提交，避免保存成功后仍未生效。
+    await syncManualChannelRule(saved, client)
+    return saved
  })

-  await syncRegistryConfigId(config.sku_key, config.id)
-  await upsertManualRegistryConfig(config)
-  // 同步渠道规则：人工配置就是「这个内部 SKU 走人工渠道」的唯一声明，
-  // 少了这一步，新配的 SKU 会一直停在待分类。
-  await syncManualChannelRule(config)
  // 配置补全后，失败的 pending 会因 locked=true 无法被定时任务重试，这里主动解锁并重试。
  const pendingAssigned = await retryPendingManualItems(config.sku_key)
  // 配置保存后立刻回填该 SKU 最近 30 天的订单项，避免必须再点一次「重新匹配」。
@@ -938,26 +963,33 @@ async function attachCostVersionsForManualConfigs() {
 }

 /** 配置保存/删除后同步登记表上的 config_id。 */
-async function syncRegistryConfigId(skuKey: string, configId: number | null) {
-  await query(
-    `UPDATE manual_sku_registry
-               SET config_id = $2, updated_at = $3
-               WHERE canonicalize_internal_sku(sku_key) = $1`,
-    [canonicalizeInternalSku(skuKey), configId, nowIso()],
-  )
+async function syncRegistryConfigId(skuKey: string, configId: number | null, client?: PoolClient) {
+  const sql = `
+    UPDATE manual_sku_registry
+    SET config_id = $2, updated_at = $3
+    WHERE canonicalize_internal_sku(sku_key) = $1
+  `
+  const params = [canonicalizeInternalSku(skuKey), configId, nowIso()]
+  if (client) {
+    await client.query(sql, params)
+    return
+  }
+  await query(sql, params)
 }

 /** 配置保存后确保登记表仍有同一 SKU 的记录，兼容旧扫描清单和运维查询。 */
-async function upsertManualRegistryConfig(config: {
-  id: number
-  sku_key: string
-  canonical_product_id: number | null
-}) {
+async function upsertManualRegistryConfig(
+  config: {
+    id: number
+    sku_key: string
+    canonical_product_id: number | null
+  },
+  client?: PoolClient,
+) {
  const key = canonicalizeInternalSku(config.sku_key)
  if (!key) return
  const now = nowIso()
-  await query(
-    `
+  const sql = `
      INSERT INTO manual_sku_registry (
        sku_key, display_name, config_id, created_at, updated_at
      )
@@ -976,9 +1008,13 @@ async function upsertManualRegistryConfig(config: {
        END,
        config_id = EXCLUDED.config_id,
        updated_at = EXCLUDED.updated_at
-    `,
-    [key, config.id, now, config.canonical_product_id],
-  )
+    `
+  const params = [key, config.id, now, config.canonical_product_id]
+  if (client) {
+    await client.query(sql, params)
+    return
+  }
+  await query(sql, params)
 }

 /** 读取单个 SKU 的配置详情（含平台标签与成本折算）。 */
@@ -1299,29 +1335,32 @@ function requiredRow<T>(row: T | undefined, message: string): T {
 /**
 * 人工配置 → internal_sku_channels（内部 SKU 精确匹配）。
 */
-async function syncManualChannelRule(config: {
-  id: number
-  sku_key: string
-  canonical_product_id: number | null
-  enabled: boolean
-  note: string
-}): Promise<void> {
+async function syncManualChannelRule(
+  config: {
+    id: number
+    sku_key: string
+    canonical_product_id: number | null
+    enabled: boolean
+    note: string
+  },
+  client?: PoolClient,
+): Promise<void> {
  const skuKey = String(config.sku_key || '').trim()
  if (!skuKey) return

-  const otherChannel = await query<{ exists: boolean }>(
-    `
+  const otherChannelSql = `
      SELECT EXISTS (
        SELECT 1 FROM internal_sku_channels
        WHERE canonicalize_internal_sku(internal_sku) = $1
          AND enabled = TRUE
          AND channel <> 'manual'
      ) AS exists
-    `,
-    [skuKey],
-  )
+    `
+  const otherChannel = client
+    ? await client.query<{ exists: boolean }>(otherChannelSql, [skuKey])
+    : await query<{ exists: boolean }>(otherChannelSql, [skuKey])
  const hasOtherChannel = otherChannel.rows[0]?.exists === true
-  await upsertInternalSkuChannel({
+  const channelInput = {
    internalSku: skuKey,
    channel: 'manual',
    priority: hasOtherChannel ? 90 : 10,
@@ -1331,5 +1370,10 @@ async function syncManualChannelRule(config: {
    source: 'manual_config',
    configJson: { manualConfigId: config.id, note: String(config.note || '').slice(0, 200) },
    replacePriority: true,
-  })
+  } as const
+  if (client) {
+    await upsertInternalSkuChannelWithClient(client, channelInput)
+    return
+  }
+  await upsertInternalSkuChannel(channelInput)
 }
@@ -17,7 +17,6 @@ import { normalizeMatchText, normalizeSkuSeriesEquivalence } from '../product-ma
 import { resolveSkuNameQuantity } from '../product-match/pricing.js'
 import { setAdminAffiliateProductPrice } from './affiliate-product-service.js'
 import { resolvePriceEffectiveFrom } from './cost-center-normalizers.js'
-import { ensureVoucherRuleForSku } from '../fulfillment/channel-assign-service.js'
 import { syncCostSnapshotsForItems } from './cost-snapshot-service.js'
 import { attachFulfillmentConfigurations } from '../fulfillment/fulfillment-state-service.js'
 import {
@@ -319,6 +318,18 @@ export async function listAdminProducts(payload: JsonObject = {}) {

 export async function createAdminProduct(payload: JsonObject, _session: AdminSession) {
  const canonicalSku = normalizeRequiredText(payload.canonicalSku, '标准 SKU', 64)
+  if (canonicalSku.startsWith('man:')) {
+    throw createHttpError('人工商品由人工 SKU 配置自动创建，请从人工 SKU 配置入口新增', {
+      statusCode: 409,
+      errorCode: 'cost_manual_product_managed_by_config',
+    })
+  }
+  if (canonicalSku.startsWith('tpv:')) {
+    throw createHttpError('卡券商品由第三方卡券 SKU 配置自动创建，请从卡券配置入口新增', {
+      statusCode: 409,
+      errorCode: 'cost_voucher_product_managed_by_config',
+    })
+  }
  if (!/^[A-Za-z0-9._:-]+$/.test(canonicalSku)) {
    throw createHttpError('标准 SKU 只能包含字母、数字、点、下划线、冒号和短横线', {
      statusCode: 400,
@@ -355,10 +366,6 @@ export async function createAdminProduct(payload: JsonObject, _session: AdminSes
      errorCode: 'cost_product_sku_exists',
    })
  }
-  // 卡券没有载体证据，创建商品时顺手把渠道规则补上，避免新 SKU 停在待分类。
-  if (canonicalSku.startsWith('tpv:')) {
-    await ensureVoucherRuleForSku(canonicalSku.slice(4))
-  }
  return { product: mapProduct(inserted) }
 }

@@ -586,164 +593,6 @@ export async function deleteAdminVoucherProduct(productId: number | string) {
  })
 }

-/**
- * 批量归档旧人工归一商品。
- *
- * 只处理 attributes_json.source=manual_normalization、当前为 active，
- * 且所有关联订单项都已 archived 的商品。只改 status，不删除任何数据。
- */
-export async function archiveHistoricalManualProducts(payload: JsonObject = {}) {
-  const execute = payload.execute === true || payload.execute === 'true'
-  const candidates = await query<{ id: number; canonical_sku: string; canonical_name: string }>(
-    `
-      SELECT p.id, p.canonical_sku, p.canonical_name
-      FROM product_catalog p
-      WHERE p.status = 'active'
-        AND p.canonical_sku LIKE 'man:%'
-        AND p.attributes_json->>'source' = 'manual_normalization'
-        AND EXISTS (SELECT 1 FROM order_items oi WHERE oi.canonical_product_id = p.id)
-        AND NOT EXISTS (
-          SELECT 1 FROM order_items oi
-          WHERE oi.canonical_product_id = p.id
-            AND oi.fulfillment_status <> 'archived'
-        )
-      ORDER BY p.id
-    `,
-  )
-  const ids = candidates.rows.map((row) => Number(row.id))
-  if (!execute || ids.length === 0) {
-    return {
-      dryRun: !execute,
-      candidateCount: ids.length,
-      archivedCount: 0,
-      sample: candidates.rows.slice(0, 20),
-    }
-  }
-
-  const updated = await query(
-    `UPDATE product_catalog
-     SET status = 'archived', updated_at = NOW()
-     WHERE id = ANY($1::bigint[]) AND status = 'active'`,
-    [ids],
-  )
-  return {
-    dryRun: false,
-    candidateCount: ids.length,
-    archivedCount: updated.rowCount || 0,
-    sample: candidates.rows.slice(0, 20),
-  }
-}
-
-type SupersededManualProductRow = {
-  id: number
-  canonical_sku: string
-  canonical_name: string
-  primary_channel: string
-  target_product_id: number
-  target_product_sku: string
-}
-
-/**
- * 归档已被其他渠道主规则接管的旧人工归一商品。
- *
- * `man:` 只表示旧人工来源，不代表当前主渠道。同名内部 SKU 已有自动发货、
- * 接单平台或卡券主渠道后，旧 man: 商品不再参与新路由，只需从当前目录归档。
- * 历史订单、成本版本和成本快照全部保留；仍指向旧商品的手动兜底改指当前主商品。
- */
-export async function archiveSupersededManualProducts(payload: JsonObject = {}) {
-  const execute = payload.execute === true || payload.execute === 'true'
-  return withTransaction(async (client) => {
-    const candidates = await client.query<SupersededManualProductRow>(
-      `
-        SELECT
-          p.id,
-          p.canonical_sku,
-          p.canonical_name,
-          live.channel AS primary_channel,
-          target.id AS target_product_id,
-          target.canonical_sku AS target_product_sku
-        FROM product_catalog p
-        JOIN LATERAL (
-          SELECT c.channel, c.canonical_product_id
-          FROM internal_sku_channels c
-          WHERE c.enabled = TRUE
-            AND c.is_fallback = FALSE
-            AND c.channel <> 'manual'
-            AND canonicalize_internal_sku(c.internal_sku)
-                = canonicalize_internal_sku(p.canonical_name)
-          ORDER BY c.priority, c.channel
-          LIMIT 1
-        ) live ON TRUE
-        JOIN product_catalog target
-          ON target.id = live.canonical_product_id
-         AND target.status = 'active'
-        WHERE p.status = 'active'
-          AND p.canonical_sku LIKE 'man:%'
-          AND p.attributes_json->>'source' = 'manual_normalization'
-        ORDER BY p.id
-        FOR UPDATE OF p
-      `,
-    )
-    const rows = candidates.rows.map((row) => ({
-      ...row,
-      id: Number(row.id),
-      target_product_id: Number(row.target_product_id),
-    }))
-    if (!execute || rows.length === 0) {
-      return {
-        dryRun: !execute,
-        candidateCount: rows.length,
-        archivedCount: 0,
-        repointedChannelCount: 0,
-        sample: rows.slice(0, 20),
-      }
-    }
-
-    const repointed = await client.query(
-      `
-        UPDATE internal_sku_channels c
-        SET canonical_product_id = input.target_product_id,
-            updated_at = NOW()
-        FROM unnest($1::text[], $2::bigint[], $3::bigint[])
-          AS input(internal_sku, source_product_id, target_product_id)
-        WHERE canonicalize_internal_sku(c.internal_sku)
-            = canonicalize_internal_sku(input.internal_sku)
-          AND c.channel = 'manual'
-          AND c.canonical_product_id = input.source_product_id
-      `,
-      [
-        rows.map((row) => row.canonical_name),
-        rows.map((row) => row.id),
-        rows.map((row) => row.target_product_id),
-      ],
-    )
-    const archived = await client.query(
-      `
-        UPDATE product_catalog
-        SET status = 'archived',
-            attributes_json = COALESCE(attributes_json, '{}'::jsonb) || $1::jsonb,
-            updated_at = NOW()
-        WHERE id = ANY($2::bigint[]) AND status = 'active'
-      `,
-      [
-        JSON.stringify({
-          legacyManualArchived: true,
-          legacyManualArchiveReason: 'superseded_by_active_primary_channel',
-          legacyManualArchivedAt: nowIso(),
-        }),
-        rows.map((row) => row.id),
-      ],
-    )
-    return {
-      dryRun: false,
-      candidateCount: rows.length,
-      archivedCount: archived.rowCount || 0,
-      repointedChannelCount: repointed.rowCount || 0,
-      sample: rows.slice(0, 20),
-    }
-  })
-}
-
 type MappingProductGroupRow = {
  canonical_product_id: number
  canonical_sku: string
@@ -1,4 +1,5 @@
 import assert from 'node:assert/strict'
+import { readFile } from 'node:fs/promises'
 import test from 'node:test'

 import {
@@ -63,9 +64,11 @@ function item(patch: Record<string, unknown> = {}) {
    sku_name: '任意',
    normalized_sku_name: null as string | null,
    normalized_sku_series: null as string | null,
-    fulfillment_channel: 'work_order' as string | null,
+    fulfillment_channel: null as string | null,
+    canonical_product_id: null as number | null,
    work_rule_key: null as string | null,
    manual_config_product_id: null as number | null,
+    manual_config_product_sku: null as string | null,
    ...patch,
  }
 }
@@ -123,6 +126,7 @@ test('商品归一：人工渠道直接使用配置里指定的商品', () => {
        fulfillment_channel: 'manual',
        sku_code: 'unknown',
        manual_config_product_id: 44,
+        manual_config_product_sku: 'man:配置商品',
      }),
      index,
    ),
@@ -144,11 +148,65 @@ test('商品归一：第三方卡券按 tpv:SKU 直接命中', () => {
  )
 })

+test('商品归一：第三方卡券缺少 tpv 商品时不能串到同名人工商品', () => {
+  assert.equal(
+    resolveProductId(
+      item({
+        fulfillment_channel: 'third_party_voucher',
+        internal_sku: '套装-测试',
+      }),
+      index,
+    ),
+    null,
+  )
+})
+
+test('商品归一：其他已确定渠道也不能借用同名其他来源商品', () => {
+  for (const fulfillmentChannel of ['manual', 'on_account', 'affiliate_dash', 'work_order']) {
+    assert.equal(
+      resolveProductId(
+        item({
+          fulfillment_channel: fulfillmentChannel,
+          sku_code: 'unknown',
+          internal_sku: '套装-测试',
+        }),
+        index,
+      ),
+      null,
+    )
+  }
+})
+
+test('商品归一：人工配置也只能挂载 man: 商品', () => {
+  assert.equal(
+    resolveProductId(
+      item({
+        fulfillment_channel: 'manual',
+        manual_config_product_id: 99,
+        manual_config_product_sku: 'aff:错误商品',
+      }),
+      index,
+    ),
+    null,
+  )
+  assert.equal(
+    resolveProductId(
+      item({
+        fulfillment_channel: 'on_account',
+        manual_config_product_id: 99,
+        manual_config_product_sku: 'man:正确商品',
+      }),
+      index,
+    ),
+    99,
+  )
+})
+
 test('商品归一：中文内部 SKU 命中标准商品，不依赖平台 SKU', () => {
  assert.equal(
    resolveProductId(
      item({
-        fulfillment_channel: 'affiliate_dash',
+        fulfillment_channel: null,
        sku_code: 'unknown',
        internal_sku: '套装-测试',
      }),
@@ -178,3 +236,14 @@ test('三方卡券自动建商品时优先使用内部中文 SKU', () => {
  assert.equal(voucherCanonicalSku({ manual_sku_key: '', sku_name: '', sku_code: 'x' }), null)
  assert.equal(voucherCanonicalSku({}), null)
 })
+
+test('商品归一扫描会纠正所有渠道的错误来源商品', async () => {
+  const source = await readFile(new URL('./product-normalize-service.ts', import.meta.url), 'utf8')
+
+  assert.ok(source.includes("current_product.canonical_sku NOT LIKE 'aff:%'"))
+  assert.ok(source.includes("current_product.canonical_sku NOT LIKE 'man:%'"))
+  assert.ok(source.includes("current_product.canonical_sku NOT LIKE 'tpl:%'"))
+  assert.ok(source.includes("current_product.canonical_sku NOT LIKE 'tpv:%'"))
+  assert.ok(source.includes('canonicalProductId: null'))
+  assert.ok(source.includes('onlyWhenCanonicalProductDifferent: true'))
+})
@@ -3,7 +3,10 @@ import { resolveExceptionsForItems, upsertAggregatedException } from './cost-exc
 import { readWatermarkItemId } from '../watermark-store.js'
 import { upsertInternalSkuChannel } from '../fulfillment/internal-sku-channel-service.js'
 import { canonicalizeInternalSku, containsCjk } from '../fulfillment/internal-sku-resolver.js'
-import { attachFulfillmentConfigurations } from '../fulfillment/fulfillment-state-service.js'
+import {
+  attachFulfillmentConfigurations,
+  transitionFulfillmentState,
+} from '../fulfillment/fulfillment-state-service.js'

 /**
 * 管道 B：商品归一。
@@ -51,8 +54,10 @@ type ItemRow = {
  normalized_sku_name: string | null
  normalized_sku_series: string | null
  fulfillment_channel: string | null
+  canonical_product_id?: number | null
  work_rule_key: string | null
  manual_config_product_id: number | null
+  manual_config_product_sku?: string | null
 }

 function key(...parts: string[]): string {
@@ -145,6 +150,17 @@ export function voucherCanonicalSku(item: {
  return internal ? `tpv:${internal}` : null
 }

+async function clearMismatchedProduct(item: ItemRow) {
+  if (!item.fulfillment_channel || !item.canonical_product_id) return
+  await transitionFulfillmentState({
+    orderItemId: Number(item.id),
+    expected: { status: 'assigned' },
+    patch: { canonicalProductId: null },
+    eventType: 'product_source_cleared',
+    reason: `渠道 ${item.fulfillment_channel} 与当前标准商品来源不一致`,
+  })
+}
+
 /**
 * 三方卡券没有预先建库时，按内部 SKU 自动建 `tpv:` 商品并补渠道规则。
 * 建完后该商品会出现在「商品与成本中心 → 第三方卡券」，运营直接设采购价。
@@ -191,7 +207,7 @@ async function ensureVoucherProductForItem(item: ItemRow): Promise<number | null

  await attachFulfillmentConfigurations({
    mappings: [{ orderItemId: Number(item.id), canonicalProductId: productId }],
-    onlyWhenCanonicalProductNull: true,
+    onlyWhenCanonicalProductDifferent: true,
  })

  await query(
@@ -225,26 +241,32 @@ export function resolveProductId(item: ItemRow, index: MappingIndex): number | n
  const seriesKey = String(item.normalized_sku_series || '').trim()
  const internalSku = canonicalizeInternalSku(item.internal_sku || item.manual_sku_key)

-  // 人工渠道：配置里已经指定商品，直接用，不需要再建映射。
-  if (item.fulfillment_channel === 'manual' && item.manual_config_product_id) {
-    return Number(item.manual_config_product_id)
+  // 已确定渠道必须命中本渠道商品，不能借用同名的其他来源商品。
+  if (item.fulfillment_channel === 'manual' || item.fulfillment_channel === 'on_account') {
+    return item.manual_config_product_id &&
+      String(item.manual_config_product_sku || '').startsWith('man:')
+      ? Number(item.manual_config_product_id)
+      : null
  }

  // 接单平台：模板即商品来源（tpl:<rule_key>），无需渠道映射。
-  if (item.fulfillment_channel === 'work_order' && item.work_rule_key) {
+  if (item.fulfillment_channel === 'work_order') {
+    if (!item.work_rule_key) return null
    const templateProduct = index.templateRules.get(String(item.work_rule_key).trim())
-    if (templateProduct) return templateProduct
+    return templateProduct || null
  }

-  if (item.fulfillment_channel === 'affiliate_dash' && skuCode) {
-    const aff = index.affSkus.get(skuCode)
-    if (aff) return aff
+  if (item.fulfillment_channel === 'affiliate_dash') {
+    return skuCode ? index.affSkus.get(skuCode) || null : null
  }

  // 第三方卡券：与自动发货同构，商品编码 tpv:<sku>。
  if (item.fulfillment_channel === 'third_party_voucher' && internalSku) {
    const voucher = index.voucherSkus.get(internalSku)
-    if (voucher) return voucher
+    return voucher || null
+  }
+  if (item.fulfillment_channel === 'third_party_voucher') {
+    return null
  }

  // 中文内部 SKU 是标准商品键；平台 SKU 映射只作为历史数据兜底。
@@ -322,15 +344,52 @@ export async function normalizeOrderItemProducts(
          oi.normalized_sku_name,
          oi.normalized_sku_series,
          oi.fulfillment_channel,
+          oi.canonical_product_id,
          wpr.rule_key AS work_rule_key,
-          cfg.canonical_product_id AS manual_config_product_id
+          cfg.canonical_product_id AS manual_config_product_id,
+          cfg_product.canonical_sku AS manual_config_product_sku
        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 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
+        LEFT JOIN product_catalog cfg_product ON cfg_product.id = cfg.canonical_product_id
+        LEFT JOIN product_catalog current_product ON current_product.id = oi.canonical_product_id
        WHERE oi.fulfillment_status = 'assigned'
-          AND oi.canonical_product_id IS NULL
+          AND (
+            oi.canonical_product_id IS NULL
+            OR (
+              oi.fulfillment_channel = 'third_party_voucher'
+              AND (
+                current_product.id IS NULL
+                OR current_product.canonical_sku NOT LIKE 'tpv:%'
+              )
+            )
+            OR (
+              oi.fulfillment_channel = 'affiliate_dash'
+              AND (
+                current_product.id IS NULL
+                OR current_product.canonical_sku NOT LIKE 'aff:%'
+              )
+            )
+            OR (
+              oi.fulfillment_channel IN ('manual', 'on_account')
+              AND (
+                current_product.id IS NULL
+                OR current_product.canonical_sku NOT LIKE 'man:%'
+              )
+            )
+            OR (
+              oi.fulfillment_channel = 'work_order'
+              AND (
+                current_product.id IS NULL
+                OR (
+                  current_product.canonical_sku NOT LIKE 'tpl:%'
+                  AND current_product.canonical_sku NOT LIKE 'wo:%'
+                )
+              )
+            )
+          )
          AND oi.id > $1
        ORDER BY oi.id
        LIMIT $2
@@ -356,6 +415,7 @@ export async function normalizeOrderItemProducts(
      }
      if (!productId) {
        result.unmatched += 1
+        await clearMismatchedProduct(item)
        await upsertAggregatedException({
          reason: 'product_mapping_missing',
          scopeKey: `product_mapping_missing|-|-|${canonicalizeInternalSku(item.internal_sku || item.manual_sku_key || item.sku_code)}|${item.fulfillment_channel || '-'}`,
@@ -375,7 +435,7 @@ export async function normalizeOrderItemProducts(
          canonicalProductId: row.productId,
        })),
        expected: { status: 'assigned' },
-        onlyWhenCanonicalProductNull: true,
+        onlyWhenCanonicalProductDifferent: true,
      })
      await resolveExceptionsForItems(attached.map((row) => Number(row.id)))
    }
@@ -474,15 +534,52 @@ export async function normalizeProductsForItems(itemIds: number[]): Promise<{
        oi.normalized_sku_name,
        oi.normalized_sku_series,
        oi.fulfillment_channel,
+        oi.canonical_product_id,
        wpr.rule_key AS work_rule_key,
-        cfg.canonical_product_id AS manual_config_product_id
+        cfg.canonical_product_id AS manual_config_product_id,
+        cfg_product.canonical_sku AS manual_config_product_sku
      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 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
+      LEFT JOIN product_catalog cfg_product ON cfg_product.id = cfg.canonical_product_id
+      LEFT JOIN product_catalog current_product ON current_product.id = oi.canonical_product_id
      WHERE oi.fulfillment_status = 'assigned'
-        AND oi.canonical_product_id IS NULL
+        AND (
+          oi.canonical_product_id IS NULL
+          OR (
+            oi.fulfillment_channel = 'third_party_voucher'
+            AND (
+              current_product.id IS NULL
+              OR current_product.canonical_sku NOT LIKE 'tpv:%'
+            )
+          )
+          OR (
+            oi.fulfillment_channel = 'affiliate_dash'
+            AND (
+              current_product.id IS NULL
+              OR current_product.canonical_sku NOT LIKE 'aff:%'
+            )
+          )
+          OR (
+            oi.fulfillment_channel IN ('manual', 'on_account')
+            AND (
+              current_product.id IS NULL
+              OR current_product.canonical_sku NOT LIKE 'man:%'
+            )
+          )
+          OR (
+            oi.fulfillment_channel = 'work_order'
+            AND (
+              current_product.id IS NULL
+              OR (
+                current_product.canonical_sku NOT LIKE 'tpl:%'
+                AND current_product.canonical_sku NOT LIKE 'wo:%'
+              )
+            )
+          )
+        )
        AND oi.id = ANY($1::bigint[])
    `,
    [ids],
@@ -496,6 +593,7 @@ export async function normalizeProductsForItems(itemIds: number[]): Promise<{
    }
    if (!productId) {
      unmatched += 1
+      await clearMismatchedProduct(item)
      await upsertAggregatedException({
        reason: 'product_mapping_missing',
        scopeKey: `product_mapping_missing|-|-|${canonicalizeInternalSku(item.internal_sku || item.manual_sku_key || item.sku_code)}|${item.fulfillment_channel || '-'}`,
@@ -513,7 +611,7 @@ export async function normalizeProductsForItems(itemIds: number[]): Promise<{
        canonicalProductId: row.productId,
      })),
      expected: { status: 'assigned' },
-      onlyWhenCanonicalProductNull: true,
+      onlyWhenCanonicalProductDifferent: true,
    })
    await resolveExceptionsForItems(attached.map((row) => Number(row.id)))
    return { matched: attached.length, unmatched }
@@ -0,0 +1,35 @@
+import assert from 'node:assert/strict'
+import { readFile } from 'node:fs/promises'
+import test from 'node:test'
+
+test('卡券 SKU 一个入口完成商品、规则、成本版本和订单回填', async () => {
+  const service = await readFile(new URL('./voucher-product-service.ts', import.meta.url), 'utf8')
+
+  assert.ok(service.includes('withTransaction('))
+  assert.ok(service.includes("channel: 'third_party_voucher'"))
+  assert.ok(service.includes('sys_third_party_voucher_cost'))
+  assert.ok(service.includes('UPDATE order_items oi'))
+  assert.ok(service.includes('syncCostSnapshotsForItems'))
+  assert.ok(service.includes('refreshFactsForDates'))
+  assert.ok(service.includes("resolveExceptionsForItemsByReason(result.itemIds, 'cost_missing')"))
+})
+
+test('第三方卡券修复迁移和归一逻辑都禁止跨来源商品', async () => {
+  const [migration, normalizer, route] = await Promise.all([
+    readFile(
+      new URL('../../db/migrations/143_repair_third_party_voucher_products.sql', import.meta.url),
+      'utf8',
+    ),
+    readFile(new URL('./product-normalize-service.ts', import.meta.url), 'utf8'),
+    readFile(new URL('../../routes/admin/cost-center-catalog.ts', import.meta.url), 'utf8'),
+  ])
+
+  assert.ok(migration.includes('tmp_repair_voucher_items'))
+  assert.ok(migration.includes('current_product.canonical_sku IS DISTINCT FROM'))
+  assert.ok(migration.includes("product.canonical_sku = 'tpv:' || item.internal_sku"))
+  assert.ok(migration.includes("channel.channel <> 'third_party_voucher'"))
+  assert.ok(normalizer.includes('return voucher || null'))
+  assert.ok(normalizer.includes('onlyWhenCanonicalProductDifferent: true'))
+  assert.ok(!route.includes('/voucher-products/ensure-rules'))
+  assert.ok(route.includes("'/cost-center/voucher-products'"))
+})
@@ -0,0 +1,321 @@
+import { query, withTransaction } from '../../db/client.js'
+import type { JsonObject } from '../../types/json.js'
+import { createHttpError } from '../../utils/http.js'
+import { nowIso } from '../../utils/time.js'
+import type { AdminSession } from '../admin/admin-auth-service.js'
+import { orderItemBasisDates, refreshFactsForDates } from '../finance/refresh-facts.js'
+import { upsertInternalSkuChannelWithClient } from '../fulfillment/internal-sku-channel-service.js'
+import { canonicalizeInternalSku, containsCjk } from '../fulfillment/internal-sku-resolver.js'
+import { resolvePriceEffectiveFrom } from './cost-center-normalizers.js'
+import {
+  resolveExceptionsForItemsByReason,
+  resolveOrphanCostExceptions,
+} from './cost-exception-service.js'
+import { mapProduct, type ProductRow } from './cost-center-shared.js'
+import { syncCostSnapshotsForItems } from './cost-snapshot-service.js'
+
+type VoucherProductItemRow = {
+  order_item_id: number
+  basis_date: string
+}
+
+/**
+ * 新增或更新三方卡券 SKU。
+ *
+ * 一个入口同时保证 tpv 商品、卡券主渠道规则、采购成本版本和订单商品关联，
+ * 避免运营跨页面手工建商品后再补规则。
+ */
+export async function upsertAdminVoucherProduct(payload: JsonObject, session: AdminSession) {
+  const skuKey = canonicalizeInternalSku(
+    payload.skuKey ?? payload.sku_key ?? payload.internalSku ?? payload.internal_sku,
+  )
+  if (!skuKey || !containsCjk(skuKey)) {
+    throw createHttpError('卡券内部 SKU 必须包含中文', {
+      statusCode: 400,
+      errorCode: 'voucher_product_sku_required',
+    })
+  }
+
+  const canonicalSku = `tpv:${skuKey}`
+  if (canonicalSku.length > 64) {
+    throw createHttpError('内部 SKU 过长，无法生成卡券商品编码', {
+      statusCode: 400,
+      errorCode: 'voucher_product_sku_too_long',
+    })
+  }
+
+  const amountFen = Number(payload.amountFen ?? payload.amount_fen)
+  if (!Number.isFinite(amountFen) || amountFen <= 0 || !Number.isInteger(amountFen)) {
+    throw createHttpError('卡券采购价必须是大于 0 的整数分', {
+      statusCode: 400,
+      errorCode: 'voucher_product_amount_invalid',
+    })
+  }
+
+  const now = nowIso()
+  const requestedEffectiveFrom = payload.effectiveFrom ?? payload.effective_from
+  const initialEffectiveFrom = resolvePriceEffectiveFrom(payload)
+  const result = await withTransaction(async (client) => {
+    const productResult = await client.query<ProductRow>(
+      `
+        INSERT INTO product_catalog (
+          canonical_sku, canonical_name, category, unit_name, status,
+          attributes_json, created_at, updated_at
+        ) VALUES (
+          $1, $2, '卡券', '张', 'active',
+          $3::jsonb, $4, $4
+        )
+        ON CONFLICT (canonical_sku) DO UPDATE SET
+          status = 'active',
+          canonical_name = EXCLUDED.canonical_name,
+          attributes_json = product_catalog.attributes_json || EXCLUDED.attributes_json,
+          updated_at = EXCLUDED.updated_at
+        RETURNING *
+      `,
+      [
+        canonicalSku,
+        skuKey,
+        JSON.stringify({
+          source: 'third_party_voucher',
+          auto_created: true,
+          managed_by_voucher_config: true,
+        }),
+        now,
+      ],
+    )
+    const product = productResult.rows[0]
+    const productId = Number(product?.id || 0)
+    if (!product || !productId) {
+      throw createHttpError('卡券标准商品保存失败', { statusCode: 500 })
+    }
+
+    await client.query(
+      `
+        INSERT INTO product_channel_capabilities (
+          product_id, channel, priority, is_fallback, enabled, source, created_at, updated_at
+        ) VALUES ($1, 'third_party_voucher', 10, FALSE, TRUE, 'voucher', $2, $2)
+        ON CONFLICT (product_id, channel) DO UPDATE SET
+          priority = 10,
+          is_fallback = FALSE,
+          enabled = TRUE,
+          source = 'voucher',
+          updated_at = EXCLUDED.updated_at
+      `,
+      [productId, now],
+    )
+
+    await upsertInternalSkuChannelWithClient(client, {
+      internalSku: skuKey,
+      channel: 'third_party_voucher',
+      priority: 10,
+      isFallback: false,
+      enabled: true,
+      canonicalProductId: productId,
+      configJson: { voucherSku: skuKey },
+      source: 'voucher',
+      note: String(payload.note || '').slice(0, 300),
+      adminUserId: session.userId,
+      replacePriority: true,
+    })
+
+    const repairedItems = await client.query<VoucherProductItemRow>(
+      `
+        UPDATE order_items oi
+        SET canonical_product_id = $2,
+            updated_at = $3
+        FROM orders o
+        WHERE o.id = oi.order_id
+          AND oi.fulfillment_channel = 'third_party_voucher'
+          AND resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code) = $1
+          AND oi.canonical_product_id IS DISTINCT FROM $2
+        RETURNING
+          oi.id AS order_item_id,
+          (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date::text
+            AS basis_date
+      `,
+      [skuKey, productId, now],
+    )
+    const itemIds = repairedItems.rows.map((row) => Number(row.order_item_id))
+    if (itemIds.length > 0) {
+      await client.query(
+        `UPDATE cost_snapshots
+         SET canonical_product_id = $2
+         WHERE order_item_id = ANY($1::bigint[])
+           AND superseded_at IS NULL AND reversed_at IS NULL`,
+        [itemIds, productId],
+      )
+    }
+
+    const costItemResult = await client.query<{ id: number }>(
+      `SELECT id FROM cost_items
+       WHERE code = 'sys_third_party_voucher_cost' AND status = 'active'
+       LIMIT 1`,
+    )
+    const costItemId = Number(costItemResult.rows[0]?.id || 0)
+    if (!costItemId) {
+      throw createHttpError('未找到第三方卡券采购成本项', {
+        statusCode: 500,
+        errorCode: 'voucher_cost_item_missing',
+      })
+    }
+
+    let effectiveFrom = initialEffectiveFrom
+    if (!requestedEffectiveFrom && repairedItems.rows.length > 0) {
+      const missingDates = await client.query<{ basis_date: string }>(
+        `
+          SELECT MIN(
+            (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
+          WHERE oi.id = ANY($1::bigint[])
+            AND NOT EXISTS (
+              SELECT 1 FROM cost_snapshots cs
+              WHERE cs.order_item_id = oi.id
+                AND cs.cost_item_id = $2
+                AND cs.superseded_at IS NULL AND cs.reversed_at IS NULL
+            )
+        `,
+        [itemIds, costItemId],
+      )
+      const earliestDate = String(missingDates.rows[0]?.basis_date || '').trim()
+      if (earliestDate) effectiveFrom = `${earliestDate}T00:00:00+08:00`
+    }
+
+    const existingResult = await client.query<{
+      id: number
+      amount_fen: number
+      effective_from: string
+    }>(
+      `SELECT id, amount_fen, effective_from
+       FROM cost_versions
+       WHERE cost_item_id = $1
+         AND canonical_product_id = $2
+         AND status = 'active'
+       ORDER BY effective_from DESC, id DESC
+       LIMIT 1`,
+      [costItemId, productId],
+    )
+    const existing = existingResult.rows[0]
+    const targetTime = Date.parse(effectiveFrom)
+    let costVersionId = Number(existing?.id || 0)
+
+    if (!existing || targetTime < Date.parse(existing.effective_from)) {
+      if (existing) {
+        await client.query(
+          `UPDATE cost_versions
+           SET status = 'retired', updated_at = $1
+           WHERE id = $2`,
+          [now, existing.id],
+        )
+      }
+      const inserted = await client.query<{ id: number }>(
+        `INSERT INTO cost_versions (
+           cost_item_id, scope_type, provider, shop_id, canonical_product_id,
+           work_product_rule_id, amount_fen, rate_bps, effective_from, effective_to,
+           source, status, change_reason, created_by_admin_user_id,
+           approved_by_admin_user_id, created_at, updated_at
+         ) VALUES (
+           $1, 'product', '', '', $2,
+           NULL, $3, 0, $4, NULL,
+           'manual', 'active', $5, $6,
+           $6, $7, $7
+         )
+         RETURNING id`,
+        [
+          costItemId,
+          productId,
+          amountFen,
+          effectiveFrom,
+          String(payload.changeReason || '第三方卡券 SKU 新增或采购价调整'),
+          session.userId,
+          now,
+        ],
+      )
+      costVersionId = Number(inserted.rows[0]?.id || 0)
+    } else if (targetTime === Date.parse(existing.effective_from)) {
+      await client.query(
+        `UPDATE cost_versions
+         SET amount_fen = $1,
+             change_reason = $2,
+             updated_at = $3
+         WHERE id = $4`,
+        [
+          amountFen,
+          String(payload.changeReason || '第三方卡券 SKU 新增或采购价调整'),
+          now,
+          existing.id,
+        ],
+      )
+    } else {
+      await client.query(
+        `UPDATE cost_versions
+         SET status = 'retired', effective_to = $1, updated_at = $2
+         WHERE id = $3`,
+        [effectiveFrom, now, existing.id],
+      )
+      const inserted = await client.query<{ id: number }>(
+        `INSERT INTO cost_versions (
+           cost_item_id, scope_type, provider, shop_id, canonical_product_id,
+           work_product_rule_id, amount_fen, rate_bps, effective_from, effective_to,
+           source, status, change_reason, created_by_admin_user_id,
+           approved_by_admin_user_id, created_at, updated_at
+         ) VALUES (
+           $1, 'product', '', '', $2,
+           NULL, $3, 0, $4, NULL,
+           'manual', 'active', $5, $6,
+           $6, $7, $7
+         )
+         RETURNING id`,
+        [
+          costItemId,
+          productId,
+          amountFen,
+          effectiveFrom,
+          String(payload.changeReason || '第三方卡券 SKU 新增或采购价调整'),
+          session.userId,
+          now,
+        ],
+      )
+      costVersionId = Number(inserted.rows[0]?.id || 0)
+    }
+
+    if (!costVersionId) {
+      throw createHttpError('卡券采购成本版本保存失败', { statusCode: 500 })
+    }
+
+    return {
+      product: mapProduct(product),
+      itemIds,
+      dates: [...new Set(repairedItems.rows.map((row) => row.basis_date).filter(Boolean))],
+      costVersionId,
+      effectiveFrom,
+      amountFen,
+    }
+  })
+
+  let snapshotResult = { created: 0, superseded: 0, missing: 0, affectedDates: [] as string[] }
+  if (result.itemIds.length > 0) {
+    snapshotResult = await syncCostSnapshotsForItems(result.itemIds).catch(() => snapshotResult)
+    await resolveExceptionsForItemsByReason(result.itemIds, 'product_mapping_missing').catch(
+      () => undefined,
+    )
+    await resolveExceptionsForItemsByReason(result.itemIds, 'cost_missing').catch(() => undefined)
+  }
+  const affectedDates = [
+    ...new Set([
+      ...result.dates,
+      ...snapshotResult.affectedDates,
+      ...(await orderItemBasisDates(result.itemIds).catch(() => [])),
+    ]),
+  ]
+  await refreshFactsForDates(affectedDates).catch(() => undefined)
+  await resolveOrphanCostExceptions().catch(() => undefined)
+
+  return {
+    ...result,
+    linkedItemCount: result.itemIds.length,
+    snapshotCreatedCount: snapshotResult.created,
+    refreshedDays: affectedDates.length,
+  }
+}
@@ -115,6 +115,14 @@ test('人工确认前必须自动匹配履约配置，配置保存/定时任务
      configSource.includes('dateFrom: shiftDate(today, -29)'),
    '保存人工 SKU 配置后必须立即回填最近 30 天订单项',
  )
+  const saveStart = configSource.indexOf('export async function saveManualFulfillmentConfig')
+  const saveEnd = configSource.indexOf('async function retryPendingManualItems', saveStart)
+  const saveBody = configSource.slice(saveStart, saveEnd)
+  assert.ok(
+    saveBody.indexOf('withTransaction(') <
+      saveBody.indexOf('await syncManualChannelRule(saved, client)'),
+    '标准商品、人工配置和人工渠道规则必须同一事务提交',
+  )
  assert.ok(
    jobSource.includes("fulfillment_channel = 'manual'") &&
      jobSource.includes('fulfillment_config_id'),
@@ -157,6 +165,15 @@ test('人工确认前必须自动匹配履约配置，配置保存/定时任务
  )
 })

+test('统一结算金额未知时只生成待对账账单，不写 0 元实际成本', async () => {
+  const source = await readFile(new URL('./manual-fulfillment-service.ts', import.meta.url), 'utf8')
+
+  assert.ok(source.includes("platformKey === 'on_account' && costFen <= 0"))
+  assert.ok(source.includes('if (!onAccountCostPending)'))
+  assert.ok(source.includes("reason: 'on_account_cost_pending'"))
+  assert.ok(source.includes('costPending: onAccountCostPending'))
+})
+
 test('人工成本缺失由管道按配置与快照生成并收口', async () => {
  const [pipelineSource, manualSource] = await Promise.all([
    readFile(new URL('../scheduler/finance-pipeline-jobs.ts', import.meta.url), 'utf8'),
@@ -21,6 +21,7 @@ import {
  attachFulfillmentConfigurationsWithExecutor,
  transitionFulfillmentStateWithExecutor,
 } from '../fulfillment/fulfillment-state-service.js'
+import { upsertAggregatedException } from '../cost-center/cost-exception-service.js'
 import {
  calculateManualCostFen,
  normalizeManualCreditQuantity,
@@ -411,6 +412,20 @@ export async function confirmManualFulfillment(payload: JsonObject, session: Adm
        errorCode: 'manual_fulfillment_platform_required',
      })
    }
+    if (selectedChannel.platformKey === 'on_account') {
+      const existingBill = await client.query<{ id: number }>(
+        `SELECT id FROM on_account_bills
+         WHERE order_item_id = $1 AND status IN ('pending', 'reconciled')
+         LIMIT 1`,
+        [orderItemId],
+      )
+      if (existingBill.rows[0]) {
+        throw createHttpError('该订单项已生成统一结算账单，不能重复操作', {
+          statusCode: 409,
+          errorCode: 'manual_fulfillment_on_account_already_exists',
+        })
+      }
+    }

    // 锁内复查：并发/双击的第二个请求会等锁，醒来后看到快照直接拒绝，不会重复扣款。
    const existing = await client.query<{ id: number }>(
@@ -447,6 +462,7 @@ export async function confirmManualFulfillment(payload: JsonObject, session: Adm
            providedCostFen,
          })
    const costFen = costPlan.costFen
+    const onAccountCostPending = platformKey === 'on_account' && costFen <= 0
    const payeeName = optionalText(payload.payeeName ?? payload.payee_name, 120)
    const note = optionalText(payload.note, 300)
    const fulfillmentReference = buildManualFulfillmentReference({
@@ -455,16 +471,8 @@ export async function confirmManualFulfillment(payload: JsonObject, session: Adm
      orderItemId,
    })

-    // 先把成本项/版本解析完再扣钱：版本缺失不能发生在扣款之后。
-    const costItemId = await resolveManualCostItemIdWithClient(client)
-    const versionId = await resolveCostVersionIdWithClient(client, costItemId, row, now)
-    if (!versionId) {
-      throw createHttpError('该成本项没有可用的成本版本，请先在成本中心配置后重试', {
-        statusCode: 409,
-        errorCode: 'manual_fulfillment_cost_version_missing',
-      })
-    }
-
+    // 扣款前先确定最终成本模式。统一结算未填金额时只生成待对账账单，
+    // 不写 0 元 actual 成本快照，避免利润被虚增。
    let fundExpenseId: number | null = null
    let creditEntryId: number | null = null
    let balanceAfter: number | null = null
@@ -522,11 +530,22 @@ export async function confirmManualFulfillment(payload: JsonObject, session: Adm
      subjectSource = 'platform_credit'
    }

-    // subject_id 固定为 order_item_id：不再与 manual_expenses.id /
-    // platform_credit_entries.id 混用，避免两张表序列撞车。
-    // subject_type 仍为 manual_case，资金流水只放在 metadata 里。
-    const snapshot = await client.query<{ id: number }>(
-      `
+    let snapshotId: number | null = null
+    if (!onAccountCostPending) {
+      const costItemId = await resolveManualCostItemIdWithClient(client)
+      const versionId = await resolveCostVersionIdWithClient(client, costItemId, row, now)
+      if (!versionId) {
+        throw createHttpError('该成本项没有可用的成本版本，请先在成本中心配置后重试', {
+          statusCode: 409,
+          errorCode: 'manual_fulfillment_cost_version_missing',
+        })
+      }
+
+      // subject_id 固定为 order_item_id：不再与 manual_expenses.id /
+      // platform_credit_entries.id 混用，资金流水只放在 metadata 里。
+      // subject_type 仍为 manual_case。
+      const snapshot = await client.query<{ id: number }>(
+        `
        INSERT INTO cost_snapshots (
          subject_type, subject_id, order_id, order_item_id, work_order_id,
          cost_item_id, cost_version_id, canonical_product_id,
@@ -540,50 +559,51 @@ export async function confirmManualFulfillment(payload: JsonObject, session: Adm
        )
        ON CONFLICT DO NOTHING
        RETURNING id
-      `,
-      [
-        orderItemId,
-        row.order_id_resolved,
-        row.work_order_id,
-        costItemId,
-        versionId,
-        row.canonical_product_id,
-        costFen,
-        JSON.stringify({
-          source: 'manual_fulfillment_confirm',
-          subjectSource,
-          configId: Number(row.config_id),
-          platformKey,
-          platformLabel: PLATFORM_LABELS[platformKey] || platformKey,
-          creditOwnerAdminUserId:
-            platformKey === 'fund_cash'
-              ? null
-              : Number(payload.creditOwnerAdminUserId || 0) || null,
-          creditQuantity,
-          creditUnitPriceFen: unitPriceFen,
-          costMode: costPlan.costMode,
+        `,
+        [
+          orderItemId,
+          row.order_id_resolved,
+          row.work_order_id,
+          costItemId,
+          versionId,
+          row.canonical_product_id,
          costFen,
-          fundExpenseId,
-          creditEntryId,
-          balanceAfter,
-          payeeName,
-          note,
-          fulfilledBy: {
-            userId: Number(session.userId || 0),
-            displayName: String(session.displayName || ''),
-            role: String(session.role || ''),
-          },
-        }),
-        now,
-      ],
-    )
-    const snapshotId = snapshot.rows[0]?.id ?? null
-    if (!snapshotId) {
-      // 订单项唯一索引兜底：任何竞态都会走到这里，抛错回滚本次扣款。
-      throw createHttpError('该订单项已有人工成本快照，不能重复录入', {
-        statusCode: 409,
-        errorCode: 'manual_fulfillment_snapshot_conflict',
-      })
+          JSON.stringify({
+            source: 'manual_fulfillment_confirm',
+            subjectSource,
+            configId: Number(row.config_id),
+            platformKey,
+            platformLabel: PLATFORM_LABELS[platformKey] || platformKey,
+            creditOwnerAdminUserId:
+              platformKey === 'fund_cash'
+                ? null
+                : Number(payload.creditOwnerAdminUserId || 0) || null,
+            creditQuantity,
+            creditUnitPriceFen: unitPriceFen,
+            costMode: costPlan.costMode,
+            costFen,
+            fundExpenseId,
+            creditEntryId,
+            balanceAfter,
+            payeeName,
+            note,
+            fulfilledBy: {
+              userId: Number(session.userId || 0),
+              displayName: String(session.displayName || ''),
+              role: String(session.role || ''),
+            },
+          }),
+          now,
+        ],
+      )
+      snapshotId = snapshot.rows[0]?.id ?? null
+      if (!snapshotId) {
+        // 订单项唯一索引兜底：任何竞态都会走到这里，抛错回滚本次扣款。
+        throw createHttpError('该订单项已有人工成本快照，不能重复录入', {
+          statusCode: 409,
+          errorCode: 'manual_fulfillment_snapshot_conflict',
+        })
+      }
    }

    if (platformKey === 'on_account') {
@@ -641,30 +661,46 @@ export async function confirmManualFulfillment(payload: JsonObject, session: Adm
      reason: platformKey,
    })

-    // 人工缺成本异常立即收口，不等下一轮定时任务。
-    await client.query(
-      `UPDATE cost_exceptions
-       SET status = 'resolved', resolved_at = $2
-       WHERE subject_type = 'order_item'
-         AND subject_id = $1
-         AND reason = 'manual_cost_missing'
-         AND status = 'open'`,
-      [orderItemId, now],
-    )
+    if (onAccountCostPending) {
+      await upsertAggregatedException({
+        reason: 'on_account_cost_pending',
+        scopeKey: `on_account_cost_pending|-|-|${row.internal_sku}|on_account`,
+        orderItemId,
+        detail: {
+          billId: onAccountBillId,
+          internalSku: row.internal_sku,
+          platformOrderId: row.platform_order_id,
+        },
+        client,
+      })
+    } else {
+      // 人工缺成本异常立即收口，不等下一轮定时任务。
+      await client.query(
+        `UPDATE cost_exceptions
+         SET status = 'resolved', resolved_at = $2
+         WHERE subject_type = 'order_item'
+           AND subject_id = $1
+           AND reason = 'manual_cost_missing'
+           AND status = 'open'`,
+        [orderItemId, now],
+      )
+    }

-    // v2 聚合异常：删除本条明细，没人再引用的异常头自动关闭。
-    await client.query(`DELETE FROM cost_exception_items WHERE order_item_id = $1`, [orderItemId])
-    await client.query(
-      `
-        UPDATE cost_exceptions ce
-        SET status = 'resolved', resolved_at = $1, item_count = 0, last_seen_at = $1
-        WHERE ce.status <> 'resolved'
-          AND NOT EXISTS (
-            SELECT 1 FROM cost_exception_items i WHERE i.exception_id = ce.id
-          )
-      `,
-      [now],
-    )
+    if (!onAccountCostPending) {
+      // v2 聚合异常：成本已经落账后，删除本条明细并自动关闭空异常头。
+      await client.query(`DELETE FROM cost_exception_items WHERE order_item_id = $1`, [orderItemId])
+      await client.query(
+        `
+          UPDATE cost_exceptions ce
+          SET status = 'resolved', resolved_at = $1, item_count = 0, last_seen_at = $1
+          WHERE ce.status <> 'resolved'
+            AND NOT EXISTS (
+              SELECT 1 FROM cost_exception_items i WHERE i.exception_id = ce.id
+            )
+        `,
+        [now],
+      )
+    }

    return {
      orderItemId,
@@ -678,6 +714,7 @@ export async function confirmManualFulfillment(payload: JsonObject, session: Adm
      creditEntryId,
      balanceAfter,
      onAccountBillId,
+      costPending: onAccountCostPending,
    }
  })

@@ -693,7 +730,7 @@ export async function confirmManualFulfillment(payload: JsonObject, session: Adm
 * 事实任务默认只回看最近几天，确认一笔较早的订单时不会自动覆盖，
 * 所以这里按该订单自己的业务日期精确重算。
 */
-async function refreshFactsForOrderItem(orderItemId: number): Promise<void> {
+export async function refreshFactsForOrderItem(orderItemId: number): Promise<void> {
  try {
    const basis = await query<{ stat_date: string | null }>(
      `
@@ -841,7 +878,7 @@ async function matchManualFulfillmentConfigWithClient(client: PoolClient, orderI
  return true
 }

-async function resolveManualCostItemIdWithClient(client: PoolClient) {
+export async function resolveManualCostItemIdWithClient(client: PoolClient) {
  const result = await client.query<{ id: number }>(
    `SELECT id FROM cost_items WHERE code = 'sys_manual_service' AND status = 'active' LIMIT 1`,
  )
@@ -855,7 +892,7 @@ async function resolveManualCostItemIdWithClient(client: PoolClient) {
  return id
 }

-async function resolveCostVersionIdWithClient(
+export async function resolveCostVersionIdWithClient(
  client: PoolClient,
  costItemId: number,
  row: {
@@ -0,0 +1,14 @@
+import assert from 'node:assert/strict'
+import { readFile } from 'node:fs/promises'
+import test from 'node:test'
+
+test('统一结算对账用实际金额替代成本快照并重算事实', async () => {
+  const source = await readFile(new URL('./on-account-bill-service.ts', import.meta.url), 'utf8')
+
+  assert.ok(source.includes('withTransaction('))
+  assert.ok(source.includes("status = 'reconciled'"))
+  assert.ok(source.includes('SET superseded_at = $2'))
+  assert.ok(source.includes("source: 'on_account_reconcile'"))
+  assert.ok(source.includes("'on_account_cost_pending'"))
+  assert.ok(source.includes('refreshFactsForOrderItem'))
+})
@@ -0,0 +1,306 @@
+import { query, withTransaction } from '../../db/client.js'
+import type { JsonObject } from '../../types/json.js'
+import { createHttpError } from '../../utils/http.js'
+import { nowIso } from '../../utils/time.js'
+import type { AdminSession } from '../admin/admin-auth-service.js'
+import { resolveExceptionsForItemsByReason } from '../cost-center/cost-exception-service.js'
+import {
+  refreshFactsForOrderItem,
+  resolveCostVersionIdWithClient,
+  resolveManualCostItemIdWithClient,
+} from './manual-fulfillment-service.js'
+
+type OnAccountBillRow = {
+  id: number
+  bill_no: string
+  order_id: number
+  order_item_id: number
+  sku_key: string
+  canonical_product_id: number | null
+  platform_order_id: string
+  amount_fen: string | number
+  status: 'pending' | 'reconciled' | 'void'
+  payee_name: string
+  note: string
+  metadata_json: Record<string, unknown>
+  created_by_admin_user_id: number | null
+  created_at: string
+  updated_at: string
+  reconciled_at: string | null
+  provider?: string
+  shop_id?: string
+  sku_name?: string
+  revenue_amount_fen?: string | number
+}
+
+function mapBill(row: OnAccountBillRow) {
+  return {
+    id: Number(row.id),
+    billNo: row.bill_no,
+    orderId: Number(row.order_id),
+    orderItemId: Number(row.order_item_id),
+    skuKey: row.sku_key,
+    skuName: row.sku_name || row.sku_key,
+    canonicalProductId: row.canonical_product_id ? Number(row.canonical_product_id) : null,
+    platformOrderId: row.platform_order_id,
+    provider: row.provider || '',
+    shopId: row.shop_id || '',
+    revenueFen: Number(row.revenue_amount_fen || 0),
+    amountFen: Number(row.amount_fen || 0),
+    status: row.status,
+    payeeName: row.payee_name || '',
+    note: row.note || '',
+    metadata: row.metadata_json || {},
+    createdByAdminUserId: row.created_by_admin_user_id
+      ? Number(row.created_by_admin_user_id)
+      : null,
+    createdAt: row.created_at,
+    updatedAt: row.updated_at,
+    reconciledAt: row.reconciled_at,
+  }
+}
+
+const BILL_SELECT = `
+  SELECT
+    bill.*,
+    order_row.provider,
+    order_row.shop_id,
+    item.sku_name,
+    item.revenue_amount_fen
+  FROM on_account_bills bill
+  JOIN orders order_row ON order_row.id = bill.order_id
+  JOIN order_items item ON item.id = bill.order_item_id
+`
+
+export async function listOnAccountBills(input: JsonObject = {}) {
+  const page = Math.max(1, Math.trunc(Number(input.page) || 1))
+  const pageSize = Math.min(100, Math.max(1, Math.trunc(Number(input.pageSize) || 20)))
+  const keyword = String(input.keyword || '').trim()
+  const status = String(input.status || '').trim()
+  const where: string[] = []
+  const params: unknown[] = []
+
+  if (keyword) {
+    params.push(`%${keyword}%`)
+    where.push(`(
+      bill.bill_no ILIKE $${params.length}
+      OR bill.platform_order_id ILIKE $${params.length}
+      OR bill.sku_key ILIKE $${params.length}
+      OR item.sku_name ILIKE $${params.length}
+    )`)
+  }
+  if (['pending', 'reconciled', 'void'].includes(status)) {
+    params.push(status)
+    where.push(`bill.status = $${params.length}`)
+  }
+
+  const whereSql = where.length > 0 ? `WHERE ${where.join(' AND ')}` : ''
+  const total = await query<{ total: number }>(
+    `SELECT COUNT(*)::int AS total
+     FROM on_account_bills bill
+     JOIN order_items item ON item.id = bill.order_item_id
+     ${whereSql}`,
+    params,
+  )
+  const listParams = [...params, pageSize, (page - 1) * pageSize]
+  const result = await query<OnAccountBillRow>(
+    `${BILL_SELECT}
+     ${whereSql}
+     ORDER BY
+       CASE bill.status WHEN 'pending' THEN 0 WHEN 'reconciled' THEN 1 ELSE 2 END,
+       bill.created_at DESC,
+       bill.id DESC
+     LIMIT $${params.length + 1} OFFSET $${params.length + 2}`,
+    listParams,
+  )
+
+  return {
+    items: result.rows.map(mapBill),
+    pagination: {
+      page,
+      pageSize,
+      total: Number(total.rows[0]?.total || 0),
+    },
+  }
+}
+
+export async function reconcileOnAccountBill(
+  billIdInput: unknown,
+  payload: JsonObject,
+  session: AdminSession,
+) {
+  const billId = Number(billIdInput)
+  if (!Number.isInteger(billId) || billId <= 0) {
+    throw createHttpError('统一结算账单 ID 无效', { statusCode: 400 })
+  }
+  const amountFen = Number(payload.amountFen ?? payload.amount_fen)
+  if (!Number.isFinite(amountFen) || amountFen <= 0 || !Number.isInteger(amountFen)) {
+    throw createHttpError('实际结算金额必须是大于 0 的整数分', {
+      statusCode: 400,
+      errorCode: 'on_account_amount_invalid',
+    })
+  }
+
+  const now = nowIso()
+  const result = await withTransaction(async (client) => {
+    const billResult = await client.query<OnAccountBillRow>(
+      `${BILL_SELECT}
+       WHERE bill.id = $1
+       LIMIT 1
+       FOR UPDATE OF bill, item`,
+      [billId],
+    )
+    const bill = billResult.rows[0]
+    if (!bill) {
+      throw createHttpError('统一结算账单不存在', { statusCode: 404 })
+    }
+    if (bill.status !== 'pending') {
+      throw createHttpError('只有待对账账单可以确认结算', {
+        statusCode: 409,
+        errorCode: 'on_account_bill_not_pending',
+      })
+    }
+
+    const orderResult = await client.query<{
+      provider: string
+      shop_id: string
+      paid_at: string | null
+      created_at: string
+    }>(
+      `SELECT provider, shop_id, paid_at, created_at
+       FROM orders WHERE id = $1`,
+      [bill.order_id],
+    )
+    const order = orderResult.rows[0]
+    if (!order) {
+      throw createHttpError('统一结算账单关联订单不存在', { statusCode: 409 })
+    }
+    const basisAt = order.paid_at || order.created_at
+    const period = await client.query<{ status: string }>(
+      `SELECT status FROM finance_periods
+       WHERE period_key = to_char($1::timestamptz AT TIME ZONE 'Asia/Shanghai', 'YYYY-MM')
+       LIMIT 1`,
+      [basisAt],
+    )
+    if (period.rows[0]?.status === 'closed') {
+      throw createHttpError('该订单所属期间已关账，请先反关账后再对账', {
+        statusCode: 409,
+        errorCode: 'on_account_period_closed',
+      })
+    }
+
+    const workOrder = await client.query<{ product_rule_id: number | null }>(
+      `SELECT product_rule_id FROM work_orders
+       WHERE order_item_id = $1 AND status <> 'cancelled'
+       ORDER BY id DESC LIMIT 1`,
+      [bill.order_item_id],
+    )
+    const costItemId = await resolveManualCostItemIdWithClient(client)
+    const versionId = await resolveCostVersionIdWithClient(
+      client,
+      costItemId,
+      {
+        order_id_resolved: Number(bill.order_id),
+        provider: order.provider,
+        shop_id: order.shop_id,
+        canonical_product_id: bill.canonical_product_id,
+        work_product_rule_id: workOrder.rows[0]?.product_rule_id || null,
+        credit_unit_price_fen: '1',
+      },
+      basisAt,
+    )
+    if (!versionId) {
+      throw createHttpError('人工履约成本版本不可用，无法完成对账', {
+        statusCode: 409,
+        errorCode: 'on_account_cost_version_missing',
+      })
+    }
+
+    await client.query(
+      `UPDATE cost_snapshots
+       SET superseded_at = $2
+       WHERE order_item_id = $1
+         AND subject_type = 'manual_case'
+         AND superseded_at IS NULL
+         AND reversed_at IS NULL`,
+      [bill.order_item_id, now],
+    )
+    const snapshot = await client.query<{ id: number }>(
+      `INSERT INTO cost_snapshots (
+         subject_type, subject_id, order_id, order_item_id, work_order_id,
+         cost_item_id, cost_version_id, canonical_product_id,
+         quantity, unit_amount_fen, total_amount_fen, snapshot_stage, source,
+         metadata_json, cost_basis_at, resolved_at, created_at
+       ) VALUES (
+         'manual_case', $1, $2, $1, NULL,
+         $3, $4, $5,
+         1, $6, $6, 'actual', 'manual_input',
+         $7::jsonb, $8, $9, $9
+       )
+       RETURNING id`,
+      [
+        bill.order_item_id,
+        bill.order_id,
+        costItemId,
+        versionId,
+        bill.canonical_product_id,
+        amountFen,
+        JSON.stringify({
+          source: 'on_account_reconcile',
+          billId: bill.id,
+          billNo: bill.bill_no,
+          platformKey: 'on_account',
+          previousAmountFen: Number(bill.amount_fen || 0),
+          reconciledBy: {
+            userId: Number(session.userId || 0),
+            displayName: String(session.displayName || ''),
+          },
+        }),
+        basisAt,
+        now,
+      ],
+    )
+    const snapshotId = Number(snapshot.rows[0]?.id || 0)
+    if (!snapshotId) {
+      throw createHttpError('统一结算成本快照保存失败', { statusCode: 500 })
+    }
+
+    const updated = await client.query<OnAccountBillRow>(
+      `UPDATE on_account_bills
+       SET amount_fen = $1,
+           status = 'reconciled',
+           payee_name = $2,
+           note = $3,
+           metadata_json = metadata_json || $4::jsonb,
+           reconciled_at = $5,
+           updated_at = $5
+       WHERE id = $6
+       RETURNING *`,
+      [
+        amountFen,
+        String(payload.payeeName ?? payload.payee_name ?? bill.payee_name ?? '').slice(0, 120),
+        String(payload.note ?? bill.note ?? '').slice(0, 300),
+        JSON.stringify({
+          reconciledByAdminUserId: Number(session.userId || 0) || null,
+          costSnapshotId: snapshotId,
+        }),
+        now,
+        bill.id,
+      ],
+    )
+    return {
+      bill: mapBill({ ...bill, ...updated.rows[0]!, amount_fen: amountFen }),
+      orderItemId: Number(bill.order_item_id),
+      snapshotId,
+      amountFen,
+    }
+  })
+
+  await Promise.all([
+    resolveExceptionsForItemsByReason([result.orderItemId], 'on_account_cost_pending'),
+    resolveExceptionsForItemsByReason([result.orderItemId], 'manual_cost_missing'),
+    resolveExceptionsForItemsByReason([result.orderItemId], 'cost_missing'),
+  ]).catch(() => undefined)
+  await refreshFactsForOrderItem(result.orderItemId).catch(() => undefined)
+  return result
+}
@@ -434,7 +434,10 @@ export async function assignPendingOrderItemChannels(
        .filter((id) => id > 0)
        .slice(0, 2000)
      result.assigned += assignedItemIds.length
-      result.assignedItemIds = assignedItemIds
+      result.assignedItemIds = [...new Set([...result.assignedItemIds, ...assignedItemIds])].slice(
+        0,
+        2000,
+      )
      await scheduleVoucherConsumeForOrderItems(assignedItemIds, 'order_received').catch(
        (error) => {
          logWarn('[channel-assign]', '订单收到后安排自动核销失败', {
@@ -2388,60 +2391,3 @@ export async function ensureVoucherRuleForSku(skuCode: string): Promise<boolean>
  })
  return true
 }
-
-/**
- * 为卡券商品（tpv: 前缀）批量补渠道规则。
- *
- * 卡券没有履约载体证据，只能靠规则判定；商品目录建好后一次补齐，
- * 避免运营按 SKU 逐条点「指定渠道」。
- */
-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 canonicalize_internal_sku(channel.internal_sku) =
-           canonicalize_internal_sku(substring(product.canonical_sku from 5))
-       AND channel.channel <> 'third_party_voucher'
-       AND channel.enabled = TRUE
-       AND channel.is_fallback = FALSE`,
-  )
-  const created = await 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
-      )
-      SELECT
-        canonicalize_internal_sku(substring(p.canonical_sku from 5)),
-        '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 canonicalize_internal_sku(substring(p.canonical_sku from 5)) <> ''
-      ON CONFLICT (internal_sku, channel) DO UPDATE SET
-        priority = 10,
-        is_fallback = FALSE,
-        enabled = TRUE,
-        canonical_product_id = EXCLUDED.canonical_product_id,
-        config_json = internal_sku_channels.config_json || EXCLUDED.config_json,
-        source = EXCLUDED.source,
-        updated_at = NOW()
-      RETURNING id
-    `,
-  )
-  const total = await query<{ count: number }>(
-    `SELECT COUNT(*)::int AS count FROM product_catalog
-     WHERE status = 'active' AND canonical_sku LIKE 'tpv:%'`,
-  )
-  return { created: created.rowCount || 0, productCount: Number(total.rows[0]?.count || 0) }
-}
@@ -16,6 +16,10 @@ const currentAttemptMigration = readFileSync(
  new URL('../../db/migrations/138_fulfillment_current_attempt.sql', import.meta.url),
  'utf8',
 )
+const currentLatestMigration = readFileSync(
+  new URL('../../db/migrations/145_fulfillment_current_attempt_latest.sql', import.meta.url),
+  'utf8',
+)

 test('自动发货历史非活跃任务默认阻止后台自动重开', () => {
  assert.ok(carrierCreation.includes('INACTIVE_CARRIER_TASK_STATUSES.includes'))
@@ -55,3 +59,9 @@ test('创建新履约尝试时自动退休旧当前尝试', () => {
  assert.ok(taskRepo.includes('$26, TRUE'))
  assert.ok(taskRepo.includes('SELECT id FROM order_items WHERE id = $1 FOR UPDATE'))
 })
+
+test('当前尝试必须是同单元最新任务，统一视图不能读到已退休尝试', () => {
+  assert.ok(currentLatestMigration.includes('assert_fulfillment_current_is_latest'))
+  assert.ok(currentLatestMigration.includes('newer.id > COALESCE(NEW.id, 0)'))
+  assert.ok(currentLatestMigration.includes("ERRCODE = '23514'"))
+})
@@ -10,6 +10,10 @@ const auditScript = readFileSync(
  new URL('../../../scripts/audit-fulfillment-state.ts', import.meta.url),
  'utf8',
 )
+const configSyncMigration = readFileSync(
+  new URL('../../db/migrations/144_fulfillment_shadow_configuration_sync.sql', import.meta.url),
+  'utf8',
+)

 test('正式履约影子模型包含履约、尝试和异常三张表', () => {
  assert.ok(migration.includes('CREATE TABLE IF NOT EXISTS fulfillments'))
@@ -27,6 +31,21 @@ test('影子模型从现有真源回填并持续双写', () => {
  assert.ok(migration.includes('assign_fulfillment_event_shadow'))
 })

+test('人工配置和统一渠道规则变化会刷新履约影子', () => {
+  assert.ok(configSyncMigration.includes('trg_manual_fulfillment_config_shadow'))
+  assert.ok(configSyncMigration.includes('trg_internal_sku_channel_shadow'))
+  assert.ok(configSyncMigration.includes('sync_fulfillment_shadow(item.id)'))
+})
+
+test('人工履约配置不能关联其他来源商品', () => {
+  assert.ok(configSyncMigration.includes('trg_manual_config_product_source'))
+  assert.ok(configSyncMigration.includes("product.canonical_sku LIKE 'man:%'"))
+  assert.ok(configSyncMigration.includes("ERRCODE = '23514'"))
+  assert.ok(configSyncMigration.includes('BEFORE INSERT OR UPDATE ON manual_fulfillment_configs'))
+  assert.ok(configSyncMigration.includes('config.enabled = TRUE'))
+  assert.ok(configSyncMigration.includes("SET status = 'active'"))
+})
+
 test('明确渠道即视为已路由，不用渠道配置缺失覆盖已选渠道', () => {
  assert.ok(migration.includes('v_fulfillment_item_state_raw'))
  assert.ok(migration.includes("WHEN raw.intended_channel IS NOT NULL THEN 'configured'"))
@@ -43,6 +43,10 @@ test('待分类查询使用统一读模型并返回异常数量', () => {
  assert.ok(channelAssign.includes('issueCategoryInput'))
 })

+test('批量判渠保留所有批次的即时成本同步订单项', () => {
+  assert.ok(channelAssign.includes('...new Set([...result.assignedItemIds, ...assignedItemIds])'))
+})
+
 test('重试必须按可重试 issueCode 限定，不能按整个 SKU 处理售后和复核项', () => {
  assert.ok(channelAssign.includes('RETRYABLE_CARRIER_ISSUES'))
  assert.ok(channelAssign.includes('issueCodes?: string[]'))
@@ -1,3 +1,5 @@
+import type { PoolClient } from 'pg'
+
 import { query, withTransaction } from '../../db/client.js'
 import type { FulfillmentChannel } from './channel-types.js'
 import { canonicalizeInternalSku, compareFulfillmentChannel } from './internal-sku-resolver.js'
@@ -127,7 +129,7 @@ export function planInternalSkuChannels(input: InternalSkuChannelRow[]): Interna
  }
 }

-export async function upsertInternalSkuChannel(input: {
+type InternalSkuChannelUpsertInput = {
  internalSku: string
  channel: FulfillmentChannel
  priority?: number
@@ -139,25 +141,29 @@ export async function upsertInternalSkuChannel(input: {
  note?: string | undefined
  adminUserId?: number | null | undefined
  replacePriority?: boolean
-}): Promise<number> {
+}
+
+async function upsertInternalSkuChannelWithExecutor(
+  client: Pick<PoolClient, 'query'>,
+  input: InternalSkuChannelUpsertInput,
+): Promise<number> {
  const internalSku = canonicalizeInternalSku(input.internalSku)
  if (!internalSku) return 0
-  return withTransaction(async (client) => {
-    if (input.enabled !== false && input.isFallback !== true) {
-      await client.query(
-        `UPDATE internal_sku_channels
-         SET is_fallback = TRUE,
-             priority = GREATEST(priority, 90),
-             updated_at = NOW()
-         WHERE canonicalize_internal_sku(internal_sku) = $1
-           AND channel <> $2
-           AND enabled = TRUE
-           AND is_fallback = FALSE`,
-        [internalSku, input.channel],
-      )
-    }
-    const result = await client.query<{ id: number }>(
-      `
+  if (input.enabled !== false && input.isFallback !== true) {
+    await client.query(
+      `UPDATE internal_sku_channels
+       SET is_fallback = TRUE,
+           priority = GREATEST(priority, 90),
+           updated_at = NOW()
+       WHERE canonicalize_internal_sku(internal_sku) = $1
+         AND channel <> $2
+         AND enabled = TRUE
+         AND is_fallback = FALSE`,
+      [internalSku, input.channel],
+    )
+  }
+  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, note, created_by_admin_user_id,
@@ -177,23 +183,35 @@ export async function upsertInternalSkuChannel(input: {
        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,
-        input.channel,
-        Number.isFinite(Number(input.priority)) ? Number(input.priority) : 100,
-        input.isFallback === true,
-        input.enabled !== false,
-        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)
-  })
+    `,
+    [
+      internalSku,
+      input.channel,
+      Number.isFinite(Number(input.priority)) ? Number(input.priority) : 100,
+      input.isFallback === true,
+      input.enabled !== false,
+      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)
+}
+
+export async function upsertInternalSkuChannelWithClient(
+  client: PoolClient,
+  input: InternalSkuChannelUpsertInput,
+): Promise<number> {
+  return upsertInternalSkuChannelWithExecutor(client, input)
+}
+
+export async function upsertInternalSkuChannel(
+  input: InternalSkuChannelUpsertInput,
+): Promise<number> {
+  return withTransaction((client) => upsertInternalSkuChannelWithExecutor(client, input))
 }

 /**
@@ -3,6 +3,10 @@ import { readFileSync } from 'node:fs'
 import test from 'node:test'

 const source = readFileSync(new URL('./voucher-consume-queue-service.ts', import.meta.url), 'utf8')
+const migration = readFileSync(
+  new URL('../../db/migrations/142_voucher_consume_job_codes.sql', import.meta.url),
+  'utf8',
+)

 test('核销任务列表支持关键词、平台、渠道和状态组合筛选', () => {
  assert.ok(source.includes('keyword?: string'))
@@ -14,6 +18,7 @@ test('核销任务列表支持关键词、平台、渠道和状态组合筛选',
  assert.ok(source.includes('status = $'))
  assert.ok(source.includes('platform_order_id ILIKE'))
  assert.ok(source.includes('target_key ILIKE'))
+  assert.ok(source.includes('fulfillment_voucher_codes_search_text(voucher_codes) ILIKE'))
 })

 test('核销任务列表使用服务端总数和分页参数', () => {
@@ -22,3 +27,15 @@ test('核销任务列表使用服务端总数和分页参数', () => {
  assert.ok(source.includes('OFFSET $'))
  assert.ok(source.includes('ORDER BY updated_at DESC, id DESC'))
 })
+
+test('核销任务保存并展示核销码，执行时刷新候选券码', () => {
+  assert.ok(migration.includes('voucher_codes TEXT[]'))
+  assert.ok(migration.includes('fulfillment_voucher_codes_search_text(voucher_codes)'))
+  assert.ok(migration.includes('gin_trgm_ops'))
+  assert.ok(migration.includes('kuaishou_industry_vouchers'))
+  assert.ok(migration.includes('redeem_certificates'))
+  assert.ok(source.includes('INSERT INTO voucher_consume_jobs'))
+  assert.ok(source.includes('candidates.voucherCodes'))
+  assert.ok(source.includes('updateVoucherConsumeJobCodes'))
+  assert.ok(source.includes('voucher_codes = $1::text[]'))
+})
@@ -6,6 +6,7 @@ import { listKuaishouIndustryVouchersByOid } from '../../repositories/kuaishou-i
 import {
  findRedeemCertificateByOrder,
  listRedeemCertificatesByLocalOrder,
+  resolveRedeemVerifyCertNo,
  type RedeemCertificateRow,
 } from '../../repositories/redeem-certificate-repo.js'
 import type { WorkOrderRow } from '../../repositories/worker-platform/index.js'
@@ -70,6 +71,7 @@ export type VoucherConsumeJobRow = {
  max_attempts: number
  locked_until: string | null
  last_error: string
+  voucher_codes: string[]
  result_json: string | Record<string, unknown>
  created_at: string
  updated_at: string
@@ -331,13 +333,14 @@ async function scheduleVoucherConsumeTarget(
        order_item_id,
        task_id,
        work_order_id,
+        voucher_codes,
        due_at,
        status,
        attempt_count,
        max_attempts,
        updated_at
      ) VALUES (
-        $1, $2, $3, $4, $5, $6, $7, $8, $9, $10,
+        $1, $2, $3, $4, $5, $6, $7, $8, $9, $10::text[], $11,
        'pending', 0, 8, NOW()
      )
      ON CONFLICT (provider, target_key)
@@ -355,6 +358,7 @@ async function scheduleVoucherConsumeTarget(
      target.orderItemId,
      target.taskId,
      target.workOrderId,
+      candidates.voucherCodes,
      dueAt,
    ],
  )
@@ -488,7 +492,8 @@ export async function listVoucherConsumeJobs(
      `(CAST(id AS TEXT) ILIKE $${params.length}
        OR platform_order_id ILIKE $${params.length}
        OR target_key ILIKE $${params.length}
-        OR CAST(order_id AS TEXT) ILIKE $${params.length})`,
+        OR CAST(order_id AS TEXT) ILIKE $${params.length}
+        OR fulfillment_voucher_codes_search_text(voucher_codes) ILIKE $${params.length})`,
    )
  }
  if (provider) {
@@ -580,6 +585,10 @@ async function executeKuaishouConsumeJob(job: VoucherConsumeJobRow) {
  if (vouchers.length === 0) {
    throw new Error('未找到待核销的快手电子凭证')
  }
+  await updateVoucherConsumeJobCodes(
+    job.id,
+    vouchers.map((voucher) => String(voucher.voucher_code || '').trim()).filter(Boolean),
+  )

  const task = job.task_id ? await getTaskById(job.task_id) : null
  let consumedCount = 0
@@ -631,6 +640,18 @@ async function executeDouyinConsumeJob(job: VoucherConsumeJobRow) {
  if (certificates.length === 0) {
    throw new Error('未找到待核销的抖音券')
  }
+  await updateVoucherConsumeJobCodes(
+    job.id,
+    Array.from(
+      new Set(
+        certificates.flatMap((certificate) => [
+          String(certificate.cert_no || '').trim(),
+          String(certificate.platform_cert_no || '').trim(),
+          resolveRedeemVerifyCertNo(certificate),
+        ]),
+      ),
+    ).filter(Boolean),
+  )

  let consumedCount = 0
  let alreadyConsumedCount = 0
@@ -779,6 +800,7 @@ function buildSyntheticJob(
    max_attempts: 8,
    locked_until: null,
    last_error: '',
+    voucher_codes: [],
    result_json: {},
    created_at: '',
    updated_at: '',
@@ -837,6 +859,23 @@ async function finishVoucherConsumeJob(
  )
 }

+async function updateVoucherConsumeJobCodes(jobId: number, voucherCodes: string[]) {
+  const codes = Array.from(new Set(voucherCodes.map((code) => String(code || '').trim()))).filter(
+    Boolean,
+  )
+  if (codes.length === 0) {
+    return
+  }
+  await query(
+    `
+      UPDATE voucher_consume_jobs
+      SET voucher_codes = $1::text[], updated_at = NOW()
+      WHERE id = $2
+    `,
+    [codes, jobId],
+  )
+}
+
 async function failVoucherConsumeJob(job: VoucherConsumeJobRow, error: unknown, retry: boolean) {
  const message = error instanceof Error ? error.message : String(error)
  if (!retry) {
@@ -20,6 +20,7 @@ import type {
  AdminKuaishouIndustryVoucherCheckAvailableInput,
  AdminKuaishouIndustryVoucherConsumeInput,
  AdminKuaishouIndustryVoucherDestroyInput,
+  AdminKuaishouIndustryVoucherReissueConsumeInput,
  AdminKuaishouIndustryVoucherResendInput,
  AdminKuaishouIndustryVoucherReverseInput,
  AdminCloudtentaclesTestLoginInput,
@@ -49,6 +50,8 @@ export type AdminKuaishouIndustryVoucherCheckAvailableRouteBody =
 export type AdminKuaishouIndustryVoucherReverseRouteBody = AdminKuaishouIndustryVoucherReverseInput
 export type AdminKuaishouIndustryVoucherConsumeRouteBody = AdminKuaishouIndustryVoucherConsumeInput
 export type AdminKuaishouIndustryVoucherResendRouteBody = AdminKuaishouIndustryVoucherResendInput
+export type AdminKuaishouIndustryVoucherReissueConsumeRouteBody =
+  AdminKuaishouIndustryVoucherReissueConsumeInput
 export type AdminKuaishouIndustryVoucherDestroyRouteBody = AdminKuaishouIndustryVoucherDestroyInput
 export type AdminNotificationConfigRouteBody = AdminNotificationConfigInput
 export type AdminNotificationTestRouteBody = AdminNotificationTestInput
@@ -354,6 +354,25 @@ export type AdminKuaishouIndustryVoucherResendInput = {
  voucherCode?: string
 }

+/**
+ * 历史券补偿：临时重建有效期，重发原券码发码回调，成功后立即核销。
+ * validMinutes 只允许较短窗口，避免发码成功但核销失败时长期暴露可用券。
+ */
+export type AdminKuaishouIndustryVoucherReissueConsumeInput = {
+  oid?: string
+  orderId?: string
+  taskId?: number | string
+  voucherCode?: string
+  validMinutes?: number | string
+  reason?: string
+  consumeType?: string
+  serialNum?: string
+  storeName?: string
+  storeAddress?: string
+  expressCode?: string
+  expressNo?: string
+}
+
 export type AdminKuaishouIndustryVoucherDestroyInput = {
  oid?: string
  orderId?: string