feat(fulfillment): 收敛履约状态写入口并追加状态事件

apps/backend/src/db/migrations/140_fulfillment_state_events.sql

@@ -0,0 +1,212 @@
+-- 140_fulfillment_state_events.sql
+--
+-- 履约状态事件：
+--   1. 订单项履约核心字段每次真实变化都追加一条事件；
+--   2. 状态写入口可写入操作者、动作和原因，遗留直写至少留下 system 事件；
+--   3. 事件表不允许应用更新；父订单删除时允许随外键级联清理。
+
+CREATE TABLE IF NOT EXISTS fulfillment_state_events (
+  id BIGSERIAL PRIMARY KEY,
+  order_item_id BIGINT NOT NULL REFERENCES order_items(id) ON DELETE CASCADE,
+  order_id BIGINT NOT NULL REFERENCES orders(id) ON DELETE CASCADE,
+  event_type TEXT NOT NULL,
+  actor_type TEXT NOT NULL DEFAULT 'system',
+  actor_id TEXT NOT NULL DEFAULT '',
+  reason TEXT NOT NULL DEFAULT '',
+  previous_status TEXT,
+  next_status TEXT,
+  previous_channel TEXT,
+  next_channel TEXT,
+  previous_source TEXT,
+  next_source TEXT,
+  previous_locked BOOLEAN,
+  next_locked BOOLEAN,
+  previous_pending_reason TEXT,
+  next_pending_reason TEXT,
+  previous_pending_detail JSONB,
+  next_pending_detail JSONB,
+  payload_json JSONB NOT NULL DEFAULT '{}'::jsonb,
+  created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+);
+
+CREATE INDEX IF NOT EXISTS idx_fulfillment_state_events_item
+  ON fulfillment_state_events (order_item_id, id DESC);
+
+CREATE INDEX IF NOT EXISTS idx_fulfillment_state_events_order
+  ON fulfillment_state_events (order_id, created_at DESC, id DESC);
+
+COMMENT ON TABLE fulfillment_state_events IS
+  '履约状态追加事件：记录订单项履约状态、渠道、锁定和待办原因的真实变化';
+COMMENT ON COLUMN fulfillment_state_events.payload_json IS
+  '本次变化字段的前后值；空对象表示仅有未知扩展字段变化';
+
+CREATE OR REPLACE FUNCTION record_fulfillment_state_event()
+RETURNS TRIGGER
+LANGUAGE plpgsql
+AS $$
+DECLARE
+  event_type_value TEXT;
+  actor_type_value TEXT;
+  actor_id_value TEXT;
+  reason_value TEXT;
+  changed_fields JSONB;
+BEGIN
+  IF ROW(
+    OLD.fulfillment_status,
+    OLD.fulfillment_channel,
+    OLD.fulfillment_source,
+    OLD.fulfillment_locked,
+    OLD.fulfillment_pending_reason,
+    OLD.fulfillment_pending_detail,
+    OLD.fulfillment_reason,
+    OLD.fulfillment_decided_by,
+    OLD.fulfillment_decided_at,
+    OLD.original_channel,
+    OLD.channel_switch_reason,
+    OLD.channel_switched_at
+  ) IS NOT DISTINCT FROM ROW(
+    NEW.fulfillment_status,
+    NEW.fulfillment_channel,
+    NEW.fulfillment_source,
+    NEW.fulfillment_locked,
+    NEW.fulfillment_pending_reason,
+    NEW.fulfillment_pending_detail,
+    NEW.fulfillment_reason,
+    NEW.fulfillment_decided_by,
+    NEW.fulfillment_decided_at,
+    NEW.original_channel,
+    NEW.channel_switch_reason,
+    NEW.channel_switched_at
+  ) THEN
+    RETURN NEW;
+  END IF;
+
+  event_type_value := COALESCE(
+    NULLIF(BTRIM(current_setting('app.fulfillment_event_type', TRUE)), ''),
+    'fulfillment_state_changed'
+  );
+  actor_type_value := COALESCE(
+    NULLIF(BTRIM(current_setting('app.fulfillment_actor_type', TRUE)), ''),
+    'system'
+  );
+  actor_id_value := COALESCE(
+    current_setting('app.fulfillment_actor_id', TRUE),
+    ''
+  );
+  reason_value := COALESCE(
+    current_setting('app.fulfillment_reason', TRUE),
+    ''
+  );
+
+  changed_fields := jsonb_strip_nulls(jsonb_build_object(
+    'fulfillmentStatus', CASE
+      WHEN OLD.fulfillment_status IS DISTINCT FROM NEW.fulfillment_status
+        THEN jsonb_build_object('from', OLD.fulfillment_status, 'to', NEW.fulfillment_status)
+    END,
+    'fulfillmentChannel', CASE
+      WHEN OLD.fulfillment_channel IS DISTINCT FROM NEW.fulfillment_channel
+        THEN jsonb_build_object('from', OLD.fulfillment_channel, 'to', NEW.fulfillment_channel)
+    END,
+    'fulfillmentSource', CASE
+      WHEN OLD.fulfillment_source IS DISTINCT FROM NEW.fulfillment_source
+        THEN jsonb_build_object('from', OLD.fulfillment_source, 'to', NEW.fulfillment_source)
+    END,
+    'fulfillmentLocked', CASE
+      WHEN OLD.fulfillment_locked IS DISTINCT FROM NEW.fulfillment_locked
+        THEN jsonb_build_object('from', OLD.fulfillment_locked, 'to', NEW.fulfillment_locked)
+    END,
+    'fulfillmentPendingReason', CASE
+      WHEN OLD.fulfillment_pending_reason IS DISTINCT FROM NEW.fulfillment_pending_reason
+        THEN jsonb_build_object(
+          'from', OLD.fulfillment_pending_reason,
+          'to', NEW.fulfillment_pending_reason
+        )
+    END,
+    'fulfillmentPendingDetail', CASE
+      WHEN OLD.fulfillment_pending_detail IS DISTINCT FROM NEW.fulfillment_pending_detail
+        THEN jsonb_build_object(
+          'from', OLD.fulfillment_pending_detail,
+          'to', NEW.fulfillment_pending_detail
+        )
+    END
+  ));
+
+  INSERT INTO fulfillment_state_events (
+    order_item_id,
+    order_id,
+    event_type,
+    actor_type,
+    actor_id,
+    reason,
+    previous_status,
+    next_status,
+    previous_channel,
+    next_channel,
+    previous_source,
+    next_source,
+    previous_locked,
+    next_locked,
+    previous_pending_reason,
+    next_pending_reason,
+    previous_pending_detail,
+    next_pending_detail,
+    payload_json
+  ) VALUES (
+    NEW.id,
+    NEW.order_id,
+    LEFT(event_type_value, 100),
+    LEFT(actor_type_value, 32),
+    LEFT(actor_id_value, 200),
+    LEFT(reason_value, 500),
+    OLD.fulfillment_status,
+    NEW.fulfillment_status,
+    OLD.fulfillment_channel,
+    NEW.fulfillment_channel,
+    OLD.fulfillment_source,
+    NEW.fulfillment_source,
+    OLD.fulfillment_locked,
+    NEW.fulfillment_locked,
+    OLD.fulfillment_pending_reason,
+    NEW.fulfillment_pending_reason,
+    OLD.fulfillment_pending_detail,
+    NEW.fulfillment_pending_detail,
+    changed_fields
+  );
+
+  RETURN NEW;
+END;
+$$;
+
+DROP TRIGGER IF EXISTS trg_order_items_fulfillment_state_event ON order_items;
+CREATE TRIGGER trg_order_items_fulfillment_state_event
+AFTER UPDATE OF
+  fulfillment_status,
+  fulfillment_channel,
+  fulfillment_source,
+  fulfillment_locked,
+  fulfillment_pending_reason,
+  fulfillment_pending_detail,
+  fulfillment_reason,
+  fulfillment_decided_by,
+  fulfillment_decided_at,
+  original_channel,
+  channel_switch_reason,
+  channel_switched_at
+ON order_items
+FOR EACH ROW
+EXECUTE FUNCTION record_fulfillment_state_event();
+
+CREATE OR REPLACE FUNCTION reject_fulfillment_state_event_mutation()
+RETURNS TRIGGER
+LANGUAGE plpgsql
+AS $$
+BEGIN
+  RAISE EXCEPTION 'fulfillment_state_events is append-only';
+END;
+$$;
+
+DROP TRIGGER IF EXISTS trg_fulfillment_state_events_append_only ON fulfillment_state_events;
+CREATE TRIGGER trg_fulfillment_state_events_append_only
+BEFORE UPDATE ON fulfillment_state_events
+FOR EACH ROW
+EXECUTE FUNCTION reject_fulfillment_state_event_mutation();

apps/backend/src/repositories/task-repo.ts

