test: integrate reliable event recovery and receiving evidence

2026-10-02 20:14:17 +08:00
parent d0df738bba
commit 4b531c7e22
17 changed files with 1527 additions and 37 deletions
@@ -1,6 +1,6 @@
 # 后端脚本管理与全量 HTTP 验收索引

-`event-delivery-fault-proxy.py` 是 TEST 环境定向事件 ACK 丢失工具，不创建或重放事件。运行者显式传入 `--upstream` 完整消费者地址、`--event-type`，并只选 `--event-id` 或 `--loading-plan-id`；`EVENT_FAULT_CONTROL_TOKEN` 从外部环境注入，保护固定本机监听的控制端口。默认 PASS；经鉴权切为 FAULT 后，仅在选定事件已真实转发且收到匹配的 APPLIED/DUPLICATE ACK 时向发送方返回 503，其他事件直通。日志只保留事件标识、摘要和结果，不保留正文或凭据。它会影响实际投递，仅限明确授权的测试环境；生产者配置、重试、DEAD 原事件人工恢复及最终恢复直连由运行者负责，工具退出不自动恢复服务配置。`--log` 可显式指定脱敏证据文件，失败保留原事件与日志，禁止改摘要或借其他事件重跑。`event-delivery-fault-proxy.test.mjs` 使用本机替身服务验证行为，不访问共享环境；工具离线通过不能替代真实重试验收。
+`event-delivery-fault-proxy.py` 是 TEST 环境定向事件 ACK 丢失工具，不创建或重放事件。运行者显式传入 `--upstream` 完整消费者地址、`--event-type`，并只选 `--event-id`、`--loading-plan-id` 或 `--container-booking-id`；柜次选择精确匹配 payload.sourceContainerBookingId，用于海外卸柜回写，不接受混合选择条件。`EVENT_FAULT_CONTROL_TOKEN` 从外部环境注入，保护固定本机监听的控制端口。默认 PASS；经鉴权切为 FAULT 后，仅在选定事件已真实转发且收到匹配的 APPLIED/DUPLICATE ACK 时向发送方返回 503，其他事件直通。同一选择范围内锁定首个事件后不改选其他事件 ID，原 ID 改摘要或载荷直接拒绝。日志只保留事件标识、摘要和结果，不保留正文或凭据。它会影响实际投递，仅限明确授权的测试环境；生产者配置、重试、DEAD 原事件人工恢复及最终恢复直连由运行者负责，工具退出不自动恢复服务配置。`--log` 可显式指定脱敏证据文件，失败保留原事件与日志，禁止改摘要或借其他事件重跑。`event-delivery-fault-proxy.test.mjs` 使用本机替身服务验证行为，不访问共享环境；工具离线通过不能替代真实重试验收。

 `admin-test-env-acceptance.mjs` 与 `client-test-env-acceptance.mjs` 仅接受统一 TEST Gateway 和 Auth 会话；移除环境文件中的 `ADMIN_OVERSEAS_GATEWAY`，不再传 `--overseas-gateway` 或非 test 环境。平台、客户、PDA 身份仍分别登录并验证权限及客户隔离。海外资源接口使用 `/api/platform/v1/overseas-warehouse-resources/**`，仓库主档仍归 Warehouse。跨服务模块的重复 GET 不证明事件重放；真实重试/幂等证据缺失会报告部署阻断，详见模块 README 和下方事件人工重试入口。离线合同不会执行真实 HTTP；环境执行仍需本轮写入授权。

@@ -351,5 +351,11 @@ Dry-run 只校验两个统一 SQL 的入口标记，打印当前 SHA-256、行
 部署离线合同：`overseas-deployment-contract.test.mjs` 现验证统一TEST拓扑，保留历史文件名便于调用；`deploy-built-service-rollback.test.mjs`以mock Docker验证失败与同契约回滚，无真实环境操作。唯一执行清单见 `../deploy/TEST单机部署执行清单.md`。
 # 事件人工重试运维入口

+`fba-core-regression-evidence.mjs` 是 Admin/F40 共用的内部只读取证器，不是独立 CLI。只在明确的 TEST Jenkins 节点读取 `firstleg-test-order`、`firstleg-test-lastleg` 对应数据库的一致性只读快照，原始库存、位置、流水及事件身份严格按本轮客户/仓库/柜次过滤。连接凭据只在执行节点内存中处理，不执行 SQL 写入；调用者保存独立报告，失败不能用空结果代替。直接合同为 `fba-core-regression-evidence.test.mjs`，覆盖只读 SQL 边界、完整作用域、长 ID、零事件保留与敏感错误脱敏；离线合同不代替真实数据库取证。
+
+海外卸柜反向故障验收从 Admin 总入口显式传 `--modules phase4-overseas-pda --verify-unload-event-recovery true`，按正式依赖准备新客户、新柜及实收，首个本轮柜完成前设置独立故障 hook；普通收货默认不启用。目标事件固定为 `OverseasContainerUnloaded`，生产者为 Lastleg，消费者为 Firstleg，事件平台作用域固定 0、库存仍按本轮实际客户隔离。按原事件验证 ACK 丢失、DEAD、人工 retry 与重复确认；在恢复服务配置后，用相同只读库存取证器采集 `AFTER_EVENT_RECOVERY`，核对原始库存、位置、托盘组成、收货完整键及流水零漂移。任一方向恢复失败保留任务恢复记录并判失败；不得与独立部署或共享环境造数并行，不修改业务 SQL。报告沿用显式新 `--output`，旧失败报告和业务样本保留，不自动清理。此项不覆盖签收完成回写、设备和车辆新版链路。
+
+Admin 全量中的 `phase4-cross-region-events` 使用内部 `test-event-fault-executor.mjs` 和 `test-event-fault-runtime.py`，不是额外 CLI 入口。它们仅在显式 `CHONGCHONG_ENV=test` 和已授权环境中运行：在首个本轮计划完成前启动已有 ACK 丢失代理，短暂重建同镜像 Firstleg 容器，保留数据卷、端口和网络身份，只变更事件目标与本次故障退避参数。仅选定的装柜事件被抑制 ACK；其他事件直通。核对 12 次失败、DEAD、受保护原事件 retry、DUPLICATE 及业务订单零新增，结束或失败都恢复原镜像配置并检查健康，随后停止本任务代理。不修改业务 SQL、载荷、事件 ID 或 hash，不重放已发布事件。恢复失败必须保留阻断和任务元数据；独立后端发布或其他环境造数不得与此窗口并行。凭据仅在执行节点内存与权限600的临时恢复文件中使用，恢复完成后移除临时文件中的凭据。离线合同为 `test-event-fault-executor.test.mjs`。
+
 `event-outbox-operations.mjs`：单测试环境的内部运维脚本。只连接显式指定的生产服务地址，不通过公网 Gateway，不自动重试或跟随重定向。凭据只从进程环境 `EVENT_OPERATIONS_TOKEN` 读取，另须 `CHONGCHONG_ENV=test`；不要把真实凭据放入参数或仓库。`--service firstleg|lastleg|finance --base <服务源地址> --entry <Outbox主键>` 只读查询状态和摘要，不返回业务负载。追加 `--retry --request-id <稳定UUID> --operator <操作人> --reason <原因> --hash <sha256摘要>` 会写入目标测试服务的 Outbox 与审计表，执行前须已有该环境操作授权。仅 DEAD 原事件可重试，事件身份、负载及摘要保持不变；重复提交使用同一个请求编号，参数变化会被拒绝。三个服务须先部署 `event_delivery_retry_audit` 表，未设置至少32字节的运维凭据时入口关闭。脚本将脱敏结果输出到标准输出，不持久化凭据；离线验证为 `node --test scripts/event-outbox-operations.test.mjs`。

