feat(finance): 人工成本单事务/幂等 + 配置自动匹配 + 缺成本监控

- 确认履约改为单事务：锁订单项 → 复查快照 → 扣积分/备用金 → 写成本快照；
  快照失败整体回滚，不再出现「钱扣了、成本没记」
- manual_case 的 subject_id 统一为 order_item_id，新增订单项级唯一索引（迁移 111），
  消除 manual_expenses.id / platform_credit_entries.id 跨表撞车
- 人工 SKU 匹配增加 sku_code / sku_name 兜底；保存配置立即回填最近 30 天，
  定时任务按窗口 rematch，确认时再按 SKU 兜底匹配
- 新增 manual_cost_missing 异常队列：从首次人工配置起算，确认履约立即收口
- 补单元测试与真实 PostgreSQL 集成测试（并发、重复、强制快照失败回滚、双资金链路）
- 校准财务文档并新增上线初始化/验收手册（历史人工不补录，只保证配置后的新单）

验证：backend typecheck/lint/format/check:sql/test 843（839 pass / 4 skip）；
frontend typecheck/lint/test 172；迁移 111 在 scratch 库全量执行 + 重跑幂等。

apps/backend/src/services/admin/admin-fund-service.ts

@@ -1538,11 +1538,15 @@ export async function createAdminFundReturn(payload: JsonObject, session: AdminS
   })
 }
 