@@ -94,6 +94,26 @@ export async function listTasksByOrderIds(orderIds: number[]): Promise<TaskRow[]
 
 export async function createTask(input: TaskCreateInput): Promise<TaskRow | null> {
   return withTransaction(async (client) => {
+    // 同一订单项单元的创建必须串行，否则并发调度会同时算出同一尝试号，
+    // 再一起绕过“仅一个当前尝试”的唯一索引。
+    await client.query(`SELECT id FROM order_items WHERE id = $1 FOR UPDATE`, [input.orderItemId])
+    const active = await client.query<{ id: number }>(
+      `SELECT id
+       FROM fulfillment_tasks
+       WHERE order_item_id = $1
+         AND unit_index = $2
+         AND task_status NOT IN (
+           'closed', 'expired', 'failed', 'manual_review', 'retry_pending', 'cancelled'
+         )
+       ORDER BY is_current DESC, id DESC
+       LIMIT 1`,
+      [input.orderItemId, input.unitIndex],
+    )
+    const activeTaskId = Number(active.rows[0]?.id || 0)
+    if (activeTaskId > 0) {
+      return getTaskByIdWithExecutor(client.query.bind(client) as TaskQueryExecutor, activeTaskId)
+    }
+
     const nextAttempt = await client.query<{ attempt_no: number }>(
       `SELECT COALESCE(MAX(attempt_no), 0) + 1 AS attempt_no
        FROM fulfillment_tasks
@@ -106,8 +126,11 @@ export async function createTask(input: TaskCreateInput): Promise<TaskRow | null
        WHERE order_item_id = $1 AND unit_index = $2 AND is_current = TRUE`,
       [input.orderItemId, input.unitIndex],
     )
-    const taskResult = await client.query<{ id: number }>(
-      `
+    const nextAttemptNo = Number(nextAttempt.rows[0]?.attempt_no || 1)
+    let taskResult: { rows: Array<{ id: number }> }
+    try {
+      taskResult = await client.query<{ id: number }>(
+        `
         INSERT INTO fulfillment_tasks (
           order_id,
           order_item_id,
@@ -143,35 +166,44 @@ export async function createTask(input: TaskCreateInput): Promise<TaskRow | null
         )
         RETURNING id
       `,
-      [
-        input.orderId,
-        input.orderItemId,
-        input.unitIndex,
-        input.taskNo,
-        input.provider,
-        input.platform,
-        input.shopId || '',
-        input.shopName || '',
-        input.platformOrderId,
-        input.profileId,
-        input.executorKey,
-        input.taskStatus || 'pending_payment',
-        input.deliveryStatus || 'pending',
-        input.resultCode || '',
-        input.resultMessage || '',
-        input.claimToken || '',
-        input.claimExpiresAt || null,
-        input.automationMode || 'manual',
-        Boolean(input.requiresClaim),
-        input.userActionStatus || 'not_required',
-        input.attemptCount || 0,
-        input.lastError || '',
-        input.contextJson || '{}',
-        input.createdAt,
-        input.updatedAt,
-        Number(nextAttempt.rows[0]?.attempt_no || 1),
-      ],
-    )
+        [
+          input.orderId,
+          input.orderItemId,
+          input.unitIndex,
+          input.taskNo,
+          input.provider,
+          input.platform,
+          input.shopId || '',
+          input.shopName || '',
+          input.platformOrderId,
+          input.profileId,
+          input.executorKey,
+          input.taskStatus || 'pending_payment',
+          input.deliveryStatus || 'pending',
+          input.resultCode || '',
+          input.resultMessage || '',
+          input.claimToken || '',
+          input.claimExpiresAt || null,
+          input.automationMode || 'manual',
+          Boolean(input.requiresClaim),
+          input.userActionStatus || 'not_required',
+          input.attemptCount || 0,
+          input.lastError || '',
+          input.contextJson || '{}',
+          input.createdAt,
+          input.updatedAt,
+          nextAttemptNo,
+        ],
+      )
+    } catch (error) {
+      const message = error instanceof Error ? error.message : String(error)
+      throw new Error(
+        `履约任务创建失败 orderItemId=${input.orderItemId} ` +
+          `unitIndex=${input.unitIndex} taskStatus=${input.taskStatus || 'pending_payment'} ` +
+          `attemptNo=${nextAttemptNo}: ${message}`,
+        { cause: error },
+      )
+    }
 
     const taskId = Number(taskResult.rows[0]?.id || 0)
     if (input.runtimeContext) {

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

@@ -6,6 +6,10 @@ 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 { reconcilePendingCarriers } from '../fulfillment/carrier-reconciliation-service.js'
+import {
+  transitionFulfillmentStateBatch,
+  transitionFulfillmentStateBatchWithExecutor,
+} from '../fulfillment/fulfillment-state-service.js'
 import { canonicalizeInternalSku, containsCjk } from '../fulfillment/internal-sku-resolver.js'
 import { refreshFactsForDates } from '../finance/refresh-facts.js'
 import {
@@ -581,12 +585,13 @@ async function retryPendingManualItems(skuKey: string): Promise<number> {
   )
   const ids = candidates.rows.map((row) => Number(row.id))
   if (ids.length === 0) return 0
-  await query(
-    `UPDATE order_items
-     SET fulfillment_locked = FALSE, updated_at = NOW()
-     WHERE id = ANY($1::bigint[]) AND fulfillment_status = 'pending'`,
-    [ids],
-  )
+  await transitionFulfillmentStateBatch({
+    orderItemIds: ids,
+    expected: { status: 'pending', locked: true },
+    patch: { locked: false },
+    eventType: 'manual_config_retry_unlocked',
+    actor: { type: 'system' },
+  })
   const result = await reconcilePendingCarriers({
     orderItemIds: ids,
     limit: ids.length,
@@ -763,29 +768,28 @@ export async function deleteManualSkuClassification(
     )
 
     if (itemIds.length > 0) {
-      await client.query(
-        `
-          UPDATE order_items
-          SET fulfillment_channel = NULL,
-              fulfillment_status = 'pending',
-              fulfillment_source = '',
-              fulfillment_locked = FALSE,
-              fulfillment_config_id = NULL,
-              fulfillment_pending_reason = 'no_channel_capability',
-              fulfillment_pending_detail = jsonb_build_object(
-                'reason', 'manual_classification_removed',
-                'removedAt', NOW()
-              ),
-              fulfillment_reason = '运营移除错误的人工分类',
-              fulfillment_decided_by = NULL,
-              fulfillment_decided_at = NULL,
-              updated_at = NOW()
-          WHERE id = ANY($1::bigint[])
-            AND fulfillment_status = 'assigned'
-            AND fulfillment_channel = 'manual'
-        `,
-        [itemIds],
-      )
+      await transitionFulfillmentStateBatchWithExecutor(client, {
+        orderItemIds: itemIds,
+        expected: { status: 'assigned', channel: 'manual' },
+        patch: {
+          status: 'pending',
+          channel: null,
+          source: '',
+          locked: false,
+          configId: null,
+          pendingReason: 'no_channel_capability',
+          pendingDetail: {
+            reason: 'manual_classification_removed',
+            removedAt: nowIso(),
+          },
+          reason: '运营移除错误的人工分类',
+          decidedBy: null,
+          decidedAt: null,
+        },
+        eventType: 'manual_classification_removed',
+        actor: { type: 'operator' },
+        reason: key,
+      })
     }
 
     return {

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

@@ -17,6 +17,7 @@ import {
 import { rebuildChannelSalesDailyFacts } from './sales-facts-channel-service.js'
 import { syncFinanceLedgerFromFacts } from './finance-ledger-service.js'
 import { scheduleVoucherConsumeForOrderItems } from '../fulfillment/voucher-consume-queue-service.js'
+import { transitionFulfillmentStateWithExecutor } from '../fulfillment/fulfillment-state-service.js'
 import {
   calculateManualCostFen,
   normalizeManualCreditQuantity,
@@ -621,19 +622,21 @@ export async function confirmManualFulfillment(payload: JsonObject, session: Adm
 
     // 人工确认即渠道成立：写新三列，避免订单一直停在 pending。
     const finalChannel = platformKey === 'on_account' ? 'on_account' : 'manual'
-    await client.query(
-      `
-        UPDATE order_items
-        SET fulfillment_channel = $3,
-            fulfillment_status = 'assigned',
-            fulfillment_source = 'manual_confirm',
-            fulfillment_decided_at = $2,
-            updated_at = $2
-        WHERE id = $1
-          AND fulfillment_status <> 'archived'
-      `,
-      [orderItemId, now, finalChannel],
-    )
+    await transitionFulfillmentStateWithExecutor(client, {
+      orderItemId,
+      patch: {
+        channel: finalChannel,
+        status: 'assigned',
+        source: 'manual_confirm',
+        decidedAt: now,
+        decidedBy: Number(session.userId || 0) || null,
+        pendingReason: null,
+        pendingDetail: null,
+      },
+      eventType: 'manual_fulfillment_confirmed',
+      actor: { type: 'finance', id: session.userId },
+      reason: platformKey,
+    })
 
     // 人工缺成本异常立即收口，不等下一轮定时任务。
     await client.query(

apps/backend/src/services/fulfillment/affiliate-fallback-service.ts

@@ -2,6 +2,7 @@ import { query, withTransaction } from '../../db/client.js'
 import { syncCostSnapshotsForItems } from '../cost-center/cost-snapshot-service.js'
 import { refreshFactsForDates } from '../finance/refresh-facts.js'
 import { scheduleVoucherConsumeForOrderItems } from './voucher-consume-queue-service.js'
+import { transitionFulfillmentStateWithExecutor } from './fulfillment-state-service.js'
 
 /**
  * 自动发货失败 → 人工兜底。
@@ -99,50 +100,51 @@ export async function runAffiliateFallback(
 
   for (const row of candidates.rows) {
     const updated = await withTransaction(async (client) => {
-      const result = await client.query(
-        `
-          UPDATE order_items
-          SET
-            original_channel = COALESCE(original_channel, 'affiliate_dash'),
-            fulfillment_channel = 'manual',
-            fulfillment_status = 'assigned',
-            fulfillment_source = 'affiliate_fallback',
-            channel_switch_reason = 'affiliate_failed',
-            channel_switched_at = NOW(),
-            fulfillment_config_id = COALESCE(fulfillment_config_id, $2),
-            canonical_product_id = COALESCE($3, canonical_product_id),
-            fulfillment_pending_reason = NULL,
-            fulfillment_pending_detail = NULL,
-            fulfillment_decided_at = NOW(),
-            item_snapshot_json = COALESCE(item_snapshot_json, '{}'::jsonb) || jsonb_build_object(
-              'internalSkuChannel', 'manual',
-              'internalSkuChannelSource', 'affiliate_fallback',
-              'internalSkuChannelReason', '自动发货失败，转人工兜底',
-              'isConfigured', TRUE,
-              'affiliateDash', NULL,
-              'fulfillmentRoute', jsonb_build_object(
-                'selectedExecutorKey', 'manual_dispatch',
-                'selectedRuleId', '',
-                'reason', '自动发货失败，转人工兜底',
-                'matchedRule', NULL,
-                'candidates', '[]'::jsonb,
-                'skipped', '[]'::jsonb
-              ),
-              'classification', jsonb_build_object(
-                'channel', 'manual',
-                'canonicalSku', resolve_internal_sku(manual_sku_key, sku_name, sku_code)
-              )
-            ),
-            updated_at = NOW()
-          WHERE id = $1
-            AND fulfillment_status = 'assigned'
-            AND fulfillment_channel = 'affiliate_dash'
-            AND fulfillment_locked = FALSE
-          RETURNING id
-        `,
-        [row.id, row.config_id, row.config_product_id],
-      )
-      if ((result.rowCount || 0) === 0) return false
+      const transitioned = await transitionFulfillmentStateWithExecutor(client, {
+        orderItemId: Number(row.id),
+        expected: {
+          status: 'assigned',
+          channel: 'affiliate_dash',
+          locked: false,
+        },
+        patch: {
+          originalChannelIfNull: 'affiliate_dash',
+          channel: 'manual',
+          status: 'assigned',
+          source: 'affiliate_fallback',
+          switchReason: 'affiliate_failed',
+          switchedAt: new Date().toISOString(),
+          ...(row.config_id ? { configIdIfNull: Number(row.config_id) } : {}),
+          ...(row.config_product_id
+            ? { canonicalProductIdIfNull: Number(row.config_product_id) }
+            : {}),
+          pendingReason: null,
+          pendingDetail: null,
+          decidedAt: new Date().toISOString(),
+          itemSnapshotPatch: {
+            internalSkuChannel: 'manual',
+            internalSkuChannelSource: 'affiliate_fallback',
+            internalSkuChannelReason: '自动发货失败，转人工兜底',
+            isConfigured: true,
+            affiliateDash: null,
+            fulfillmentRoute: {
+              selectedExecutorKey: 'manual_dispatch',
+              selectedRuleId: '',
+              reason: '自动发货失败，转人工兜底',
+              matchedRule: null,
+              candidates: [],
+              skipped: [],
+            },
+            classification: {
+              channel: 'manual',
+              canonicalSku: row.internal_sku,
+            },
+          },
+        },
+        eventType: 'affiliate_failed_fallback_to_manual',
+        reason: 'affiliate_failed',
+      })
+      if (!transitioned) return false
 
       await client.query(
         `

apps/backend/src/services/fulfillment/carrier-reconciliation-service.ts

@@ -1,4 +1,4 @@
-import { query } from '../../db/client.js'
+import { query, withTransaction } from '../../db/client.js'
 import { getOrderById } from '../../repositories/order-repo.js'
 import type { OrderItemRow } from '../../types/repository/rows.js'
 import { logWarn } from '../../utils/logger.js'
@@ -8,6 +8,10 @@ import {
   type FulfillmentChannel,
 } from './channel-types.js'
 import { ensureCarrierForOrderItem } from './carrier-creation-service.js'
+import {
+  transitionFulfillmentState,
+  transitionFulfillmentStateWithExecutor,
+} from './fulfillment-state-service.js'
 import {
   planInternalSkuChannels,
   type InternalSkuChannelRow,
@@ -145,24 +149,21 @@ async function markPendingWithoutCarrier(input: {
   attempts: Array<{ channel: FulfillmentChannel; reason: string; message: string }>
 }): Promise<void> {
   const carrierIssue = input.attempts[0]?.reason || 'carrier_attempt_failed'
-  await query(
-    `
-      UPDATE order_items
-      SET fulfillment_pending_reason = 'no_carrier',
-          fulfillment_pending_detail = $2::jsonb,
-          updated_at = NOW()
-      WHERE id = $1 AND fulfillment_status = 'pending'
-    `,
-    [
-      input.itemId,
-      JSON.stringify({
+  await transitionFulfillmentState({
+    orderItemId: input.itemId,
+    expected: { status: 'pending' },
+    patch: {
+      pendingReason: 'no_carrier',
+      pendingDetail: {
         internalSku: input.internalSku,
         carrierIssue,
         carrierMessage: input.attempts[0]?.message || '',
         attempts: input.attempts,
-      }),
-    ],
-  )
+      },
+    },
+    eventType: 'carrier_attempt_failed',
+    reason: carrierIssue,
+  })
 }
 
 async function markAssignedMissingCarrierPending(input: {
@@ -172,29 +173,26 @@ async function markAssignedMissingCarrierPending(input: {
   attempts: Array<{ channel: FulfillmentChannel; reason: string; message: string }>
 }): Promise<void> {
   const carrierIssue = input.attempts[0]?.reason || 'carrier_attempt_failed'
-  await query(
-    `
-      UPDATE order_items
-      SET fulfillment_channel = NULL,
-          fulfillment_status = 'pending',
-          fulfillment_source = 'auto_mapping',
-          fulfillment_locked = FALSE,
-          fulfillment_pending_reason = 'no_carrier',
-          fulfillment_pending_detail = $2::jsonb,
-          updated_at = NOW()
-      WHERE id = $1 AND fulfillment_status = 'assigned'
-    `,
-    [
-      input.itemId,
-      JSON.stringify({
+  await transitionFulfillmentState({
+    orderItemId: input.itemId,
+    expected: { status: 'assigned' },
+    patch: {
+      status: 'pending',
+      channel: null,
+      source: 'auto_mapping',
+      locked: false,
+      pendingReason: 'no_carrier',
+      pendingDetail: {
         internalSku: input.internalSku,
         previousChannel: input.previousChannel,
         carrierIssue,
         carrierMessage: input.attempts[0]?.message || '',
         attempts: input.attempts,
-      }),
-    ],
-  )
+      },
+    },
+    eventType: 'assigned_carrier_missing',
+    reason: carrierIssue,
+  })
 }
 
 async function selectSafeFallback(input: {
@@ -204,39 +202,41 @@ async function selectSafeFallback(input: {
   reason: string
   includeAssignedMissingCarrier: boolean
 }): Promise<boolean> {
-  const result = await query(
-    `
-      UPDATE order_items
-      SET fulfillment_channel = $2,
-          fulfillment_status = 'assigned',
-          fulfillment_source = 'auto_mapping',
-          fulfillment_reason = $3,
-          channel_switch_reason = $3,
-          channel_switched_at = NOW(),
-          canonical_product_id = NULL,
-          fulfillment_config_id = NULL,
-          fulfillment_pending_reason = NULL,
-          fulfillment_pending_detail = NULL,
-          updated_at = NOW()
-      WHERE id = $1
-        AND (
-          (fulfillment_status = 'pending' AND fulfillment_locked = FALSE)
-          OR (
-            $4::boolean
-            AND fulfillment_status = 'assigned'
-            AND fulfillment_locked = FALSE
-            AND fulfillment_source NOT IN ('operator_assigned', 'manual_confirm')
-          )
-        )
-    `,
-    [
-      input.orderItemId,
-      input.channel,
-      `主渠道 ${input.primaryChannel} 无法创建载体：${input.reason}`,
-      input.includeAssignedMissingCarrier,
-    ],
-  )
-  const updated = (result.rowCount || 0) > 0
+  const reason = `主渠道 ${input.primaryChannel} 无法创建载体：${input.reason}`
+  const updated = await withTransaction(async (client) => {
+    const common = {
+      orderItemId: input.orderItemId,
+      patch: {
+        status: 'assigned' as const,
+        channel: input.channel,
+        source: 'auto_mapping',
+        reason,
+        switchReason: reason,
+        switchedAt: new Date().toISOString(),
+        canonicalProductId: null,
+        configId: null,
+        pendingReason: null,
+        pendingDetail: null,
+      },
+      eventType: 'fallback_carrier_selected',
+      reason,
+    }
+    const pending = await transitionFulfillmentStateWithExecutor(client, {
+      ...common,
+      expected: { status: 'pending', locked: false },
+    })
+    if (pending) return true
+    if (!input.includeAssignedMissingCarrier) return false
+    const assigned = await transitionFulfillmentStateWithExecutor(client, {
+      ...common,
+      expected: {
+        status: 'assigned',
+        locked: false,
+        sourceNotIn: ['operator_assigned', 'manual_confirm'],
+      },
+    })
+    return Boolean(assigned)
+  })
   if (updated) {
     await scheduleVoucherConsumeForOrderItems([input.orderItemId], 'order_received').catch(
       () => undefined,
@@ -320,20 +320,19 @@ export async function reconcilePendingCarriers(
       if (attempts.length === 0) {
         failed += 1
         if (!input.includeAssignedMissingCarrier) {
-          await query(
-            `UPDATE order_items
-             SET fulfillment_pending_reason = 'routing_conflict',
-                 fulfillment_pending_detail = $2::jsonb,
-                 updated_at = NOW()
-             WHERE id = $1 AND fulfillment_status = 'pending'`,
-            [
-              item.id,
-              JSON.stringify({
+          await transitionFulfillmentState({
+            orderItemId: Number(item.id),
+            expected: { status: 'pending' },
+            patch: {
+              pendingReason: 'routing_conflict',
+              pendingDetail: {
                 internalSku: candidate.internal_sku,
                 channels: plan.conflicts.map((channel) => channel.channel),
-              }),
-            ],
-          )
+              },
+            },
+            eventType: 'carrier_routing_conflict',
+            reason: 'routing_conflict',
+          })
         }
         continue
       }

apps/backend/src/services/fulfillment/channel-assign-service.ts

@@ -35,6 +35,13 @@ import { reconcilePendingCarriers } from './carrier-reconciliation-service.js'
 import { TASK_STATUS } from '../../domain/task-status.js'
 import { canonicalizeInternalSku, containsCjk } from './internal-sku-resolver.js'
 import { scheduleVoucherConsumeForOrderItems } from './voucher-consume-queue-service.js'
+import {
+  assignFulfillmentChannelsWithExecutor,
+  transitionFulfillmentState,
+  transitionFulfillmentStateBatch,
+  transitionFulfillmentStateBatchWithExecutor,
+  transitionFulfillmentStateWithExecutor,
+} from './fulfillment-state-service.js'
 
 /**
  * 管道 A：渠道判定。
@@ -365,20 +372,19 @@ export async function assignPendingOrderItemChannels(
       if (decision.channel) affectedDates.add(item.basis_date)
       if (decision.conflict) {
         result.conflicted += 1
-        await query(
-          `UPDATE order_items
-           SET fulfillment_pending_reason = 'routing_conflict',
-               fulfillment_pending_detail = $2::jsonb,
-               updated_at = NOW()
-           WHERE id = $1 AND fulfillment_status = 'pending'`,
-          [
-            item.id,
-            JSON.stringify({
+        await transitionFulfillmentState({
+          orderItemId: Number(item.id),
+          expected: { status: 'pending' },
+          patch: {
+            pendingReason: 'routing_conflict',
+            pendingDetail: {
               channels: decision.conflictChannels,
               ruleIds: decision.matchedRuleIds,
-            }),
-          ],
-        )
+            },
+          },
+          eventType: 'routing_conflict_detected',
+          reason: 'routing_conflict',
+        })
         await upsertAggregatedException({
           reason: 'routing_conflict',
           scopeKey: `routing_conflict|-|-|${item.internal_sku}|-`,
@@ -391,20 +397,18 @@ export async function assignPendingOrderItemChannels(
         result.stillPending += 1
         const pending = resolvePendingReason(item)
         result.pendingReasons[pending.reason] = (result.pendingReasons[pending.reason] || 0) + 1
-        await query(
-          `UPDATE order_items
-           SET fulfillment_pending_reason = $2,
-               fulfillment_pending_detail = CASE
-                 WHEN $2 = 'no_carrier'
-                   AND COALESCE(fulfillment_pending_detail->>'carrierIssue', '')
-                       NOT IN ('', 'carrier_not_attempted')
-                 THEN fulfillment_pending_detail
-                 ELSE $3::jsonb
-               END,
-               updated_at = NOW()
-           WHERE id = $1 AND fulfillment_status = 'pending'`,
-          [item.id, pending.reason, JSON.stringify(pending.detail)],
-        )
+        await transitionFulfillmentState({
+          orderItemId: Number(item.id),
+          expected: { status: 'pending' },
+          patch: {
+            pendingReason: pending.reason,
+            ...(pending.reason === 'no_carrier'
+              ? { pendingDetailUnlessCarrierIssue: pending.detail }
+              : { pendingDetail: pending.detail }),
+          },
+          eventType: 'routing_pending',
+          reason: pending.reason,
+        })
         continue
       }
       assignments.push({ id: item.id, channel: decision.channel, source: decision.source })
@@ -412,32 +416,19 @@ export async function assignPendingOrderItemChannels(
     }
 
     if (assignments.length > 0) {
-      const updated = await query<{ id: number }>(
-        `
-          UPDATE order_items oi
-          SET
-            fulfillment_channel = input.channel,
-            fulfillment_status = 'assigned',
-            fulfillment_source = input.source,
-            fulfillment_decided_at = NOW(),
-            fulfillment_pending_reason = NULL,
-            fulfillment_pending_detail = NULL,
-            updated_at = NOW()
-          FROM unnest($1::bigint[], $2::text[], $3::text[])
-            AS input(id, channel, source)
-          WHERE oi.id = input.id
-            AND oi.fulfillment_status = 'pending'
-            AND (oi.fulfillment_locked = FALSE OR $4::boolean)
-          RETURNING oi.id
-        `,
-        [
-          assignments.map((row) => row.id),
-          assignments.map((row) => row.channel),
-          assignments.map((row) => row.source),
-          options.includeLocked === true,
-        ],
+      const updated = await withTransaction((client) =>
+        assignFulfillmentChannelsWithExecutor(client, {
+          assignments: assignments.map((row) => ({
+            orderItemId: row.id,
+            channel: row.channel,
+            source: row.source,
+          })),
+          includeLocked: options.includeLocked === true,
+          eventType: 'routing_channel_assigned',
+          reason: 'automatic_routing',
+        }),
       )
-      const assignedItemIds = updated.rows
+      const assignedItemIds = updated
         .map((row) => Number(row.id))
         .filter((id) => id > 0)
         .slice(0, 2000)
@@ -503,12 +494,13 @@ export async function assignOrderItemsAfterCarrierReady(input: {
     ) {
       // 旧载体尚未撤销时禁止直接换渠道，否则会同时留下两个履约执行方。
       carrierFailed += 1
-      await query(
-        `UPDATE order_items
-         SET fulfillment_reason = $2, updated_at = NOW()
-         WHERE id = $1 AND fulfillment_status = 'assigned'`,
-        [item.id, '切换渠道前需先撤销现有履约载体'],
-      )
+      await transitionFulfillmentState({
+        orderItemId: Number(item.id),
+        expected: { status: 'assigned' },
+        patch: { reason: '切换渠道前需先撤销现有履约载体' },
+        eventType: 'channel_switch_blocked_by_existing_carrier',
+        reason: String(item.fulfillment_channel || ''),
+      })
       continue
     }
 
@@ -538,77 +530,71 @@ export async function assignOrderItemsAfterCarrierReady(input: {
       })
       if (item.fulfillment_status === 'assigned') {
         // 运营切换渠道失败时恢复原渠道，宁可保留旧状态，也不制造渠道与载体不一致。
-        await query(
-          `UPDATE order_items
-           SET fulfillment_channel = $2,
-               fulfillment_status = $3,
-               fulfillment_source = $4,
-               fulfillment_locked = $5,
-               fulfillment_reason = $6,
-               fulfillment_pending_reason = $7,
-               fulfillment_pending_detail = $8::jsonb,
-               updated_at = NOW()
-           WHERE id = $1 AND fulfillment_status <> 'archived'`,
-          [
-            item.id,
-            item.fulfillment_channel,
-            item.fulfillment_status,
-            item.fulfillment_source,
-            item.fulfillment_locked,
-            item.fulfillment_reason,
-            item.fulfillment_pending_reason,
-            JSON.stringify(item.fulfillment_pending_detail || null),
-          ],
-        )
+        await transitionFulfillmentState({
+          orderItemId: Number(item.id),
+          expected: { status: 'assigned' },
+          patch: {
+            channel: item.fulfillment_channel as FulfillmentChannel,
+            status: 'assigned',
+            source: item.fulfillment_source || '',
+            locked: Boolean(item.fulfillment_locked),
+            reason: item.fulfillment_reason || null,
+            pendingReason: item.fulfillment_pending_reason || null,
+            pendingDetail:
+              item.fulfillment_pending_detail && typeof item.fulfillment_pending_detail === 'object'
+                ? (item.fulfillment_pending_detail as Record<string, unknown>)
+                : null,
+          },
+          eventType: 'channel_switch_failed_existing_state_kept',
+          reason: carrier.reason,
+        })
       } else {
-        await query(
-          `UPDATE order_items
-           SET fulfillment_channel = NULL,
-               fulfillment_status = 'pending',
-               fulfillment_source = 'operator_assigned',
-               fulfillment_locked = TRUE,
-               fulfillment_pending_reason = 'no_carrier',
-               fulfillment_pending_detail = $2::jsonb,
-               canonical_product_id = COALESCE($3, canonical_product_id),
-               updated_at = NOW()
-           WHERE id = $1 AND fulfillment_status <> 'archived'`,
-          [
-            item.id,
-            JSON.stringify({
+        await transitionFulfillmentState({
+          orderItemId: Number(item.id),
+          expected: { status: 'pending' },
+          patch: {
+            channel: null,
+            status: 'pending',
+            source: 'operator_assigned',
+            locked: true,
+            pendingReason: 'no_carrier',
+            pendingDetail: {
               channel: input.channel,
               reason: carrier.reason,
               message: carrier.message,
               carrierIssue: carrier.reason,
-            }),
-            input.canonicalProductId || null,
-          ],
-        )
+            },
+            ...(input.canonicalProductId
+              ? { canonicalProductIdIfNull: Number(input.canonicalProductId) }
+              : {}),
+          },
+          eventType: 'operator_channel_assign_missing_carrier',
+          actor: { type: 'operator' },
+          reason: carrier.reason,
+        })
       }
       continue
     }
 
-    const updated = await query(
-      `UPDATE order_items
-       SET fulfillment_channel = $2,
-           fulfillment_status = 'assigned',
-           fulfillment_source = $3,
-           fulfillment_locked = $4,
-           canonical_product_id = COALESCE($6, canonical_product_id),
-           fulfillment_reason = COALESCE(NULLIF($5, ''), fulfillment_reason),
-           fulfillment_pending_reason = NULL,
-           fulfillment_pending_detail = NULL,
-           updated_at = NOW()
-       WHERE id = $1 AND fulfillment_status <> 'archived'`,
-      [
-        item.id,
-        input.channel,
-        input.source,
-        input.lock,
-        String(input.note || '').slice(0, 200),
-        input.canonicalProductId || null,
-      ],
-    )
-    if ((updated.rowCount || 0) > 0) assignedIds.push(Number(item.id))
+    const updated = await transitionFulfillmentState({
+      orderItemId: Number(item.id),
+      patch: {
+        channel: input.channel,
+        status: 'assigned',
+        source: input.source,
+        locked: input.lock,
+        ...(input.canonicalProductId
+          ? { canonicalProductIdIfNull: Number(input.canonicalProductId) }
+          : {}),
+        ...(String(input.note || '').trim() ? { reason: String(input.note).slice(0, 200) } : {}),
+        pendingReason: null,
+        pendingDetail: null,
+      },
+      eventType: 'operator_carrier_ready',
+      actor: { type: 'operator' },
+      reason: input.note || '',
+    })
+    if (updated) assignedIds.push(Number(item.id))
   }
 
   if (assignedIds.length > 0) {
@@ -625,25 +611,22 @@ export async function assignOrderItemsAfterCarrierReady(input: {
     ]
     await refreshFactsForDates(dates).catch(() => undefined)
   }
-  const final = await query<{ count: number }>(
-    `WITH updated AS (
-       UPDATE order_items
-       SET fulfillment_source = $3,
-           fulfillment_locked = $4,
-           fulfillment_pending_reason = NULL,
-           fulfillment_pending_detail = NULL,
-           updated_at = NOW()
-       WHERE id = ANY($1::bigint[])
-         AND fulfillment_status = 'assigned'
-         AND fulfillment_channel = $2
-       RETURNING id
-     )
-     SELECT count(*)::int AS count FROM updated`,
-    [ids, input.channel, input.source, input.lock],
-  )
+  const finalized = await transitionFulfillmentStateBatch({
+    orderItemIds: ids,
+    expected: { status: 'assigned', channel: input.channel },
+    patch: {
+      source: input.source,
+      locked: input.lock,
+      pendingReason: null,
+      pendingDetail: null,
+    },
+    eventType: 'assigned_channel_snapshot_updated',
+    actor: { type: 'operator' },
+    reason: input.note || '',
+  })
   await syncAssignedChannelSnapshots(ids, input.channel, input.source)
   return {
-    updated: Number(final.rows[0]?.count || assignedIds.length),
+    updated: finalized.length || assignedIds.length,
     carrierFailed,
     carrierFailures,
   }
@@ -867,24 +850,29 @@ export async function switchChannelForOrderItems(input: {
            WHERE order_item_id = $1 AND superseded_at IS NULL AND reversed_at IS NULL`,
           [item.id, now],
         )
-        await client.query(
-          `UPDATE order_items
-           SET original_channel = COALESCE(original_channel, fulfillment_channel),
-               fulfillment_channel = NULL,
-               fulfillment_status = 'pending',
-               fulfillment_source = 'operator_assigned',
-               fulfillment_locked = FALSE,
-               fulfillment_reason = $2,
-               channel_switch_reason = 'operator',
-               channel_switched_at = $3,
-               canonical_product_id = NULL,
-               fulfillment_config_id = NULL,
-               fulfillment_pending_reason = NULL,
-               fulfillment_pending_detail = NULL,
-               updated_at = $3
-           WHERE id = $1`,
-          [item.id, String(input.note || '运营切换履约渠道').slice(0, 200), now],
-        )
+        await transitionFulfillmentStateWithExecutor(client, {
+          orderItemId: Number(item.id),
+          expected: { status: item.fulfillment_status as 'pending' | 'assigned' | 'archived' },
+          patch: {
+            ...(isFulfillmentChannel(item.fulfillment_channel)
+              ? { originalChannelIfNull: item.fulfillment_channel }
+              : {}),
+            channel: null,
+            status: 'pending',
+            source: 'operator_assigned',
+            locked: false,
+            reason: String(input.note || '运营切换履约渠道').slice(0, 200),
+            switchReason: 'operator',
+            switchedAt: now,
+            canonicalProductId: null,
+            configId: null,
+            pendingReason: null,
+            pendingDetail: null,
+          },
+          eventType: 'operator_channel_switched',
+          actor: { type: 'operator', id: input.adminUserId },
+          reason: input.note || '',
+        })
       })
 
       for (const workOrder of workOrders.rows) {
@@ -925,12 +913,16 @@ export async function switchChannelForOrderItems(input: {
     note: input.note,
     canonicalProductId: draftProductId,
   })
-  await query(
-    `UPDATE order_items
-     SET fulfillment_decided_by = $2, fulfillment_decided_at = NOW()
-     WHERE id = ANY($1::bigint[])`,
-    [readyIds, input.adminUserId],
-  )
+  await transitionFulfillmentStateBatch({
+    orderItemIds: readyIds,
+    patch: {
+      decidedBy: input.adminUserId,
+      decidedAt: nowIso(),
+    },
+    eventType: 'operator_channel_decision_recorded',
+    actor: { type: 'operator', id: input.adminUserId },
+    reason: input.note || '',
+  })
   if (assigned.updated > 0) {
     const carrier = await resolveInternalSkuCarrierConfig(targetInternalSku, input.channel)
     await setInternalSkuPrimaryChannel({
@@ -1000,11 +992,14 @@ export async function assignChannelToOrderItems(input: {
 export async function unlockChannel(orderItemIds: number[]): Promise<{ updated: number }> {
   const ids = [...new Set(orderItemIds.map(Number).filter((id) => Number.isInteger(id) && id > 0))]
   if (ids.length === 0) return { updated: 0 }
-  const result = await query(
-    `UPDATE order_items SET fulfillment_locked = FALSE, updated_at = NOW() WHERE id = ANY($1::bigint[])`,
-    [ids],
-  )
-  return { updated: result.rowCount || 0 }
+  const updated = await transitionFulfillmentStateBatch({
+    orderItemIds: ids,
+    expected: { locked: true },
+    patch: { locked: false },
+    eventType: 'fulfillment_unlocked',
+    actor: { type: 'operator' },
+  })
+  return { updated: updated.length }
 }
 
 export type PendingGroupRow = {
@@ -1494,16 +1489,21 @@ export async function unlockPendingChannelGroup(input: {
     params.push(issueCategory)
     where.push(`state.issue_category = $${params.length}`)
   }
-  const result = await query(
-    `UPDATE order_items item
-     SET fulfillment_locked = FALSE,
-         updated_at = NOW()
+  const matched = await query<{ order_item_id: number }>(
+    `SELECT state.order_item_id
      FROM v_fulfillment_item_state state
-     WHERE item.id = state.order_item_id
-       AND ${where.join(' AND ')}`,
+     WHERE ${where.join(' AND ')}
+     ORDER BY state.order_item_id`,
     params,
   )
-  return { updated: result.rowCount || 0 }
+  const updated = await transitionFulfillmentStateBatch({
+    orderItemIds: matched.rows.map((row) => Number(row.order_item_id)),
+    expected: { status: 'pending', locked: true },
+    patch: { locked: false },
+    eventType: 'pending_group_unlocked',
+    actor: { type: 'operator' },
+  })
+  return { updated: updated.length }
 }
 
 export type DeletePendingChannelGroupResult = {
@@ -1650,27 +1650,26 @@ export async function deletePendingChannelGroup(input: {
       [itemIds, exceptionIds],
     )
 
-    const archived = await client.query(
-      `
-        UPDATE order_items
-        SET fulfillment_channel = NULL,
-            fulfillment_status = 'archived',
-            fulfillment_source = 'cutover_archive',
-            fulfillment_locked = TRUE,
-            fulfillment_reason = '运营从待分类队列删除',
-            fulfillment_pending_reason = NULL,
-            fulfillment_pending_detail = NULL,
-            fulfillment_decided_by = $2,
-            fulfillment_decided_at = NOW(),
-            updated_at = NOW()
-        WHERE id = ANY($1::bigint[])
-          AND fulfillment_status = 'pending'
-      `,
-      [itemIds, input.adminUserId],
-    )
+    const archived = await transitionFulfillmentStateBatchWithExecutor(client, {
+      orderItemIds: itemIds,
+      expected: { status: 'pending' },
+      patch: {
+        channel: null,
+        status: 'archived',
+        source: 'cutover_archive',
+        locked: true,
+        reason: '运营从待分类队列删除',
+        pendingReason: null,
+        pendingDetail: null,
+        decidedBy: input.adminUserId,
+        decidedAt: nowIso(),
+      },
+      eventType: 'pending_group_archived',
+      actor: { type: 'operator', id: input.adminUserId },
+    })
 
     return {
-      archivedItems: archived.rowCount || 0,
+      archivedItems: archived.length,
       preservedOrders: new Set(items.rows.map((row) => Number(row.order_id))).size,
       deletedCostSnapshots: snapshots.rowCount || 0,
       resolvedExceptions: resolved.rowCount || 0,
@@ -1866,12 +1865,16 @@ export async function assignChannelByInternalSku(input: {
     note: input.note,
     canonicalProductId: draftProductId,
   })
-  await query(
-    `UPDATE order_items
-     SET fulfillment_decided_by = $2, fulfillment_decided_at = NOW()
-     WHERE id = ANY($1::bigint[])`,
-    [ids, input.adminUserId],
-  )
+  await transitionFulfillmentStateBatch({
+    orderItemIds: ids,
+    patch: {
+      decidedBy: input.adminUserId,
+      decidedAt: nowIso(),
+    },
+    eventType: 'operator_channel_decision_recorded',
+    actor: { type: 'operator', id: input.adminUserId },
+    reason: input.note || '',
+  })
 
   return {
     ...applied,

apps/backend/src/services/fulfillment/fulfillment-retry-guard.test.ts

@@ -27,8 +27,9 @@ test('自动发货历史非活跃任务默认阻止后台自动重开', () => {
 
 test('已分配但载体失效的订单会退回 pending/no_carrier', () => {
   assert.ok(carrierReconciliation.includes('markAssignedMissingCarrierPending'))
-  assert.ok(carrierReconciliation.includes("fulfillment_status = 'pending'"))
-  assert.ok(carrierReconciliation.includes("fulfillment_pending_reason = 'no_carrier'"))
+  assert.ok(carrierReconciliation.includes('transitionFulfillmentState'))
+  assert.ok(carrierReconciliation.includes("status: 'pending'"))
+  assert.ok(carrierReconciliation.includes("pendingReason: 'no_carrier'"))
   assert.ok(carrierReconciliation.includes("'USER_APPLY_REFUND', 'affiliate_dash_cancelled'"))
 })
 
@@ -52,4 +53,5 @@ test('创建新履约尝试时自动退休旧当前尝试', () => {
   assert.ok(taskRepo.includes('MAX(attempt_no)'))
   assert.ok(taskRepo.includes('SET is_current = FALSE'))
   assert.ok(taskRepo.includes('$26, TRUE'))
+  assert.ok(taskRepo.includes('SELECT id FROM order_items WHERE id = $1 FOR UPDATE'))
 })

apps/backend/src/services/fulfillment/fulfillment-state-service.integration.test.ts

@@ -0,0 +1,162 @@
+/**
+ * 履约状态写入口集成测试（需要已执行完全部迁移的 PostgreSQL）。
+ *
+ * FULFILLMENT_STATE_TEST_DATABASE_URL=postgres://... \
+ *   npx tsx --test src/services/fulfillment/fulfillment-state-service.integration.test.ts
+ */
+import assert from 'node:assert/strict'
+import test from 'node:test'
+
+const TEST_DATABASE_URL = String(process.env.FULFILLMENT_STATE_TEST_DATABASE_URL || '').trim()
+
+test(
+  '履约状态事件覆盖状态写入口、遗留直写，并保持变更幂等',
+  { skip: !TEST_DATABASE_URL ? '未设置 FULFILLMENT_STATE_TEST_DATABASE_URL' : false },
+  async () => {
+    process.env.DATABASE_URL = TEST_DATABASE_URL
+    const { closeDb, query } = await import('../../db/client.js')
+    const {
+      assignFulfillmentChannelsWithExecutor,
+      listFulfillmentStateEvents,
+      runWithFulfillmentStateTransaction,
+      transitionFulfillmentState,
+    } = await import('./fulfillment-state-service.js')
+    const suffix = `${Date.now()}_${Math.floor(Math.random() * 1_000_000)}`
+    let orderId = 0
+    let orderItemId = 0
+
+    try {
+      const order = await query<{ id: number }>(
+        `INSERT INTO orders (
+           provider, platform, shop_id, shop_name, platform_order_id,
+           order_status, pay_status, total_amount, paid_at, created_at, updated_at
+         ) VALUES (
+           '91kaquan', 'kuaishou-91', $1, '状态事件集成测试', $2,
+           'paid', 'paid', 1000, NOW(), NOW(), NOW()
+         )
+         RETURNING id`,
+        [`ff-state-${suffix}`, `FF-STATE-${suffix}`],
+      )
+      orderId = Number(order.rows[0]!.id)
+      const item = await query<{ id: number }>(
+        `INSERT INTO order_items (
+           order_id, sku_code, sku_name, quantity, spec_json,
+           revenue_amount_fen, revenue_basis, created_at, updated_at
+         ) VALUES ($1, $2, $2, 1, '{}'::jsonb, 1000, 'order_total', NOW(), NOW())
+         RETURNING id`,
+        [orderId, `状态事件测试-${suffix}`],
+      )
+      orderItemId = Number(item.rows[0]!.id)
+
+      const pendingDetail = {
+        internalSku: `状态事件测试-${suffix}`,
+        channel: 'manual',
+        carrierIssue: 'manual_config_missing',
+      }
+      const first = await transitionFulfillmentState({
+        orderItemId,
+        expected: { status: 'pending' },
+        patch: {
+          pendingReason: 'no_carrier',
+          pendingDetail,
+        },
+        eventType: 'test_carrier_pending',
+        actor: { type: 'operator', id: 42 },
+        reason: 'integration_test',
+      })
+      assert.equal(first?.fulfillment_pending_reason, 'no_carrier')
+
+      let events = await listFulfillmentStateEvents(orderItemId)
+      assert.equal(events.length, 1)
+      assert.equal(events[0]?.event_type, 'test_carrier_pending')
+      assert.equal(events[0]?.actor_type, 'operator')
+      assert.equal(events[0]?.actor_id, '42')
+      assert.equal(events[0]?.reason, 'integration_test')
+      assert.deepEqual(events[0]?.next_pending_detail, pendingDetail)
+
+      await transitionFulfillmentState({
+        orderItemId,
+        expected: { status: 'pending' },
+        patch: {
+          pendingReason: 'no_carrier',
+          pendingDetail,
+        },
+        eventType: 'test_same_state_replay',
+      })
+      events = await listFulfillmentStateEvents(orderItemId)
+      assert.equal(events.length, 1, '相同状态重复写入不能制造第二条事件')
+
+      const stale = await transitionFulfillmentState({
+        orderItemId,
+        expected: { status: 'assigned' },
+        patch: {
+          channel: 'manual',
+          status: 'assigned',
+        },
+        eventType: 'test_stale_transition',
+      })
+      assert.equal(stale, null, '期望状态不一致时必须拒绝写入')
+      assert.equal((await listFulfillmentStateEvents(orderItemId)).length, 1)
+
+      await query(
+        `UPDATE order_items
+         SET fulfillment_pending_detail = jsonb_build_object('legacyWrite', TRUE),
+             updated_at = NOW()
+         WHERE id = $1`,
+        [orderItemId],
+      )
+      events = await listFulfillmentStateEvents(orderItemId)
+      assert.equal(events.length, 2, '遗留直写也必须由触发器留痕')
+      assert.equal(events[0]?.event_type, 'fulfillment_state_changed')
+      assert.equal(events[0]?.actor_type, 'system')
+
+      await query(
+        `UPDATE order_items
+         SET fulfillment_pending_detail = jsonb_build_object('legacyWrite', TRUE),
+             updated_at = NOW()
+         WHERE id = $1`,
+        [orderItemId],
+      )
+      events = await listFulfillmentStateEvents(orderItemId)
+      assert.equal(events.length, 2, '无字段变化的更新不能再记事件')
+
+      await runWithFulfillmentStateTransaction(async (client) => {
+        const assigned = await assignFulfillmentChannelsWithExecutor(client, {
+          assignments: [
+            {
+              orderItemId,
+              channel: 'third_party_voucher',
+              source: 'auto_mapping',
+            },
+          ],
+          eventType: 'test_channel_assigned',
+          reason: 'integration_test',
+        })
+        assert.equal(assigned.length, 1)
+        assert.equal(assigned[0]?.fulfillment_status, 'assigned')
+        assert.equal(assigned[0]?.fulfillment_channel, 'third_party_voucher')
+      })
+      events = await listFulfillmentStateEvents(orderItemId)
+      assert.equal(events[0]?.event_type, 'test_channel_assigned')
+      assert.equal(events[0]?.previous_status, 'pending')
+      assert.equal(events[0]?.next_status, 'assigned')
+      assert.equal(events[0]?.previous_channel, null)
+      assert.equal(events[0]?.next_channel, 'third_party_voucher')
+
+      await assert.rejects(
+        query(`UPDATE fulfillment_state_events SET reason = 'tampered' WHERE order_item_id = $1`, [
+          orderItemId,
+        ]),
+        /append-only/,
+      )
+    } finally {
+      if (orderItemId > 0) {
+        await query(`DELETE FROM order_items WHERE id = $1`, [orderItemId])
+      }
+      if (orderId > 0) {
+        await query(`DELETE FROM orders WHERE id = $1`, [orderId])
+      }
+      await closeDb()
+    }
+  },
+)

apps/backend/src/services/fulfillment/fulfillment-state-service.test.ts

@@ -0,0 +1,47 @@
+import assert from 'node:assert/strict'
+import { readFileSync } from 'node:fs'
+import test from 'node:test'
+
+function read(relativePath: string) {
+  return readFileSync(new URL(relativePath, import.meta.url), 'utf8')
+}
+
+const migration = read('../../db/migrations/140_fulfillment_state_events.sql')
+const stateService = read('./fulfillment-state-service.ts')
+const channelAssign = read('./channel-assign-service.ts')
+const carrierReconciliation = read('./carrier-reconciliation-service.ts')
+const affiliateFallback = read('./affiliate-fallback-service.ts')
+const taskRepo = read('../../repositories/task-repo.ts')
+
+test('履约状态事件表记录前后状态并禁止应用修改', () => {
+  assert.ok(migration.includes('CREATE TABLE IF NOT EXISTS fulfillment_state_events'))
+  assert.ok(migration.includes('previous_pending_detail JSONB'))
+  assert.ok(migration.includes('next_pending_detail JSONB'))
+  assert.ok(migration.includes('record_fulfillment_state_event'))
+  assert.ok(migration.includes("current_setting('app.fulfillment_actor_type', TRUE)"))
+  assert.ok(migration.includes('BEFORE UPDATE ON fulfillment_state_events'))
+  assert.ok(migration.includes('fulfillment_state_events is append-only'))
+})
+
+test('状态写入口串行锁订单项，并在写入前校验期望状态', () => {
+  assert.ok(stateService.includes('SELECT * FROM order_items WHERE id = $1 FOR UPDATE'))
+  assert.ok(stateService.includes('expected.status'))
+  assert.ok(stateService.includes('expected.channel'))
+  assert.ok(stateService.includes('sourceNotIn'))
+  assert.ok(stateService.includes("set_config('app.fulfillment_event_type'"))
+  assert.ok(stateService.includes('itemSnapshotPatch'))
+})
+
+test('高风险履约流程统一通过状态服务写入', () => {
+  for (const source of [channelAssign, carrierReconciliation, affiliateFallback]) {
+    assert.ok(source.includes('fulfillment-state-service'))
+    assert.ok(!source.includes('SET fulfillment_status ='))
+    assert.ok(!source.includes('fulfillment_status = $'))
+  }
+})
+
+test('任务创建先锁订单项并复用已有活跃尝试', () => {
+  assert.ok(taskRepo.includes('SELECT id FROM order_items WHERE id = $1 FOR UPDATE'))
+  assert.ok(taskRepo.includes('task_status NOT IN ('))
+  assert.ok(taskRepo.includes('ORDER BY is_current DESC, id DESC'))
+})

apps/backend/src/services/fulfillment/fulfillment-state-service.ts

@@ -0,0 +1,454 @@
+import type { PoolClient, QueryResult, QueryResultRow } from 'pg'
+
+import { query, withTransaction } from '../../db/client.js'
+import type { JsonObject } from '../../types/json.js'
+import type { OrderItemRow } from '../../types/repository/rows.js'
+import {
+  isFulfillmentChannel,
+  type FulfillmentChannel,
+  type FulfillmentSource,
+  type FulfillmentStatus,
+} from './channel-types.js'
+
+export type FulfillmentStateExecutor = {
+  query: <T extends QueryResultRow = QueryResultRow>(
+    text: string,
+    params?: unknown[],
+  ) => Promise<QueryResult<T>>
+}
+
+export type FulfillmentStateActor = {
+  type?: 'system' | 'operator' | 'worker' | 'finance' | 'api'
+  id?: number | string | null
+}
+
+export type FulfillmentStatePatch = {
+  status?: FulfillmentStatus
+  channel?: FulfillmentChannel | null
+  source?: FulfillmentSource | string
+  locked?: boolean
+  pendingReason?: string | null
+  pendingDetail?: JsonObject | null
+  pendingDetailUnlessCarrierIssue?: JsonObject
+  reason?: string | null
+  decidedBy?: number | null
+  decidedAt?: string | null
+  originalChannelIfNull?: FulfillmentChannel
+  switchReason?: string | null
+  switchedAt?: string | null
+  configId?: number | null
+  configIdIfNull?: number
+  canonicalProductId?: number | null
+  canonicalProductIdIfNull?: number
+  itemSnapshotJson?: JsonObject
+  itemSnapshotPatch?: JsonObject
+}
+
+export type FulfillmentStateExpected = {
+  status?: FulfillmentStatus | null
+  channel?: FulfillmentChannel | null
+  source?: string | null
+  sourceNotIn?: string[]
+  locked?: boolean | null
+}
+
+export type FulfillmentStateTransition = {
+  orderItemId: number
+  patch: FulfillmentStatePatch
+  expected?: FulfillmentStateExpected
+  eventType: string
+  actor?: FulfillmentStateActor
+  reason?: string
+}
+
+export type FulfillmentStateEventRow = {
+  [column: string]: unknown
+  id: number
+  order_item_id: number
+  order_id: number
+  event_type: string
+  actor_type: string
+  actor_id: string
+  reason: string
+  previous_status: string | null
+  next_status: string | null
+  previous_channel: string | null
+  next_channel: string | null
+  previous_source: string | null
+  next_source: string | null
+  previous_locked: boolean | null
+  next_locked: boolean | null
+  previous_pending_reason: string | null
+  next_pending_reason: string | null
+  previous_pending_detail: JsonObject | null
+  next_pending_detail: JsonObject | null
+  payload_json: JsonObject
+  created_at: string
+}
+
+type StateColumn = {
+  column: string
+  value: unknown
+  cast?: 'jsonb'
+  mergeSnapshot?: boolean
+  coalesceCurrent?: boolean
+  preserveCarrierIssue?: boolean
+}
+
+function normalizeId(value: unknown): number {
+  const id = Number(value)
+  if (!Number.isInteger(id) || id <= 0) {
+    throw new TypeError('orderItemId must be a positive integer')
+  }
+  return id
+}
+
+function buildPatchColumns(patch: FulfillmentStatePatch): StateColumn[] {
+  if (!patch || typeof patch !== 'object') {
+    throw new TypeError('fulfillment state patch is required')
+  }
+  const columns: StateColumn[] = []
+  if (patch.status !== undefined) {
+    if (!['pending', 'assigned', 'archived'].includes(patch.status)) {
+      throw new TypeError(`invalid fulfillment status: ${patch.status}`)
+    }
+    columns.push({ column: 'fulfillment_status', value: patch.status })
+  }
+  if (patch.channel !== undefined) {
+    if (patch.channel !== null && !isFulfillmentChannel(patch.channel)) {
+      throw new TypeError(`invalid fulfillment channel: ${patch.channel}`)
+    }
+    columns.push({ column: 'fulfillment_channel', value: patch.channel })
+  }
+  if (patch.source !== undefined) {
+    columns.push({ column: 'fulfillment_source', value: patch.source })
+  }
+  if (patch.locked !== undefined) {
+    columns.push({ column: 'fulfillment_locked', value: Boolean(patch.locked) })
+  }
+  if (patch.pendingReason !== undefined) {
+    columns.push({ column: 'fulfillment_pending_reason', value: patch.pendingReason || null })
+  }
+  if (patch.pendingDetail !== undefined) {
+    columns.push({
+      column: 'fulfillment_pending_detail',
+      value: patch.pendingDetail,
+      cast: 'jsonb',
+    })
+  }
+  if (patch.pendingDetailUnlessCarrierIssue !== undefined) {
+    columns.push({
+      column: 'fulfillment_pending_detail',
+      value: patch.pendingDetailUnlessCarrierIssue,
+      cast: 'jsonb',
+      preserveCarrierIssue: true,
+    })
+  }
+  if (patch.reason !== undefined) {
+    columns.push({ column: 'fulfillment_reason', value: patch.reason || null })
+  }
+  if (patch.decidedBy !== undefined) {
+    columns.push({ column: 'fulfillment_decided_by', value: patch.decidedBy })
+  }
+  if (patch.decidedAt !== undefined) {
+    columns.push({ column: 'fulfillment_decided_at', value: patch.decidedAt })
+  }
+  if (patch.originalChannelIfNull !== undefined) {
+    if (!isFulfillmentChannel(patch.originalChannelIfNull)) {
+      throw new TypeError(`invalid original fulfillment channel: ${patch.originalChannelIfNull}`)
+    }
+    columns.push({
+      column: 'original_channel',
+      value: patch.originalChannelIfNull,
+      coalesceCurrent: true,
+    })
+  }
+  if (patch.switchReason !== undefined) {
+    columns.push({ column: 'channel_switch_reason', value: patch.switchReason || null })
+  }
+  if (patch.switchedAt !== undefined) {
+    columns.push({ column: 'channel_switched_at', value: patch.switchedAt })
+  }
+  if (patch.configId !== undefined) {
+    columns.push({ column: 'fulfillment_config_id', value: patch.configId })
+  }
+  if (patch.configIdIfNull !== undefined) {
+    columns.push({
+      column: 'fulfillment_config_id',
+      value: patch.configIdIfNull,
+      coalesceCurrent: true,
+    })
+  }
+  if (patch.canonicalProductId !== undefined) {
+    columns.push({ column: 'canonical_product_id', value: patch.canonicalProductId })
+  }
+  if (patch.canonicalProductIdIfNull !== undefined) {
+    columns.push({
+      column: 'canonical_product_id',
+      value: patch.canonicalProductIdIfNull,
+      coalesceCurrent: true,
+    })
+  }
+  if (patch.itemSnapshotJson !== undefined) {
+    columns.push({
+      column: 'item_snapshot_json',
+      value: patch.itemSnapshotJson,
+      cast: 'jsonb',
+    })
+  }
+  if (patch.itemSnapshotPatch !== undefined) {
+    columns.push({
+      column: 'item_snapshot_json',
+      value: patch.itemSnapshotPatch,
+      cast: 'jsonb',
+      mergeSnapshot: true,
+    })
+  }
+  if (columns.length === 0) {
+    throw new TypeError('fulfillment state patch must change at least one column')
+  }
+  return columns
+}
+
+async function setEventContext(
+  executor: FulfillmentStateExecutor,
+  transition: Pick<FulfillmentStateTransition, 'eventType'> & {
+    actor?: FulfillmentStateActor
+    reason?: string
+  },
+) {
+  const eventType = String(transition.eventType || '').trim()
+  if (!eventType) throw new TypeError('fulfillment state eventType is required')
+  await executor.query(
+    `SELECT
+       set_config('app.fulfillment_event_type', $1, TRUE),
+       set_config('app.fulfillment_actor_type', $2, TRUE),
+       set_config('app.fulfillment_actor_id', $3, TRUE),
+       set_config('app.fulfillment_reason', $4, TRUE)`,
+    [
+      eventType,
+      transition.actor?.type || 'system',
+      String(transition.actor?.id ?? ''),
+      String(transition.reason || '').slice(0, 500),
+    ],
+  )
+}
+
+async function clearEventContext(executor: FulfillmentStateExecutor) {
+  await executor
+    .query(
+      `SELECT
+         set_config('app.fulfillment_event_type', '', TRUE),
+         set_config('app.fulfillment_actor_type', '', TRUE),
+         set_config('app.fulfillment_actor_id', '', TRUE),
+         set_config('app.fulfillment_reason', '', TRUE)`,
+    )
+    .catch(() => undefined)
+}
+
+export async function transitionFulfillmentStateWithExecutor(
+  executor: FulfillmentStateExecutor,
+  input: FulfillmentStateTransition,
+): Promise<OrderItemRow | null> {
+  const orderItemId = normalizeId(input.orderItemId)
+  const columns = buildPatchColumns(input.patch)
+  await setEventContext(executor, input)
+  try {
+    const lockResult = await executor.query<OrderItemRow>(
+      `SELECT * FROM order_items WHERE id = $1 FOR UPDATE`,
+      [orderItemId],
+    )
+    const current = lockResult.rows[0]
+    if (!current) return null
+
+    if (
+      input.expected?.status !== undefined &&
+      current.fulfillment_status !== input.expected.status
+    ) {
+      return null
+    }
+    if (
+      input.expected?.channel !== undefined &&
+      (current.fulfillment_channel ?? null) !== input.expected.channel
+    ) {
+      return null
+    }
+    if (
+      input.expected?.source !== undefined &&
+      (current.fulfillment_source || '') !== (input.expected.source || '')
+    ) {
+      return null
+    }
+    if (
+      input.expected?.sourceNotIn &&
+      input.expected.sourceNotIn.includes(String(current.fulfillment_source || ''))
+    ) {
+      return null
+    }
+    if (
+      input.expected?.locked !== undefined &&
+      Boolean(current.fulfillment_locked) !== Boolean(input.expected.locked)
+    ) {
+      return null
+    }
+
+    const params: unknown[] = []
+    const assignments = columns.map((entry) => {
+      params.push(entry.value)
+      let expression = `$${params.length}`
+      if (entry.cast) expression += `::${entry.cast}`
+      if (entry.mergeSnapshot) {
+        expression = `COALESCE(item_snapshot_json, '{}'::jsonb) || ${expression}`
+      } else if (entry.preserveCarrierIssue) {
+        expression =
+          `CASE WHEN COALESCE(fulfillment_pending_detail->>'carrierIssue', '') ` +
+          `NOT IN ('', 'carrier_not_attempted') ` +
+          `THEN fulfillment_pending_detail ELSE ${expression} END`
+      } else if (entry.coalesceCurrent) {
+        expression = `COALESCE(${entry.column}, ${expression})`
+      }
+      return `${entry.column} = ${expression}`
+    })
+    params.push(new Date().toISOString())
+    assignments.push(`updated_at = $${params.length}::timestamptz`)
+    params.push(orderItemId)
+
+    const updated = await executor.query<OrderItemRow>(
+      `UPDATE order_items
+       SET ${assignments.join(', ')}
+       WHERE id = $${params.length}
+       RETURNING *`,
+      params,
+    )
+    return updated.rows[0] || null
+  } finally {
+    await clearEventContext(executor)
+  }
+}
+
+export async function transitionFulfillmentState(
+  input: FulfillmentStateTransition,
+): Promise<OrderItemRow | null> {
+  return withTransaction((client) =>
+    transitionFulfillmentStateWithExecutor(client as FulfillmentStateExecutor, input),
+  )
+}
+
+export async function transitionFulfillmentStateBatchWithExecutor(
+  executor: FulfillmentStateExecutor,
+  input: {
+    orderItemIds: number[]
+    patch: FulfillmentStatePatch
+    expected?: FulfillmentStateExpected
+    eventType: string
+    actor?: FulfillmentStateActor
+    reason?: string
+  },
+): Promise<OrderItemRow[]> {
+  const ids = [
+    ...new Set(
+      (input.orderItemIds || []).map(Number).filter((id) => Number.isInteger(id) && id > 0),
+    ),
+  ]
+  const results: OrderItemRow[] = []
+  for (const orderItemId of ids) {
+    const updated = await transitionFulfillmentStateWithExecutor(executor, {
+      orderItemId,
+      patch: input.patch,
+      eventType: input.eventType,
+      ...(input.expected ? { expected: input.expected } : {}),
+      ...(input.actor ? { actor: input.actor } : {}),
+      ...(input.reason ? { reason: input.reason } : {}),
+    })
+    if (updated) results.push(updated)
+  }
+  return results
+}
+
+export async function transitionFulfillmentStateBatch(
+  input: Parameters<typeof transitionFulfillmentStateBatchWithExecutor>[1],
+): Promise<OrderItemRow[]> {
+  return withTransaction((client) =>
+    transitionFulfillmentStateBatchWithExecutor(client as FulfillmentStateExecutor, input),
+  )
+}
+
+export async function assignFulfillmentChannelsWithExecutor(
+  executor: FulfillmentStateExecutor,
+  input: {
+    assignments: Array<{
+      orderItemId: number
+      channel: FulfillmentChannel
+      source: FulfillmentSource | string
+    }>
+    includeLocked?: boolean
+    eventType?: string
+    actor?: FulfillmentStateActor
+    reason?: string
+  },
+): Promise<OrderItemRow[]> {
+  const assignments = (input.assignments || [])
+    .map((row) => ({
+      orderItemId: normalizeId(row.orderItemId),
+      channel: row.channel,
+      source: row.source,
+    }))
+    .filter((row) => isFulfillmentChannel(row.channel))
+  if (assignments.length === 0) return []
+
+  await setEventContext(executor, {
+    eventType: input.eventType || 'fulfillment_channel_assigned',
+    ...(input.actor ? { actor: input.actor } : {}),
+    ...(input.reason ? { reason: input.reason } : {}),
+  })
+  try {
+    const result = await executor.query<OrderItemRow>(
+      `UPDATE order_items item
+       SET fulfillment_channel = input.channel,
+           fulfillment_status = 'assigned',
+           fulfillment_source = input.source,
+           fulfillment_decided_at = NOW(),
+           fulfillment_pending_reason = NULL,
+           fulfillment_pending_detail = NULL,
+           updated_at = NOW()
+       FROM unnest($1::bigint[], $2::text[], $3::text[])
+         AS input(order_item_id, channel, source)
+       WHERE item.id = input.order_item_id
+         AND item.fulfillment_status = 'pending'
+         AND (item.fulfillment_locked = FALSE OR $4::boolean)
+       RETURNING item.*`,
+      [
+        assignments.map((row) => row.orderItemId),
+        assignments.map((row) => row.channel),
+        assignments.map((row) => row.source),
+        input.includeLocked === true,
+      ],
+    )
+    return result.rows
+  } finally {
+    await clearEventContext(executor)
+  }
+}
+
+export async function listFulfillmentStateEvents(
+  orderItemId: number | string,
+  options: { limit?: number } = {},
+): Promise<FulfillmentStateEventRow[]> {
+  const id = normalizeId(orderItemId)
+  const limit = Math.min(500, Math.max(1, Math.trunc(Number(options.limit) || 100)))
+  const result = await query<FulfillmentStateEventRow>(
+    `SELECT *
+     FROM fulfillment_state_events
+     WHERE order_item_id = $1
+     ORDER BY id DESC
+     LIMIT $2`,
+    [id, limit],
+  )
+  return result.rows
+}
+
+export async function runWithFulfillmentStateTransaction<T>(
+  fn: (client: PoolClient) => Promise<T>,
+): Promise<T> {
+  return withTransaction(fn)
+}

apps/backend/src/services/order/delivery-task-service.test.ts

@@ -66,7 +66,7 @@ function stubCreatedTask(id: number, input: CreatedTaskInput): TaskRow {
     unit_index: Number(input.unitIndex || 1),
     platform_order_id: '',
     profile_id: 0,
-    task_no: 'DT-test',
+    task_no: String(input.taskNo || 'DT-test'),
     executor_key: String(input.executorKey || ''),
     task_status: String(input.taskStatus || 'pending'),
     delivery_status: 'pending',
@@ -428,6 +428,51 @@ test('syncDeliveryTasksForOrder 并发创建冲突时复用已落库任务', asy
   assert.equal(eventCount, 0)
 })
 
+test('createTask 幂等返回已有活跃任务时不重复记计划事件', async () => {
+  const existingTask = {
+    ...stubCreatedTask(402, {
+      orderId: 61,
+      orderItemId: 51,
+      unitIndex: 1,
+      executorKey: 'affiliate_dash',
+    }),
+    task_no: 'DT-existing',
+  }
+  let eventCount = 0
+
+  const tasks = await syncDeliveryTasksForOrderWithDeps(
+    stubOrder({ id: 61, platform_order_id: 'idempotent-create-order' }),
+    [
+      stubOrderItem({
+        id: 51,
+        order_id: 61,
+        sku_code: 'SKU-idempotent',
+        item_snapshot_json: buildAffiliateSnapshot('AFF-IDEMPOTENT', '商品 IDEMPOTENT'),
+      }),
+    ],
+    {
+      loadActiveCarrierEvidence: async () => new Map(),
+      listTasksByOrderId: async () => [],
+      getFulfillmentProfileByKey: async () => buildAffiliateProfile(),
+      createTask: async () => existingTask,
+      createTaskEvent: async () => {
+        eventCount += 1
+        return null
+      },
+      logInfo: () => null,
+      logWarn: () => null,
+      nowIso: () => '2026-05-29T09:05:46.658Z',
+      randomId: () => 'DT-reused',
+    },
+  )
+
+  assert.deepEqual(
+    tasks.map((task) => task.id),
+    [402],
+  )
+  assert.equal(eventCount, 0)
+})
+
 test('来源重推时已分配其它渠道的订单不会补建 affiliate 任务', async () => {
   let createdCount = 0
   const tasks = await syncDeliveryTasksForOrderWithDeps(

apps/backend/src/services/order/delivery-task-service.ts

@@ -181,8 +181,9 @@ export async function syncDeliveryTasksForOrderWithDeps(
           ? resolvePaidTaskStatus(plan.profile)
           : TASK_STATUS.PENDING_PAYMENT
 
-      let createdFresh = true
+      let createdFresh = false
       let task: TaskRow | null
+      const plannedTaskNo = createRandomId('DT')
       try {
         task = await createDeliveryTask({
           orderId: order.id,
@@ -193,7 +194,7 @@ export async function syncDeliveryTasksForOrderWithDeps(
           shopId: order.shop_id,
           shopName: order.shop_name,
           platformOrderId: order.platform_order_id,
-          taskNo: createRandomId('DT'),
+          taskNo: plannedTaskNo,
           profileId: plan.profileId,
           executorKey: plan.executorKey,
           taskStatus: initialStatus,
@@ -211,6 +212,7 @@ export async function syncDeliveryTasksForOrderWithDeps(
           createdAt,
           updatedAt: createdAt,
         })
+        createdFresh = task?.task_no === plannedTaskNo
       } catch (error) {
         const concurrentTasks = await listTasks(order.id)
         task =
@@ -229,6 +231,14 @@ export async function syncDeliveryTasksForOrderWithDeps(
             profileId: plan.profileId,
             profileKey: plan.profileKey,
             executorKey: plan.executorKey,
+            candidateTasks: concurrentTasks
+              .filter((candidate) => candidate.order_item_id === item.id)
+              .map((candidate) => ({
+                id: candidate.id,
+                unitIndex: candidate.unit_index,
+                status: candidate.task_status,
+                isCurrent: candidate.is_current,
+              })),
             error: error instanceof Error ? error.message : String(error),
           })
           throw error

apps/backend/src/types/repository/rows.ts

@@ -84,6 +84,8 @@ export type TaskRow = {
   screenshot_path: string
   last_error: string
   retry_count: number
+  attempt_no?: number
+  is_current?: boolean
   created_at: string
   updated_at: string
   kuaishou_ticket_status?: string