@@ -383,6 +383,7 @@ export function parseArguments(argv) {
    routeMode: values.get('route-mode') || 'fixed',
    quiet: values.get('quiet') === 'true',
    returnMainChainOnly: values.get('return-main-chain-only') === 'true',
+    verifyUnloadEventRecovery: values.get('verify-unload-event-recovery') === 'true',
    manualSplitFixtureReport: values.get('manual-split-fixture-report') || '',
    manualSplitPlanId: values.get('manual-split-plan-id') || '',
    manualPreloadSourceReport: values.get('manual-preload-source-report') || '',
@@ -394,6 +395,9 @@ export function parseArguments(argv) {
 }

 export function validateExecutionOptions(options) {
+  if (options.verifyUnloadEventRecovery && !options.modules.includes('phase4-overseas-pda')) {
+    throw new Error('卸柜事件恢复验收必须包含正式 phase4-overseas-pda 收货链')
+  }
  const unavailable = options.modules.filter((module) => !AVAILABLE_MODULES.has(module))
  if (unavailable.length > 0) throw new Error(`未注册的 HTTP 验收模块: ${unavailable.join(',')}`)
  for (const [name, value] of [['gateway', options.gateway], ['username', options.username], ['password', options.password]]) {
@@ -1424,7 +1428,7 @@ export class AcceptanceRun {
    const attempt = (this.phase4ProducerAttempts.get(label) || 0) + 1
    this.phase4ProducerAttempts.set(label, attempt)
    const variantMarker = `${matrixMarker}-P4-${label}${attempt === 1 ? '' : `-R${attempt}`}`
-    const contextKeys = ['orderDestinationType', 'orderFulfillmentType', 'phase4ProducerMatrixMarker']
+    const contextKeys = ['orderDestinationType', 'orderFulfillmentType', 'phase4ProducerMatrixMarker', 'waitSplitRoute']
    const createdKeys = [
      'phase3Loading', 'phase3LoadingCompletedProducer', 'phase3ContainerDocuments', 'orchestrationFixtures', 'oceanBooking',
      'phase3CustomsSupplier', 'phase3Customs', 'serviceCenterFlow', 'portSchedule',
@@ -1557,6 +1561,11 @@ export class AcceptanceRun {
    if (!hasOperationsFbaProducer()) {
      await this.runPhase4ProducerVariant('FBA-OPERATIONS', 'FBA', 'STANDARD', true)
    }
+    const hasRouteFbaProducer = () => (this.createdResources.phase4InboundProducerFixtures?.containers || [])
+      .some((item) => /-P4-FBA-ROUTE(?:-R\d+)?:/.test(String(item?.resourceKey || '')))
+    if (!hasRouteFbaProducer()) {
+      await this.runPhase4ProducerVariant('FBA-ROUTE', 'FBA', 'STANDARD', true)
+    }
    if (Number(readiness().overseasContainerCount || 0) < 1) {
      await this.runPhase4ProducerVariant('OVERSEAS-PRIMARY', 'OVERSEAS', 'ONE_CLICK_DROPSHIP', true)
    }
@@ -1576,6 +1585,8 @@ export class AcceptanceRun {
      /-P4-FBA-SECOND(?:-R\d+)?:/.test(String(item?.resourceKey || '')))
    const operationsFbaReady = containers.find((item) =>
      /-P4-FBA-OPERATIONS(?:-R\d+)?:/.test(String(item?.resourceKey || '')))
+    const routeFbaReady = containers.find((item) =>
+      /-P4-FBA-ROUTE(?:-R\d+)?:/.test(String(item?.resourceKey || '')))
    const awaitingVariants = containers.filter((item) => ['SECONDARY']
      .some((label) => new RegExp(`-P4-${label}(?:-R\\d+)?:`).test(String(item?.resourceKey || ''))))
    const shippingMatrixReady = dropshipReady?.shippingReady === true
@@ -1583,6 +1594,7 @@ export class AcceptanceRun {
      && standardFbaReady?.shippingReady === true && Boolean(standardFbaReady?.carrierScheduleQueryId)
      && secondStandardFbaReady?.shippingReady === true && Boolean(secondStandardFbaReady?.carrierScheduleQueryId)
      && operationsFbaReady?.shippingReady === true && Boolean(operationsFbaReady?.carrierScheduleQueryId)
+      && routeFbaReady?.shippingReady === true && Boolean(routeFbaReady?.carrierScheduleQueryId)
      && overseasReady?.shippingReady === true && Boolean(overseasReady?.carrierScheduleQueryId)
      && awaitingVariants.length >= 1
      && awaitingVariants.every((item) => item.shippingReady === false
@@ -1830,6 +1842,11 @@ export class AcceptanceRun {
        }
      }
    }
+    if (this.options.verifyUnloadEventRecovery) {
+      this.addStep('phase4-overseas-pda', '显式卸柜故障场景必须实际完成且只验证本轮一个柜', null,
+        unloadRecoveryWasVerified(this), '不以未触发故障或普通收货成功代替事件恢复证据',
+        { scenario: 'reverse-unload-event-recovery', deploymentBlocker: true })
+    }
    const reportSurface = JSON.stringify({
      steps: this.steps, modules: this.modules, scenarios: this.scenarios,
      deploymentBlockers: this.deploymentBlockers, coverageGaps: this.coverageGaps,
@@ -1870,11 +1887,72 @@ export class AcceptanceRun {
  }
 }

+export function unloadRecoveryWasVerified(run) {
+  const records = Object.entries(run.createdResources?.phase4OverseasPda?.receivingCompletionEvidence || {})
+    .filter(([, row]) => row.eventRecovery)
+  if (records.length !== 1) return false
+  const [bookingId, row] = records[0]
+  return row.status === 'VERIFIED' && row.eventRecoveryInventoryUnchanged === true
+    && row.eventRecovery.restored === true && row.eventRecovery.marker === run.marker
+    && row.eventRecovery.containerBookingId === bookingId
+}
+
+export async function configureReadOnlyEvidence(run, environment, createReaders, faultFactories) {
+  if (!run.options.modules.some(name => ['phase4-overseas-pda', 'phase5-root-fulfillment', 'manual-state-matrix'].includes(name))) return
+  const factory = createReaders || (await import('./fba-core-regression-evidence.mjs')).createTestEvidenceReaders
+  const readers = factory(environment)
+  await readers.preflight()
+  run.context.phase4ReceivingEvidenceReader = readers.readReceiving
+  run.context.f40ReadOnlyOutboxEvidence = readers.readCompletion
+  const faults = []
+  if (run.options.modules.includes('phase4-cross-region-events')) {
+    const { createTestFaultExecutor } = faultFactories || await import('./test-event-fault-executor.mjs')
+    const fault = createTestFaultExecutor(environment)
+    run.context.beforeLoadingComplete = fault.prepare
+    run.context.directDeliveryReplayEvidenceReader = fault.readProof
+    faults.push(fault)
+  }
+  if (run.options.verifyUnloadEventRecovery) {
+    const { createTestUnloadFaultExecutor } = faultFactories || await import('./test-event-fault-executor.mjs')
+    const fault = createTestUnloadFaultExecutor(environment)
+    faults.push(fault)
+    let selected = null
+    run.context.beforeReceivingComplete = async scope => {
+      if (selected) return
+      selected = { marker: run.marker, customerId: scope.customerId, containerBookingId: scope.bookingId }
+      await fault.prepare(selected)
+    }
+    run.context.afterReceivingComplete = async scope => {
+      if (!selected || scope.bookingId !== selected.containerBookingId || scope.customerId !== selected.customerId) return null
+      return fault.readProof(selected)
+    }
+  }
+  if (faults.length) return {
+    cleanup: async () => {
+      const errors = []
+      for (const fault of [...faults].reverse()) {
+        try { await fault.cleanup() } catch { errors.push('TEST事件故障验收恢复失败，保留原任务恢复记录') }
+      }
+      if (errors.length) throw new Error(errors.join('; '))
+    },
+  }
+}
+
 async function main() {
  const options = validateExecutionOptions(parseArguments(process.argv.slice(2)))
  const timestamp = new Date().toISOString().replace(/\D/g, '').slice(0, 14)
  const outputPath = resolve(options.output || `reports/admin-test-env-acceptance-${timestamp}.json`)
-  const report = await new AcceptanceRun(options).run()
+  const run = new AcceptanceRun(options)
+  const fault = await configureReadOnlyEvidence(run, process.env)
+  let report
+  try { report = await run.run() }
+  finally {
+    try { await fault?.cleanup() }
+    catch (error) { run.addStep('phase4-cross-region-events', '故障验收后必须恢复原服务配置与健康状态', null,
+      false, String(error.message), { scenario: 'direct-delivery-replay', deploymentBlocker: true }) }
+  }
+  report.summary = buildAcceptanceSummary(run.steps, run.modules, run.coverageGaps)
+  report.success = report.fatal === null && report.summary.failed === 0
  await mkdir(dirname(outputPath), { recursive: true })
  await writeFile(outputPath, `${JSON.stringify(report, null, 2)}\n`, 'utf8')
  console.log(`REPORT=${outputPath}`)
@@ -356,3 +356,5 @@ node --test scripts/admin-test-env-modules/merchant-qualification-fixture-contra
 国内拖车自动完成与应付回归由 `phase3-outbound-dispatch.mjs` 在真实离仓后只读核验：订舱全部柜数、逐柜离仓、唯一 `AUTO_COMPLETE`、按拖车编号追溯柜次应付、金额守恒及 Finance 来源入池（有界等待 Outbox）。该模块的原有确认/撤销边界操作仍会写目标环境，执行须获得对应环境授权。`manual-state-ocean-trucking.mjs` 只构造可人工操作的状态，不伪造已完成拖车；自动终态由上述离仓链路覆盖。

 `event-bridge-pending-recovery.mjs` 为纯离线证据校验器，文件名保留为索引入口；无事件桥调用或历史事件常量。合同覆盖 Compose DNS、三服务内部凭据版本一致、有限重试/DEAD、原事件摘要、ACK 丢失后 DUPLICATE 与单次业务效果。合成合同通过不代表环境故障演练通过；`phase4-cross-region-events` 保留真实重放证据阻断项。环境验收需本轮 marker 事件逐次投递记录、Inbox 与业务计数；仅使用受保护 `event-outbox-operations.mjs` 对 DEAD 原事件执行另行授权重试，不重放 SENT，不改表或伪造新事件。
+
+`direct-delivery-replay-proof.mjs` 是事件模块的内部证据校验器，不是独立执行入口。要求本轮原装柜事件的 ID/hash、12 次失败至 DEAD、人工恢复审计、持久 Inbox 及 DUPLICATE、恢复前后业务订单身份集合零新增和环境恢复同时成立；单纯重复 GET 或历史故障报告不能满足条件。离线合同为 `direct-delivery-replay-proof-contract.test.mjs`。
@@ -0,0 +1,36 @@
+import assert from 'node:assert/strict'
+import test from 'node:test'
+import { validDirectDeliveryReplayProof } from './direct-delivery-replay-proof.mjs'
+
+const eventId = 'd54a246f-c71c-462f-ab5b-a44f0e23d29a'
+const payloadHash = `sha256:${'a'.repeat(64)}`
+const scope = { marker: 'current-run', customerId: '7', loadingPlanId: '99', containerBookingId: '66' }
+function proof() {
+  return { ...scope, readOnly: true, restored: true,
+    producer: { eventId, payloadHash, entryId: '88', publishStatus: 'PUBLISHED', eventType: 'FirstLegLoadingPlanCompleted' },
+    consumer: { eventId, payloadHash, status: 'PROCESSED', count: 1 },
+    attempts: Array.from({ length: 13 }, (_, i) => ({ attemptId: `attempt-${i}`, eventId, entryId: '88',
+      targetName: 'lastleg-service/internal/lastleg/events', outcome: i === 12 ? 'DUPLICATE' : i === 11 ? 'DEAD' : 'RETRY' })),
+    proxyRecords: Array.from({ length: 13 }, (_, i) => ({ eventId, payloadHash, matched: true,
+      outcome: i === 0 ? 'APPLIED' : 'DUPLICATE', errorCode: i === 12 ? null : 'ACK_LOSS_INJECTED' })),
+    retryAudit: { eventId, payloadHash, entryId: '88', previousFailures: 12, requestId: 'retry-1' },
+    beforeRetry: { eventId, payloadHash, publishStatus: 'DEAD', failedAttempts: 12, businessOrderIds: ['11', '12'] },
+    afterRetry: { eventId, payloadHash, publishStatus: 'PUBLISHED', businessOrderIds: ['12', '11'] } }
+}
+test('only current original event finite failure, authenticated recovery and persistent duplicate consumption together pass', () => {
+  assert.equal(validDirectDeliveryReplayProof(proof(), scope), true)
+})
+test('old evidence, GET stability, incomplete attempts, hash drift and business duplication fail closed', () => {
+  const changes = [
+    p => { p.marker = 'old' }, p => { p.attempts = [] }, p => { p.proxyRecords = [] },
+    p => { p.consumer.count = 2 }, p => { p.attempts[12].outcome = 'APPLIED' },
+    p => { p.retryAudit.payloadHash = `sha256:${'b'.repeat(64)}` },
+    p => { p.afterRetry.businessOrderIds.push('13') }, p => { p.restored = false },
+    p => { p.attempts[11].outcome = 'RETRY' }, p => { p.beforeRetry.failedAttempts = 11 },
+    p => { p.attempts[12].targetName = 'lastleg-service/internal/finance/events' },
+  ]
+  for (const change of changes) {
+    const p = proof(); change(p)
+    assert.equal(validDirectDeliveryReplayProof(p, scope), false)
+  }
+})
@@ -0,0 +1,44 @@
+// This checks live, scoped evidence supplied by the TEST fault executor.
+// It never accepts historical reports or initiates event replay itself.
+const identifier = value => typeof value === 'string' && /^[1-9][0-9]{0,19}$/.test(value)
+const eventIdentifier = value => typeof value === 'string'
+  && /^(?:[0-9a-f]{8}(?:-[0-9a-f]{4}){3}-[0-9a-f]{12}|[0-9A-HJKMNP-TV-Z]{26})$/i.test(value)
+const hash = value => typeof value === 'string' && /^sha256:[0-9a-f]{64}$/i.test(value)
+const sortedIds = rows => [...rows].sort()
+
+export function validDirectDeliveryReplayProof(proof, scope) {
+  if (!proof || proof.readOnly !== true || proof.marker !== scope.marker
+      || !identifier(scope.loadingPlanId) || !identifier(scope.customerId)
+      || proof.loadingPlanId !== scope.loadingPlanId || proof.customerId !== scope.customerId
+      || proof.containerBookingId !== scope.containerBookingId) return false
+  const { producer, consumer, attempts, proxyRecords, beforeRetry, afterRetry, retryAudit } = proof
+  if (!eventIdentifier(producer?.eventId) || !hash(producer?.payloadHash)
+      || !identifier(producer?.entryId) || producer.publishStatus !== 'PUBLISHED'
+      || producer.eventType !== 'FirstLegLoadingPlanCompleted'
+      || consumer?.eventId !== producer.eventId || consumer?.payloadHash !== producer.payloadHash
+      || consumer?.status !== 'PROCESSED' || consumer?.count !== 1) return false
+  if (!Array.isArray(attempts) || attempts.length !== 13
+      || new Set(attempts.map(row => row.attemptId)).size !== attempts.length
+      || attempts.some(row => !row.attemptId || row.eventId !== producer.eventId
+        || row.entryId !== producer.entryId || row.targetName !== 'lastleg-service/internal/lastleg/events')
+      || attempts.at(-1)?.outcome !== 'DUPLICATE'
+      || attempts[11]?.outcome !== 'DEAD'
+      || attempts.slice(0, 11).some(row => row.outcome !== 'RETRY')) return false
+  if (!Array.isArray(proxyRecords) || proxyRecords.length !== 13
+      || proxyRecords.some(row => row.matched !== true || row.eventId !== producer.eventId
+        || row.payloadHash !== producer.payloadHash)
+      || !proxyRecords.some(row => row.outcome === 'APPLIED' && row.errorCode === 'ACK_LOSS_INJECTED')
+      || !proxyRecords.some(row => row.outcome === 'DUPLICATE' && row.errorCode === 'ACK_LOSS_INJECTED')) return false
+  if (retryAudit?.eventId !== producer.eventId || retryAudit?.payloadHash !== producer.payloadHash
+      || retryAudit?.entryId !== producer.entryId || !retryAudit.requestId
+      || retryAudit.previousFailures !== 12 || beforeRetry?.publishStatus !== 'DEAD'
+      || beforeRetry?.failedAttempts !== 12 || afterRetry?.publishStatus !== 'PUBLISHED'
+      || beforeRetry.eventId !== producer.eventId || afterRetry.eventId !== producer.eventId
+      || beforeRetry.payloadHash !== producer.payloadHash || afterRetry.payloadHash !== producer.payloadHash) return false
+  const before = beforeRetry.businessOrderIds, after = afterRetry.businessOrderIds
+  return Array.isArray(before) && Array.isArray(after) && before.length > 0
+    && before.every(identifier) && after.every(identifier)
+    && new Set(before).size === before.length && new Set(after).size === after.length
+    && JSON.stringify(sortedIds(before)) === JSON.stringify(sortedIds(after))
+    && proof.restored === true
+}
@@ -229,6 +229,10 @@ export async function run(context) {
      packageCount: Number(item?.packageCount), picked: Number(item?.pickedPackageCount) })),
  })
  const operationsBeforeSignature = JSON.stringify(operationsBefore || [])
+  if (typeof context.context?.beforeLoadingComplete === 'function') {
+    await context.context.beforeLoadingComplete({ marker: context.marker, customerId: tenantId,
+      loadingPlanId, containerBookingId })
+  }
  const completeAttempts = [1, 2].map((index) => ({
    body: {
      expectedTaskVersion: Number(taskBefore.version),
@@ -1,3 +1,5 @@
+import { validDirectDeliveryReplayProof } from './direct-delivery-replay-proof.mjs'
+
 const MODULE = 'phase4-cross-region-events'
 const OPTIONS = { scenario: 'phase4-cross-region-events', deploymentBlocker: true }

@@ -139,7 +141,18 @@ export async function run(context) {
    success(writtenBackAgain) && frozen.every((field) => (replaySnapshot[field] ?? null) === (snapshot[field] ?? null))
      && traced(writtenBackAgain),
    evidence(writtenBackAgain, `containerId=${id(domesticMatches[0].id)}, sourceVersion=${snapshot.emptyReportSourceVersion ?? 'missing'}`), OPTIONS)
-  blockingGap(context, '真实投递重试与同事件幂等证据',
+  const reader = context.context?.directDeliveryReplayEvidenceReader
+  if (typeof reader !== 'function') blockingGap(context, '真实投递重试与同事件幂等证据',
    '两次GET仅证明读取稳定；需按event-outbox-operations.mjs的受保护内部运维流程，在本轮事件上验证故障后有限重试/DEAD及同eventId+payloadHash重复消费，保存逐次尝试与业务计数证据。不得重放SENT、改表或伪造新事件。',
    'missing-direct-delivery-replay-evidence')
+  const scope = { marker: context.marker, loadingPlanId: id(phase3.loadingPlanId),
+    customerId: id(phase3.tenantId), containerBookingId: id(phase3.containerBookingId) }
+  const proof = await reader(scope)
+  context.createdResources.directDeliveryReplayProof = proof
+  const valid = validDirectDeliveryReplayProof(proof, scope)
+  context.assert(MODULE, '本轮原事件12次失败DEAD人工恢复DUPLICATE且业务零新增', null, valid,
+    `loadingPlanId=${scope.loadingPlanId}, attempts=${proof?.attempts?.length || 0}, restored=${proof?.restored === true}`, OPTIONS)
+  if (!valid) blockingGap(context, '真实投递重试证据不完整或身份漂移',
+    '本轮事件身份/摘要、12次有限重试、DEAD、人工恢复审计、重复消费及业务计数和恢复直连必须同时核平',
+    'invalid-direct-delivery-replay-evidence')
 }
@@ -6,7 +6,54 @@ import { fileURLToPath } from 'node:url'

 import { expandOverseasCartonCandidates, resolveCartonFixtures, run, uniqueContainerProducer,
  crossContainerCartonProbe, normalizeReceivingEvidence, receivingEvidenceValid,
-  receivingCompletionProof, captureReceivingEvidence, probeCartonResolveNegatives } from './phase4-overseas-pda.mjs'
+  receivingCompletionProof, captureReceivingEvidence, receivingEventRecoveryInventoryUnchanged,
+  probeCartonResolveNegatives, awaitReceivingCandidate } from './phase4-overseas-pda.mjs'
+
+function candidateContext(responses) {
+  const calls=[],steps=[]
+  return {calls,steps,createdResources:{},addStep:(_module,name,_response,passed)=>steps.push({name,passed}),
+    request:async(method,path,_body,token)=>{calls.push({method,path,token});return responses[Math.min(calls.length-1,responses.length-1)]}}
+}
+const candidateResponse = data => ({http:200,body:{code:'200',data},requestId:'offline-request',traceId:'offline-trace'})
+const exactCandidate = {containerNo:'TSTU6403402',sourceContainerBookingId:'2105838069516079106',operable:true,totalPackages:1}
+const exactProducer = {containerNo:exactCandidate.containerNo,containerBookingId:exactCandidate.sourceContainerBookingId}
+
+test('original r2 async shipping gate race waits same booking and container until visible without writing',async()=>{
+  const context=candidateContext([candidateResponse([]),candidateResponse([]),candidateResponse([exactCandidate])]),delays=[]
+  const result=await awaitReceivingCandidate(context,'offline-picker',exactProducer,{attempts:3,delay:async ms=>delays.push(ms)})
+  assert.deepEqual(result.candidate,exactCandidate)
+  assert.deepEqual(delays,[1000,1000])
+  assert.equal(context.calls.length,3)
+  assert.ok(context.calls.every(call=>call.method==='GET'&&call.token==='offline-picker'&&call.path.endsWith('/candidates')))
+  const proof=context.createdResources.phase4OverseasPda.candidateWaitEvidence[0]
+  assert.equal(proof.outcome,'READY');assert.deepEqual(proof.attempts.map(v=>v.exactMatches),[0,0,1])
+  assert.ok(proof.attempts.every(v=>v.requestId&&v.traceId))
+})
+test('same container from another booking, other container and nonoperable original cannot satisfy wait',async()=>{
+  for(const candidate of [{...exactCandidate,sourceContainerBookingId:'999'},
+    {...exactCandidate,containerNo:'ANOTHER'},{...exactCandidate,operable:false}]) {
+    const context=candidateContext([candidateResponse([candidate])]);let waits=0
+    const result=await awaitReceivingCandidate(context,'offline',exactProducer,{attempts:2,delay:async()=>waits++})
+    assert.equal(result.candidate,null);assert.equal(result.reason,'TIMED_OUT')
+    assert.equal(context.calls.length,2);assert.equal(waits,1)
+  }
+})
+test('ambiguous exact candidates and authentication or malformed replies stop without masking server errors',async()=>{
+  for(const response of [candidateResponse([exactCandidate,exactCandidate]),{http:401,body:{code:'401'}},candidateResponse({records:[exactCandidate]})]) {
+    const context=candidateContext([response]);let waits=0
+    const result=await awaitReceivingCandidate(context,'offline',exactProducer,{attempts:3,delay:async()=>waits++})
+    assert.equal(result.candidate,null);assert.equal(context.calls.length,1);assert.equal(waits,0)
+    assert.ok(['AMBIGUOUS','QUERY_FAILED'].includes(result.reason))
+  }
+})
+test('candidate waits are per invocation with no cross-mode cache and enforce exact scope/bounds',async()=>{
+  const context=candidateContext([candidateResponse([exactCandidate]),candidateResponse([])])
+  assert.ok((await awaitReceivingCandidate(context,'first',exactProducer,{attempts:1})).candidate)
+  assert.equal((await awaitReceivingCandidate(context,'second',exactProducer,{attempts:1})).candidate,null)
+  for(const producer of [{containerNo:'TSTU6403402'},{}])await assert.rejects(awaitReceivingCandidate(context,'offline',producer))
+  await assert.rejects(awaitReceivingCandidate(context,'offline',exactProducer,{attempts:31}))
+  assert.equal(context.calls.length,2)
+})

 test('single-container continuation retains malformed/unknown probes without requiring unshipped cancelled CHILD; default chain still blocks missing real other container', async () => {
  const requests = [], gaps = [], checks = []
@@ -73,6 +120,28 @@ test('receiving completion freezes all raw balances when group MIN represents tw
  assert.equal(stages.AFTER_COMPLETE.positions.length, 2)
 })

+test('unrelated canonical UNBOUND stock remains in all snapshots but new receiving still requires a real pallet', () => {
+  const { scope, stages } = receivingProofFixture()
+  const unboundPosition = { ...stages.AFTER_COMPLETE.positions[0], id: '301', balanceId: '30',
+    palletId: '0', palletNo: 'UNBOUND', availablePackages: 1 }
+  for (const snapshot of Object.values(stages)) {
+    const balance = { ...snapshot.balances[0], id: '30', inboundOrderId: '30',
+      inboundCode: 'UNBOUND-SOURCE', skuCode: 'UNRELATED', availablePackages: 1, inTransitPackages: 0 }
+    snapshot.balances.push(balance)
+    snapshot.httpGroups.push({ ...balance })
+    snapshot.positions.push({ ...unboundPosition })
+    snapshot.palletComposition.push({ ...unboundPosition })
+    assert.equal(receivingEvidenceValid(snapshot, scope), true)
+  }
+  assert.equal(receivingCompletionProof(scope, stages).passed, true)
+  stages.AFTER_COMPLETE.stockReceipts[0].warehousePalletId = '0'
+  assert.equal(receivingEvidenceValid(stages.AFTER_COMPLETE, scope), false)
+  const malformed = structuredClone(stages.BEFORE_COMPLETE)
+  malformed.positions[0].palletNo = 'UNKNOWN'
+  malformed.palletComposition[0].palletNo = 'UNKNOWN'
+  assert.equal(receivingEvidenceValid(malformed, scope), false)
+})
+
 test('receiving proof rejects missing raw source, wrong customer/warehouse, prefix source, missing bucket, ledger key, duplicate receipt and operation mismatch', () => {
  const mutations = [
    row => row.balances.pop(), row => row.positions[0].customerId = 'OTHER',
@@ -111,6 +180,21 @@ test('normalization is deterministic, preserves zero buckets, and excludes token
  assert.equal(normalizeReceivingEvidence({ ...after, stockReceipts: undefined }), null)
 })

+test('reverse event recovery requires complete owned inventory evidence and every immutable fact unchanged', () => {
+  const { scope, stages } = receivingProofFixture()
+  const before = stages.AFTER_FRESH_KEY_REJECTION
+  assert.equal(receivingEventRecoveryInventoryUnchanged(scope, before, structuredClone(before)), true)
+  assert.equal(receivingEventRecoveryInventoryUnchanged(scope, before, null), false)
+  assert.equal(receivingEventRecoveryInventoryUnchanged({ ...scope, customerId: 'OTHER' }, before, before), false)
+  for (const mutate of [row => row.balances[0].availablePackages++, row => row.stockReceipts.pop(),
+    row => row.positions[0].warehouseLocationCode = 'ANOTHER', row => row.stockInLedger[0].sourceEventId = 'OTHER',
+    row => row.receivingOperations.push({ ...row.receivingOperations[0], id: '999' })]) {
+    const changed = structuredClone(before)
+    mutate(changed)
+    assert.equal(receivingEventRecoveryInventoryUnchanged(scope, before, changed), false)
+  }
+})
+
 test('missing read-only receipt/key reader is a blocker and retains an unproven boundary without inventing HTTP', async () => {
  const checks = [], target = { stages: {} }
  const context = { context: {}, addCoverageGap() {}, assert(m, n, r, passed) { checks.push(passed) },
@@ -166,7 +250,7 @@ test('physical OVERSEAS and dropship probes still require another real loaded pr
 })

 test('receiving selects one current booking and supplies mandatory booking identity', () => {
-  assert.match(source, /id\(item.sourceContainerBookingId\) === preferredBookingId/)
+  assert.match(source, /item.containerNo === containerNo && id\(item.sourceContainerBookingId\) === bookingId/)
  assert.match(source, /matchingCandidates.length === 1/)
  assert.match(source, /if \(!longId\(sourceContainerBookingId\)\)/)
  assert.match(source, /sourceContainerBookingId=\$\{encodeURIComponent\(sourceContainerBookingId\)\}/)
@@ -314,7 +398,8 @@ test('PDA module covers the container-number receiving contract and excludes out
  assert.match(source, /standardFbaProducers = \(Array\.isArray\(matrix\?\.containers\) \? matrix\.containers : \[\]\)/)
  assert.match(source, /item\?\.fulfillmentType !== 'ONE_CLICK_DROPSHIP'/)
  assert.match(source, /FBA-OPERATIONS/)
-  assert.match(source, /\.slice\(0, 3\)/)
+  assert.match(source, /FBA-ROUTE/)
+  assert.match(source, /\.slice\(0, 4\)/)
  assert.match(source, /const producers = \[dropship, \.\.\.standardFbaProducers,/)
  assert.doesNotMatch(source, /创建当前柜实收托盘|创建当前柜实收库位|async function createPallet/)
  assert.match(source, /retained as acceptance evidence/)
@@ -417,7 +502,7 @@ test('real-device deployed FBA resolve A0404 is retained until server-side inbou

 test('PDA receiving lifecycle is ordered from candidate claim through photos, mode-specific receipts, pallet and complete', () => {
  const orderedTokens = [
-    "'查询当前客户待收柜号'", "'选择柜号并建立WAIT_SIGN会话'",
+    'const candidates = await awaitReceivingCandidate', "'选择柜号并建立WAIT_SIGN会话'",
    'uploadPhotos(context, token)', "'绑定车牌封签开柜三角色照片并签收'",
    'resolveGeneratedPallet(context, inbound)', "`解析${inbound.destinationType}实物托盘`", 'receiveFba(context, token',
    'receiveOverseas(context, token', "'回查柜内逐入仓码实收进度'",
@@ -464,7 +549,7 @@ test('missing OVERSEAS_WAREHOUSE_PDA_OPERATOR overseas authority is a deployment
  assert.ok(steps.some((step) => step?.passed === false && step?.options?.deploymentBlocker))
 })

-test('empty candidate response cannot become a zero-business-request pass', async () => {
+test('missing original producer identity blocks before candidate lookup and still cleans temporary identities', async () => {
  const calls = []; const gaps = []; const steps = []
  const response = (data, requestPath, method = 'GET') => ({ method, path: requestPath, http: 200,
    body: { code: '200', data }, requestId: 'req-contract', traceId: 'trace-contract' })
@@ -509,9 +594,9 @@ test('empty candidate response cannot become a zero-business-request pass', asyn
    addCoverageGap(...args) { gaps.push(args) },
  }
  await run(context)
-  assert.ok(calls.some((call) => call.requestPath.endsWith('/candidates')))
+  assert.equal(calls.some((call) => call.requestPath.endsWith('/candidates')),false)
  assert.ok(steps.some((step) => step.name.includes('正常空态') && step.passed))
-  assert.ok(steps.some((step) => step.name.includes('PDA柜号候选覆盖前置') && !step.passed && step.options.deploymentBlocker))
+  assert.ok(steps.some((step) => step.name.includes('本轮原柜候选身份覆盖前置') && !step.passed && step.options.deploymentBlocker))
  assert.equal(gaps.length, 1)
  assert.equal(context.createdResources.phase4OverseasPda.finalStatus, null)
  assert.equal(context.createdResources.phase4OverseasPda.pickerMembershipIds.length, 2)
@@ -99,7 +99,8 @@ export function receivingEvidenceValid(snapshot, scope) {
      || !snapshot.httpGroups.every(row => owned(row) && bucketsValid(row, RECEIVING_BALANCE_BUCKETS))
      || !snapshot.httpLedger.every(owned)
      || !snapshot.positions.every(row => owned(row) && bucketsValid(row, RECEIVING_POSITION_BUCKETS)
-        && snapshot.balances.some(balance => balance.id === row.balanceId) && longId(row.palletId) && nonBlank(row.palletNo)
+        && snapshot.balances.some(balance => balance.id === row.balanceId)
+        && ((longId(row.palletId) && nonBlank(row.palletNo)) || (row.palletId === '0' && row.palletNo === 'UNBOUND'))
        && longId(row.warehouseLocationId) && nonBlank(row.warehouseLocationCode))
      || !sameEvidence(snapshot.positions, snapshot.palletComposition)) return false
  if (!snapshot.balances.every(balance => RECEIVING_POSITION_BUCKETS.every(field =>
@@ -116,6 +117,7 @@ export function receivingEvidenceValid(snapshot, scope) {
      || !snapshot.httpGroups.every(row => groups.has(balanceGroupKey(row)) && RECEIVING_BALANCE_BUCKETS.every(field => row[field] === groups.get(balanceGroupKey(row))[field]))) return false
  if (!snapshot.stockReceipts.every(row => row.customerId === scope.customerId && scope.inboundCodes.includes(row.inboundCode)
      && longId(row.planId) && longId(row.taskId) && longId(row.allocationId) && longId(row.lineId) && longId(row.orderId)
+      && longId(row.warehousePalletId) && nonBlank(row.palletNo)
      && nonBlank(row.skuCode) && Number.isSafeInteger(row.receivedPackages) && row.receivedPackages > 0)) return false
  if (snapshot.inventoryOperations.length !== snapshot.stockReceipts.length || snapshot.stockInLedger.length !== snapshot.stockReceipts.length) return false
  for (const receipt of snapshot.stockReceipts) {
@@ -220,6 +222,11 @@ export async function captureReceivingEvidence(context, scope, stage, target) {
  }
 }

+export function receivingEventRecoveryInventoryUnchanged(scope, before, after) {
+  return receivingEvidenceValid(before, scope) && receivingEvidenceValid(after, scope)
+    && sameEvidence(before, after)
+}
+
 export function expandOverseasCartonCandidates(inbounds, producerCartonMarks = []) {
  const marks = (Array.isArray(producerCartonMarks) ? producerCartonMarks : [producerCartonMarks])
    .map(id).filter(nonBlank)
@@ -339,7 +346,17 @@ export async function createPicker(context, index, overseasWarehouseId, access)
  }
  context.addStep(MODULE, `海外PDA操作员${index}安全投影已收敛`, projectionResponse, projectionReady,
    `attempts=${projectionAttempt + 1}; ${trace(projectionResponse)}`, OPTIONS)
-  return { membershipId, username, token: projectionReady ? token : null }
+  return { membershipId, username, token: projectionReady ? token : null,
+    refreshToken: login.body?.data?.refreshToken, password,
+    userUpdate: { displayName: payload.displayName, phone, email: payload.email, tenantId: '0',
+      roleIds: payload.roleIds, overseasWarehouseId, domesticWarehouseId: null } }
+}
+
+export function pickerSessionHandoff(picker) {
+  if (!picker) return null
+  const { membershipId, username, token, refreshToken, password, userUpdate } = picker
+  return { membershipId, username, token, refreshToken, password,
+    userUpdate: userUpdate ? { ...userUpdate, roleIds: [...userUpdate.roleIds] } : undefined }
 }

 export async function disablePicker(context, picker, index) {
@@ -827,7 +844,37 @@ async function receiveDropship(context, token, producer, inbounds) {
  return accepted === expectedPending
 }

-export async function runReceiving(context, token, secondaryToken, merchantToken, preferredProducer = null, runAuthorizationProbes = true) {
+export async function awaitReceivingCandidate(context, token, producer, { attempts = 30, intervalMs = 1000,
+  delay = ms => new Promise(resolve => setTimeout(resolve, ms)) } = {}) {
+  const containerNo = id(producer?.containerNo), bookingId = id(producer?.containerBookingId)
+  if (!nonBlank(containerNo) || !longId(bookingId) || !Number.isInteger(attempts) || attempts < 1 || attempts > 30
+    || !Number.isInteger(intervalMs) || intervalMs < 0 || intervalMs > 1000) throw Error('Invalid exact receiving candidate scope or wait bound')
+  const proof = { containerNo, sourceContainerBookingId:bookingId, attempts:[], outcome:'PENDING' }
+  const resource = context.createdResources.phase4OverseasPda ||= {}
+  ;(resource.candidateWaitEvidence ||= []).push(proof)
+  let result
+  for (let attempt = 1; attempt <= attempts; attempt++) {
+    result = await pda(context, token, 'GET', `查询本轮原柜候选收敛第${attempt}次`, '/api/pda/v1/last-leg/receiving-containers/candidates')
+    const rows = Array.isArray(result.data) ? result.data : []
+    const matchingCandidates = rows.filter(item => item.containerNo === containerNo && id(item.sourceContainerBookingId) === bookingId)
+    proof.attempts.push({ attempt, http:result.response?.http, code:code(result.response), requestId:result.response?.requestId,
+      traceId:result.response?.traceId, candidateCount:rows.length, exactMatches:matchingCandidates.length,
+      sameNumberCandidates:rows.filter(item => item.containerNo === containerNo).map(item => ({
+        containerNo:item.containerNo, sourceContainerBookingId:id(item.sourceContainerBookingId),
+        operable:item.operable, blockReason:item.blockReason })) })
+    if (!success(result.response) || !Array.isArray(result.data)) { proof.outcome='QUERY_FAILED'; break }
+    if (matchingCandidates.length > 1) { proof.outcome='AMBIGUOUS'; break }
+    if (matchingCandidates.length === 1 && matchingCandidates[0].operable === true) {
+      proof.outcome='READY'
+      return {...result,candidate:matchingCandidates[0]}
+    }
+    if (attempt < attempts) await delay(intervalMs)
+  }
+  if (proof.outcome === 'PENDING') proof.outcome='TIMED_OUT'
+  return {...result,candidate:null,reason:proof.outcome}
+}
+
+export async function runReceiving(context, token, secondaryToken, merchantToken, preferredProducer = null, runAuthorizationProbes = true, candidateWait = {}) {
  if (runAuthorizationProbes && context.token) {
    const denied = await context.request('GET', '/api/pda/v1/last-leg/receiving-containers/candidates', undefined, context.token, {})
    context.addStep(MODULE, '平台管理员访问令牌不能冒充仓库PDA操作员', denied, forbidden(denied) && traced(denied), trace(denied), OPTIONS)
@@ -844,20 +891,19 @@ export async function runReceiving(context, token, secondaryToken, merchantToken
    blockerGap(context, '可重复的完整柜号验收', `账号已有未完成柜 ${before.body?.data?.containerNo || 'unknown'}，本轮不覆盖或篡改遗留现场会话`, 'dirty-fixture', before)
    return
  }
-  const candidates = await pda(context, token, 'GET', '查询当前客户待收柜号', '/api/pda/v1/last-leg/receiving-containers/candidates')
-  if (!Array.isArray(candidates.data) || candidates.data.length === 0) {
-    blockerGap(context, 'PDA柜号候选', `当前客户没有由三期装柜完成事件生成的待收入仓计划；${trace(candidates.response)}`, 'producer-chain', candidates.response)
-    return
-  }
  const preferredContainer = id(preferredProducer?.containerNo)
    || context.createdResources?.phase3Loading?.containerNo
-  const preferredBookingId = id(preferredProducer?.containerBookingId)
-  const matchingCandidates = candidates.data.filter(item => item.containerNo === preferredContainer
-    && (!preferredBookingId || id(item.sourceContainerBookingId) === preferredBookingId))
-  const candidate = matchingCandidates.length === 1 ? matchingCandidates[0] : null
+  const preferredBookingId = id(preferredProducer?.containerBookingId || context.createdResources?.phase3Loading?.containerBookingId)
+  if (!nonBlank(preferredContainer) || !longId(preferredBookingId)) {
+    blockerGap(context, '本轮原柜候选身份', '缺少本轮containerNo或containerBookingId，禁止仅按柜号匹配历史柜', 'producer-chain')
+    return
+  }
+  const candidates = await awaitReceivingCandidate(context,token,
+    {containerNo:preferredContainer,containerBookingId:preferredBookingId},candidateWait)
+  const candidate = candidates.candidate
  if (!candidate) {
    blockerGap(context, `本轮柜号${preferredContainer || 'missing'}候选`,
-      '三期装柜完成生产者已落地，但海外候选尚未出现该柜；禁止回退处理历史或其他柜', 'producer-chain', candidates.response)
+      `本轮原柜候选未在有界等待内唯一可操作：${candidates.reason}；禁止回退处理历史或其他柜`, 'producer-chain', candidates.response)
    return
  }
  const containerNo = id(candidate?.containerNo)
@@ -1015,6 +1061,9 @@ export async function runReceiving(context, token, secondaryToken, merchantToken
  const receivingProof = { scope, stages: {}, proof: null, status: 'UNPROVEN' }
  context.createdResources.phase4OverseasPda.receivingCompletionEvidence ||= {}
  context.createdResources.phase4OverseasPda.receivingCompletionEvidence[sourceContainerBookingId] = receivingProof
+  if (typeof context.context.beforeReceivingComplete === 'function') {
+    await context.context.beforeReceivingComplete(scope)
+  }
  await captureReceivingEvidence(context, scope, 'BEFORE_COMPLETE', receivingProof)
  const completed = await pda(context, token, 'POST', '完成当前柜并写入库存与卸柜事件', '/api/pda/v1/last-leg/receiving-containers/current/complete', completeBody)
  await captureReceivingEvidence(context, scope, 'AFTER_COMPLETE', receivingProof)
@@ -1033,6 +1082,20 @@ export async function runReceiving(context, token, secondaryToken, merchantToken
  receivingProof.status = receivingProof.proof.passed && Object.values(receivingProof.commandResults).every(Boolean) ? 'VERIFIED' : 'UNPROVEN'
  for (const [check, passed] of Object.entries(receivingProof.proof.checks)) context.assert(MODULE, `收货完成库存守恒 ${check}`, null, passed,
    `bookingId=${sourceContainerBookingId}, proofStatus=${receivingProof.status}`, OPTIONS)
+  if (typeof context.context.afterReceivingComplete === 'function') {
+    if (receivingProof.status !== 'VERIFIED') throw new Error('收货完成库存证据未闭合，禁止继续故障恢复验收')
+    const eventRecovery = await context.context.afterReceivingComplete(scope)
+    if (eventRecovery) {
+      receivingProof.eventRecovery = eventRecovery
+      const afterRecovery = await captureReceivingEvidence(context, scope, 'AFTER_EVENT_RECOVERY', receivingProof)
+      const unchanged = eventRecovery.restored === true && receivingEventRecoveryInventoryUnchanged(scope,
+        receivingProof.stages.AFTER_FRESH_KEY_REJECTION.evidence, afterRecovery)
+      receivingProof.eventRecoveryInventoryUnchanged = unchanged
+      context.assert(MODULE, '原卸柜事件DEAD恢复后库存位置托盘组成及不可变流水零漂移', null, unchanged,
+        `bookingId=${sourceContainerBookingId}`, OPTIONS)
+      if (!unchanged) throw new Error('卸柜事件恢复后库存证据变化或服务恢复未验证')
+    }
+  }
  const after = await context.request('GET', '/api/pda/v1/last-leg/receiving-containers/current', undefined, token, {})
  context.addStep(MODULE, '完成后current返回404正常空态', after, notFound(after), trace(after), OPTIONS)
  context.createdResources.phase4OverseasPda.finalStatus = completed.data?.status || null
@@ -1110,8 +1173,8 @@ export async function run(context) {
  }
  context.context.phase4OverseasPdaOperations = {
    marker: context.marker,
-    primary: primary ? { membershipId: primary.membershipId, username: primary.username, token: primary.token } : null,
-    secondary: secondary ? { membershipId: secondary.membershipId, username: secondary.username, token: secondary.token } : null,
+    primary: pickerSessionHandoff(primary),
+    secondary: pickerSessionHandoff(secondary),
    overseasWarehouseId: sharedWarehouse.id,
    merchantToken,
    completedFixtures: [],
@@ -1127,9 +1190,9 @@ export async function run(context) {
    const standardFbaProducers = (Array.isArray(matrix?.containers) ? matrix.containers : [])
      .filter((item) => item?.destinationType === 'FBA')
      .filter((item) => item?.fulfillmentType !== 'ONE_CLICK_DROPSHIP')
-      .filter((item) => /-P4-(?:FBA-STANDARD|FBA-SECOND|FBA-OPERATIONS)(?:-R\d+)?:/.test(String(item?.resourceKey || '')))
+      .filter((item) => /-P4-(?:FBA-STANDARD|FBA-SECOND|FBA-OPERATIONS|FBA-ROUTE)(?:-R\d+)?:/.test(String(item?.resourceKey || '')))
      .sort((left, right) => String(left?.resourceKey).localeCompare(String(right?.resourceKey)))
-      .slice(0, 3)
+      .slice(0, 4)
    // The sealed dropship pallet proves the indivisible handover. A separate standard FBA
    // fixture supplies multiple cartons for reversible operations. The second standard FBA
    // fixture remains available for the multi-destination dispatch and delivery split proof.
@@ -32,10 +32,11 @@ def encoded(value):


 class State:
-    def __init__(self, event_type, event_id, loading_plan_id, log_path=None):
+    def __init__(self, event_type, event_id, loading_plan_id, log_path=None, container_booking_id=None):
        self.event_type = event_type
        self.selector_id = event_id
        self.plan_id = loading_plan_id
+        self.booking_id = container_booking_id
        self.log_path = log_path
        self.lock = threading.Lock()
        self.mode = "PASS"
@@ -65,9 +66,13 @@ class State:
            return None, False, None, 0, self.mode
        identity = {"eventId": event_id, "payloadHash": payload_hash.lower()}
        payload = event.get("payload")
-        qualifies = event.get("eventType") == self.event_type and (
-            event_id == self.selector_id if self.selector_id else
-            isinstance(payload, dict) and str(payload.get("loadingPlanId")) == self.plan_id)
+        if self.selector_id:
+            selected = event_id == self.selector_id
+        elif self.booking_id:
+            selected = isinstance(payload, dict) and str(payload.get("sourceContainerBookingId")) == self.booking_id
+        else:
+            selected = isinstance(payload, dict) and str(payload.get("loadingPlanId")) == self.plan_id
+        qualifies = event.get("eventType") == self.event_type and selected
        envelope_hash = hashlib.sha256(json.dumps(event, sort_keys=True, ensure_ascii=False,
                                                separators=(",", ":")).encode("utf-8")).hexdigest()
        with self.lock:
@@ -254,6 +259,7 @@ def main():
    selector = parser.add_mutually_exclusive_group(required=True)
    selector.add_argument("--event-id")
    selector.add_argument("--loading-plan-id")
+    selector.add_argument("--container-booking-id")
    parser.add_argument("--bind", default="127.0.0.1")
    parser.add_argument("--port", type=int, default=0)
    parser.add_argument("--control-port", type=int, default=0)
@@ -270,9 +276,11 @@ def main():
        parser.error("event-id must be a UUID or ULID")
    if args.loading_plan_id and not re.fullmatch(r"[1-9][0-9]{0,19}", args.loading_plan_id):
        parser.error("loading-plan-id must be a positive decimal identifier")
+    if args.container_booking_id and not re.fullmatch(r"[1-9][0-9]{0,19}", args.container_booking_id):
+        parser.error("container-booking-id must be a positive decimal identifier")
    if not re.fullmatch(r"[A-Za-z][A-Za-z0-9]{0,79}", args.event_type) or not 0 < args.timeout <= 120:
        parser.error("invalid event type or timeout")
-    state = State(args.event_type, args.event_id, args.loading_plan_id, args.log)
+    state = State(args.event_type, args.event_id, args.loading_plan_id, args.log, args.container_booking_id)
    data = ThreadingHTTPServer((args.bind, args.port), data_handler(state, args.upstream, args.timeout))
    control = ThreadingHTTPServer(("127.0.0.1", args.control_port), control_handler(state, token))
    for server in (data, control):
@@ -1,5 +1,5 @@
 import assert from 'node:assert/strict'
-import { spawn } from 'node:child_process'
+import { spawn, spawnSync } from 'node:child_process'
 import { createServer } from 'node:http'
 import { mkdtemp, readFile, rm } from 'node:fs/promises'
 import { tmpdir } from 'node:os'
@@ -19,7 +19,7 @@ const event = (id = targetId, planId = '2100000000000000001') => ({ eventId: id,
  payload: { loadingPlanId: planId, syntheticPrivateField: secret } })
 const close = server => new Promise(resolve => server.close(resolve))

-async function fixture(selector = ['--loading-plan-id', '2100000000000000001']) {
+async function fixture(selector = ['--loading-plan-id', '2100000000000000001'], eventType = 'LoadingCompleted') {
  const calls = []
  const applied = new Set()
  let invalidAck = false
@@ -40,7 +40,7 @@ async function fixture(selector = ['--loading-plan-id', '2100000000000000001'])
  const directory = await mkdtemp(join(tmpdir(), 'event-fault-offline-'))
  const log = join(directory, 'evidence.jsonl')
  const child = spawn(process.env.PYTHON || 'python', [script, '--upstream',
-    `http://127.0.0.1:${consumer.address().port}/internal/events`, '--event-type', 'LoadingCompleted',
+    `http://127.0.0.1:${consumer.address().port}/internal/events`, '--event-type', eventType,
    ...selector, '--log', log], { env: { ...process.env, EVENT_FAULT_CONTROL_TOKEN: controlToken },
    stdio: ['ignore', 'pipe', 'pipe'] })
  let output = ''; let errors = ''
@@ -179,3 +179,42 @@ test('explicit event-ID selector requires exact event type and forwards unrelate
  assert.equal(status.attempt, 1)
  assert.equal(status.eventId, targetId)
 })
+
+test('unloading selector pins only the exact source booking and preserves reverse-event ACK recovery', async t => {
+  const bookingId = '2100000000000000066'
+  const proxy = await fixture(['--container-booking-id', bookingId], 'OverseasContainerUnloaded')
+  t.after(() => proxy.stop())
+  await proxy.control('/mode', { mode: 'FAULT' })
+  const unloaded = { ...event(), eventType: 'OverseasContainerUnloaded',
+    payload: { sourceContainerBookingId: bookingId, containerNo: 'TEST-CONTAINER' } }
+  for (const unrelated of [
+    { ...unloaded, eventId: otherId, payload: { sourceContainerBookingId: '2100000000000000067' } },
+    { ...unloaded, eventId: otherId, payload: { loadingPlanId: bookingId } },
+    { ...unloaded, eventId: otherId, eventType: 'OverseasDeliveryCompleted' },
+  ]) assert.equal((await proxy.send(unrelated)).status, 200)
+  assert.equal((await proxy.send(unloaded)).status, 503)
+  assert.equal((await proxy.send(unloaded)).status, 503)
+  assert.equal((await proxy.send({ ...unloaded, eventId: samePlanOtherId })).status, 200)
+  const before = proxy.calls.length
+  assert.equal((await proxy.send({ ...unloaded, payload: { sourceContainerBookingId: '2100000000000000067' } })).status, 409)
+  assert.equal(proxy.calls.length, before)
+  await proxy.control('/mode', { mode: 'PASS' })
+  const recovered = await proxy.send(unloaded)
+  assert.equal(recovered.status, 200)
+  assert.equal(JSON.parse(recovered.raw).data.outcome, 'DUPLICATE')
+  const status = await (await proxy.control('/status')).json()
+  assert.equal(status.eventId, targetId)
+  assert.deepEqual(status.records.filter(row => row.errorCode !== 'TARGET_EVENT_HASH_CONFLICT').map(row => row.outcome),
+    ['APPLIED', 'DUPLICATE', 'DUPLICATE'])
+})
+
+test('booking selector rejects malformed identities and mixed selection before starting listeners', () => {
+  for (const args of [['--container-booking-id', '0'], ['--container-booking-id', '../66'],
+    ['--container-booking-id', '66', '--loading-plan-id', '77'], ['--container-booking-id', '66', '--event-id', targetId]]) {
+    const result = spawnSync(process.env.PYTHON || 'python', [script, '--upstream', 'http://127.0.0.1:1/internal/events',
+      '--event-type', 'OverseasContainerUnloaded', ...args], { encoding: 'utf8', timeout: 5000,
+      env: { ...process.env, EVENT_FAULT_CONTROL_TOKEN: controlToken } })
+    assert.equal(result.status, 2)
+    assert.ok(!result.stdout.includes('"ready"'))
+  }
+})
@@ -0,0 +1,185 @@
+// Internal explicit TEST evidence queries; no network or environment reads on import.
+const stages = new Set(['BEFORE_COMPLETE', 'AFTER_COMPLETE', 'AFTER_SAME_KEY_REPLAY', 'AFTER_FRESH_KEY_REJECTION', 'AFTER_EVENT_RECOVERY'])
+const numeric = new Set(['availablePackages', 'holdPackages', 'allocatedPackages', 'pickedPackages', 'inTransitPackages', 'receivingPackages', 'procurementPackages', 'receivedPackages', 'beforePackages', 'deltaPackages', 'afterPackages', 'packageCount'])
+const fields = {
+  sources: 'id customerId overseasWarehouseId sourceContainerBookingId containerNo sourceInboundId inboundCode sourceRegion packageCount',
+  sessions: 'id sourceContainerBookingId containerNo overseasWarehouseId',
+  balances: 'id customerId overseasWarehouseId destinationWarehouseId destinationWarehouseCode inboundOrderId inboundOrderNo inboundCode skuCode availablePackages holdPackages allocatedPackages pickedPackages inTransitPackages receivingPackages procurementPackages',
+  positions: 'id balanceId customerId overseasWarehouseId destinationWarehouseCode palletId palletNo batchNo warehouseLocationId warehouseLocationCode status availablePackages holdPackages allocatedPackages pickedPackages receivingPackages',
+  stockReceipts: 'id customerId planId taskId allocationId lineId orderId inboundCode skuCode warehousePalletId palletNo warehouseLocationId warehouseLocationCode batchNo receivedPackages inventoryBalanceId inventoryPositionId inventoryOperationKey',
+  inventoryOperations: 'id customerId idempotencyKey operationType requestHash status',
+  stockInLedger: 'id customerId balanceId positionId overseasWarehouseId inboundOrderId inboundCode palletId palletNo batchNo skuCode businessType businessOrderNo sourceEventId idempotencyKey statusFrom statusTo beforePackages deltaPackages afterPackages',
+  receivingOperations: 'id sessionId action idempotencyKey requestHash',
+}
+const snake = value => value.replace(/[A-Z]/g, letter => '_' + letter.toLowerCase())
+function identifier(value) {
+  if (typeof value !== 'string' || !/^[1-9][0-9]{0,19}$/.test(value)) throw Error('Invalid receiving evidence identifier')
+  return value
+}
+// UTF-8 hex literals avoid quote, backslash and SQL-mode ambiguity.
+const literal = value => `(CONVERT(0x${Buffer.from(value, 'utf8').toString('hex')} USING utf8mb4) COLLATE utf8mb4_bin)`
+function validateScope(scope) {
+  for (const name of ['customerId', 'warehouseId', 'bookingId']) identifier(scope?.[name])
+  if (!stages.has(scope.stage) || typeof scope.containerNo !== 'string' || !scope.containerNo.trim()
+      || !Array.isArray(scope.inboundCodes) || !scope.inboundCodes.length
+      || scope.inboundCodes.some(code => typeof code !== 'string' || !code.trim() || code.length > 128)
+      || new Set(scope.inboundCodes).size !== scope.inboundCodes.length
+      || !Array.isArray(scope.sourceInboundIds) || scope.sourceInboundIds.length !== scope.inboundCodes.length
+      || new Set(scope.sourceInboundIds).size !== scope.sourceInboundIds.length) throw Error('Invalid receiving evidence scope')
+  scope.sourceInboundIds.forEach(identifier)
+  for (const code of scope.inboundCodes) if (!Number.isSafeInteger(scope.expectedPackagesByInbound?.[code])
+      || scope.expectedPackagesByInbound[code] <= 0) throw Error('Invalid receiving evidence quantity')
+}
+export function buildReceivingEvidenceSql(scope) {
+  validateScope(scope)
+  const owned = alias => `${alias}.customer_id=${literal(scope.customerId)} AND ${alias}.overseas_warehouse_id=${literal(scope.warehouseId)}`
+  const source = `o.customer_id=${literal(scope.customerId)} AND o.source_container_booking_id=${literal(scope.bookingId)}`
+  const orderIds = `SELECT o.id FROM ll_order o WHERE ${source}`
+  const session = `s.source_container_booking_id=${literal(scope.bookingId)} AND s.container_no=${literal(scope.containerNo)} AND s.overseas_warehouse_id=${scope.warehouseId}`
+  const taskIds = `SELECT t.id FROM ll_inbound_pda_task t JOIN ll_inbound_plan p ON p.id=t.plan_id JOIN ll_order o ON o.id=p.source_order_id WHERE ${source} AND BINARY t.customer_id=BINARY o.customer_id AND BINARY p.customer_id=BINARY o.customer_id`
+  const query = (name, alias, from, where) => {
+    const pairs = fields[name].split(' ').map(field => {
+      const column = `${alias}.${snake(field)}`
+      return `'${field}',${numeric.has(field) ? column : `CAST(${column} AS CHAR)`}`
+    }).join(',')
+    return `SELECT JSON_OBJECT('collection','${name}','row',JSON_OBJECT(${pairs})) FROM ${from} WHERE ${where} ORDER BY ${alias}.id;`
+  }
+  return [
+    'SET NAMES utf8mb4 COLLATE utf8mb4_bin;',
+    'SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ;',
+    'START TRANSACTION WITH CONSISTENT SNAPSHOT, READ ONLY;',
+    query('sources', 'o', 'll_order o', source),
+    query('sessions', 's', 'll_receiving_container_session s', session),
+    query('balances', 'b', 'll_inventory_balance b', owned('b')),
+    query('positions', 'p', 'll_inventory_position p', owned('p')),
+    query('stockReceipts', 'r', 'll_inbound_stock_receipt r', `r.customer_id=${literal(scope.customerId)} AND r.order_id IN (${orderIds})`),
+    query('inventoryOperations', 'i', 'll_inventory_operation i', `i.customer_id=${literal(scope.customerId)} AND i.operation_type=${literal('STOCK_IN')} AND EXISTS (SELECT 1 FROM ll_inbound_pda_task t WHERE t.id IN (${taskIds}) AND i.idempotency_key LIKE (CONCAT('PDA_STOCK_IN:',t.id,':%') COLLATE utf8mb4_bin))`),
+    query('stockInLedger', 'l', 'll_inventory_ledger l', `${owned('l')} AND l.business_type='STOCK_IN' AND l.inbound_order_id IN (${orderIds})`),
+    query('receivingOperations', 'r', 'll_receiving_container_operation r', `r.action='COMPLETE' AND r.session_id IN (SELECT s.id FROM ll_receiving_container_session s WHERE ${session})`),
+    'COMMIT;',
+  ].join('\n')
+}
+export function receivingEvidenceFromRows(scope, rows) {
+  validateScope(scope)
+  const result = Object.fromEntries(Object.keys(fields).map(name => [name, []]))
+  if (!Array.isArray(rows)) throw Error('Receiving evidence rows missing')
+  for (const item of rows) {
+    if (!Object.hasOwn(result, item?.collection) || !item.row) throw Error('Invalid receiving evidence collection')
+    result[item.collection].push(item.row)
+  }
+  const sources = result.sources
+  if (sources.length !== scope.inboundCodes.length
+      || new Set(sources.map(row => row.inboundCode)).size !== sources.length
+      || new Set(sources.map(row => row.sourceInboundId)).size !== sources.length
+      || sources.some(row => row.customerId !== scope.customerId || row.overseasWarehouseId !== scope.warehouseId
+        || row.sourceContainerBookingId !== scope.bookingId || row.containerNo !== scope.containerNo
+        || row.sourceRegion !== 'FIRSTLEG' || !scope.sourceInboundIds.includes(row.sourceInboundId)
+        || !scope.inboundCodes.includes(row.inboundCode) || row.packageCount !== scope.expectedPackagesByInbound[row.inboundCode])
+      || result.sessions.length !== 1 || result.sessions.some(row => row.sourceContainerBookingId !== scope.bookingId
+        || row.containerNo !== scope.containerNo || row.overseasWarehouseId !== scope.warehouseId)) throw Error('Receiving evidence source/session scope mismatch')
+  delete result.sources
+  delete result.sessions
+  for (const rows of Object.values(result)) rows.sort((a, b) => String(a.id).localeCompare(String(b.id)))
+  result.palletComposition = result.positions.map(row => ({ ...row }))
+  return result
+}
+export function createReceivingEvidenceReader(executeReadOnly) {
+  if (typeof executeReadOnly !== 'function') throw Error('Receiving evidence transport missing')
+  return async scope => receivingEvidenceFromRows(scope, await executeReadOnly(buildReceivingEvidenceSql(scope)))
+}
+
+const transactionStart = ['SET NAMES utf8mb4 COLLATE utf8mb4_bin;',
+  'SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ;',
+  'START TRANSACTION WITH CONSISTENT SNAPSHOT, READ ONLY;']
+export function assertReadOnlySql(sql) {
+  if (typeof sql !== 'string' || /--|\/\*|\b(?:INSERT|UPDATE|DELETE|REPLACE|DROP|ALTER|TRUNCATE|CALL|EXECUTE|PREPARE|INTO|OUTFILE|DUMPFILE|LOAD_FILE|SLEEP|BENCHMARK|GET_LOCK)\b/i.test(sql))
+    throw Error('Read-only SQL rejected')
+  const lines = sql.split('\n').filter(Boolean)
+  if (!transactionStart.every((line,i) => lines[i] === line) || lines.at(-1) !== 'COMMIT;'
+    || lines.length < 5 || lines.slice(3,-1).some(line => !/^SELECT [^;]+;$/.test(line))) throw Error('Read-only transaction required')
+  return sql
+}
+export function buildCompletionEvidenceSql(scope) {
+  identifier(scope?.customerId); identifier(scope?.rootOrderId)
+  if (scope.eventType !== 'FIRST_LEG_ORDER_FULFILLMENT_COMPLETED' || !Array.isArray(scope.sourceOrderIds)
+    || scope.sourceOrderIds.length !== 2 || new Set(scope.sourceOrderIds).size !== 2
+    || !scope.sourceOrderIds.includes(scope.rootOrderId)) throw Error('Invalid completion evidence scope')
+  scope.sourceOrderIds.forEach(identifier)
+  return [...transactionStart,
+    `SELECT JSON_OBJECT('orderId',CAST(o.id AS CHAR),'customerId',CAST(o.tenant_id AS CHAR),'rootOrderId',CAST(o.root_order_id AS CHAR),'eventType','FIRST_LEG_ORDER_FULFILLMENT_COMPLETED','count',COUNT(x.id),'publishedCount',COALESCE(SUM(x.publish_status='PUBLISHED'),0),'pendingCount',COALESCE(SUM(x.publish_status='PENDING'),0),'failedCount',COALESCE(SUM(x.publish_status NOT IN ('PENDING','PUBLISHED')),0)) FROM ord_order o LEFT JOIN ord_outbox_event x ON x.aggregate_id=o.id AND x.tenant_id=o.tenant_id AND x.event_type='FIRST_LEG_ORDER_FULFILLMENT_COMPLETED' WHERE o.tenant_id=${scope.customerId} AND o.root_order_id=${scope.rootOrderId} AND o.id IN (${scope.sourceOrderIds.join(',')}) GROUP BY o.id,o.tenant_id,o.root_order_id ORDER BY o.id;`,
+    'COMMIT;'].join('\n')
+}
+export function completionEvidenceFromRows(scope, rows) {
+  buildCompletionEvidenceSql(scope)
+  if (!Array.isArray(rows) || rows.length !== 2 || new Set(rows.map(row => row.orderId)).size !== 2
+    || rows.some(row => !scope.sourceOrderIds.includes(row.orderId) || row.customerId !== scope.customerId
+      || row.rootOrderId !== scope.rootOrderId || row.eventType !== scope.eventType
+      || ['count','publishedCount','pendingCount','failedCount'].some(key => !Number.isSafeInteger(row[key]) || row[key] < 0)))
+    throw Error('Completion evidence scope or counts mismatch')
+  return {readOnly:true,customerId:scope.customerId,rootOrderId:scope.rootOrderId,orders:rows}
+}
+export const EVIDENCE_ENV_KEYS = ['JENKINS_URL','JENKINS_USER','JENKINS_API_TOKEN','FBA_CORE_JENKINS_AGENT']
+export function validateEvidenceConfig(env) {
+  if (EVIDENCE_ENV_KEYS.some(key => typeof env[key] !== 'string' || !env[key].trim())) throw Error('Missing TEST evidence configuration')
+  const url = new URL(env.JENKINS_URL)
+  if (!['http:','https:'].includes(url.protocol) || url.username || url.password || url.search || url.hash
+    || !/^[A-Za-z0-9_.-]{1,128}$/.test(env.FBA_CORE_JENKINS_AGENT)) throw Error('Invalid TEST evidence configuration')
+}
+const base64Expression = value => '['+Buffer.from(value).toString('base64').match(/.{1,4000}/g).map(part=>`'${part}'`).join(',')+"].join('')"
+// Credentials stay in the executor's memory. No deploy, SQL mutation, or arbitrary SQL entry point.
+export function buildEvidenceRemoteScript(service, sql, agent) {
+  assertReadOnlySql(sql)
+  if (!['order','lastleg'].includes(service) || !/^[A-Za-z0-9_.-]{1,128}$/.test(agent)) throw Error('Invalid TEST evidence target')
+  const prefix=service.toUpperCase(),container=`firstleg-test-${service}`
+  const python = `import json,os,subprocess,urllib.parse,base64,re
+try:
+ c=json.loads(subprocess.check_output(['docker','inspect','${container}'],text=True,stderr=subprocess.DEVNULL))[0]
+ if c.get('Name')!='/${container}' or c.get('Config',{}).get('Labels',{}).get('com.docker.compose.project')!='firstleg-test': raise RuntimeError('target')
+ e=dict(v.split('=',1) for v in c['Config']['Env'])
+ u=urllib.parse.urlsplit(e['${prefix}_DB_URL'].removeprefix('jdbc:'))
+ if not u.hostname or not u.path.lstrip('/'): raise RuntimeError('database')
+ childEnv=dict(os.environ,MYSQL_PWD=e['${prefix}_DB_PASSWORD'])
+ host='127.0.0.1' if u.hostname=='host.docker.internal' else u.hostname
+ sql=base64.b64decode('${Buffer.from(sql).toString('base64')}').decode('utf-8')
+ p=subprocess.run(['mysql','--host='+host,'--port='+str(u.port or 3306),'--user='+e['${prefix}_DB_USERNAME'],'--batch','--raw','--skip-column-names',u.path.lstrip('/')],input=sql,capture_output=True,text=True,env=childEnv,timeout=40)
+ if p.returncode:
+  code=re.search(r'ERROR ([0-9]{3,5})',p.stderr)
+  print('FBA_CORE_JSON='+json.dumps({'ok':False,'error':'MYSQL_'+(code.group(1) if code else 'UNKNOWN')}))
+ else: print('FBA_CORE_JSON='+json.dumps({'ok':True,'rows':[json.loads(line) for line in p.stdout.splitlines() if line.strip()]}))
+except Exception:
+ print('FBA_CORE_JSON='+json.dumps({'ok':False,'error':'REMOTE_QUERY_FAILED'}))
+`
+  const groovy=`def p=new ProcessBuilder(['python3','-c',new String(java.util.Base64.decoder.decode(${base64Expression(python)}),'UTF-8')]).redirectErrorStream(true).start(); def output=p.inputStream.getText('UTF-8'); p.waitFor(); print(output)`
+  return `def c=jenkins.model.Jenkins.instance.getComputer('${agent}'); if(c==null || c.channel==null) throw new RuntimeException('TEST evidence agent unavailable'); print(hudson.util.RemotingDiagnostics.executeGroovy(new String(java.util.Base64.decoder.decode(${base64Expression(groovy)}),'UTF-8'),c.channel));`
+}
+export function createTestEvidenceReaders(env, fetcher = globalThis.fetch) {
+  validateEvidenceConfig(env)
+  async function execute(service, sql) {
+    const script=buildEvidenceRemoteScript(service,sql,env.FBA_CORE_JENKINS_AGENT)
+    let response
+    try {
+      response=await fetcher(new URL('scriptText',env.JENKINS_URL.replace(/\/?$/,'/')),{
+        method:'POST',headers:{Authorization:'Basic '+Buffer.from(env.JENKINS_USER+':'+env.JENKINS_API_TOKEN).toString('base64'),
+          'Content-Type':'application/x-www-form-urlencoded'},body:new URLSearchParams({script}),signal:AbortSignal.timeout(55000)})
+    } catch { throw Error('TEST evidence transport failed') }
+    if (!response.ok) throw Error('TEST evidence HTTP '+response.status)
+    let raw
+    try { raw=await response.text() } catch { throw Error('TEST evidence response failed') }
+    const lines=raw.split(/\r?\n/).filter(line=>line.startsWith('FBA_CORE_JSON='))
+    if(lines.length!==1) throw Error('TEST evidence JSON missing or ambiguous')
+    let result
+    try { result=JSON.parse(lines[0].slice('FBA_CORE_JSON='.length)) } catch { throw Error('TEST evidence JSON invalid') }
+    if(result?.ok!==true || !Array.isArray(result.rows)) throw Error(/^MYSQL_[0-9]{3,5}$/.test(result?.error)?result.error:'TEST evidence query failed')
+    return result.rows
+  }
+  return {
+    readReceiving: createReceivingEvidenceReader(sql => execute('lastleg',sql)),
+    readCompletion: async scope => completionEvidenceFromRows(scope,await execute('order',buildCompletionEvidenceSql(scope))),
+    preflight: async () => {
+      for(const service of ['order','lastleg']) {
+        const rows=await execute(service,[...transactionStart,"SELECT JSON_OBJECT('testProbe',1);",'COMMIT;'].join('\n'))
+        if(rows.length!==1 || rows[0].testProbe!==1)throw Error('TEST evidence preflight failed')
+      }
+    },
+  }
+}
@@ -0,0 +1,60 @@
+import test from 'node:test'
+import assert from 'node:assert/strict'
+import { assertReadOnlySql, buildReceivingEvidenceSql, receivingEvidenceFromRows, buildCompletionEvidenceSql,
+  completionEvidenceFromRows, buildEvidenceRemoteScript, createTestEvidenceReaders } from './fba-core-regression-evidence.mjs'
+
+const env=()=>({JENKINS_URL:'https://ci.example.invalid/',JENKINS_USER:'offline-ci',JENKINS_API_TOKEN:'offline-api-secret',FBA_CORE_JENKINS_AGENT:'java-test-agent'})
+const scope=()=>({customerId:'2105209721220861954',warehouseId:'193000000000200003',bookingId:'2105220190013755394',
+  stage:'BEFORE_COMPLETE',containerNo:'TEST-ONE',inboundCodes:['IN-1'],sourceInboundIds:['2105220155339444226'],
+  expectedPackagesByInbound:{'IN-1':1}})
+const completion=()=>({customerId:'90',rootOrderId:'100',sourceOrderIds:['100','101'],eventType:'FIRST_LEG_ORDER_FULFILLMENT_COMPLETED'})
+const completionRows=()=>['100','101'].map(orderId=>({customerId:'90',rootOrderId:'100',orderId,
+  eventType:'FIRST_LEG_ORDER_FULFILLMENT_COMPLETED',count:orderId==='100'?1:0,publishedCount:orderId==='100'?1:0,pendingCount:0,failedCount:0}))
+
+test('both queries use consistent read-only transactions and exact scope without losing long identifiers',()=>{
+  const sql=buildReceivingEvidenceSql(scope())
+  assert.equal(buildReceivingEvidenceSql({...scope(),stage:'AFTER_EVENT_RECOVERY'}),sql)
+  assert.throws(()=>buildReceivingEvidenceSql({...scope(),stage:'UNKNOWN_RECOVERY'}))
+  assert.equal(assertReadOnlySql(sql),sql)
+  assert.match(sql,/START TRANSACTION WITH CONSISTENT SNAPSHOT, READ ONLY/)
+  assert.ok(sql.includes(Buffer.from(scope().customerId).toString('hex')))
+  assert.doesNotMatch(sql,/customer_id=2105209721220861954/)
+  assert.match(sql,/CAST\(b.id AS CHAR\)/)
+  assert.match(assertReadOnlySql(buildCompletionEvidenceSql(completion())),/LEFT JOIN ord_outbox_event/)
+})
+test('read-only guard rejects mutation, multiple statements, lock or unsafe select effects',()=>{
+  const sql=buildCompletionEvidenceSql(completion())
+  for(const statement of ['DELETE FROM ord_order;','SELECT 1 INTO OUTFILE \'x\';','SELECT SLEEP(1);',
+    'SELECT 1; DROP TABLE x;','SET SESSION autocommit=1;'])assert.throws(()=>assertReadOnlySql(sql.replace(/SELECT .+;/,statement)))
+  assert.throws(()=>assertReadOnlySql('SELECT 1;'))
+  assert.throws(()=>buildEvidenceRemoteScript('prod',sql,'test'))
+})
+test('invalid and injected IDs fail before transport; cancelled child zero evidence cannot disappear',()=>{
+  assert.throws(()=>buildReceivingEvidenceSql({...scope(),bookingId:'1 OR 1=1'}))
+  assert.throws(()=>buildCompletionEvidenceSql({...completion(),sourceOrderIds:['100','100']}))
+  assert.equal(completionEvidenceFromRows(completion(),completionRows()).orders[1].count,0)
+  for(const mutate of [r=>r.pop(),r=>r[1].customerId='91',r=>r[1].count=null,r=>r[1].orderId='100']) {
+    const rows=completionRows();mutate(rows);assert.throws(()=>completionEvidenceFromRows(completion(),rows))
+  }
+})
+test('receiving rows require exact source and session; empty or foreign scope fails',()=>{
+  assert.throws(()=>receivingEvidenceFromRows(scope(),[]))
+  const s=scope(),rows=[{collection:'sources',row:{id:'1',customerId:s.customerId,overseasWarehouseId:s.warehouseId,
+    sourceContainerBookingId:s.bookingId,containerNo:s.containerNo,sourceInboundId:s.sourceInboundIds[0],inboundCode:'IN-1',sourceRegion:'FIRSTLEG',packageCount:1}},
+    {collection:'sessions',row:{id:'2',sourceContainerBookingId:s.bookingId,containerNo:s.containerNo,overseasWarehouseId:s.warehouseId}}]
+  assert.deepEqual(receivingEvidenceFromRows(s,rows).balances,[])
+  rows[0].row.customerId='99';assert.throws(()=>receivingEvidenceFromRows(s,rows))
+})
+test('transport frames valid empty rows and suppresses raw remote and connection secrets',async()=>{
+  let calls=0
+  const transport=createTestEvidenceReaders(env(),async()=>{calls++;return {ok:true,text:async()=> 'FBA_CORE_JSON='+JSON.stringify({ok:true,rows:[{testProbe:1}]})}})
+  await transport.preflight();assert.equal(calls,2)
+  for(const output of ['secret-leak','FBA_CORE_JSON={bad','FBA_CORE_JSON={"ok":false,"error":"password=secret-leak"}']) {
+    const reader=createTestEvidenceReaders(env(),async()=>({ok:true,text:async()=>output}))
+    await assert.rejects(reader.preflight(),error=>!error.message.includes('secret-leak'))
+  }
+  const reader=createTestEvidenceReaders(env(),async()=>{throw Error('offline-api-secret')})
+  await assert.rejects(reader.preflight(),/TEST evidence transport failed/)
+  const script=buildEvidenceRemoteScript('order',buildCompletionEvidenceSql(completion()),'java-test-agent')
+  assert.equal(script.includes(env().JENKINS_API_TOKEN),false)
+})
@@ -2,10 +2,69 @@ import test from 'node:test'
 import assert from 'node:assert/strict'
 import { readFileSync } from 'node:fs'
 import { resolve } from 'node:path'
-import { AcceptanceRun, parseArguments, validateExecutionOptions } from './admin-test-env-acceptance.mjs'
+import { AcceptanceRun, configureReadOnlyEvidence, unloadRecoveryWasVerified, parseArguments, validateExecutionOptions } from './admin-test-env-acceptance.mjs'

 const source = readFileSync(resolve('scripts/admin-test-env-acceptance.mjs'), 'utf8')

+test('full CLI evidence preflight attaches readers and fails before business writes when unavailable', async () => {
+  const run = { options: { modules: ['manual-state-matrix'] }, context: {} }
+  const readReceiving = () => {}, readCompletion = () => {}
+  let checked = false
+  await configureReadOnlyEvidence(run, {}, () => ({ readReceiving, readCompletion,
+    preflight: async () => { checked = true } }))
+  assert.equal(checked, true)
+  assert.equal(run.context.phase4ReceivingEvidenceReader, readReceiving)
+  assert.equal(run.context.f40ReadOnlyOutboxEvidence, readCompletion)
+  const unavailable = { options: run.options, context: {} }
+  await assert.rejects(configureReadOnlyEvidence(unavailable, {}, () => ({
+    preflight: async () => { throw Error('evidence unavailable') } })), /unavailable/)
+  assert.deepEqual(unavailable.context, {})
+})
+
+test('explicit unloading fault uses a separate first-container hook and restores both directions on failure', async () => {
+  const calls = []
+  const run = { marker: 'TEST-UNLOAD', options: { modules: ['phase4-overseas-pda', 'phase4-cross-region-events'],
+    verifyUnloadEventRecovery: true }, context: {} }
+  const fault = name => ({ prepare: async scope => calls.push([name, 'prepare', scope]),
+    readProof: async scope => ({ name, scope, restored: true }),
+    cleanup: async () => { calls.push([name, 'cleanup']); if (name === 'reverse') throw Error('recovery failed') } })
+  const cleanup = await configureReadOnlyEvidence(run, {}, () => ({ readReceiving() {}, readCompletion() {}, preflight: async () => {} }),
+    { createTestFaultExecutor: () => fault('forward'), createTestUnloadFaultExecutor: () => fault('reverse') })
+  assert.equal(calls.length, 0)
+  await run.context.beforeReceivingComplete({ customerId: '7', bookingId: '66' })
+  await run.context.beforeReceivingComplete({ customerId: '8', bookingId: '77' })
+  assert.deepEqual(calls, [['reverse', 'prepare', { marker: 'TEST-UNLOAD', customerId: '7', containerBookingId: '66' }]])
+  assert.equal(await run.context.afterReceivingComplete({ customerId: '8', bookingId: '77' }), null)
+  const proof = await run.context.afterReceivingComplete({ customerId: '7', bookingId: '66' })
+  assert.equal(proof.name, 'reverse')
+  await assert.rejects(cleanup.cleanup(), /恢复失败/)
+  assert.deepEqual(calls.slice(-2), [['reverse', 'cleanup'], ['forward', 'cleanup']])
+})
+
+test('unloading recovery is opt-in and cannot run against unrelated modules', () => {
+  const args = ['--modules', 'phase4-overseas-return-followups', '--gateway', 'http://localhost:9000',
+    '--username', 'contract', '--password', 'contract']
+  assert.equal(parseArguments(args).verifyUnloadEventRecovery, false)
+  assert.throws(() => validateExecutionOptions(parseArguments([...args, '--verify-unload-event-recovery', 'true'])), /收货链/)
+})
+
+test('requested unloading fault cannot pass without exact current-run recovery and inventory proof', () => {
+  const run = { marker: 'CURRENT', createdResources: {} }
+  assert.equal(unloadRecoveryWasVerified(run), false)
+  const row = { status: 'VERIFIED', eventRecoveryInventoryUnchanged: true,
+    eventRecovery: { restored: true, marker: 'CURRENT', containerBookingId: '66' } }
+  run.createdResources.phase4OverseasPda = { receivingCompletionEvidence: { '66': row } }
+  assert.equal(unloadRecoveryWasVerified(run), true)
+  for (const change of [r => r.eventRecoveryInventoryUnchanged = false, r => r.status = 'UNPROVEN',
+    r => r.eventRecovery.restored = false, r => r.eventRecovery.marker = 'OLD', r => r.eventRecovery.containerBookingId = '77']) {
+    const changed = structuredClone(run)
+    change(changed.createdResources.phase4OverseasPda.receivingCompletionEvidence['66'])
+    assert.equal(unloadRecoveryWasVerified(changed), false)
+  }
+  run.createdResources.phase4OverseasPda.receivingCompletionEvidence['77'] = structuredClone(row)
+  assert.equal(unloadRecoveryWasVerified(run), false)
+})
+
 test('F40 optional chain always creates its customer and rejects absent domestic PDA credentials', () => {
  const options = parseArguments(['--modules', 'phase5-root-fulfillment',
    '--gateway', 'http://localhost:9000', '--username', 'contract-user', '--password', 'contract-password',
@@ -0,0 +1,157 @@
+// Internal single TEST fault lifecycle. Uses the existing ACK-loss proxy and
+// authenticated original-DEAD retry endpoint, never direct business SQL writes.
+import { readFileSync } from 'node:fs'
+import { createHash } from 'node:crypto'
+import { isDeepStrictEqual } from 'node:util'
+import { validateEvidenceConfig } from './fba-core-regression-evidence.mjs'
+
+const runtime = readFileSync(new URL('./test-event-fault-runtime.py', import.meta.url), 'utf8')
+const proxySource = readFileSync(new URL('./event-delivery-fault-proxy.py', import.meta.url)).toString('base64')
+const encode = value => Buffer.from(value).toString('base64')
+const expression = value => '[' + encode(value).match(/.{1,4000}/g).map(part => `'${part}'`).join(',') + "].join('')"
+
+export function renderTestFaultAction(action, scope, agent) {
+  return renderAction(action,scope,agent,'LOADING')
+}
+export function renderTestUnloadFaultAction(action, scope, agent) {
+  return renderAction(action,scope,agent,'UNLOAD')
+}
+function renderAction(action,scope,agent,direction) {
+  const keys=direction==='UNLOAD'?['marker','customerId','containerBookingId']:['marker','customerId','loadingPlanId','containerBookingId']
+  if (!['prepare', 'activate', 'proxy-ready', 'health', 'status', 'recover', 'proof', 'restore', 'finish'].includes(action)
+      || !/^[A-Za-z0-9_.-]{1,128}$/.test(agent) || !/^[A-Za-z0-9_-]{1,96}$/.test(scope?.marker || '')
+      || keys.filter(key=>key!=='marker').some(key => !/^[1-9][0-9]{0,19}$/.test(scope?.[key] || ''))
+      || (direction==='UNLOAD'&&Object.keys(scope).some(key=>!keys.includes(key)))) {
+    throw Error('Invalid TEST fault action or scope')
+  }
+  const config = encode(JSON.stringify({ action, scope, direction, ...(action === 'prepare' ? { proxySource } : {}) }))
+  const groovy = `def p=new ProcessBuilder(['python3','-c',new String(java.util.Base64.decoder.decode(${expression(runtime)}),'UTF-8'),'${config}']).redirectErrorStream(true).start();def output=p.inputStream.getText('UTF-8');p.waitFor();print(output)`
+  return `def c=jenkins.model.Jenkins.instance.getComputer('${agent}');if(c==null || c.channel==null)throw new RuntimeException('TEST fault agent unavailable');print(hudson.util.RemotingDiagnostics.executeGroovy(new String(java.util.Base64.decoder.decode(${expression(groovy)}),'UTF-8'),c.channel))`
+}
+
+export function createTestFaultExecutor(env, fetcher = globalThis.fetch) {
+  return createExecutor(env,fetcher,'LOADING')
+}
+export function createTestUnloadFaultExecutor(env, fetcher = globalThis.fetch) {
+  return createExecutor(env,fetcher,'UNLOAD')
+}
+const canonical=value=>Array.isArray(value)?value.map(canonical):value&&typeof value==='object'
+  ?Object.fromEntries(Object.keys(value).sort().map(key=>[key,canonical(value[key])])):value
+export function mysqlDatetimeMillisFromInstant(value) {
+  const match=typeof value==='string'&&value.match(/^(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2})(?:\.(\d{1,9}))?Z$/)
+  if(!match)return NaN
+  const seconds=Date.parse(match[1]+'Z')
+  const nanos=Number((match[2]||'').padEnd(9,'0'))
+  // DATETIME(3) rounds fractional seconds; Date.parse alone truncates them.
+  return seconds+Math.floor((nanos+500000)/1000000)
+}
+export function assertTestUnloadFaultProof(proof,scope) {
+  const require=(passed,code)=>{if(!passed)throw Error('Invalid unload fault proof: '+code)}
+  require(proof?.readOnly===true&&Object.keys(scope).every(key=>proof[key]===scope[key]),'SCOPE')
+  const before=proof.beforeRetry,after=proof.afterRetry,row=proof.producer,env=row?.envelope
+  require(row&&env&&/^[1-9][0-9]{0,19}$/.test(row.entryId)&&/^(?:[0-9a-fA-F-]{36}|[0-9A-HJKMNP-TV-Z]{26})$/.test(row.eventId),'IDENTITY')
+  require(row.eventType==='OverseasContainerUnloaded'&&row.customerId==='0'&&row.aggregateId===scope.containerBookingId
+    &&row.eventKey===`OverseasContainerUnloaded:${scope.containerBookingId}`&&row.aggregateType==='CONTAINER_BOOKING','PRODUCER')
+  require(env.eventId===row.eventId&&env.eventType===row.eventType&&Number(env.eventVersion)===1&&env.tenantId==='0'
+    &&env.aggregateId===scope.containerBookingId&&env.aggregateType==='CONTAINER_BOOKING'
+    &&env.payload?.sourceContainerBookingId===scope.containerBookingId&&Boolean(env.payload.containerNo),'ENVELOPE')
+  const hash='sha256:'+createHash('sha256').update(JSON.stringify(canonical(env.payload))).digest('hex')
+  require(row.payloadHash===hash&&env.payloadHash===hash,'HASH')
+  require(before?.publishStatus==='DEAD'&&before.failedAttempts===12&&row.publishStatus==='PUBLISHED'
+    &&before.entryId===row.entryId&&before.eventId===row.eventId&&before.payloadHash===hash
+    &&isDeepStrictEqual(before.envelope,env)&&after?.publishStatus==='PUBLISHED'&&after.entryId===row.entryId
+    &&after.eventId===row.eventId&&after.payloadHash===hash&&isDeepStrictEqual(after.envelope,env),'ORIGINAL_RETRY')
+  const inbox=proof.consumer
+  require(inbox?.count===1&&inbox.status==='PROCESSED'&&inbox.eventId===row.eventId&&inbox.payloadHash===hash
+    &&inbox.eventType===row.eventType&&inbox.aggregateId===scope.containerBookingId&&isDeepStrictEqual(inbox.payload,env.payload),'INBOX')
+  const attempts=proof.attempts||[],proxy=proof.proxyRecords||[]
+  require(attempts.length===13&&new Set(attempts.map(a=>a.attemptId)).size===13&&attempts.every(a=>a.eventId===row.eventId&&a.entryId===row.entryId&&a.serviceName==='lastleg')
+    &&attempts.slice(0,11).every(a=>a.outcome==='RETRY')&&attempts[11].outcome==='DEAD'&&attempts[12].outcome==='DUPLICATE','ATTEMPTS')
+  const matches=proxy.filter(p=>p.matched===true)
+  require(matches.length===13&&matches.every((p,i)=>p.eventId===row.eventId&&p.payloadHash===hash&&p.attempt===i+1)
+    &&matches.slice(0,12).every((p,i)=>p.errorCode==='ACK_LOSS_INJECTED'&&p.outcome===(i===0?'APPLIED':'DUPLICATE'))
+    &&matches[12].outcome==='DUPLICATE'&&matches[12].errorCode==null,'REAL_ACK_LOSS')
+  const audit=proof.retryAudit
+  require(audit&&proof.retryAuditCount===1&&audit.requestId===proof.retryRequestId&&audit.entryId===row.entryId
+    &&audit.eventId===row.eventId&&audit.payloadHash===hash&&audit.previousFailures===12,'RETRY_AUDIT')
+  const container=before.container
+  require(container&&container.id===scope.containerBookingId&&container.containerNo===env.payload.containerNo
+    &&isDeepStrictEqual(container,after.container)&&container.version!=null&&container.emptyReportSourceVersion!=null,'CONTAINER_UNCHANGED')
+  const time=value=>Date.parse(String(value).replace(' ','T').replace(/Z$/,'')+'Z')
+  require(time(container.unloadingCompletedAt)===mysqlDatetimeMillisFromInstant(env.payload.unloadingCompletedAt)
+    &&time(container.unloadingPhotoUploadedAt)===mysqlDatetimeMillisFromInstant(env.payload.unloadingPhotoUploadedAt)
+    &&time(container.estimatedEmptyReportedAt)-time(container.unloadingCompletedAt)===72*3600000
+    &&(env.payload.actualEmptyReportedAt==null?container.actualEmptyReportedAt===null:time(container.actualEmptyReportedAt)===Date.parse(env.payload.actualEmptyReportedAt))
+    &&String(container.emptyReportSourceVersion)===String(env.aggregateVersion),'CONTAINER_FACTS')
+  return proof
+}
+function createExecutor(env,fetcher,direction) {
+  validateEvidenceConfig(env)
+  if (env.CHONGCHONG_ENV !== 'test') throw Error('TEST fault requires CHONGCHONG_ENV=test')
+  let scope = null, finished = false, preparedEvidence=null, restoration=null
+  async function action(name) {
+    const script = renderAction(name, scope, env.FBA_CORE_JENKINS_AGENT,direction)
+    const response = await fetcher(new URL('scriptText', env.JENKINS_URL.replace(/\/?$/, '/')), {
+      method: 'POST', redirect: 'error', headers: { Authorization: 'Basic ' + encode(env.JENKINS_USER + ':' + env.JENKINS_API_TOKEN),
+        'Content-Type': 'application/x-www-form-urlencoded' }, body: new URLSearchParams({ script }), signal: AbortSignal.timeout(55000) })
+    if (!response.ok) throw Error('TEST fault HTTP ' + response.status)
+    const rows = (await response.text()).split(/\r?\n/).filter(row => row.startsWith('TEST_EVENT_JSON='))
+    if (rows.length !== 1) throw Error('TEST fault evidence missing or ambiguous')
+    const result = JSON.parse(rows[0].slice('TEST_EVENT_JSON='.length))
+    if (!result?.ok) throw Error('TEST fault action failed: ' + (/^[A-Z_0-9]{1,80}$/.test(result?.error) ? result.error : 'UNKNOWN'))
+    return result.data
+  }
+  async function wait(name, check, attempts = 120, tolerateStartup = false) {
+    for (let index = 0; index < attempts; index++) {
+      let value
+      try { value = await action(name) } catch (error) {
+        if (!tolerateStartup || !String(error.message).endsWith('TEST_EVENT_ACTION_FAILED')) throw error
+      }
+      if (value && check(value)) return value
+      await new Promise(resolve => setTimeout(resolve, 1000))
+    }
+    throw Error('TEST fault observation timed out: ' + name)
+  }
+  async function cleanup() {
+    if (!scope || finished) return restoration
+    await action('restore')
+    await wait('health', row => row.healthy)
+    restoration=await action('finish')
+    if(direction==='UNLOAD'&&!(/^sha256:[0-9a-f]{64}$/.test(preparedEvidence?.originalImageId||'')
+      &&/^[0-9a-f]{64}$/.test(preparedEvidence?.originalFingerprint||'')
+      &&restoration?.restored===true&&restoration.healthy===true&&restoration.proxyStopped===true
+      &&restoration.producerName==='firstleg-test-lastleg'&&restoration.originalImageId===preparedEvidence?.originalImageId
+      &&restoration.currentImageId===preparedEvidence?.originalImageId&&restoration.originalFingerprint===preparedEvidence?.originalFingerprint
+      &&restoration.currentFingerprint===preparedEvidence?.originalFingerprint))throw Error('Unload restoration evidence mismatch')
+    finished = true
+    return restoration
+  }
+  return {
+    async prepare(next) {
+      if (scope) {
+        if(direction==='UNLOAD'&&!isDeepStrictEqual(scope,next))throw Error('TEST unload fault scope drift')
+        return // Pin only the first actual current-run plan; other events pass through.
+      }
+      if(direction==='UNLOAD')renderAction('prepare',next,env.FBA_CORE_JENKINS_AGENT,direction)
+      scope = direction==='UNLOAD'?{...next}:{ marker: next.marker, customerId: next.customerId,
+        loadingPlanId: next.loadingPlanId, containerBookingId: next.containerBookingId }
+      preparedEvidence=await action('prepare')
+      await wait('proxy-ready', row => row.ready, 10, true)
+      await action('activate')
+      await wait('health', row => row.healthy)
+    },
+    async readProof(expected) {
+      if (!scope || (direction==='UNLOAD'?!isDeepStrictEqual(scope,expected):Object.keys(scope).some(key => expected?.[key] !== scope[key]))) throw Error('TEST fault scope drift')
+      try {
+        await wait('status', row => row.producer?.publishStatus === 'DEAD' && row.producer.failedAttempts === 12, 180)
+        await action('recover')
+        await wait('status', row => row.producer?.publishStatus === 'PUBLISHED', 90)
+        const proof = await action('proof')
+        if(direction==='UNLOAD')assertTestUnloadFaultProof(proof,scope)
+        await cleanup()
+        return { ...proof, restored: true, ...(direction==='UNLOAD'?{restoration}:{}) }
+      } finally { if(direction==='UNLOAD')await cleanup() }
+    },
+    cleanup,
+  }
+}
@@ -0,0 +1,221 @@
+import assert from 'node:assert/strict'
+import test from 'node:test'
+import { createHash } from 'node:crypto'
+import { spawnSync } from 'node:child_process'
+import { createTestFaultExecutor, createTestUnloadFaultExecutor, assertTestUnloadFaultProof, renderTestFaultAction } from './test-event-fault-executor.mjs'
+
+const scope = { marker: 'TEST-ROOT', customerId: '7', loadingPlanId: '99', containerBookingId: '66' }
+const environment = { CHONGCHONG_ENV: 'test', JENKINS_URL: 'https://jenkins.example.invalid/',
+  JENKINS_USER: 'offline', JENKINS_API_TOKEN: 'offline-secret', FBA_CORE_JENKINS_AGENT: 'java-test-agent' }
+function requestedAction(options) {
+  const script = options.body.get('script')
+  const chunks = [...script.matchAll(/'([A-Za-z0-9+/=]+)'/g)].map(match => match[1]).join('')
+  const groovy = Buffer.from(chunks, 'base64').toString('utf8')
+  const argument = groovy.match(/'UTF-8'\),'([A-Za-z0-9+/=]+)'\]\)/)?.[1]
+  return JSON.parse(Buffer.from(argument, 'base64').toString('utf8'))
+}
+function transport(calls, failActivate = false) {
+  let recovered = false
+  return async (_url, options) => {
+    assert.equal(options.redirect, 'error')
+    const { action } = requestedAction(options)
+    calls.push(action)
+    if (action === 'recover') recovered = true
+    const data = action === 'proxy-ready' ? { ready: true } : action === 'health' ? { healthy: true }
+      : action === 'status' ? { producer: { publishStatus: recovered ? 'PUBLISHED' : 'DEAD', failedAttempts: 12 } }
+        : action === 'proof' ? { ...scope, restored: false } : {}
+    const result = action === 'activate' && failActivate ? { ok: false, error: 'DOCKER_409' } : { ok: true, data }
+    return { ok: true, text: async () => 'TEST_EVENT_JSON=' + JSON.stringify(result) }
+  }
+}
+test('scope and actions are bounded and renderer never embeds caller credentials', () => {
+  const rendered = renderTestFaultAction('status', scope, environment.FBA_CORE_JENKINS_AGENT)
+  assert.ok(!rendered.includes(environment.JENKINS_API_TOKEN))
+  for (const change of [{ ...scope, customerId: '7;DROP' }, { ...scope, marker: '../other' }]) {
+    assert.throws(() => renderTestFaultAction('status', change, environment.FBA_CORE_JENKINS_AGENT))
+  }
+  assert.throws(() => renderTestFaultAction('arbitrary', scope, 'java-test-agent'))
+  assert.throws(() => createTestFaultExecutor({ ...environment, CHONGCHONG_ENV: 'prod' }))
+})
+
+const unloadScope = { marker: 'TEST-ROOT', customerId: '7', containerBookingId: '66' }
+const originalImageId = 'sha256:' + 'a'.repeat(64), originalFingerprint = 'b'.repeat(64)
+function unloadProof(nanoseconds = false) {
+  const payload = { actualEmptyReportedAt: null, containerNo: 'OFFLINE66', sourceContainerBookingId: '66',
+    unloadingCompletedAt: nanoseconds ? '2026-10-02T01:00:00.136827515Z' : '2026-10-02T01:00:00Z', unloadingPhotoUploadedAt: '2026-10-02T00:59:00Z' }
+  const payloadHash = 'sha256:' + createHash('sha256').update(JSON.stringify(payload)).digest('hex')
+  const eventId = '00000000-0000-4000-8000-000000000001', entryId = '123'
+  const envelope = { eventId, eventType: 'OverseasContainerUnloaded', eventVersion: 1, tenantId: '0',
+    aggregateId: '66', aggregateType: 'CONTAINER_BOOKING', aggregateVersion: 3, payloadHash, payload }
+  const row = { entryId, eventId, payloadHash, envelope, eventType: envelope.eventType, customerId: '0',
+    aggregateId: '66', aggregateType: 'CONTAINER_BOOKING', eventKey: 'OverseasContainerUnloaded:66', publishStatus: 'PUBLISHED' }
+  const container = { id: '66', containerNo: 'OFFLINE66', version: '9', emptyReportSourceVersion: '3',
+    unloadingCompletedAt: nanoseconds ? '2026-10-02 01:00:00.137' : '2026-10-02 01:00:00', unloadingPhotoUploadedAt: '2026-10-02 00:59:00',
+    estimatedEmptyReportedAt: nanoseconds ? '2026-10-05 01:00:00.137' : '2026-10-05 01:00:00', actualEmptyReportedAt: null }
+  return JSON.parse(JSON.stringify({ ...unloadScope, readOnly: true, producer: row,
+    beforeRetry: { ...row, publishStatus: 'DEAD', failedAttempts: 12, container }, afterRetry: { ...row, container },
+    consumer: { count: 1, status: 'PROCESSED', eventId, payloadHash, eventType: envelope.eventType, aggregateId: '66', payload },
+    attempts: Array.from({ length: 13 }, (_, i) => ({ attemptId: `attempt-${i}`, entryId, eventId, serviceName: 'lastleg', outcome: i < 11 ? 'RETRY' : i === 11 ? 'DEAD' : 'DUPLICATE' })),
+    proxyRecords: Array.from({ length: 13 }, (_, i) => ({ eventId, payloadHash, attempt: i + 1, matched: true,
+      outcome: i ? 'DUPLICATE' : 'APPLIED', errorCode: i < 12 ? 'ACK_LOSS_INJECTED' : null })),
+    retryRequestId: '00000000-0000-4000-8000-000000000002', retryAuditCount: 1,
+    retryAudit: { requestId: '00000000-0000-4000-8000-000000000002', entryId, eventId, payloadHash, previousFailures: 12 } }))
+}
+function unloadTransport(calls, proof = unloadProof(), finish = {}) {
+  let recovered = false
+  return async (_url, options) => {
+    const config = requestedAction(options), { action } = config
+    assert.equal(config.direction, 'UNLOAD')
+    assert.deepEqual(config.scope, unloadScope)
+    calls.push(action)
+    if (action === 'recover') recovered = true
+    const data = action === 'prepare' ? { originalImageId, originalFingerprint }
+      : action === 'proxy-ready' ? { ready: true } : action === 'health' ? { healthy: true }
+        : action === 'status' ? { producer: { publishStatus: recovered ? 'PUBLISHED' : 'DEAD', failedAttempts: 12 } }
+          : action === 'proof' ? proof : action === 'finish' ? { restored: true, healthy: true, proxyStopped: true,
+            producerName: 'firstleg-test-lastleg', originalImageId, currentImageId: originalImageId,
+            originalFingerprint, currentFingerprint: originalFingerprint, ...finish } : {}
+    return { ok: true, text: async () => 'TEST_EVENT_JSON=' + JSON.stringify({ ok: true, data }) }
+  }
+}
+test('unload factory pins exactly three scope fields and verifies restoration before success', async () => {
+  const calls = [], executor = createTestUnloadFaultExecutor(environment, unloadTransport(calls))
+  await assert.rejects(executor.prepare({ ...unloadScope, eventType: 'Other' }), /Invalid/)
+  await executor.prepare(unloadScope)
+  await assert.rejects(executor.prepare({ ...unloadScope, containerBookingId: '67' }), /scope drift/)
+  await assert.rejects(executor.readProof({ ...unloadScope, eventType: 'Other' }), /scope drift/)
+  const proof = await executor.readProof(unloadScope)
+  assert.equal(proof.restored, true)
+  assert.equal(proof.restoration.currentImageId, originalImageId)
+  await executor.cleanup()
+  assert.deepEqual(calls, ['prepare', 'proxy-ready', 'activate', 'health', 'status', 'recover', 'status', 'proof', 'restore', 'health', 'finish'])
+})
+test('unload proof rejects changed original event, fabricated ACK, nonunique audit and changed business facts', () => {
+  assertTestUnloadFaultProof(unloadProof(), unloadScope)
+  for (const mutate of [
+    p => { p.producer.customerId = '7' }, p => { p.producer.eventKey = 'other' },
+    p => { p.producer.envelope.tenantId = '7' }, p => { p.producer.envelope.payload.containerNo = 'CHANGED' },
+    p => { p.beforeRetry.entryId = '124' }, p => { p.beforeRetry.envelope.aggregateVersion = 4 },
+    p => { p.beforeRetry.failedAttempts = 11 }, p => { p.afterRetry.eventId = 'OTHER' },
+    p => { p.consumer.count = 2 }, p => { p.consumer.status = 'RECEIVED' }, p => { p.consumer.payload = {} },
+    p => { p.attempts.pop() }, p => { p.attempts[0].serviceName = 'firstleg' },
+    p => { p.proxyRecords[0].outcome = 'DUPLICATE' }, p => { p.proxyRecords[0].errorCode = null },
+    p => { p.proxyRecords[12].attempt = 1 }, p => { p.retryAuditCount = 2 },
+    p => { p.retryAudit.requestId = 'OTHER' }, p => { p.afterRetry.container.version = '10' },
+    p => { p.beforeRetry.container.estimatedEmptyReportedAt = p.afterRetry.container.estimatedEmptyReportedAt = '2026-10-04 01:00:00' },
+  ]) {
+    const proof = unloadProof(); mutate(proof)
+    assert.throws(() => assertTestUnloadFaultProof(proof, unloadScope), /Invalid unload fault proof/)
+  }
+})
+
+test('unload timestamps follow DATETIME(3) rounding rather than allowing time drift', () => {
+  const proof=unloadProof(true)
+  assertTestUnloadFaultProof(proof,unloadScope)
+  for(const row of [proof.beforeRetry,proof.afterRetry]) {
+    row.container.unloadingCompletedAt='2026-10-02 01:00:00.138'
+    row.container.estimatedEmptyReportedAt='2026-10-05 01:00:00.138'
+  }
+  assert.throws(()=>assertTestUnloadFaultProof(proof,unloadScope),/CONTAINER_FACTS/)
+})
+test('invalid unload proof still restores and incomplete restoration never becomes success', async () => {
+  const invalid = unloadProof(); invalid.consumer.count = 0
+  const calls = [], executor = createTestUnloadFaultExecutor(environment, unloadTransport(calls, invalid))
+  await executor.prepare(unloadScope)
+  await assert.rejects(executor.readProof(unloadScope), /INBOX/)
+  assert.deepEqual(calls.slice(-3), ['restore', 'health', 'finish'])
+  for (const change of [{ restored: false }, { healthy: false }, { proxyStopped: false },
+    { currentImageId: 'sha256:' + 'c'.repeat(64) }, { currentFingerprint: 'c'.repeat(64) }]) {
+    const failed = createTestUnloadFaultExecutor(environment, unloadTransport([], unloadProof(), change))
+    await failed.prepare(unloadScope)
+    await assert.rejects(failed.readProof(unloadScope), /restoration evidence mismatch/)
+  }
+})
+test('Python reverse SQL routing, scope isolation and full configuration recovery fail closed offline', () => {
+  const result = spawnSync('python', ['-c', String.raw`
+import ast, copy, importlib.util, io, pathlib, sys
+from unittest.mock import patch
+sys.dont_write_bytecode = True
+path = pathlib.Path('scripts/test-event-fault-runtime.py')
+ast.parse(path.read_text(encoding='utf-8'))
+spec = importlib.util.spec_from_file_location('fault_runtime', path)
+m = importlib.util.module_from_spec(spec); spec.loader.exec_module(m)
+scope = {'marker':'TEST-ROOT','customerId':'7','containerBookingId':'66'}
+assert m.target_setting({'direction':'UNLOAD'}) == ('FIRSTLEG_SERVICE_URL','http://firstleg:9202')
+assert m.scope_directory(scope, 'UNLOAD') != m.scope_directory(dict(scope, loadingPlanId='99'))
+try: m.scope_directory(dict(scope, eventType='Other'), 'UNLOAD'); assert False
+except RuntimeError as error: assert str(error) == 'INVALID_SCOPE'
+queries=[]
+with patch.object(m, 'inspect', side_effect=lambda name: name), patch.object(m, 'read_sql', side_effect=lambda name, sql: queries.append((name, sql)) or []):
+    assert m.producer(scope, 'UNLOAD') is None
+assert queries[0][0] == 'firstleg-test-lastleg'
+assert "customer_id='0'" in queries[0][1] and "event_key='OverseasContainerUnloaded:66'" in queries[0][1]
+assert "JSON_EXTRACT(payload_json,'$')" in queries[0][1] and "publishStatus',status" in queries[0][1]
+row={'entryId':'123','eventId':'00000000-0000-4000-8000-000000000001','payloadHash':'sha256:'+'a'*64,'publishStatus':'DEAD','failedAttempts':12,'envelope':{'eventType':'OverseasContainerUnloaded'}}
+retry_state={'direction':'UNLOAD','operationsRequestId':'00000000-0000-4000-8000-000000000002'}
+retry_container={'Config':{'Env':['EVENT_OPERATIONS_TOKEN=offline']},'NetworkSettings':{'Networks':{'firstleg-test-network':{'IPAddress':'172.20.0.3'}}}}
+with patch.object(m,'producer',return_value=row),patch.object(m,'control',return_value=row),patch.object(m,'container_facts',return_value={'version':'9'}),patch.object(m,'business_ids',side_effect=AssertionError('reverse must not query inventory or orders')),patch.object(m,'private_write'),patch.object(m,'inspect',return_value=retry_container),patch.object(m.urllib.request,'urlopen',return_value=io.BytesIO(b'{"code":"200"}')) as http:
+    assert m.recover(scope,pathlib.Path('/offline'),retry_state)['retrySubmitted']
+    assert http.call_args.args[0].full_url == 'http://172.20.0.3:9303/internal/lastleg/event-operations/123/retry'
+    assert retry_state['beforeRetry']['container']=={'version':'9'}
+    assert retry_state['beforeRetry']['envelope']==row['envelope']
+row['publishStatus']='PUBLISHED'
+queries=[]
+with patch.object(m,'producer',return_value=row),patch.object(m,'control',return_value={'records':[]}),patch.object(m,'container_facts',return_value={'version':'9'}),patch.object(m,'business_ids',side_effect=AssertionError('reverse must not query inventory or orders')),patch.object(m,'inspect',side_effect=lambda name:name),patch.object(m,'read_sql',side_effect=lambda name,sql: queries.append((name,sql)) or []),patch.object(m,'private_write') as save:
+    proof=m.proof(scope,retry_state)
+    save.assert_called_once_with(m.scope_directory(scope,'UNLOAD')/'proof.json',proof)
+assert proof['retryAuditCount']==0 and proof['retryRequestId']==retry_state['operationsRequestId']
+assert [item[0] for item in queries]==['firstleg-test-lastleg','firstleg-test-lastleg','firstleg-test-business']
+assert "WHERE entry_id='123' AND event_type='OverseasContainerUnloaded'" in queries[1][1]
+assert 'fl_container_event_inbox' in queries[2][1]
+container={'Name':'/firstleg-test-lastleg','Id':'container123456789','Image':'sha256:'+'a'*64,
+ 'Config':{'Image':'lastleg:test','Env':['FIRSTLEG_SERVICE_URL=http://firstleg:9202','A=1'],'Labels':{'com.docker.compose.project':'firstleg-test'}},
+ 'HostConfig':{'Binds':['source:/data:rw'],'PortBindings':{'9303/tcp':[{'HostPort':'9303'}]}},
+ 'Mounts':[{'Type':'bind','Source':'/source','Destination':'/data','RW':True}],
+ 'NetworkSettings':{'Networks':{'firstleg-test-network':{'IPAddress':'172.20.0.3','Aliases':['lastleg-service','container123']}}},
+ 'State':{'Health':{'Status':'healthy'}}}
+fingerprint=m.fingerprint(container)
+multi=copy.deepcopy(container)
+multi['NetworkSettings']['Networks']['firstleg-test-network']['MacAddress']='02:42:ac:15:00:04'
+multi['NetworkSettings']['Networks']['second-test-network']={'IPAddress':'172.22.0.7','MacAddress':'02:42:ac:16:00:07','Aliases':['lastleg']}
+legacy=copy.deepcopy(multi); legacy['Config']['MacAddress']='02:42:ac:16:00:07'
+assert m.fingerprint(legacy)==m.fingerprint(multi)
+changed=copy.deepcopy(legacy); changed['Config']['MacAddress']='02:42:ac:99:00:01'
+assert m.fingerprint(changed)!=m.fingerprint(multi)
+changed=copy.deepcopy(legacy); changed['NetworkSettings']['Networks']['second-test-network']['MacAddress']='02:42:ac:16:00:08'
+assert m.fingerprint(changed)!=m.fingerprint(multi)
+state={'direction':'UNLOAD','restored':False,'image':container['Config']['Image'],'originalImageId':container['Image'],
+ 'originalFingerprint':fingerprint,'originalBody':m.recreation(container,m.env_of(container)),'proxyPid':123,'controlToken':'offline'}
+for mutate in [lambda c:c['HostConfig'].update(Binds=['other:/data:rw']),lambda c:c['NetworkSettings']['Networks']['firstleg-test-network'].update(IPAddress='172.20.0.4'),lambda c:c.update(Image='sha256:'+'b'*64),lambda c:c['State']['Health'].update(Status='unhealthy')]:
+    changed=copy.deepcopy(container); mutate(changed)
+    with patch.object(m,'inspect',return_value=changed), patch.object(m,'private_write') as write:
+        try: m.finish(pathlib.Path('/offline'),copy.deepcopy(state)); assert False
+        except RuntimeError as error: assert str(error).startswith('RESTORED_')
+        write.assert_not_called()
+with patch.object(m,'inspect',return_value=container),patch.object(m.Path,'exists',return_value=False),patch.object(m,'private_write'):
+    restored=m.finish(pathlib.Path('/offline'),copy.deepcopy(state))
+assert restored['proxyStopped'] and restored['currentFingerprint']==fingerprint and restored['originalImageId']==restored['currentImageId']
+with patch.object(m,'inspect',return_value=container),patch.object(m.Path,'exists',return_value=True),patch.object(m.Path,'read_bytes',return_value=b'python3\x00'+str(pathlib.Path('/offline')/'proxy.py').encode()+b'\x00'),patch.object(m.os,'kill'),patch.object(m.time,'sleep'),patch.object(m,'private_write') as write:
+    try: m.finish(pathlib.Path('/offline'),copy.deepcopy(state)); assert False
+    except RuntimeError as error: assert str(error)=='PROXY_NOT_STOPPED'
+    write.assert_not_called()
+`], { encoding: 'utf8', cwd: new URL('..', import.meta.url), timeout: 10000 })
+  assert.equal(result.status, 0, result.stderr || result.error?.message)
+})
+test('only original scoped event is recovered and restoration precedes a successful proof', async () => {
+  const calls = [], executor = createTestFaultExecutor(environment, transport(calls))
+  await executor.prepare(scope)
+  await executor.prepare({ ...scope, loadingPlanId: '100' })
+  const proof = await executor.readProof({ containerBookingId: '66', loadingPlanId: '99', customerId: '7', marker: 'TEST-ROOT' })
+  assert.equal(proof.restored, true)
+  await executor.cleanup()
+  assert.deepEqual(calls, ['prepare', 'proxy-ready', 'activate', 'health', 'status', 'recover',
+    'status', 'proof', 'restore', 'health', 'finish'])
+})
+test('failed activation retains task ownership for explicit restoration, never retries activation', async () => {
+  const calls = [], executor = createTestFaultExecutor(environment, transport(calls, true))
+  await assert.rejects(executor.prepare(scope), /DOCKER_409/)
+  await executor.cleanup()
+  assert.equal(calls.filter(name => name === 'activate').length, 1)
+  assert.deepEqual(calls.slice(-3), ['restore', 'health', 'finish'])
+})
@@ -0,0 +1,430 @@
+"""Internal TEST executor. Receives one bounded task action from Jenkins.
+
+Never edits business SQL. Docker configuration is captured before the scoped
+fault and restored to the same image, mounts, ports and network identity.
+Only the original DEAD event can be retried through its protected HTTP API.
+"""
+import base64
+import copy
+import hashlib
+import http.client
+import json
+import os
+import re
+import secrets
+import signal
+import socket
+import subprocess
+import sys
+import tempfile
+import time
+import urllib.parse
+import urllib.request
+import uuid
+from pathlib import Path
+
+
+class DockerConnection(http.client.HTTPConnection):
+    def connect(self):
+        self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+        self.sock.settimeout(self.timeout)
+        self.sock.connect('/var/run/docker.sock')
+
+
+def docker(method, path, body=None):
+    conn = DockerConnection('localhost', timeout=20)
+    data = None if body is None else json.dumps(body).encode()
+    conn.request(method, path, data, {'Content-Type': 'application/json'})
+    response = conn.getresponse()
+    raw = response.read()
+    conn.close()
+    if response.status >= 300:
+        raise RuntimeError('DOCKER_' + str(response.status))
+    return json.loads(raw) if raw else None
+
+
+def inspect(name):
+    return docker('GET', '/containers/' + urllib.parse.quote(name, safe='') + '/json')
+
+
+def env_of(container):
+    return dict(item.split('=', 1) for item in container['Config']['Env'])
+
+
+def require_test(container, name):
+    if container['Name'] != '/' + name or container['Config']['Labels'].get('com.docker.compose.project') != 'firstleg-test':
+        raise RuntimeError('NOT_TEST_CONTAINER')
+
+
+def recreation(container, environment, full=False):
+    body = copy.deepcopy(container['Config'])
+    body['Env'] = [key + '=' + value for key, value in environment.items()]
+    body['HostConfig'] = copy.deepcopy(container['HostConfig'])
+    if body['HostConfig'].get('AutoRemove') or any(mount['Type'] not in ('bind', 'volume') for mount in container['Mounts']):
+        raise RuntimeError('UNSUPPORTED_CONTAINER_STORAGE')
+    # Preserve named volumes explicitly, never create a replacement anonymous volume.
+    if any(mount['Type'] == 'volume' and not mount.get('Name') for mount in container['Mounts']):
+        raise RuntimeError('UNNAMED_VOLUME')
+    if body['HostConfig'].get('VolumesFrom'):
+        raise RuntimeError('UNSUPPORTED_VOLUME_REFERENCE')
+    endpoints = {}
+    for name, network in container['NetworkSettings']['Networks'].items():
+        aliases = [alias for alias in network.get('Aliases') or [] if alias != container['Id'][:12]]
+        endpoints[name] = {'Aliases': aliases, 'IPAMConfig': {'IPv4Address': network['IPAddress']}}
+        if full:
+            ipam = copy.deepcopy(network.get('IPAMConfig') or {})
+            ipam['IPv4Address'] = network['IPAddress']
+            if network.get('GlobalIPv6Address'):
+                ipam['IPv6Address'] = network['GlobalIPv6Address']
+            endpoints[name]['IPAMConfig'] = ipam
+            for key in ('Links', 'DriverOpts', 'GwPriority', 'MacAddress'):
+                if network.get(key) is not None:
+                    endpoints[name][key] = copy.deepcopy(network[key])
+    body['NetworkingConfig'] = {'EndpointsConfig': endpoints}
+    return body
+
+
+def replace(container, body):
+    name = container['Name'].lstrip('/')
+    current = inspect(name)
+    if current['Id'] != container['Id']:
+        raise RuntimeError('CONCURRENT_CONTAINER_CHANGE')
+    docker('POST', '/containers/' + current['Id'] + '/stop?t=10')
+    docker('DELETE', '/containers/' + current['Id'] + '?v=false')
+    created = docker('POST', '/containers/create?name=' + name, body)
+    docker('POST', '/containers/' + created['Id'] + '/start')
+    return created['Id']
+
+
+def private_write(path, value):
+    fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
+    with os.fdopen(fd, 'w') as file:
+        json.dump(value, file)
+
+
+def scope_directory(scope, direction='LOADING'):
+    if direction not in ('LOADING', 'UNLOAD'):
+        raise RuntimeError('INVALID_DIRECTION')
+    marker = scope.get('marker', '')
+    if not re.fullmatch(r'[A-Za-z0-9_-]{1,96}', marker):
+        raise RuntimeError('INVALID_SCOPE')
+    keys = ('customerId', 'containerBookingId') if direction == 'UNLOAD' else ('customerId', 'loadingPlanId', 'containerBookingId')
+    if direction == 'UNLOAD' and set(scope) != {'marker', *keys}:
+        raise RuntimeError('INVALID_SCOPE')
+    for key in keys:
+        if not re.fullmatch(r'[1-9][0-9]{0,19}', scope.get(key, '')):
+            raise RuntimeError('INVALID_SCOPE')
+    identity = 'UNLOAD:' + marker if direction == 'UNLOAD' else marker
+    return Path(tempfile.gettempdir()) / ('chongchong-event-' + hashlib.sha256(identity.encode()).hexdigest()[:24])
+
+
+def unload(state):
+    return state.get('direction', 'LOADING') == 'UNLOAD'
+
+
+def producer_name(state):
+    return 'firstleg-test-lastleg' if unload(state) else 'firstleg-test-business'
+
+
+def target_setting(state):
+    return ('FIRSTLEG_SERVICE_URL', 'http://firstleg:9202') if unload(state) else ('LASTLEG_SERVICE_URL', 'http://lastleg-service:9303')
+
+
+def fingerprint(container):
+    body = recreation(container, env_of(container), full=True)
+    # Docker may add the legacy Config.MacAddress after explicit per-network MAC
+    # restoration. It is redundant only when it equals an actual preserved MAC;
+    # every network MAC remains in the fingerprint and unrelated values still fail.
+    network_macs = {endpoint.get('MacAddress') for endpoint in body['NetworkingConfig']['EndpointsConfig'].values()}
+    if body.get('MacAddress') and body['MacAddress'] in network_macs:
+        body.pop('MacAddress')
+    body['Env'].sort()
+    for endpoint in body['NetworkingConfig']['EndpointsConfig'].values():
+        endpoint['Aliases'].sort()
+    body['ResolvedMounts'] = sorted(container['Mounts'], key=lambda mount: mount['Destination'])
+    body['NetworkIds'] = {name: network.get('NetworkID') for name, network in container['NetworkSettings']['Networks'].items()}
+    return hashlib.sha256(json.dumps(body, sort_keys=True, separators=(',', ':')).encode()).hexdigest()
+
+
+def control(state, mode=None):
+    url = 'http://127.0.0.1:' + str(state['controlPort']) + ('/mode' if mode else '/status')
+    request = urllib.request.Request(url, None if mode is None else json.dumps({'mode': mode}).encode(),
+        {'X-Event-Fault-Control-Token': state['controlToken'], 'Content-Type': 'application/json'})
+    with urllib.request.urlopen(request, timeout=5) as response:
+        return json.load(response)
+
+
+def read_sql(container, sql):
+    # SQL is generated here from fixed templates and decimal/UUID/hash identities.
+    if re.search(r'\b(INSERT|UPDATE|DELETE|ALTER|DROP|TRUNCATE|CALL|INTO|OUTFILE)\b', sql, re.I):
+        raise RuntimeError('NON_READ_ONLY_SQL')
+    e = env_of(container)
+    prefix = 'FIRSTLEG' if container['Name'] == '/firstleg-test-business' else 'LASTLEG'
+    url = urllib.parse.urlsplit(e[prefix + '_DB_URL'].removeprefix('jdbc:'))
+    host = '127.0.0.1' if url.hostname == 'host.docker.internal' else url.hostname
+    transaction = 'SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ; START TRANSACTION WITH CONSISTENT SNAPSHOT, READ ONLY;\n' + sql + '\nCOMMIT;'
+    result = subprocess.run(['mysql', '--host=' + host, '--port=' + str(url.port or 3306),
+        '--user=' + e[prefix + '_DB_USERNAME'], '--batch', '--raw', '--skip-column-names', url.path.lstrip('/')],
+        input=transaction, text=True, capture_output=True, timeout=20,
+        env=dict(os.environ, MYSQL_PWD=e[prefix + '_DB_PASSWORD']))
+    if result.returncode:
+        code = re.search(r'ERROR (\d+)', result.stderr)
+        raise RuntimeError('MYSQL_' + (code.group(1) if code else 'UNKNOWN'))
+    return [json.loads(line) for line in result.stdout.splitlines() if line.strip()]
+
+
+def producer(scope, direction='LOADING'):
+    if direction == 'UNLOAD':
+        rows = read_sql(inspect('firstleg-test-lastleg'),
+            "SELECT JSON_OBJECT('entryId',CAST(id AS CHAR),'eventId',JSON_UNQUOTE(JSON_EXTRACT(payload_json,'$.eventId')),'payloadHash',JSON_UNQUOTE(JSON_EXTRACT(payload_json,'$.payloadHash')),'eventType',event_type,'customerId',customer_id,'aggregateId',aggregate_id,'aggregateType',aggregate_type,'eventKey',event_key,'envelope',JSON_EXTRACT(payload_json,'$'),'publishStatus',status,'failedAttempts',retry_count) FROM ll_event_outbox WHERE customer_id='0' AND aggregate_id='"
+            + scope['containerBookingId'] + "' AND event_type='OverseasContainerUnloaded' AND event_key='OverseasContainerUnloaded:" + scope['containerBookingId'] + "';")
+    else:
+        rows = read_sql(inspect('firstleg-test-business'),
+        "SELECT JSON_OBJECT('entryId',CAST(id AS CHAR),'eventId',JSON_UNQUOTE(JSON_EXTRACT(payload_json,'$.eventId')),'payloadHash',payload_hash,'eventType',event_type,'publishStatus',publish_status,'failedAttempts',retry_count) FROM fl_inbound_outbox WHERE tenant_id=" + scope['customerId']
+        + " AND aggregate_id=" + scope['loadingPlanId'] + " AND event_type='FirstLegLoadingPlanCompleted';")
+    if len(rows) > 1:
+        raise RuntimeError('AMBIGUOUS_EVENT')
+    return rows[0] if rows else None
+
+
+def container_facts(scope):
+    rows = read_sql(inspect('firstleg-test-business'),
+        "SELECT JSON_OBJECT('id',CAST(id AS CHAR),'containerNo',container_no,'version',CAST(version AS CHAR),'unloadingCompletedAt',CAST(unloading_completed_at AS CHAR),'unloadingPhotoUploadedAt',CAST(unloading_photo_uploaded_at AS CHAR),'estimatedEmptyReportedAt',CAST(estimated_empty_reported_at AS CHAR),'actualEmptyReportedAt',CAST(actual_empty_reported_at AS CHAR),'emptyReportSourceVersion',CAST(empty_report_source_version AS CHAR),'updatedAt',CAST(updated_at AS CHAR),'updatedBy',CAST(updated_by AS CHAR),'updatedByName',updated_by_name) FROM fl_container_booking WHERE id=" + scope['containerBookingId'] + ";")
+    if len(rows) != 1:
+        raise RuntimeError('CONTAINER_SCOPE_MISMATCH')
+    return rows[0]
+
+
+def business_ids(scope):
+    rows = read_sql(inspect('firstleg-test-lastleg'), "SELECT JSON_OBJECT('id',CAST(id AS CHAR)) FROM ll_order WHERE customer_id='"
+        + scope['customerId'] + "' AND source_container_booking_id='" + scope['containerBookingId'] + "' ORDER BY id;")
+    return [row['id'] for row in rows]
+
+
+def prepare(scope, directory, config):
+    directory.mkdir(mode=0o700)
+    original = inspect(producer_name(config))
+    require_test(original, producer_name(config))
+    consumer_name = 'firstleg-test-business' if unload(config) else 'firstleg-test-lastleg'
+    consumer = inspect(consumer_name)
+    require_test(consumer, consumer_name)
+    old_env = env_of(original)
+    setting, direct = target_setting(config)
+    if old_env.get(setting) != direct:
+        raise RuntimeError('PRODUCER_NOT_DIRECT')
+    original_body = recreation(original, old_env, full=unload(config))
+    token = secrets.token_urlsafe(32)
+    # Reserve task-specific ports and let the existing proxy select exactly this plan.
+    sockets = [socket.socket() for _ in range(2)]
+    for sock in sockets:
+        sock.bind(('127.0.0.1', 0))
+    ports = [sock.getsockname()[1] for sock in sockets]
+    for sock in sockets:
+        sock.close()
+    proxy = directory / 'proxy.py'
+    proxy.write_bytes(base64.b64decode(config['proxySource']))
+    ip = consumer['NetworkSettings']['Networks']['firstleg-test-network']['IPAddress']
+    log = open(directory / 'proxy-output.log', 'wb')
+    upstream = ':9202/internal/firstleg/v1/container-events' if unload(config) else ':9303/internal/lastleg/events'
+    selector = ['--event-type', 'OverseasContainerUnloaded', '--container-booking-id', scope['containerBookingId']] if unload(config) else ['--event-type', 'FirstLegLoadingPlanCompleted', '--loading-plan-id', scope['loadingPlanId']]
+    process = subprocess.Popen(['python3', str(proxy), '--upstream', 'http://' + ip + upstream, *selector,
+        '--bind', '0.0.0.0', '--port', str(ports[0]), '--control-port', str(ports[1]),
+        '--log', str(directory / 'proxy-evidence.jsonl')],
+        env=dict(os.environ, EVENT_FAULT_CONTROL_TOKEN=token), stdout=log, stderr=log, start_new_session=True)
+    state = {'scope': scope, 'originalBody': original_body, 'originalId': original['Id'], 'image': original['Config']['Image'],
+        'direction': config.get('direction', 'LOADING'), 'originalImageId': original['Image'], 'originalFingerprint': fingerprint(original),
+        'proxyPid': process.pid, 'proxyPort': ports[0], 'controlPort': ports[1], 'controlToken': token,
+        'operationsRequestId': str(uuid.uuid4()), 'prepared': False, 'restored': False}
+    private_write(directory / 'state.json', state)
+    return {'created': True, 'originalImageId': state['originalImageId'], 'originalFingerprint': state['originalFingerprint']}
+
+
+def activate(scope, directory, state):
+    snapshot = control(state, 'FAULT')
+    if snapshot.get('eventId'):
+        raise RuntimeError('EVENT_ALREADY_PINNED')
+    original = inspect(producer_name(state))
+    require_test(original, producer_name(state))
+    setting, direct = target_setting(state)
+    if original['Id'] != state['originalId'] or original['Config']['Image'] != state['image'] or env_of(original).get(setting) != direct:
+        raise RuntimeError('CONCURRENT_PRODUCER_CHANGE')
+    if unload(state) and (original['Image'] != state['originalImageId'] or fingerprint(original) != state['originalFingerprint']):
+        raise RuntimeError('CONCURRENT_PRODUCER_CHANGE')
+    body = copy.deepcopy(state['originalBody'])
+    e = env_of(original)
+    e[setting] = 'http://host.docker.internal:' + str(state['proxyPort'])
+    e.update(EVENT_MAX_ATTEMPTS='12', EVENT_BASE_DELAY_SECONDS='5', EVENT_MAX_DELAY_SECONDS='5', EVENT_JITTER='0')
+    if len(e.get('EVENT_OPERATIONS_TOKEN', '')) < 32:
+        e['EVENT_OPERATIONS_TOKEN'] = secrets.token_urlsafe(32)
+    body['Env'] = [key + '=' + value for key, value in e.items()]
+    state['prepared'] = True
+    private_write(directory / 'state.json', state)
+    replace(original, body)
+    return {'activated': True}
+
+
+def recover(scope, directory, state):
+    row = producer(scope, state.get('direction', 'LOADING'))
+    if not row or row['publishStatus'] != 'DEAD' or row['failedAttempts'] != 12:
+        raise RuntimeError('ORIGINAL_EVENT_NOT_DEAD')
+    pinned = control(state)
+    if pinned['eventId'] != row['eventId'] or pinned['payloadHash'] != row['payloadHash']:
+        raise RuntimeError('PINNED_EVENT_MISMATCH')
+    state['beforeRetry'] = {**row, **({'container': container_facts(scope)} if unload(state) else {'businessOrderIds': business_ids(scope)})}
+    private_write(directory / 'state.json', state)
+    control(state, 'PASS')
+    container = inspect(producer_name(state))
+    e = env_of(container)
+    ip = container['NetworkSettings']['Networks']['firstleg-test-network']['IPAddress']
+    body = {'requestId': state['operationsRequestId'], 'operator': 'admin-full-test',
+        'reason': scope['marker'] + ': scoped ACK-loss recovery', 'expectedPayloadHash': row['payloadHash']}
+    endpoint = ':9303/internal/lastleg/event-operations/' if unload(state) else ':9202/internal/firstleg/event-operations/'
+    url = 'http://' + ip + endpoint + row['entryId'] + '/retry'
+    request = urllib.request.Request(url, json.dumps(body).encode(),
+        {'X-Event-Operations-Token': e['EVENT_OPERATIONS_TOKEN'], 'Content-Type': 'application/json'})
+    with urllib.request.urlopen(request, timeout=15) as response:
+        result = json.load(response)
+    if result.get('code') != '200':
+        raise RuntimeError('OPERATIONS_RETRY_REJECTED')
+    return {'retrySubmitted': True}
+
+
+def restore(directory, state):
+    if state['restored']:
+        return {'restored': True}
+    try:
+        control(state, 'PASS')
+    except Exception:
+        if not unload(state):
+            raise
+        # A dead proxy must not prevent restoring the producer's direct target.
+    if state['prepared']:
+        try:
+            current = inspect(producer_name(state))
+        except RuntimeError as error:
+            if str(error) != 'DOCKER_404':
+                raise
+            created = docker('POST', '/containers/create?name=' + producer_name(state), state['originalBody'])
+            docker('POST', '/containers/' + created['Id'] + '/start')
+            state['prepared'] = False
+            private_write(directory / 'state.json', state)
+            return {'restoreStarted': True}
+        if current['Id'] == state['originalId'] or (unload(state) and current['Image'] == state['originalImageId'] and fingerprint(current) == state['originalFingerprint']):
+            if unload(state) and not current['State'].get('Running'):
+                docker('POST', '/containers/' + current['Id'] + '/start')
+            state['prepared'] = False
+            private_write(directory / 'state.json', state)
+            return {'restoreStarted': True}
+        if current['Config']['Image'] != state['image'] or env_of(current).get(target_setting(state)[0]) != 'http://host.docker.internal:' + str(state['proxyPort']):
+            raise RuntimeError('CONCURRENT_PRODUCER_CHANGE')
+        replace(current, state['originalBody'])
+        state['prepared'] = False
+        private_write(directory / 'state.json', state)
+    return {'restoreStarted': True}
+
+
+def proof(scope, state):
+    row = producer(scope, state.get('direction', 'LOADING'))
+    if not row or row['publishStatus'] != 'PUBLISHED' or not state.get('beforeRetry'):
+        raise RuntimeError('RECOVERY_NOT_PUBLISHED')
+    event_id, entry_id = row['eventId'], row['entryId']
+    if not re.fullmatch(r'[0-9a-fA-F-]{36}|[0-9A-HJKMNP-TV-Z]{26}', event_id) or not re.fullmatch(r'[1-9][0-9]{0,19}', entry_id):
+        raise RuntimeError('INVALID_EVENT_IDENTITY')
+    event_type = 'OverseasContainerUnloaded' if unload(state) else 'FirstLegLoadingPlanCompleted'
+    attempts = read_sql(inspect(producer_name(state)),
+        "SELECT JSON_OBJECT('attemptId',attempt_id,'serviceName',service_name,'entryId',entry_id,'eventId',event_id,'targetName',target_name,'outcome',outcome) FROM event_delivery_attempt WHERE entry_id='"
+        + entry_id + "' AND event_type='" + event_type + "' ORDER BY started_at,attempt_id;")
+    audit = read_sql(inspect(producer_name(state)),
+        "SELECT JSON_OBJECT('requestId',request_id,'entryId',entry_id,'eventId',event_id,'payloadHash',payload_hash,'previousFailures',previous_failures) FROM event_delivery_retry_audit WHERE request_id='"
+        + state['operationsRequestId'] + "';" if not unload(state) else
+        "SELECT JSON_OBJECT('requestId',request_id,'entryId',entry_id,'eventId',event_id,'payloadHash',payload_hash,'previousFailures',previous_failures) FROM event_delivery_retry_audit WHERE entry_id='" + entry_id + "' AND event_type='OverseasContainerUnloaded';")
+    if unload(state):
+        consumers = read_sql(inspect('firstleg-test-business'),
+            "SELECT JSON_OBJECT('eventId',event_id,'payloadHash',payload_hash,'status',status,'eventType',event_type,'aggregateId',aggregate_id,'payload',JSON_EXTRACT(payload_json,'$')) FROM fl_container_event_inbox WHERE event_id='" + event_id + "';")
+    else:
+        consumers = read_sql(inspect('firstleg-test-lastleg'),
+        "SELECT JSON_OBJECT('eventId',event_id,'payloadHash',payload_hash,'status',status) FROM ll_loading_event_inbox WHERE event_id='"
+        + event_id + "' AND customer_id='" + scope['customerId'] + "';")
+    consumer = {**(consumers[0] if len(consumers) == 1 else {}), 'count': len(consumers)}
+    result = {**scope, 'readOnly': True, 'producer': row, 'consumer': consumer, 'attempts': attempts,
+        'proxyRecords': control(state)['records'], 'beforeRetry': state['beforeRetry'],
+        'afterRetry': {**row, **({'container': container_facts(scope)} if unload(state) else {'businessOrderIds': business_ids(scope)})},
+        'retryAudit': audit[0] if len(audit) == 1 else None, 'retryAuditCount': len(audit), 'retryRequestId': state['operationsRequestId'], 'restored': False}
+    private_write(scope_directory(scope, state.get('direction', 'LOADING')) / 'proof.json', result)
+    return result
+
+
+def finish(directory, state):
+    if state['restored']:
+        return state.get('restoration', {'restored': True})
+    current = inspect(producer_name(state))
+    require_test(current, producer_name(state))
+    if current['State'].get('Health', {}).get('Status') != 'healthy' or current['Config']['Image'] != state['image']:
+        raise RuntimeError('RESTORED_SERVICE_NOT_HEALTHY')
+    if sorted(current['Config']['Env']) != sorted(state['originalBody']['Env']):
+        raise RuntimeError('RESTORED_CONFIGURATION_MISMATCH')
+    if unload(state) and (current['Image'] != state['originalImageId'] or fingerprint(current) != state['originalFingerprint']):
+        raise RuntimeError('RESTORED_CONFIGURATION_MISMATCH')
+    cmdline = Path('/proc/' + str(state['proxyPid']) + '/cmdline')
+    if cmdline.exists():
+        if str(directory / 'proxy.py').encode() not in cmdline.read_bytes().split(b'\x00'):
+            raise RuntimeError('PROXY_PROCESS_OWNERSHIP_MISMATCH')
+        os.kill(state['proxyPid'], signal.SIGTERM)
+        if unload(state):
+            for _ in range(50):
+                if not cmdline.exists() or not cmdline.read_bytes():
+                    break
+                time.sleep(0.1)
+            else:
+                raise RuntimeError('PROXY_NOT_STOPPED')
+    result = {'restored': True}
+    if unload(state):
+        result.update(healthy=True, proxyStopped=True, producerName=producer_name(state),
+            originalImageId=state['originalImageId'], currentImageId=current['Image'],
+            originalFingerprint=state['originalFingerprint'], currentFingerprint=fingerprint(current))
+    state['restoration'] = result
+    state['restored'] = True
+    # Proof is already captured. Remove credentials from the retained task metadata.
+    state.pop('originalBody', None)
+    state.pop('controlToken', None)
+    private_write(directory / 'state.json', state)
+    return result
+
+
+def main(config):
+    scope, action = config['scope'], config['action']
+    direction = config.get('direction', 'LOADING')
+    directory = scope_directory(scope, direction)
+    if action == 'prepare':
+        return prepare(scope, directory, config)
+    state = json.loads((directory / 'state.json').read_text())
+    if state['scope'] != scope or state.get('direction', 'LOADING') != direction:
+        raise RuntimeError('TASK_SCOPE_MISMATCH')
+    if action == 'activate':
+        return activate(scope, directory, state)
+    if action == 'health':
+        current = inspect(producer_name(state))
+        return {'healthy': current['State'].get('Health', {}).get('Status') == 'healthy'}
+    if action == 'proxy-ready':
+        return {'ready': control(state)['mode'] == 'PASS'}
+    if action == 'status':
+        return {'producer': producer(scope, direction), 'proxy': control(state)}
+    if action == 'recover':
+        return recover(scope, directory, state)
+    if action == 'proof':
+        return proof(scope, state)
+    if action == 'restore':
+        return restore(directory, state)
+    if action == 'finish':
+        return finish(directory, state)
+    raise RuntimeError('INVALID_ACTION')
+
+
+if __name__ == '__main__':
+    try:
+        value = main(json.loads(base64.b64decode(sys.argv[1])))
+        print('TEST_EVENT_JSON=' + json.dumps({'ok': True, 'data': value}))
+    except Exception as error:
+        message = str(error)
+        safe = message if re.fullmatch(r'[A-Z_0-9]{1,80}', message) else 'TEST_EVENT_ACTION_FAILED'
+        print('TEST_EVENT_JSON=' + json.dumps({'ok': False, 'error': safe}))