-export async function getOwnAgentAccountId(userId: number): Promise<number> {
-  const own = await query<FundAccountRow>(
-    "SELECT * FROM fund_accounts WHERE owner_admin_user_id = $1 AND account_type = 'agent' AND status <> 'closed' ORDER BY id LIMIT 1",
-    [userId],
-  )
+export async function getOwnAgentAccountId(
+  userId: number,
+  client?: Pick<PoolClient, 'query'>,
+): Promise<number> {
+  const sql =
+    "SELECT * FROM fund_accounts WHERE owner_admin_user_id = $1 AND account_type = 'agent' AND status <> 'closed' ORDER BY id LIMIT 1"
+  const own = client
+    ? await client.query<FundAccountRow>(sql, [userId])
+    : await query<FundAccountRow>(sql, [userId])
   return own.rows[0] ? Number(own.rows[0].id) : 0
 }
 

apps/backend/src/services/cost-center/cost-snapshot-sync-service.ts

@@ -672,6 +672,7 @@ async function insertAndResolveCostMissingExceptions(
         SELECT
           oi.id AS order_item_id,
           oi.canonical_product_id,
+          oi.fulfillment_config_id,
           COALESCE(o.paid_at, o.created_at) AS basis_at,
           wo.id AS work_order_id,
           COALESCE(oi.fulfillment_mode, 'manual') AS fulfillment_mode
@@ -714,6 +715,58 @@ async function insertAndResolveCostMissingExceptions(
     [fromDate, toDate],
   )
 
+  const manualInserted = await client.query(
+    `
+      WITH cutoff AS (
+        SELECT MIN(created_at) AS enforce_from
+        FROM manual_fulfillment_configs
+      ),
+      item_scope AS (
+        SELECT
+          oi.id AS order_item_id,
+          oi.fulfillment_config_id,
+          COALESCE(o.paid_at, o.created_at) AS basis_at
+        FROM order_items oi
+        JOIN orders o ON o.id = oi.order_id
+        WHERE o.pay_status IN ('paid', 'refunded')
+          AND (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date
+            BETWEEN $1::date AND $2::date
+          AND oi.fulfillment_mode = 'manual'
+      )
+      INSERT INTO cost_exceptions (
+        subject_type, subject_id, order_item_id, work_order_id,
+        reason, status, detail_json, created_at
+      )
+      SELECT
+        'order_item', s.order_item_id, s.order_item_id, NULL,
+        'manual_cost_missing', 'open',
+        jsonb_build_object(
+          'fulfillmentMode', 'manual',
+          'basisAt', s.basis_at,
+          'configMissing', s.fulfillment_config_id IS NULL,
+          'message', CASE
+            WHEN s.fulfillment_config_id IS NULL
+              THEN '人工订单项未配置履约渠道，请先在成本中心配置人工 SKU'
+            ELSE '人工订单项尚未确认履约并记录成本'
+          END
+        ),
+        NOW()
+      FROM item_scope s
+      CROSS JOIN cutoff c
+      WHERE c.enforce_from IS NOT NULL
+        -- 只监控「第一次配置人工履约之后」的订单项：历史人工单按既定口径不补录。
+        AND s.basis_at >= c.enforce_from
+        AND NOT EXISTS (
+          SELECT 1 FROM cost_snapshots cs
+          WHERE cs.subject_type = 'manual_case'
+            AND cs.order_item_id = s.order_item_id
+            AND cs.superseded_at IS NULL
+        )
+      ON CONFLICT (subject_type, subject_id, reason) DO NOTHING
+    `,
+    [fromDate, toDate],
+  )
+
   const resolved = await client.query(
     `
       UPDATE cost_exceptions ce
@@ -730,5 +783,24 @@ async function insertAndResolveCostMissingExceptions(
     `,
   )
 
-  return { created: inserted.rowCount || 0, resolved: resolved.rowCount || 0 }
+  const manualResolved = await client.query(
+    `
+      UPDATE cost_exceptions ce
+      SET status = 'resolved', resolved_at = NOW()
+      WHERE ce.status = 'open'
+        AND ce.subject_type = 'order_item'
+        AND ce.reason = 'manual_cost_missing'
+        AND EXISTS (
+          SELECT 1 FROM cost_snapshots cs
+          WHERE cs.subject_type = 'manual_case'
+            AND cs.order_item_id = ce.subject_id
+            AND cs.superseded_at IS NULL
+        )
+    `,
+  )
+
+  return {
+    created: (inserted.rowCount || 0) + (manualInserted.rowCount || 0),
+    resolved: (resolved.rowCount || 0) + (manualResolved.rowCount || 0),
+  }
 }

apps/backend/src/services/cost-center/manual-fulfillment-config-service.ts

@@ -1,14 +1,15 @@
 import { query, withTransaction } from '../../db/client.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'
 
 /**
  * 人工履约 SKU 配置服务。
  *
  * 口径（2026-10-24 确认）：
- *   * 配置主键是**内部 SKU**（快手 spec_json.skuNick / 拼多多 spec_json.outSkuSn），
- *     不是商品标题；同名 SKU 算同一个商品，跨渠道共用一份配置；
+ *   * 配置主键优先取内部 SKU（快手 spec_json.skuNick / 拼多多 spec_json.outSkuSn），
+ *     缺失时按 sku_code → sku_name 兜底；同名/同码算同一个商品，跨渠道共用一份配置；
  *   * 配置项：履约渠道（南航/乐玩/ROCK/备用金现金）+ 消耗积分 + 成本模式；
  *   * 换算率固定：1 积分 = 1 分钱；
  *   * 消耗积分 0 = 成本不固定，由客服在确认履约时手填实际成本；
@@ -125,7 +126,9 @@ export async function scanManualSkus(options: ScanOptions) {
           (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date AS stat_date,
           COALESCE(
             NULLIF(oi.spec_json->>'skuNick', ''),
-            NULLIF(oi.spec_json->>'outSkuSn', '')
+            NULLIF(oi.spec_json->>'outSkuSn', ''),
+            NULLIF(oi.sku_code, ''),
+            NULLIF(oi.sku_name, '')
           ) AS sku_key
         FROM order_items oi
         JOIN orders o ON o.id = oi.order_id
@@ -380,6 +383,15 @@ export async function saveManualFulfillmentConfig(payload: JsonObject) {
 
   const config = requiredRow(result.rows[0], '保存人工履约配置失败')
   await syncRegistryConfigId(config.sku_key, config.id)
+  // 配置保存后立刻回填该 SKU 最近 30 天的订单项，避免必须再点一次「重新匹配」。
+  // 历史更早的订单项不自动回填（历史人工成本不补录）；需要时再显式点「重新匹配」。
+  const today = formatShanghaiDate(new Date())
+  await rematchManualFulfillments({
+    skuKey: config.sku_key,
+    onlyUnmatched: true,
+    dateFrom: shiftDate(today, -29),
+    dateTo: today,
+  })
   return mapManualConfig(config, unitPrices)
 }
 
@@ -408,6 +420,9 @@ export async function deleteManualFulfillmentConfig(skuKey: string) {
 export async function rematchManualFulfillments(input: Record<string, unknown> = {}) {
   const skuKey = String(input.skuKey ?? input.sku_key ?? '').trim()
   const onlyUnmatched = input.onlyUnmatched !== false
+  // 定时任务只回填最近窗口，避免一次性把 10 万历史人工单变成「待履约」。
+  const dateFrom = normalizeOptionalDate(input.dateFrom ?? input.date_from, '开始日期')
+  const dateTo = normalizeOptionalDate(input.dateTo ?? input.date_to, '结束日期')
 
   const now = nowIso()
   const result = await query<{ updated_count: number }>(
@@ -422,22 +437,32 @@ export async function rematchManualFulfillments(input: Record<string, unknown> =
         JOIN manual_fulfillment_configs cfg
           ON cfg.sku_key = COALESCE(
             NULLIF(oi.spec_json->>'skuNick', ''),
-            NULLIF(oi.spec_json->>'outSkuSn', '')
+            NULLIF(oi.spec_json->>'outSkuSn', ''),
+            NULLIF(oi.sku_code, ''),
+            NULLIF(oi.sku_name, '')
           )
         WHERE oi.fulfillment_mode = 'manual'
           AND cfg.enabled = TRUE
           AND ($1 = '' OR cfg.sku_key = $1)
           AND ($2 = FALSE OR oi.fulfillment_config_id IS NULL)
+          AND (
+            $3::date IS NULL
+            OR (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date >= $3::date
+          )
+          AND (
+            $4::date IS NULL
+            OR (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date <= $4::date
+          )
       )
       UPDATE order_items oi
       SET fulfillment_config_id = target.config_id,
           canonical_product_id = COALESCE(target.canonical_product_id, oi.canonical_product_id),
-          updated_at = $3
+          updated_at = $5
       FROM target
       WHERE oi.id = target.order_item_id
       RETURNING 1
     `,
-    [skuKey, onlyUnmatched, now],
+    [skuKey, onlyUnmatched, dateFrom, dateTo, now],
   )
 
   // 配置生效后，把 SKU 对应的成本版本挂上（若已配置商品）。
@@ -446,6 +471,8 @@ export async function rematchManualFulfillments(input: Record<string, unknown> =
   return {
     updatedCount: result.rowCount || 0,
     skuKey,
+    dateFrom,
+    dateTo,
   }
 }
 
@@ -629,6 +656,12 @@ function normalizeCreditQuantity(value: unknown) {
   return parsed
 }
 
+function normalizeOptionalDate(value: unknown, label: string) {
+  const text = String(value || '').trim()
+  if (!text) return null
+  return normalizeDate(text, label)
+}
+
 function normalizeDate(value: unknown, label: string) {
   const text = String(value || '').trim()
   if (!/^\d{4}-\d{2}-\d{2}$/.test(text)) {

apps/backend/src/services/cost-center/platform-credit-service.ts

@@ -290,55 +290,73 @@ export async function consumePlatformCredit(input: {
   platformRef?: string
   note?: string
 }): Promise<{ entryId: number; balanceAfter: number; amountFen: number }> {
+  return withTransaction((client) => consumePlatformCreditWithClient(client, input))
+}
+
+/**
+ * 在调用方事务内消耗积分。
+ *
+ * 人工履约的「扣积分 + 写成本快照」必须在同一事务里完成，
+ * 否则快照失败会留下已扣积分、未记成本的资金洞。
+ */
+export async function consumePlatformCreditWithClient(
+  client: PoolClient,
+  input: {
+    ownerAdminUserId: number
+    platformKey: PlatformCreditKey
+    quantity: number
+    relatedTaskId?: number | null
+    platformRef?: string
+    note?: string
+  },
+): Promise<{ entryId: number; balanceAfter: number; amountFen: number }> {
   const unitPrices = await loadUnitPrices()
   const unitPriceFen = unitPrices[input.platformKey]
   const now = nowIso()
 
-  return withTransaction(async (client) => {
-    await ensureAccountRow(client, input.ownerAdminUserId, input.platformKey, unitPriceFen, now)
-    const account = await lockAccount(client, input.ownerAdminUserId, input.platformKey)
-    const balance = Number(account.balance_quantity)
-    if (balance < input.quantity) {
-      throw createHttpError(
-        `${PLATFORM_LABELS[input.platformKey]}积分不足（当前 ${balance}，需要 ${input.quantity}），请联系主管充值`,
-        {
-          statusCode: 409,
-          errorCode: 'platform_credit_balance_insufficient',
-        },
-      )
-    }
-    const nextBalance = balance - input.quantity
-    await client.query(
-      `UPDATE platform_credit_accounts
-       SET balance_quantity = $2, updated_at = $3
-       WHERE id = $1`,
-      [account.id, nextBalance, now],
+  await ensureAccountRow(client, input.ownerAdminUserId, input.platformKey, unitPriceFen, now)
+  const account = await lockAccount(client, input.ownerAdminUserId, input.platformKey)
+  const balance = Number(account.balance_quantity)
+  if (balance < input.quantity) {
+    throw createHttpError(
+      `${PLATFORM_LABELS[input.platformKey]}积分不足（当前 ${balance}，需要 ${input.quantity}），请联系主管充值`,
+      {
+        statusCode: 409,
+        errorCode: 'platform_credit_balance_insufficient',
+      },
     )
-    const entryResult = await client.query<CreditEntryRow>(
-      `
-        INSERT INTO platform_credit_entries (
-          account_id, entry_type, quantity_delta, balance_after, amount_fen,
-          platform_ref, related_task_id, note, created_at
-        ) VALUES ($1, 'consume', $2, $3, $4, $5, $6, $7, $8)
-        RETURNING *
-      `,
-      [
-        account.id,
-        -input.quantity,
-        nextBalance,
-        input.quantity * unitPriceFen,
-        input.platformRef || '',
-        input.relatedTaskId ?? null,
-        input.note || '',
-        now,
-      ],
-    )
-    return {
-      entryId: Number(entryResult.rows[0]?.id || 0),
-      balanceAfter: nextBalance,
-      amountFen: input.quantity * unitPriceFen,
-    }
-  })
+  }
+  const nextBalance = balance - input.quantity
+  await client.query(
+    `UPDATE platform_credit_accounts
+     SET balance_quantity = $2, updated_at = $3
+     WHERE id = $1`,
+    [account.id, nextBalance, now],
+  )
+  const entryResult = await client.query<CreditEntryRow>(
+    `
+      INSERT INTO platform_credit_entries (
+        account_id, entry_type, quantity_delta, balance_after, amount_fen,
+        platform_ref, related_task_id, note, created_at
+      ) VALUES ($1, 'consume', $2, $3, $4, $5, $6, $7, $8)
+      RETURNING *
+    `,
+    [
+      account.id,
+      -input.quantity,
+      nextBalance,
+      input.quantity * unitPriceFen,
+      input.platformRef || '',
+      input.relatedTaskId ?? null,
+      input.note || '',
+      now,
+    ],
+  )
+  return {
+    entryId: Number(entryResult.rows[0]?.id || 0),
+    balanceAfter: nextBalance,
+    amountFen: input.quantity * unitPriceFen,
+  }
 }
 
 /** 积分流水列表。 */

apps/backend/src/services/finance/manual-fulfillment-integrity.test.ts

@@ -0,0 +1,145 @@
+import assert from 'node:assert/strict'
+import { readFile } from 'node:fs/promises'
+import test from 'node:test'
+
+import { resolveManualFulfillmentCost } from './manual-fulfillment-service.js'
+
+test('人工成本：固定成本按配置积分 × 平台换算率', () => {
+  assert.deepEqual(
+    resolveManualFulfillmentCost({
+      creditQuantity: 4,
+      unitPriceFen: 100,
+      providedCostFen: undefined,
+    }),
+    { costFen: 400, costMode: 'fixed' },
+  )
+  // 固定成本时手填金额不生效（客服零输入）。
+  assert.equal(
+    resolveManualFulfillmentCost({ creditQuantity: 2, unitPriceFen: 100, providedCostFen: 999 })
+      .costFen,
+    200,
+  )
+})
+
+test('人工成本：成本不固定必须手填正数金额', () => {
+  assert.deepEqual(
+    resolveManualFulfillmentCost({
+      creditQuantity: 0,
+      unitPriceFen: 1,
+      providedCostFen: 1234,
+    }),
+    { costFen: 1234, costMode: 'variable' },
+  )
+  assert.throws(
+    () => resolveManualFulfillmentCost({ creditQuantity: 0, unitPriceFen: 1 }),
+    /成本不固定/,
+  )
+  assert.throws(
+    () => resolveManualFulfillmentCost({ creditQuantity: 0, unitPriceFen: 1, providedCostFen: 0 }),
+    /成本不固定/,
+  )
+  assert.throws(
+    () =>
+      resolveManualFulfillmentCost({
+        creditQuantity: 0,
+        unitPriceFen: 1,
+        providedCostFen: -1,
+      }),
+    /金额无效/,
+  )
+})
+
+test('人工确认必须单事务：扣款与快照同事务，且使用 with-client 版本', async () => {
+  const source = await readFile(new URL('./manual-fulfillment-service.ts', import.meta.url), 'utf8')
+  const start = source.indexOf('export async function confirmManualFulfillment')
+  const end = source.indexOf('async function resolveFundAccountId')
+  const body = source.slice(start, end)
+
+  assert.ok(body.includes('withTransaction('), '人工确认必须包在事务里')
+  assert.ok(
+    body.indexOf('withTransaction(') < body.indexOf('writeFundExpense(client'),
+    '备用金扣款必须在同一事务内执行',
+  )
+  assert.ok(
+    body.includes('consumePlatformCreditWithClient(client'),
+    '积分扣减必须复用调用方事务（不能再用自带事务的 consumePlatformCredit）',
+  )
+  assert.ok(
+    !body.includes('consumePlatformCredit('),
+    'confirm 内不得调用自带事务的 consumePlatformCredit',
+  )
+  assert.ok(
+    body.includes('FOR UPDATE OF oi') && body.includes("subject_type = 'manual_case'"),
+    '必须先锁订单项并复查快照，避免并发重复扣款',
+  )
+  assert.ok(
+    body.includes("'manual_case', $1") && body.includes('subjectSource'),
+    'manual_case 的 subject_id 必须是 order_item_id，资金来源只写 metadata',
+  )
+})
+
+test('人工确认前必须自动匹配履约配置，配置保存/定时任务也要回填', async () => {
+  const [manualSource, configSource, jobSource] = await Promise.all([
+    readFile(new URL('./manual-fulfillment-service.ts', import.meta.url), 'utf8'),
+    readFile(
+      new URL('../cost-center/manual-fulfillment-config-service.ts', import.meta.url),
+      'utf8',
+    ),
+    readFile(new URL('../scheduler/cost-finance-sync-job.ts', import.meta.url), 'utf8'),
+  ])
+  assert.ok(
+    manualSource.includes('matchManualFulfillmentConfigWithClient(client, orderItemId, now)'),
+    '确认时必须按 SKU 自动挂渠道',
+  )
+  assert.ok(
+    configSource.includes('rematchManualFulfillments({') &&
+      configSource.includes('skuKey: config.sku_key') &&
+      configSource.includes('dateFrom: shiftDate(today, -29)'),
+    '保存人工 SKU 配置后必须立即回填最近 30 天订单项',
+  )
+  assert.ok(
+    jobSource.includes('rematchManualFulfillments({') &&
+      jobSource.includes('onlyUnmatched: true') &&
+      jobSource.includes('dateFrom:'),
+    '定时同步必须自动回填未匹配的人工订单项，并限制在最近窗口',
+  )
+  assert.ok(
+    manualSource.includes('NULLIF(oi.sku_code') && configSource.includes('NULLIF(oi.sku_code'),
+    'SKU 匹配必须有 sku_code / sku_name 兜底，不能只认 spec_json',
+  )
+})
+
+test('人工成本缺失必须进入异常队列，且只监控首次配置之后的订单项', async () => {
+  const source = await readFile(
+    new URL('../cost-center/cost-snapshot-sync-service.ts', import.meta.url),
+    'utf8',
+  )
+  assert.ok(source.includes("'manual_cost_missing'"), '缺少人工成本缺失异常')
+  assert.ok(
+    source.includes('FROM manual_fulfillment_configs') &&
+      source.includes('s.basis_at >= c.enforce_from'),
+    '人工异常必须以首次配置时间为起点，避免历史人工单刷屏',
+  )
+  assert.ok(
+    source.includes("cs.subject_type = 'manual_case'") &&
+      source.includes('cs.order_item_id = s.order_item_id'),
+    '人工异常回填/收口必须按订单项关联人工快照',
+  )
+})
+
+test('迁移 111 提供订单项级人工成本唯一约束与 subject 语义约束', async () => {
+  const source = await readFile(
+    new URL('../../db/migrations/111_manual_cost_integrity.sql', import.meta.url),
+    'utf8',
+  )
+  assert.ok(source.includes('uq_cost_snapshots_manual_order_item'), '缺少订单项级人工快照唯一索引')
+  assert.ok(
+    source.includes("WHERE subject_type = 'manual_case' AND superseded_at IS NULL"),
+    '唯一索引必须只约束有效人工快照',
+  )
+  assert.ok(
+    source.includes('subject_id <> order_item_id') &&
+      source.includes('chk_cost_snapshots_manual_subject'),
+    '必须把历史 subject_id 归一并加语义约束',
+  )
+})

apps/backend/src/services/finance/manual-fulfillment-service.ts

@@ -1,3 +1,5 @@
+import type { PoolClient } from 'pg'
+
 import { query, withTransaction } from '../../db/client.js'
 import type { JsonObject } from '../../types/json.js'
 import { createHttpError } from '../../utils/http.js'
@@ -8,7 +10,8 @@ import {
   buildManualFulfillmentReference,
 } from './manual-fulfillment-note.js'
 import {
-  consumePlatformCredit,
+  consumePlatformCreditWithClient,
+  PLATFORM_CREDIT_KEYS,
   type PlatformCreditKey,
 } from '../cost-center/platform-credit-service.js'
 
@@ -52,6 +55,28 @@ type PendingRow = {
   fulfilled_at: string | null
 }
 
+type ConfirmContextRow = {
+  order_item_id: number
+  order_id: number
+  order_id_resolved: number
+  platform_order_id: string
+  provider: string
+  shop_id: string
+  sku_code: string
+  sku_name: string
+  revenue_fen: string
+  canonical_product_id: number | null
+  fulfillment_mode: string
+  config_id: number | null
+  platform_key: string
+  credit_quantity: string
+  cost_mode: string
+  credit_unit_price_fen: string
+  work_order_id: number | null
+  work_order_no: string | null
+  work_product_rule_id: number | null
+}
+
 /**
  * 列出客服的人工履约待办。
  *
@@ -128,7 +153,12 @@ export async function listManualFulfillmentPending(
        o.shop_id,
        oi.sku_code,
        oi.sku_name,
-       COALESCE(NULLIF(oi.spec_json->>'skuNick',''), NULLIF(oi.spec_json->>'outSkuSn','')) AS internal_sku,
+       COALESCE(
+         NULLIF(oi.spec_json->>'skuNick',''),
+         NULLIF(oi.spec_json->>'outSkuSn',''),
+         NULLIF(oi.sku_code,''),
+         NULLIF(oi.sku_name,'')
+       ) AS internal_sku,
        COALESCE(oi.revenue_amount_fen, 0)::bigint AS revenue_fen,
        COALESCE(o.paid_at, o.created_at) AS paid_at,
        cfg.id AS config_id,
@@ -167,135 +197,134 @@ export async function confirmManualFulfillment(payload: JsonObject, session: Adm
   )
   const now = nowIso()
 
-  const context = await query<{
-    order_item_id: number
-    order_id: number
-    order_id_resolved: number
-    platform_order_id: string
-    provider: string
-    shop_id: string
-    sku_code: string
-    sku_name: string
-    revenue_fen: string
-    canonical_product_id: number | null
-    fulfillment_mode: string
-    config_id: number | null
-    platform_key: string
-    credit_quantity: string
-    cost_mode: string
-    credit_unit_price_fen: string
-    work_order_id: number | null
-    work_order_no: string | null
-    work_product_rule_id: number | null
-    fulfilled_at: string | null
-  }>(
-    `
-      WITH fulfilled AS (
-        SELECT cs.order_item_id, MAX(cs.created_at) AS fulfilled_at
-        FROM cost_snapshots cs
-        WHERE cs.subject_type = 'manual_case' AND cs.superseded_at IS NULL
-        GROUP BY cs.order_item_id
-      )
-      SELECT
-        oi.id AS order_item_id,
-        oi.order_id,
-        o.id AS order_id_resolved,
-        o.platform_order_id,
-        o.provider,
-        o.shop_id,
-        oi.sku_code,
-        oi.sku_name,
-        COALESCE(oi.revenue_amount_fen, 0)::bigint AS revenue_fen,
-        oi.canonical_product_id,
-        oi.fulfillment_mode,
-        oi.fulfillment_config_id AS config_id,
-        COALESCE(cfg.platform_key, '') AS platform_key,
-        COALESCE(cfg.credit_quantity, 0)::bigint AS credit_quantity,
-        COALESCE(cfg.cost_mode, 'variable') AS cost_mode,
-        COALESCE((fp.config_json->>'rechargeUnitPriceFen')::bigint, 1) AS credit_unit_price_fen,
-        wo.id AS work_order_id,
-        wo.work_order_no,
-        wo.product_rule_id AS work_product_rule_id,
-        fulfilled.fulfilled_at
-      FROM order_items oi
-      JOIN orders o ON o.id = oi.order_id
-      LEFT JOIN manual_fulfillment_configs cfg ON cfg.id = oi.fulfillment_config_id
-      LEFT JOIN fulfillment_profiles fp ON fp.profile_key = 'manual_' || cfg.platform_key
-      LEFT JOIN LATERAL (
-        SELECT id, product_rule_id, work_order_no FROM work_orders
-        WHERE order_item_id = oi.id ORDER BY id DESC LIMIT 1
-      ) wo ON TRUE
-      LEFT JOIN fulfilled ON fulfilled.order_item_id = oi.id
-      WHERE oi.id = $1
-      LIMIT 1
-    `,
-    [orderItemId],
-  )
+  // 钱与成本必须同一事务：先锁订单项、复查未履约，再扣款、最后写快照。
+  // 任何一步失败（余额不足、成本版本缺失、重复快照）都整体回滚，不会留下「扣了钱没记成本」。
+  return withTransaction(async (client) => {
+    // 配置可能晚于订单创建：确认前按 SKU 再匹配一次（与定时 rematch 同口径），
+    // 保证「配置好之后的新单」不需要人工再点一次「重新匹配」。
+    await matchManualFulfillmentConfigWithClient(client, orderItemId, now)
 
-  const row = context.rows[0]
-  if (!row) {
-    throw createHttpError('订单项不存在', {
-      statusCode: 404,
-      errorCode: 'manual_fulfillment_order_item_not_found',
-    })
-  }
-  if (row.fulfillment_mode !== 'manual') {
-    throw createHttpError('该订单项不是人工履约', {
-      statusCode: 409,
-      errorCode: 'manual_fulfillment_not_manual',
-    })
-  }
-  if (row.fulfilled_at) {
-    throw createHttpError('该订单项已确认过履约，不能重复操作', {
-      statusCode: 409,
-      errorCode: 'manual_fulfillment_already_done',
-    })
-  }
-  if (!row.config_id) {
-    throw createHttpError('该商品还没有配置履约渠道，请先在成本中心配置', {
-      statusCode: 409,
-      errorCode: 'manual_fulfillment_config_missing',
-    })
-  }
+    const context = await client.query<ConfirmContextRow>(
+      `
+        SELECT
+          oi.id AS order_item_id,
+          oi.order_id,
+          o.id AS order_id_resolved,
+          o.platform_order_id,
+          o.provider,
+          o.shop_id,
+          oi.sku_code,
+          oi.sku_name,
+          COALESCE(oi.revenue_amount_fen, 0)::bigint AS revenue_fen,
+          oi.canonical_product_id,
+          oi.fulfillment_mode,
+          oi.fulfillment_config_id AS config_id,
+          COALESCE(cfg.platform_key, '') AS platform_key,
+          COALESCE(cfg.credit_quantity, 0)::bigint AS credit_quantity,
+          COALESCE(cfg.cost_mode, 'variable') AS cost_mode,
+          COALESCE((fp.config_json->>'rechargeUnitPriceFen')::bigint, 1) AS credit_unit_price_fen,
+          wo.id AS work_order_id,
+          wo.work_order_no,
+          wo.product_rule_id AS work_product_rule_id
+        FROM order_items oi
+        JOIN orders o ON o.id = oi.order_id
+        LEFT JOIN manual_fulfillment_configs cfg ON cfg.id = oi.fulfillment_config_id
+        LEFT JOIN fulfillment_profiles fp ON fp.profile_key = 'manual_' || cfg.platform_key
+        LEFT JOIN LATERAL (
+          SELECT id, product_rule_id, work_order_no FROM work_orders
+          WHERE order_item_id = oi.id ORDER BY id DESC LIMIT 1
+        ) wo ON TRUE
+        WHERE oi.id = $1
+        LIMIT 1
+        FOR UPDATE OF oi
+      `,
+      [orderItemId],
+    )
 
-  const platformKey = row.platform_key as PlatformCreditKey | 'fund_cash'
-  const creditQuantity = Number(row.credit_quantity || 0)
-  const unitPriceFen = Number(row.credit_unit_price_fen || 1)
-
-  // 成本：固定成本用配置值；成本不固定必须客服手填。
-  let costFen: number
-  if (creditQuantity > 0) {
-    costFen = creditQuantity * unitPriceFen
-  } else {
-    const provided = payload.costFen ?? payload.cost_fen ?? payload.amountFen
-    costFen = normalizeCostFen(provided)
-    if (costFen === 0) {
-      throw createHttpError('该商品成本不固定，请填写本次实际成本', {
-        statusCode: 400,
-        errorCode: 'manual_fulfillment_cost_required',
+    const row = context.rows[0]
+    if (!row) {
+      throw createHttpError('订单项不存在', {
+        statusCode: 404,
+        errorCode: 'manual_fulfillment_order_item_not_found',
+      })
+    }
+    if (row.fulfillment_mode !== 'manual') {
+      throw createHttpError('该订单项不是人工履约', {
+        statusCode: 409,
+        errorCode: 'manual_fulfillment_not_manual',
+      })
+    }
+    if (!row.config_id) {
+      throw createHttpError('该商品还没有配置履约渠道，请先在成本中心配置', {
+        statusCode: 409,
+        errorCode: 'manual_fulfillment_config_missing',
+      })
+    }
+    const platformKey = row.platform_key as PlatformCreditKey | 'fund_cash'
+    if (platformKey !== 'fund_cash' && !PLATFORM_CREDIT_KEYS.includes(platformKey)) {
+      throw createHttpError('该商品的人工履约平台配置无效，请重新配置', {
+        statusCode: 409,
+        errorCode: 'manual_fulfillment_config_invalid',
       })
     }
-  }
 
-  const note = optionalText(payload.note, 300)
-  const payeeName = optionalText(payload.payeeName ?? payload.payee_name, 120)
-  const fulfillmentReference = buildManualFulfillmentReference({
-    workOrderNo: row.work_order_no,
-    platformOrderId: row.platform_order_id,
-    orderItemId,
-  })
+    // 锁内复查：并发/双击的第二个请求会等锁，醒来后看到快照直接拒绝，不会重复扣款。
+    const existing = await client.query<{ id: number }>(
+      `SELECT id FROM cost_snapshots
+       WHERE subject_type = 'manual_case' AND order_item_id = $1 AND superseded_at IS NULL
+       LIMIT 1`,
+      [orderItemId],
+    )
+    if (existing.rows[0]) {
+      throw createHttpError('该订单项已确认过履约，不能重复操作', {
+        statusCode: 409,
+        errorCode: 'manual_fulfillment_already_done',
+      })
+    }
 
-  // 1. 先扣额度（积分平台走积分钱包；备用金现金走现有 fund_* 链路）。
-  let creditEntryId: number | null = null
-  let fundExpenseId: number | null = null
-  let balanceAfter: number | null = null
+    const creditQuantity = Number(row.credit_quantity || 0)
+    const unitPriceFen = Number(row.credit_unit_price_fen || 1)
+    if (platformKey !== 'fund_cash' && creditQuantity <= 0) {
+      throw createHttpError('该人工 SKU 的积分配置无效，请先在成本中心重新配置', {
+        statusCode: 409,
+        errorCode: 'manual_fulfillment_config_invalid',
+      })
+    }
 
-  if (platformKey === 'fund_cash') {
-    // 备用金现金：复用现有客服备用金支出链路（扣余额 + 写 fund_entries + manual_expenses）。
-    const { writeFundExpense } = await import('../admin/admin-fund-service.js')
-    const fundAccountId = await resolveFundAccountId(session)
-    const written = await withTransaction(async (client) => {
-      return writeFundExpense(client, {
+    const costPlan = resolveManualFulfillmentCost({
+      creditQuantity,
+      unitPriceFen,
+      providedCostFen: payload.costFen ?? payload.cost_fen ?? payload.amountFen,
+    })
+    const costFen = costPlan.costFen
+    const payeeName = optionalText(payload.payeeName ?? payload.payee_name, 120)
+    const note = optionalText(payload.note, 300)
+    const fulfillmentReference = buildManualFulfillmentReference({
+      workOrderNo: row.work_order_no,
+      platformOrderId: row.platform_order_id,
+      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',
+      })
+    }
+
+    let fundExpenseId: number | null = null
+    let creditEntryId: number | null = null
+    let balanceAfter: number | null = null
+    let subjectSource: 'manual_expense' | 'platform_credit'
+
+    if (platformKey === 'fund_cash') {
+      // 备用金现金：复用唯一写入口，但把 client 传进去，和快照共用同一个事务。
+      const fundAccountId = await resolveFundAccountId(session, client)
+      const { writeFundExpense } = await import('../admin/admin-fund-service.js')
+      const written = await writeFundExpense(client, {
         agentAccountId: fundAccountId,
         amountFen: costFen,
         category: 'other',
@@ -312,115 +341,122 @@ export async function confirmManualFulfillment(payload: JsonObject, session: Adm
           maxLength: 500,
         }),
         actorUserId: session.userId,
+        // 同一订单可能有多个同价人工项：幂等由订单项锁 + 快照唯一索引保证，
+        // 这里放开 24 小时同金额同收款方的通用防重，避免误拦第二个订单项。
+        allowDuplicate: true,
       })
-    })
-    fundExpenseId = Number(written.expense.id)
-    balanceAfter = written.balanceAfterFen
-  } else if (creditQuantity > 0) {
-    const consumed = await consumePlatformCredit({
-      ownerAdminUserId: session.userId,
-      platformKey: platformKey as PlatformCreditKey,
-      quantity: creditQuantity,
-      relatedTaskId: null,
-      platformRef: row.work_order_no || row.platform_order_id,
-      note: buildManualFulfillmentNote({
-        skuName: row.sku_name,
-        reference: fulfillmentReference,
-        note,
-        maxLength: 300,
-      }),
-    })
-    creditEntryId = consumed.entryId
-    balanceAfter = consumed.balanceAfter
-  }
-
-  // 2. 写 actual 成本快照（主体 = 备用金支出单 / 积分流水）。
-  const subjectId = fundExpenseId ?? creditEntryId ?? orderItemId
-  const costItemId = await resolveManualCostItemId()
-  const versionId = await resolveCostVersionId(costItemId, row, now)
-
-  if (!versionId) {
-    throw createHttpError('该成本项没有可用的成本版本，请先在成本中心配置后重试', {
-      statusCode: 409,
-      errorCode: 'manual_fulfillment_cost_version_missing',
-    })
-  }
-
-  const snapshot = await 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, $3, $4,
-        $5, $6, $7,
-        1, $8, $8, 'actual', 'manual_input',
-        $9::jsonb, $10, $11, $11
-      )
-      ON CONFLICT (subject_type, subject_id, cost_item_id)
-        WHERE superseded_at IS NULL
-      DO NOTHING
-      RETURNING id
-    `,
-    [
-      subjectId,
-      row.order_id_resolved,
-      orderItemId,
-      row.work_order_id,
-      costItemId,
-      versionId,
-      row.canonical_product_id,
-      costFen,
-      JSON.stringify({
-        source: 'manual_fulfillment_confirm',
+      fundExpenseId = Number(written.expense.id)
+      balanceAfter = written.balanceAfterFen
+      subjectSource = 'manual_expense'
+    } else {
+      const consumed = await consumePlatformCreditWithClient(client, {
+        ownerAdminUserId: session.userId,
         platformKey,
-        platformLabel: PLATFORM_LABELS[platformKey] || platformKey,
-        creditQuantity,
-        creditUnitPriceFen: unitPriceFen,
-        costMode: creditQuantity > 0 ? 'fixed' : 'variable',
-        costFen,
-        fundExpenseId,
-        creditEntryId,
-        payeeName,
-        note,
-        fulfilledBy: {
-          userId: Number(session.userId || 0),
-          displayName: String(session.displayName || ''),
-          role: String(session.role || ''),
-        },
-      }),
-      now,
-      nowIso(),
-    ],
-  )
-  const snapshotId = snapshot.rows[0]?.id ?? null
-  if (!snapshotId) {
-    throw createHttpError('该订单项已有人工成本快照，不能重复录入', {
-      statusCode: 409,
-      errorCode: 'manual_fulfillment_snapshot_conflict',
-    })
-  }
+        quantity: creditQuantity,
+        relatedTaskId: null,
+        platformRef: row.work_order_no || row.platform_order_id,
+        note: buildManualFulfillmentNote({
+          skuName: row.sku_name,
+          reference: fulfillmentReference,
+          note,
+          maxLength: 300,
+        }),
+      })
+      creditEntryId = consumed.entryId
+      balanceAfter = consumed.balanceAfter
+      subjectSource = 'platform_credit'
+    }
 
-  return {
-    orderItemId,
-    orderId: Number(row.order_id_resolved),
-    costSnapshotId: snapshotId,
-    platformKey,
-    platformLabel: PLATFORM_LABELS[platformKey] || platformKey,
-    creditQuantity,
-    costFen,
-    fundExpenseId,
-    creditEntryId,
-    balanceAfter,
-  }
+    // subject_id 固定为 order_item_id：不再与 manual_expenses.id /
+    // platform_credit_entries.id 混用，避免两张表序列撞车。
+    // subject_type 仍为 manual_case，资金流水只放在 metadata 里。
+    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, $3,
+          $4, $5, $6,
+          1, $7, $7, 'actual', 'manual_input',
+          $8::jsonb, $9, $9, $9
+        )
+        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,
+          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,
+      ],
+    )
+    const snapshotId = snapshot.rows[0]?.id ?? null
+    if (!snapshotId) {
+      // 订单项唯一索引兜底：任何竞态都会走到这里，抛错回滚本次扣款。
+      throw createHttpError('该订单项已有人工成本快照，不能重复录入', {
+        statusCode: 409,
+        errorCode: 'manual_fulfillment_snapshot_conflict',
+      })
+    }
+
+    // 人工缺成本异常立即收口，不等下一轮定时任务。
+    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],
+    )
+
+    return {
+      orderItemId,
+      orderId: Number(row.order_id_resolved),
+      costSnapshotId: snapshotId,
+      platformKey,
+      platformLabel: PLATFORM_LABELS[platformKey] || platformKey,
+      creditQuantity,
+      costFen,
+      fundExpenseId,
+      creditEntryId,
+      balanceAfter,
+    }
+  })
 }
 
 /** 客服自己的备用金账户（管理员代录时用指定账户）。 */
-async function resolveFundAccountId(session: AdminSession) {
+async function resolveFundAccountId(session: AdminSession, client?: Pick<PoolClient, 'query'>) {
   const { getOwnAgentAccountId } = await import('../admin/admin-fund-service.js')
-  const own = await getOwnAgentAccountId(session.userId)
+  const own = await getOwnAgentAccountId(session.userId, client)
   if (own) return own
   throw createHttpError('当前账号还没有备用金账户', {
     statusCode: 404,
@@ -428,8 +464,67 @@ async function resolveFundAccountId(session: AdminSession) {
   })
 }
 
-async function resolveManualCostItemId() {
-  const result = await query<{ id: number }>(
+/**
+ * 确认前按 SKU 匹配人工履约配置。
+ *
+ * 与 `rematchManualFulfillments` 使用同一套键：
+ * spec_json.skuNick -> spec_json.outSkuSn -> sku_code -> sku_name。
+ * 这样即使配置晚于订单创建，新单在确认时也能自动挂上渠道。
+ */
+async function matchManualFulfillmentConfigWithClient(
+  client: PoolClient,
+  orderItemId: number,
+  now: string,
+) {
+  const matched = await client.query<{
+    id: number
+    canonical_product_id: number | null
+  }>(
+    `
+      SELECT cfg.id, cfg.canonical_product_id
+      FROM order_items oi
+      JOIN manual_fulfillment_configs cfg ON cfg.enabled = TRUE
+        AND (
+          cfg.sku_key = COALESCE(
+            NULLIF(oi.spec_json->>'skuNick', ''),
+            NULLIF(oi.spec_json->>'outSkuSn', ''),
+            NULLIF(oi.sku_code, ''),
+            NULLIF(oi.sku_name, '')
+          )
+          OR (
+            cfg.canonical_product_id IS NOT NULL
+            AND cfg.canonical_product_id = oi.canonical_product_id
+          )
+        )
+      WHERE oi.id = $1
+        AND oi.fulfillment_mode = 'manual'
+        AND oi.fulfillment_config_id IS NULL
+      ORDER BY
+        (cfg.canonical_product_id IS NOT NULL
+          AND cfg.canonical_product_id = oi.canonical_product_id) DESC,
+        cfg.id ASC
+      LIMIT 1
+    `,
+    [orderItemId],
+  )
+  const match = matched.rows[0]
+  if (!match) return false
+  await client.query(
+    `
+      UPDATE order_items
+      SET fulfillment_config_id = $2,
+          canonical_product_id = COALESCE($3, canonical_product_id),
+          updated_at = $4
+      WHERE id = $1
+        AND fulfillment_config_id IS NULL
+    `,
+    [orderItemId, match.id, match.canonical_product_id, now],
+  )
+  return true
+}
+
+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`,
   )
   const id = Number(result.rows[0]?.id || 0)
@@ -442,7 +537,8 @@ async function resolveManualCostItemId() {
   return id
 }
 
-async function resolveCostVersionId(
+async function resolveCostVersionIdWithClient(
+  client: PoolClient,
   costItemId: number,
   row: {
     order_id_resolved: number
@@ -454,7 +550,7 @@ async function resolveCostVersionId(
   },
   basisAt: string,
 ) {
-  const result = await query<{ id: number }>(
+  const result = await client.query<{ id: number }>(
     `
       SELECT v.id
       FROM cost_versions v
@@ -499,6 +595,40 @@ async function resolveCostVersionId(
   return result.rows[0]?.id ?? null
 }
 
+/**
+ * 人工履约的本次成本：固定成本按「配置积分 × 平台换算率」，成本不固定必须客服手填。
+ * 抽成纯函数，便于单测覆盖固定/手填/非法金额三种分支。
+ */
+export function resolveManualFulfillmentCost(input: {
+  creditQuantity: number
+  unitPriceFen: number
+  providedCostFen?: unknown
+}): { costFen: number; costMode: 'fixed' | 'variable' } {
+  const creditQuantity = Math.max(0, Math.trunc(Number(input.creditQuantity) || 0))
+  if (creditQuantity > 0) {
+    const unitPriceFen = Math.max(0, Math.trunc(Number(input.unitPriceFen) || 0))
+    return { costFen: creditQuantity * unitPriceFen, costMode: 'fixed' }
+  }
+  if (
+    input.providedCostFen === undefined ||
+    input.providedCostFen === null ||
+    input.providedCostFen === ''
+  ) {
+    throw createHttpError('该商品成本不固定，请填写本次实际成本', {
+      statusCode: 400,
+      errorCode: 'manual_fulfillment_cost_required',
+    })
+  }
+  const costFen = normalizeCostFen(input.providedCostFen)
+  if (costFen <= 0) {
+    throw createHttpError('该商品成本不固定，请填写本次实际成本', {
+      statusCode: 400,
+      errorCode: 'manual_fulfillment_cost_required',
+    })
+  }
+  return { costFen, costMode: 'variable' }
+}
+
 function mapPending(row: PendingRow) {
   const creditQuantity = Number(row.credit_quantity || 0)
   const unitPriceFen = Number(row.credit_unit_price_fen || 1)

apps/backend/src/services/finance/manual-fulfillment.integration.test.ts

@@ -0,0 +1,406 @@
+/**
+ * 人工履约资金链路集成测试（需要真实 PostgreSQL）。
+ *
+ * 运行方式：
+ *   MANUAL_FULFILLMENT_TEST_DATABASE_URL=postgres://.../scratch npm test -- --test-name-pattern ...
+ * 或直接：
+ *   MANUAL_FULFILLMENT_TEST_DATABASE_URL=... npx tsx --test src/services/finance/manual-fulfillment.integration.test.ts
+ *
+ * 约定：目标库必须已执行到迁移 111；测试只创建带唯一前缀的临时数据，结束后按依赖顺序清理。
+ * 未设置环境变量时整个文件跳过，因此普通 `npm test` 不依赖数据库。
+ */
+import assert from 'node:assert/strict'
+import test from 'node:test'
+
+const TEST_DATABASE_URL = String(process.env.MANUAL_FULFILLMENT_TEST_DATABASE_URL || '').trim()
+
+test(
+  '人工履约：单事务、防重复扣款、subject 不再跨表撞车、缺成本入异常队列',
+  { skip: !TEST_DATABASE_URL ? '未设置 MANUAL_FULFILLMENT_TEST_DATABASE_URL' : false },
+  async () => {
+    process.env.DATABASE_URL = TEST_DATABASE_URL
+    const { query, closeDb } = await import('../../db/client.js')
+    const { confirmManualFulfillment } = await import('./manual-fulfillment-service.js')
+    const { syncEstimatedOrderItemCostSnapshots } =
+      await import('../cost-center/cost-snapshot-sync-service.js')
+    const { rematchManualFulfillments } =
+      await import('../cost-center/manual-fulfillment-config-service.js')
+
+    const suffix = `${Date.now()}_${Math.floor(Math.random() * 1_000_000)}`
+    const skuKey = `IT-MANUAL-${suffix}`
+    const creditSkuKey = `IT-CREDIT-${suffix}`
+    const unconfirmedSkuKey = `IT-PENDING-${suffix}`
+    const orderIds: number[] = []
+    const orderItemIds: number[] = []
+    const fundAccountIds: number[] = []
+    const adminIds: number[] = []
+    const creditAccountIds: number[] = []
+    const configIds: number[] = []
+    let triggerName = ''
+    let fnName = ''
+
+    try {
+      // 1. 客服 + 备用金账户（余额 100 元）
+      const admin = await query<{ id: number }>(
+        `INSERT INTO admin_users (username, password_hash, role, status, created_at, updated_at)
+         VALUES ($1, 'x', 'support', 'active', NOW(), NOW()) RETURNING id`,
+        [`it_mf_${suffix}`],
+      )
+      const adminId = Number(admin.rows[0]?.id)
+      adminIds.push(adminId)
+      const fundAccount = await query<{ id: number }>(
+        `INSERT INTO fund_accounts (
+           account_no, account_type, owner_admin_user_id, balance_fen, status, created_at, updated_at
+         ) VALUES ($1, 'agent', $2, 10000, 'active', NOW(), NOW()) RETURNING id`,
+        [`FA-IT-${suffix}`, adminId],
+      )
+      const fundAccountId = Number(fundAccount.rows[0]?.id)
+      fundAccountIds.push(fundAccountId)
+
+      const session = {
+        userId: adminId,
+        role: 'support',
+        displayName: '集成测试客服',
+      } as never
+
+      // 2. 现金人工项：配置在 1 小时前创建，订单项尚未挂 config（验证确认时自动匹配）
+      const cashSku = `${skuKey}-CASH`
+      const cashConfig = await query<{ id: number }>(
+        `INSERT INTO manual_fulfillment_configs (
+           sku_key, platform_key, credit_quantity, cost_mode, enabled, note, created_at, updated_at
+         ) VALUES ($1, 'fund_cash', 0, 'variable', TRUE, '', NOW() - interval '1 hour', NOW())
+         RETURNING id`,
+        [cashSku],
+      )
+      configIds.push(Number(cashConfig.rows[0]?.id))
+      const cashItem = await createManualItem(query, cashSku)
+      orderIds.push(cashItem.orderId)
+      orderItemIds.push(cashItem.orderItemId)
+
+      // 配置晚于下单：rematch（最近窗口）应把订单项挂上配置
+      const today = shanghaiDate()
+      const rematched = await rematchManualFulfillments({
+        onlyUnmatched: true,
+        dateFrom: today,
+        dateTo: today,
+      })
+      assert.ok(rematched.updatedCount >= 1)
+      const matchedConfig = await query<{ fulfillment_config_id: number | null }>(
+        `SELECT fulfillment_config_id FROM order_items WHERE id = $1`,
+        [cashItem.orderItemId],
+      )
+      assert.ok(Number(matchedConfig.rows[0]?.fulfillment_config_id) > 0, 'rematch 必须挂上配置')
+
+      const cashResult = await confirmManualFulfillment(
+        { orderItemId: cashItem.orderItemId, costFen: 1234 },
+        session,
+      )
+      assert.ok(cashResult.costSnapshotId > 0)
+      assert.equal(cashResult.costFen, 1234)
+
+      const cashSnapshot = await query<{
+        subject_id: number
+        order_item_id: number
+        total_amount_fen: number
+        metadata_json: Record<string, unknown>
+      }>(
+        `SELECT subject_id, order_item_id, total_amount_fen, metadata_json
+         FROM cost_snapshots WHERE id = $1`,
+        [cashResult.costSnapshotId],
+      )
+      assert.equal(
+        Number(cashSnapshot.rows[0]?.subject_id),
+        cashItem.orderItemId,
+        'manual_case 的 subject_id 必须是 order_item_id',
+      )
+      assert.equal(Number(cashSnapshot.rows[0]?.order_item_id), cashItem.orderItemId)
+      assert.equal(Number(cashSnapshot.rows[0]?.total_amount_fen), 1234)
+      assert.equal(cashSnapshot.rows[0]?.metadata_json?.fundExpenseId, cashResult.fundExpenseId)
+
+      const cashExpense = await query<{ amount_fen: number; order_item_id: number }>(
+        `SELECT amount_fen, order_item_id FROM manual_expenses WHERE id = $1`,
+        [cashResult.fundExpenseId],
+      )
+      assert.equal(Number(cashExpense.rows[0]?.amount_fen), 1234)
+      assert.equal(Number(cashExpense.rows[0]?.order_item_id), cashItem.orderItemId)
+
+      const balanceAfterCash = await query<{ balance_fen: number }>(
+        `SELECT balance_fen FROM fund_accounts WHERE id = $1`,
+        [fundAccountId],
+      )
+      assert.equal(Number(balanceAfterCash.rows[0]?.balance_fen), 10000 - 1234)
+
+      // 3. 重复确认：必须拒绝，且不再扣款
+      await assert.rejects(
+        () =>
+          confirmManualFulfillment({ orderItemId: cashItem.orderItemId, costFen: 1234 }, session),
+        /已确认过履约/,
+      )
+      const cashExpenseCount = await query<{ count: number }>(
+        `SELECT COUNT(*)::int AS count FROM manual_expenses WHERE order_item_id = $1`,
+        [cashItem.orderItemId],
+      )
+      assert.equal(Number(cashExpenseCount.rows[0]?.count), 1, '重复确认不得产生第二条支出')
+      const balanceStill = await query<{ balance_fen: number }>(
+        `SELECT balance_fen FROM fund_accounts WHERE id = $1`,
+        [fundAccountId],
+      )
+      assert.equal(Number(balanceStill.rows[0]?.balance_fen), 10000 - 1234)
+
+      // 4. 并发确认：只有一个成功，扣款也只发生一次
+      const concurrentItem = await createManualItem(query, cashSku)
+      orderIds.push(concurrentItem.orderId)
+      orderItemIds.push(concurrentItem.orderItemId)
+      const concurrent = await Promise.allSettled([
+        confirmManualFulfillment(
+          { orderItemId: concurrentItem.orderItemId, costFen: 500 },
+          session,
+        ),
+        confirmManualFulfillment(
+          { orderItemId: concurrentItem.orderItemId, costFen: 500 },
+          session,
+        ),
+      ])
+      assert.equal(concurrent.filter((item) => item.status === 'fulfilled').length, 1)
+      assert.equal(concurrent.filter((item) => item.status === 'rejected').length, 1)
+      const concurrentExpense = await query<{ count: number }>(
+        `SELECT COUNT(*)::int AS count FROM manual_expenses WHERE order_item_id = $1`,
+        [concurrentItem.orderItemId],
+      )
+      assert.equal(Number(concurrentExpense.rows[0]?.count), 1, '并发确认只能扣一次')
+      const balanceAfterConcurrent = await query<{ balance_fen: number }>(
+        `SELECT balance_fen FROM fund_accounts WHERE id = $1`,
+        [fundAccountId],
+      )
+      assert.equal(Number(balanceAfterConcurrent.rows[0]?.balance_fen), 10000 - 1234 - 500)
+
+      // 5. 积分人工项：与现金项同时存在，subject_id 都是 order_item_id，不撞唯一键
+      const creditAccount = await query<{ id: number }>(
+        `INSERT INTO platform_credit_accounts (
+           platform_key, owner_admin_user_id, balance_quantity, unit_name,
+           recharge_unit_price_fen, status, created_at, updated_at
+         ) VALUES ('nanhang', $1, 10, '积分', 1, 'active', NOW(), NOW()) RETURNING id`,
+        [adminId],
+      )
+      creditAccountIds.push(Number(creditAccount.rows[0]?.id))
+      const creditSku = `${creditSkuKey}-NANHANG`
+      const creditConfig = await query<{ id: number }>(
+        `INSERT INTO manual_fulfillment_configs (
+           sku_key, platform_key, credit_quantity, cost_mode, enabled, note, created_at, updated_at
+         ) VALUES ($1, 'nanhang', 2, 'fixed', TRUE, '', NOW() - interval '1 hour', NOW())
+         RETURNING id`,
+        [creditSku],
+      )
+      configIds.push(Number(creditConfig.rows[0]?.id))
+      const creditItem = await createManualItem(query, creditSku)
+      orderIds.push(creditItem.orderId)
+      orderItemIds.push(creditItem.orderItemId)
+
+      const creditResult = await confirmManualFulfillment(
+        { orderItemId: creditItem.orderItemId },
+        session,
+      )
+      assert.equal(creditResult.costFen, 2)
+      assert.ok(creditResult.creditEntryId)
+      const creditSnapshot = await query<{
+        subject_id: number
+        metadata_json: Record<string, unknown>
+      }>(`SELECT subject_id, metadata_json FROM cost_snapshots WHERE id = $1`, [
+        creditResult.costSnapshotId,
+      ])
+      assert.equal(Number(creditSnapshot.rows[0]?.subject_id), creditItem.orderItemId)
+      assert.equal(creditSnapshot.rows[0]?.metadata_json?.subjectSource, 'platform_credit')
+      const creditBalance = await query<{ balance_quantity: number }>(
+        `SELECT balance_quantity FROM platform_credit_accounts WHERE id = $1`,
+        [creditAccountIds[0]],
+      )
+      assert.equal(Number(creditBalance.rows[0]?.balance_quantity), 8)
+      const creditExpenseCount = await query<{ count: number }>(
+        `SELECT COUNT(*)::int AS count FROM manual_expenses WHERE order_item_id = $1`,
+        [creditItem.orderItemId],
+      )
+      assert.equal(Number(creditExpenseCount.rows[0]?.count), 0, '积分路径不得写备用金支出')
+
+      // 6. 快照写入失败必须回滚已扣款：用临时触发器强制 cost_snapshots 插入失败，
+      //    验证「钱与成本同一事务」不是只写在注释里。
+      const failSku = `${skuKey}-FAIL`
+      const failConfig = await query<{ id: number }>(
+        `INSERT INTO manual_fulfillment_configs (
+           sku_key, platform_key, credit_quantity, cost_mode, enabled, note, created_at, updated_at
+         ) VALUES ($1, 'fund_cash', 0, 'variable', TRUE, '', NOW() - interval '1 hour', NOW())
+         RETURNING id`,
+        [failSku],
+      )
+      configIds.push(Number(failConfig.rows[0]?.id))
+      const failItem = await createManualItem(query, failSku)
+      orderIds.push(failItem.orderId)
+      orderItemIds.push(failItem.orderItemId)
+      triggerName = `it_mf_fail_${suffix}`
+      fnName = `it_mf_fail_fn_${suffix}`
+      await query(
+        `CREATE OR REPLACE FUNCTION ${fnName}() RETURNS trigger AS $$
+         BEGIN
+           IF NEW.order_item_id = ${Number(failItem.orderItemId)} THEN
+             RAISE EXCEPTION 'forced snapshot failure for test';
+           END IF;
+           RETURN NEW;
+         END $$ LANGUAGE plpgsql`,
+      )
+      await query(
+        `CREATE TRIGGER ${triggerName} BEFORE INSERT ON cost_snapshots
+         FOR EACH ROW EXECUTE FUNCTION ${fnName}()`,
+      )
+      const balanceBeforeFail = await query<{ balance_fen: number }>(
+        `SELECT balance_fen FROM fund_accounts WHERE id = $1`,
+        [fundAccountId],
+      )
+      await assert.rejects(
+        () =>
+          confirmManualFulfillment({ orderItemId: failItem.orderItemId, costFen: 700 }, session),
+        /forced snapshot failure/,
+      )
+      const balanceAfterFail = await query<{ balance_fen: number }>(
+        `SELECT balance_fen FROM fund_accounts WHERE id = $1`,
+        [fundAccountId],
+      )
+      assert.equal(
+        Number(balanceAfterFail.rows[0]?.balance_fen),
+        Number(balanceBeforeFail.rows[0]?.balance_fen),
+        '快照失败必须回滚扣款',
+      )
+      const failExpense = await query<{ count: number }>(
+        `SELECT COUNT(*)::int AS count FROM manual_expenses WHERE order_item_id = $1`,
+        [failItem.orderItemId],
+      )
+      assert.equal(Number(failExpense.rows[0]?.count), 0, '快照失败不得留下支出')
+      const failSnapshot = await query<{ count: number }>(
+        `SELECT COUNT(*)::int AS count FROM cost_snapshots WHERE order_item_id = $1`,
+        [failItem.orderItemId],
+      )
+      assert.equal(Number(failSnapshot.rows[0]?.count), 0, '快照失败不得留下成本')
+      await query(`DROP TRIGGER ${triggerName} ON cost_snapshots`)
+      await query(`DROP FUNCTION ${fnName}()`)
+
+      // 7. 人工缺成本：首次配置之后、尚未确认的订单项进入 manual_cost_missing；确认后自动收口
+      const pendingSku = `${unconfirmedSkuKey}-CASH`
+      const pendingConfig = await query<{ id: number }>(
+        `INSERT INTO manual_fulfillment_configs (
+           sku_key, platform_key, credit_quantity, cost_mode, enabled, note, created_at, updated_at
+         ) VALUES ($1, 'fund_cash', 0, 'variable', TRUE, '', NOW() - interval '1 hour', NOW())
+         RETURNING id`,
+        [pendingSku],
+      )
+      configIds.push(Number(pendingConfig.rows[0]?.id))
+      const pendingItem = await createManualItem(query, pendingSku)
+      orderIds.push(pendingItem.orderId)
+      orderItemIds.push(pendingItem.orderItemId)
+
+      await syncEstimatedOrderItemCostSnapshots({ dateFrom: today, dateTo: today })
+      const openManualException = await query<{ count: number }>(
+        `SELECT COUNT(*)::int AS count FROM cost_exceptions
+         WHERE reason = 'manual_cost_missing' AND status = 'open' AND subject_id = $1`,
+        [pendingItem.orderItemId],
+      )
+      assert.equal(Number(openManualException.rows[0]?.count), 1, '人工缺成本必须进异常队列')
+
+      await confirmManualFulfillment(
+        { orderItemId: pendingItem.orderItemId, costFen: 100 },
+        session,
+      )
+      const openAfterConfirm = await query<{ count: number }>(
+        `SELECT COUNT(*)::int AS count FROM cost_exceptions
+         WHERE reason = 'manual_cost_missing' AND status = 'open' AND subject_id = $1`,
+        [pendingItem.orderItemId],
+      )
+      assert.equal(Number(openAfterConfirm.rows[0]?.count), 0, '确认后应立即收口人工异常')
+      await syncEstimatedOrderItemCostSnapshots({ dateFrom: today, dateTo: today })
+      const resolvedManualException = await query<{ count: number }>(
+        `SELECT COUNT(*)::int AS count FROM cost_exceptions
+         WHERE reason = 'manual_cost_missing' AND status = 'resolved' AND subject_id = $1`,
+        [pendingItem.orderItemId],
+      )
+      assert.equal(Number(resolvedManualException.rows[0]?.count), 1, '确认后异常必须自动收口')
+    } finally {
+      // 按依赖顺序清理，避免污染测试库。
+      if (triggerName) {
+        await query(`DROP TRIGGER IF EXISTS ${triggerName} ON cost_snapshots`)
+      }
+      if (fnName) {
+        await query(`DROP FUNCTION IF EXISTS ${fnName}()`)
+      }
+      if (orderItemIds.length > 0) {
+        await query(`DELETE FROM cost_exceptions WHERE subject_id = ANY($1::bigint[])`, [
+          orderItemIds,
+        ])
+        await query(`DELETE FROM cost_snapshots WHERE order_item_id = ANY($1::bigint[])`, [
+          orderItemIds,
+        ])
+        await query(`DELETE FROM manual_expenses WHERE order_item_id = ANY($1::bigint[])`, [
+          orderItemIds,
+        ])
+      }
+      if (fundAccountIds.length > 0) {
+        await query(`DELETE FROM fund_entries WHERE account_id = ANY($1::bigint[])`, [
+          fundAccountIds,
+        ])
+        await query(`DELETE FROM fund_accounts WHERE id = ANY($1::bigint[])`, [fundAccountIds])
+      }
+      if (creditAccountIds.length > 0) {
+        await query(`DELETE FROM platform_credit_entries WHERE account_id = ANY($1::bigint[])`, [
+          creditAccountIds,
+        ])
+        await query(`DELETE FROM platform_credit_accounts WHERE id = ANY($1::bigint[])`, [
+          creditAccountIds,
+        ])
+      }
+      if (orderItemIds.length > 0) {
+        await query(`DELETE FROM order_items WHERE id = ANY($1::bigint[])`, [orderItemIds])
+      }
+      if (orderIds.length > 0) {
+        await query(`DELETE FROM orders WHERE id = ANY($1::bigint[])`, [orderIds])
+      }
+      if (configIds.length > 0) {
+        await query(`DELETE FROM manual_fulfillment_configs WHERE id = ANY($1::bigint[])`, [
+          configIds,
+        ])
+      }
+      if (adminIds.length > 0) {
+        await query(`DELETE FROM admin_users WHERE id = ANY($1::bigint[])`, [adminIds])
+      }
+      await closeDb()
+    }
+  },
+)
+
+async function createManualItem(
+  query: (sql: string, params?: unknown[]) => Promise<{ rows: Array<Record<string, unknown>> }>,
+  skuNick: string,
+) {
+  const order = await query(
+    `INSERT INTO orders (
+       provider, platform, platform_order_id, pay_status, total_amount, paid_at,
+       created_at, updated_at
+     ) VALUES ('91kaquan', 'kuaishou-91', $1, 'paid', 50, NOW(), NOW(), NOW())
+     RETURNING id`,
+    [`IT-MF-${Date.now()}-${Math.floor(Math.random() * 1_000_000)}`],
+  )
+  const orderId = Number(order.rows[0]?.id)
+  const item = await query(
+    `INSERT INTO order_items (
+       order_id, sku_code, sku_name, fulfillment_mode, spec_json,
+       revenue_amount_fen, revenue_basis, created_at, updated_at
+     ) VALUES ($1, $2, $3, 'manual', $4::jsonb, 5000, 'order_total', NOW(), NOW())
+     RETURNING id`,
+    [orderId, skuNick, `集成测试SKU-${skuNick}`, JSON.stringify({ skuNick })],
+  )
+  return { orderId, orderItemId: Number(item.rows[0]?.id) }
+}
+
+function shanghaiDate(date = new Date()) {
+  return new Intl.DateTimeFormat('en-CA', {
+    timeZone: 'Asia/Shanghai',
+    year: 'numeric',
+    month: '2-digit',
+    day: '2-digit',
+  }).format(date)
+}

apps/backend/src/services/scheduler/config-service.ts

@@ -152,7 +152,7 @@ function normalizeCostFinanceSyncJob(rawValue: JsonObject) {
     enabled: rawValue.enabled !== false,
     intervalSeconds: normalizeRangeInteger(rawValue.intervalSeconds, 300, 60, 3600),
     config: {
-      days: normalizeRangeInteger(config.days, 7, 1, 90),
+      days: normalizeRangeInteger(config.days, 30, 1, 90),
     },
   }
 }
@@ -227,7 +227,7 @@ function createDefaultCostFinanceSyncJob() {
     enabled: true,
     intervalSeconds: 300,
     config: {
-      days: 7,
+      days: 30,
     },
   }
 }

apps/backend/src/services/scheduler/cost-finance-sync-job.ts

@@ -4,6 +4,7 @@ import {
   syncEstimatedOrderItemCostSnapshots,
 } from '../cost-center/cost-snapshot-sync-service.js'
 import { normalizeManualProducts } from '../cost-center/manual-product-normalization-service.js'
+import { rematchManualFulfillments } from '../cost-center/manual-fulfillment-config-service.js'
 import { rebuildSalesDailyFacts } from '../finance/sales-facts-service.js'
 import { reconcilePendingFulfillmentModes } from '../fulfillment/fulfillment-mode-service.js'
 
@@ -26,6 +27,13 @@ export async function runCostFinanceSyncJob(job: JsonObject = {}) {
     dateTo: today(),
     resolveExceptions: false,
   })
+  // 人工 SKU 配置可能在订单之后才补：每次同步都按配置回填未匹配订单项，
+  // 保证「配置好人工之后的新单」自动挂上渠道，不依赖管理员手动点「重新匹配」。
+  const manualMatches = await rematchManualFulfillments({
+    onlyUnmatched: true,
+    dateFrom: shiftDays(-(days - 1)),
+    dateTo: today(),
+  })
 
   const estimated = await syncEstimatedOrderItemCostSnapshots({
     dateFrom: shiftDays(-(days - 1)),
@@ -45,6 +53,7 @@ export async function runCostFinanceSyncJob(job: JsonObject = {}) {
     message:
       `履约方式 ${fulfillment.updatedCount} 项 / ` +
       `人工商品 ${manualProducts.updatedItemCount} 项 / ` +
+      `人工渠道匹配 ${manualMatches.updatedCount} 项 / ` +
       `成本快照 ${estimated.createdCount} 商品预估 / ` +
       `${actual.directRewardCount + actual.upgradedCount} 工单实际 / ` +
       `${actual.estimatedRewardCount} 工单在途，销售事实 ${facts.insertedCount} 行`,
@@ -53,6 +62,7 @@ export async function runCostFinanceSyncJob(job: JsonObject = {}) {
     fulfillmentModeRemaining: fulfillment.remainingCount,
     manualItemsNormalized: manualProducts.updatedItemCount,
     manualProductsCreated: manualProducts.createdProductCount,
+    manualItemsMatched: manualMatches.updatedCount,
     estimatedCreated: estimated.createdCount,
     actualCreated: actual.directRewardCount + actual.upgradedCount,
     estimatedRewardCreated: actual.estimatedRewardCount,