test(fleet): verify two-customer departure rollback in TEST

scripts/README.md

@@ -36,6 +36,8 @@ F40 专项可用 `--signature-mode ROOT_FIRST|CHILD_FIRST|ROOT_SIGNED_REMAINDER_
 
 ## 验收证据校验
 
+`fleet-trip-rollback-test-env.mjs` 是跨客户整车发车中途回滚的串行TEST入口，默认`--plan`零网络；实际参数沿用`--execute --environment TEST --run-id <新批次> --warehouse-id <实际海外仓ID> --destination-warehouse-id <FBA目的仓ID>`，凭据只通过显式TEST/Jenkins环境文件及`FBA_CORE_JENKINS_AGENT`提供。复用正式两客户一托一箱来货/实收/排车链，全部合法拣货和理货审核后，在第二客户原托盘锁行建立65秒SELECT FOR UPDATE租约，以默认50秒锁等待超时使原整车发车事务失败；不修改业务行、Schema、共享参数或镜像。严格证明原等待图及第一客户尚未提交的DEPARTED/OUTBOUNDED/重放记录，再按一致性完整快照证明全部回滚；READ UNCOMMITTED仅用于明确标识的中途观察，不作为最终库存事实。租约自动ROLLBACK并核对无等待，再用同一原发车键/载荷完成正式发车、重放、逐客户签收和资源释放。报告沿`reports/fleet-cross-<新批次>/`，额外`rollback.json`、源码SHA与镜像身份保留；失败停止、原现场和请求保留、不自动重造或续跑，finally仅清理自有PDA/会话。共享TEST窗口禁止并行发布或操作相同库存。对应`fleet-trip-rollback-test-env.test.mjs`验证证据反例和范围；通用跨客户入口不默认故障注入。
+
 `fleet-test-env-acceptance.mjs` 是显式 TEST 车辆专项入口，默认 `--plan` 仅输出计划；使用显式外部 TEST 与 Jenkins 配置后，以 `--execute --environment TEST --run-id <新批次>` 串行创建本轮头程/海外来源并验证原费用全覆盖、取消重排后的部分覆盖、外接承运占位、PDA真实收货、外接排车计费、逐客户PDA拣货、整车发车、逐目的地签收及资源释放。报告在 `reports/fleet-<run-id>/`；目录和锁独占，失败保留原请求/资源，不自动续跑或重造。会通过正式HTTP写入测试资源和取消未执行单据；既有收货取证使用Jenkins只读查询，不执行业务SQL写入、部署或共享配置修改。Finance费用池消费事实与生产者ACK分开记录；UI及未实现车辆产品范围另验。对应离线入口合同为 `fleet-test-env-acceptance.test.mjs`，内部助手详见模块README。
 
 该入口用 `--scenario handover` 验证整车在途交接及经营成本；`--scenario return-redispatch` 验证实际退回入库后的重派和逐来源完成，默认要求全新批次；仅下述精确候选失败检查点允许显式恢复原样本。`--output reports/<新目录>` 指定独占报告目录，已有目录禁止覆盖。所有写请求发送前保存业务载荷、版本和幂等依据；网络失败、5xx、缺少关联信息或不符合正式响应契约时立即停止，保留原请求，不自动重发。

scripts/admin-test-env-modules/README.md

@@ -14,7 +14,7 @@
 
 `fleet-cross-customer-fixture.mjs` 导出 `createFleetCrossCustomerFixture`，由上述专用 TEST 入口显式调用。默认只返回准备契约；执行时重新核对指定实际收货仓和FBA目的仓，通过正式客户资质、普通转运来货、PDA一托一箱测量/绑定/实收、客户确认及预排接口生成两套独立本轮货物，再共用一个真实车次并各自转派送，停在尚未拣货。客户上下文独立且不覆盖外层客户环境，退出本函数创建的客户会话，PDA启停归外层入口；无外接预订循环依赖、业务SQL、部署或自动恢复。`fleet-cross-customer-fixture-contract.test.mjs` 只证明生成编排及拒绝边界，真实运行另行留证。
 
-`fleet-cross-customer-acceptance.mjs` 是跨客户共享车次的内部后半链验收函数，不是独立入口。默认调用只返回 fixture 契约、零网络；显式 `execute:true` 需要同批次两个独立客户、同仓各一托一箱已实收并分别预排/转派送、关联同一合格车次且未拣货，以及调用方提供的完整库存只读取证器。逐客户审核和拣货，验证第二单未就绪时整车拒绝且全部快照不变，再验证统一发车、原键重放和分别签收。前置门禁证据与事务中途故障回滚严格区分，后者保持 `NOT_TESTED`。失败保留当前样本与原请求，不自动续跑、补造或清理。对应合同 `fleet-cross-customer-acceptance-contract.test.mjs` 仅证明编排与反例断言，不能代替真实运行。临时账号生命周期、最小来货生成及报告保存由外层显式编排负责。
+`fleet-cross-customer-acceptance.mjs` 是跨客户共享车次的内部后半链验收函数，不是独立入口。默认调用只返回 fixture 契约、零网络；显式 `execute:true` 需要同批次两个独立客户、同仓各一托一箱已实收并分别预排/转派送、关联同一合格车次且未拣货，以及调用方提供的完整库存只读取证器。逐客户审核和拣货，验证第二单未就绪时整车拒绝且全部快照不变，再验证统一发车、原键重放和分别签收。前置门禁与事务中途回滚分别记录；默认后者为 `NOT_TESTED`，只有显式注入`verifyDepartureRollback`且返回精确本轮验证、完整一致性快照恢复相同后才记VERIFIED。专用`../fleet-trip-rollback-test-env.mjs`负责租约/脏读观察与持久化，普通入口不注入故障；钩子异常、未验证或快照漂移必须停止在正常发车前。失败保留当前样本与原请求，不自动续跑、补造或清理。对应合同 `fleet-cross-customer-acceptance-contract.test.mjs` 仅证明编排与反例断言，不能代替真实运行。临时账号生命周期、最小来货生成及报告保存由外层显式编排负责。
 
 `TERMINAL_READONLY` 用于全部业务写入和结案助手均完成、仅专项入口汇总状态断言错误的已复核报告。结案助手成功状态为 `CARGO_COMPLETED_AND_RESOURCES_RELEASED`。该检查点登录后仅 GET 原车次、每张排车/派送、资源占用测试计划与精确财务来源版本，不重放任何业务写入，不覆盖原失败报告；真实状态、数量、签收或费用不符仍失败。它不补充生产者 ACK 或页面验收证据。
 

scripts/admin-test-env-modules/fleet-cross-customer-acceptance-contract.test.mjs

@@ -78,6 +78,29 @@ test('current DTO-shaped two-customer trip rejects unready gate, departs once an
   assert.ok(m.calls.filter(c=>c.raw.includes('/api/pda/')).every(c=>c.token===m.fixture.pdaToken))
   assert.ok(!JSON.stringify(proof).includes(m.fixture.pdaToken))
 })
+test('explicit rollback verifier runs after both picks and before successful departure',async()=>{
+  const m=model();let calls=0
+  const proof=await verifyFleetCrossCustomer(m.context,m.fixture,{execute:true,readInventory:m.readInventory,
+    verifyDepartureRollback:async({ready,departPath,departBody})=>{
+      calls++;assert.equal(ready.trip.trip.status,'PLANNED')
+      assert.ok(ready.members.every(row=>row.dispatch.status==='READY_TO_DEPART'))
+      assert.equal(m.calls.filter(call=>call.raw.endsWith('/depart')).length,1)
+      assert.ok(departPath.includes(m.fixture.members[0].dispatchNo));assert.equal(departBody.idempotencyKey,'RUN-CROSS-DEPART')
+      return {verified:true,scope:'OWNED_TWO_CUSTOMER_DEPARTURE_ROLLBACK'}
+    }})
+  assert.equal(calls,1);assert.equal(proof.midTransactionRollback,'VERIFIED')
+})
+for(const failure of ['unverified','throws','mutates'])test(`rollback verification ${failure} prevents successful departure`,async()=>{
+  const m=model()
+  await assert.rejects(verifyFleetCrossCustomer(m.context,m.fixture,{execute:true,readInventory:m.readInventory,
+    verifyDepartureRollback:async()=>{
+      if(failure==='throws')throw Error('Observed rollback failed')
+      if(failure==='mutates')m.state[0].stock.balances[0].availablePackages='-1'
+      return {verified:failure!=='unverified',scope:'OWNED_TWO_CUSTOMER_DEPARTURE_ROLLBACK'}
+    }}))
+  assert.equal(m.calls.filter(call=>call.raw.endsWith('/depart')).length,1)
+  assert.ok(m.state.every(row=>row.delivery.order.status==='WAIT_DELIVERY'))
+})
 for(const fault of ['wrong stock scope','missing PDA position','duplicate ship','gate mutates second customer','gate reorders route','outage disguised as rejection','unrelated gate rejection','partial departure','depart deducts again','signs both']) {
   test(`fails closed without widening the claimed evidence: ${fault}`,async()=>{
     const m=model(fault);await assert.rejects(verifyFleetCrossCustomer(m.context,m.fixture,{execute:true,readInventory:m.readInventory}),

scripts/admin-test-env-modules/fleet-cross-customer-acceptance.mjs

@@ -33,7 +33,7 @@ export const fixtureContract = Object.freeze({
 /** Import/default call is network-free. Explicit execute:true requires separately authorized TEST execution.
  * Tokens stay in the caller, never in the persisted proof. Failure retains fixture and exact request keys.
  */
-export async function verifyFleetCrossCustomer(context, fixture, { execute = false, readInventory } = {}) {
+export async function verifyFleetCrossCustomer(context, fixture, { execute = false, readInventory, verifyDepartureRollback } = {}) {
   if (!execute) return fixtureContract
   if (!context?.marker || fixture?.marker !== context.marker || fixture.stage !== fixtureContract.stage
     || !fixture.tripNo || !fixture.warehouseId || !fixture.pdaToken || typeof readInventory !== 'function'
@@ -159,6 +159,13 @@ export async function verifyFleetCrossCustomer(context, fixture, { execute = fal
     const ready=await snapshot('READY')
     const departBody={recordVersion:reviewed.trip.recordVersion,idempotencyKey:key('DEPART'),remark:context.marker}
     const departPath=`${PDA}/${encode(fixture.members[0].dispatchNo)}/depart`
+    if(verifyDepartureRollback!==undefined) {
+      check('Explicit rollback verifier is callable',typeof verifyDepartureRollback==='function')
+      const rollback=await verifyDepartureRollback({context,fixture,ready,departPath,departBody,snapshot})
+      check('Mid-transaction rollback has exact owned verification',rollback?.verified===true && rollback.scope==='OWNED_TWO_CUSTOMER_DEPARTURE_ROLLBACK')
+      equal(ready,await snapshot('ROLLBACK_RESTORED'),'Mid-transaction rollback restores both complete customer execution and inventory snapshots')
+      proof.rollbackEvidence=rollback;proof.midTransactionRollback='VERIFIED'
+    }
     const departed=await request('POST',departPath,departBody,'Depart whole trip once',true)
     check('Departure includes both customer dispatches',departed.tripNo===fixture.tripNo && departed.status==='IN_TRANSIT' && integer(departed.shippedPackages)===2
       && canonical(departed.dispatches.map(row=>row.dispatchNo).sort())===canonical(fixture.members.map(row=>row.dispatchNo).sort()))

scripts/fleet-trip-rollback-test-env.mjs

@@ -0,0 +1,133 @@
+// Explicit TEST-only row-lock fault; importing this file performs no I/O.
+import fs from 'node:fs'
+import { createHash } from 'node:crypto'
+import { resolve } from 'node:path'
+import { pathToFileURL } from 'node:url'
+import { parseCrossCustomerArguments, runCrossCustomer } from './fleet-cross-customer-test-env.mjs'
+import { verifyFleetCrossCustomer } from './admin-test-env-modules/fleet-cross-customer-acceptance.mjs'
+
+const demand=(v,message)=>{if(!v)throw Error(message)}
+const literal=v=>`CONVERT(0x${Buffer.from(v).toString('hex')} USING utf8mb4) COLLATE utf8mb4_bin`
+const encoded=v=>'['+Buffer.from(v).toString('base64').match(/.{1,4000}/g).map(x=>`'${x}'`).join(',')+"].join('')"
+const id=v=>String(v??'')
+const rr=['SET NAMES utf8mb4 COLLATE utf8mb4_bin;','SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ;','START TRANSACTION WITH CONSISTENT SNAPSHOT, READ ONLY;']
+const pause=ms=>new Promise(done=>setTimeout(done,ms))
+const canonical=v=>JSON.stringify(sortFields(v))
+function sortFields(v){return Array.isArray(v)?v.map(sortFields):v&&typeof v==='object'?Object.fromEntries(Object.keys(v).sort().map(k=>[k,sortFields(v[k])])):v}
+export function validateRollbackScope(s) {
+  demand(s&&/^[A-Za-z0-9_-]{1,64}$/.test(s.marker||'')&&/^[A-Za-z0-9_-]{1,64}$/.test(s.tripNo||''),'ROLLBACK_SCOPE_REQUIRED')
+  for(const key of ['firstCustomerId','secondCustomerId','firstDispatchId','secondDispatchId','firstOutboundId','secondPalletId'])
+    demand(typeof s[key]==='string'&&/^[1-9][0-9]{0,19}$/.test(s[key]),'ROLLBACK_IDENTIFIER_INVALID')
+  demand(s.firstCustomerId!==s.secondCustomerId&&s.firstDispatchId!==s.secondDispatchId&&/^sha256:[a-f0-9]{64}$/.test(s.expectedImage||''),'ROLLBACK_DISTINCT_CUSTOMERS_AND_RUNTIME_REQUIRED')
+  demand(/^[A-Za-z0-9_-]{1,64}$/.test(s.firstDispatchNo||''),'ROLLBACK_DISPATCH_NUMBER_INVALID')
+  return s
+}
+export function rollbackSql(scope,mode) {
+  const s=validateRollbackScope(scope),second=`customer_id=${literal(s.secondCustomerId)} AND dispatch_order_id=${s.secondDispatchId} AND warehouse_pallet_id=${s.secondPalletId} AND status='ACTIVE'`
+  if(mode==='LEASE')return `SET SESSION innodb_lock_wait_timeout=5;START TRANSACTION;SELECT JSON_OBJECT('lockId',CAST(id AS CHAR),'customerId',customer_id,'connectionId',CONNECTION_ID()) FROM ll_dispatch_pallet_lock WHERE ${second} FOR UPDATE;SELECT JSON_OBJECT('marker','ROLLBACK_${s.marker}','elapsed',SLEEP(65));ROLLBACK;SELECT JSON_OBJECT('released',JSON_EXTRACT('true','$'));`
+  const lease=`SELECT JSON_OBJECT('collection','lease','connectionId',t.trx_mysql_thread_id,'lockId',l.LOCK_DATA) FROM performance_schema.data_locks l JOIN information_schema.innodb_trx t ON t.trx_id=l.ENGINE_TRANSACTION_ID WHERE l.OBJECT_SCHEMA=DATABASE() AND l.OBJECT_NAME='ll_dispatch_pallet_lock' AND l.INDEX_NAME='PRIMARY' AND l.LOCK_STATUS='GRANTED' AND l.LOCK_DATA IN (SELECT CAST(id AS CHAR) FROM ll_dispatch_pallet_lock WHERE ${second}) AND t.trx_query LIKE ${literal('%ROLLBACK_'+s.marker+'%')};`
+  const waits=`SELECT JSON_OBJECT('collection','wait','requester',r.trx_mysql_thread_id,'blocker',b.trx_mysql_thread_id,'lockId',l.LOCK_DATA) FROM performance_schema.data_lock_waits w JOIN performance_schema.data_locks l ON l.ENGINE_LOCK_ID=w.REQUESTING_ENGINE_LOCK_ID JOIN information_schema.innodb_trx r ON r.trx_id=w.REQUESTING_ENGINE_TRANSACTION_ID JOIN information_schema.innodb_trx b ON b.trx_id=w.BLOCKING_ENGINE_TRANSACTION_ID WHERE l.OBJECT_SCHEMA=DATABASE() AND l.OBJECT_NAME='ll_dispatch_pallet_lock' AND l.INDEX_NAME='PRIMARY' AND l.LOCK_DATA IN (SELECT CAST(id AS CHAR) FROM ll_dispatch_pallet_lock WHERE ${second});`
+  if(mode==='LOCKS')return [...rr,lease,waits,'COMMIT;'].join('\n')
+  if(mode==='OBSERVE')return ['SET NAMES utf8mb4 COLLATE utf8mb4_bin;','SET SESSION TRANSACTION ISOLATION LEVEL READ UNCOMMITTED;','START TRANSACTION READ ONLY;',lease,waits,
+    `SELECT JSON_OBJECT('collection','firstDispatch','id',CAST(id AS CHAR),'status',status,'shipped',shipped_package_count) FROM ll_dispatch_order WHERE id=${s.firstDispatchId} AND customer_id=${literal(s.firstCustomerId)};`,
+    `SELECT JSON_OBJECT('collection','firstOutbound','id',CAST(id AS CHAR),'status',status) FROM ll_outbound_order WHERE id=${s.firstOutboundId} AND customer_id=${literal(s.firstCustomerId)};`,
+    `SELECT JSON_OBJECT('collection','firstReplay','count',COUNT(*)) FROM ll_dispatch_pda_operation WHERE customer_id=${literal(s.firstCustomerId)} AND operation_type='DEPART' AND idempotency_key=${literal(s.tripNo+':'+s.firstDispatchNo)};`,'COMMIT;'].join('\n')
+  if(mode==='RELEASED')return [...rr,lease,waits,'COMMIT;'].join('\n')
+  throw Error('ROLLBACK_MODE_INVALID')
+}
+export function buildRollbackRemoteScript(scope,mode,agent) {
+  demand(['RUNTIME','LEASE','LOCKS','OBSERVE','RELEASED'].includes(mode)&&/^[A-Za-z0-9_.-]{1,128}$/.test(agent||''),'ROLLBACK_REMOTE_MODE_REQUIRED')
+  const sql=mode==='RUNTIME'?"SELECT JSON_OBJECT('lockWaitSeconds',@@GLOBAL.innodb_lock_wait_timeout);":rollbackSql(scope,mode)
+  const imageGuard=mode==='RUNTIME'?'':` or c['Image']!='${scope.expectedImage}'`
+  const python=`import json,os,subprocess,urllib.parse,re
+try:
+ c=json.loads(subprocess.check_output(['docker','inspect','firstleg-test-lastleg'],text=True,stderr=subprocess.DEVNULL))[0]
+ e=dict(x.split('=',1) for x in c['Config']['Env'])
+ if c['Name']!='/firstleg-test-lastleg' or c['Config'].get('Labels',{}).get('com.docker.compose.project')!='firstleg-test' or e.get('DEPLOY_ENV')!='test' or c['State'].get('Health',{}).get('Status')!='healthy'${imageGuard}:raise RuntimeError('TEST_RUNTIME_CHANGED')
+ u=urllib.parse.urlsplit(e['LASTLEG_DB_URL'].removeprefix('jdbc:'));host='127.0.0.1' if u.hostname=='host.docker.internal' else u.hostname
+ q=subprocess.run(['mysql','--host='+host,'--port='+str(u.port or 3306),'--user='+e['LASTLEG_DB_USERNAME'],'--batch','--raw','--skip-column-names',u.path.lstrip('/')],input=${JSON.stringify(sql)},capture_output=True,text=True,env=dict(os.environ,MYSQL_PWD=e['LASTLEG_DB_PASSWORD']),timeout=${mode==='LEASE'?75:25})
+ if q.returncode:
+  code=re.search(r'ERROR ([0-9]{3,5})',q.stderr);print('TRIP_ROLLBACK_JSON='+json.dumps({'ok':False,'error':'MYSQL_'+(code.group(1) if code else 'UNKNOWN')}))
+ else:print('TRIP_ROLLBACK_JSON='+json.dumps({'ok':True,'imageId':c['Image'],'rows':[json.loads(x) for x in q.stdout.splitlines() if x.strip()]}))
+except Exception as error:
+ print('TRIP_ROLLBACK_JSON='+json.dumps({'ok':False,'error':str(error) if re.fullmatch('[A-Z_]{1,64}',str(error)) else 'REMOTE_ROLLBACK_FAILED'}))
+`
+  const groovy=`def p=new ProcessBuilder(['python3','-c',new String(java.util.Base64.decoder.decode(${encoded(python)}),'UTF-8')]).redirectErrorStream(true).start();def out=p.inputStream.getText('UTF-8');p.waitFor();print(out)`
+  return `def c=jenkins.model.Jenkins.instance.getComputer('${agent}');if(c==null||c.channel==null)throw new RuntimeException('TEST agent unavailable');print(hudson.util.RemotingDiagnostics.executeGroovy(new String(java.util.Base64.decoder.decode(${encoded(groovy)}),'UTF-8'),c.channel))`
+}
+export function assertMidTransactionObservation(rows) {
+  const one=name=>rows.filter(x=>x.collection===name)
+  const lease=one('lease'),wait=one('wait'),dispatch=one('firstDispatch'),outbound=one('firstOutbound'),replay=one('firstReplay')
+  demand(lease.length===1&&wait.length===1&&wait[0].blocker===lease[0].connectionId&&wait[0].lockId===lease[0].lockId&&wait[0].requester!==wait[0].blocker,'OWNED_LATE_ROW_WAIT_NOT_PROVED')
+  demand(dispatch.length===1&&dispatch[0].status==='DEPARTED'&&dispatch[0].shipped===1&&outbound.length===1&&outbound[0].status==='OUTBOUNDED'&&replay.length===1&&replay[0].count===1,'FIRST_CUSTOMER_UNCOMMITTED_WRITE_NOT_PROVED')
+  return {leaseConnectionId:lease[0].connectionId,requestingConnectionId:wait[0].requester,lockId:lease[0].lockId,dirtyObservationOnly:true}
+}
+async function remote(env,scope,mode) {
+  const script=buildRollbackRemoteScript(scope,mode,env.FBA_CORE_JENKINS_AGENT)
+  const response=await fetch(new URL('scriptText',env.JENKINS_URL.replace(/\/?$/,'/')),{method:'POST',redirect:'error',signal:AbortSignal.timeout(mode==='LEASE'?95000:45000),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})})
+  demand(response.ok,'ROLLBACK_EVIDENCE_HTTP_FAILED')
+  const lines=(await response.text()).split(/\r?\n/).filter(x=>x.startsWith('TRIP_ROLLBACK_JSON='));demand(lines.length===1,'ROLLBACK_REMOTE_RESULT_AMBIGUOUS')
+  const result=JSON.parse(lines[0].slice('TRIP_ROLLBACK_JSON='.length));demand(result.ok===true,/^[A-Z_0-9]+$/.test(result.error||'')?result.error:'ROLLBACK_REMOTE_FAILED')
+  return result
+}
+export function createDepartureRollbackVerifier(env,config,runtime) {
+  return async({context,fixture,ready,departPath,departBody,snapshot})=>{
+    const output=resolve('reports',`fleet-cross-${config.runId}`,'rollback.json'),fd=fs.openSync(output,'wx',0o600)
+    const report={verified:false,scope:'OWNED_TWO_CUSTOMER_DEPARTURE_ROLLBACK',startedAt:new Date().toISOString(),samples:[],command:{path:departPath,body:departBody},runtime}
+    const save=()=>{fs.ftruncateSync(fd,0);fs.writeSync(fd,JSON.stringify(report,null,2),0,'utf8');fs.fsyncSync(fd)}
+    let leasePromise,pending
+    try {
+      demand(ready.trip.dispatches?.length===2&&fixture.members.length===2,'TWO_CUSTOMER_READY_SCOPE_REQUIRED')
+      const [first,second]=ready.trip.dispatches.map(binding=>fixture.members.find(x=>x.dispatchNo===binding.dispatchNo))
+      demand(first&&second&&first.customerId!==second.customerId&&first.dispatchNo===fixture.members[0].dispatchNo,'CURRENT_DEPARTURE_ORDER_REQUIRED')
+      const firstState=ready.members[fixture.members.indexOf(first)],secondState=ready.members[fixture.members.indexOf(second)]
+      const scope=report.target=validateRollbackScope({marker:context.marker,tripNo:fixture.tripNo,expectedImage:runtime.imageId,
+        firstCustomerId:first.customerId,secondCustomerId:second.customerId,firstDispatchId:id(firstState.dispatch.id),secondDispatchId:id(secondState.dispatch.id),
+        firstOutboundId:id(firstState.outbound.outbound.id),secondPalletId:second.palletId,firstDispatchNo:first.dispatchNo})
+      save();leasePromise=remote(env,scope,'LEASE').then(value=>({value}),error=>({error:error.message}))
+      for(let i=0;i<12;i++) {
+        const rows=(await remote(env,scope,'LOCKS')).rows;report.samples.push({phase:'LEASE',rows});save()
+        if(rows.filter(x=>x.collection==='lease').length===1)break
+        demand(i!==11,'LIVE_LEASE_NOT_OBSERVED');await pause(250)
+      }
+      pending=context.request('POST',departPath,departBody,fixture.pdaToken).then(value=>({value}),error=>({error:error.message}))
+      for(let i=0;i<20;i++) {
+        const rows=(await remote(env,scope,'OBSERVE')).rows;report.samples.push({phase:'MID_TRANSACTION',rows});save()
+        if(rows.some(x=>x.collection==='wait')) {report.observation=assertMidTransactionObservation(rows);save();break}
+        demand(i!==19,'MID_TRANSACTION_NOT_OBSERVED');await pause(250)
+      }
+      report.response=await pending;save()
+      demand(!report.response.error,'DEPARTURE_OUTCOME_UNKNOWN_RETAIN_ORIGINAL_COMMAND')
+      const response=report.response.value
+      demand(response.http===500||(response.http===200&&String(response.body?.code)==='S0500'),'EXPECTED_DATABASE_FAILURE_RESPONSE_REQUIRED')
+      report.lease=await leasePromise;save()
+      demand(!report.lease.error&&report.lease.value.rows.at(-1)?.released===true,'LEASE_RELEASE_NOT_PROVED')
+      const released=report.released=(await remote(env,scope,'RELEASED')).rows
+      demand(released.length===0,'OWNED_ROW_LOCK_OR_WAITER_REMAINS')
+      const restored=await snapshot('ROLLBACK_AFTER_FAULT')
+      report.consistentSnapshot={beforeSha256:createHash('sha256').update(canonical(ready)).digest('hex'),afterSha256:createHash('sha256').update(canonical(restored)).digest('hex')}
+      demand(report.consistentSnapshot.beforeSha256===report.consistentSnapshot.afterSha256,'FULL_TWO_CUSTOMER_ROLLBACK_SNAPSHOT_CHANGED')
+      // The caller also checks restoration before sending the same original command.
+      report.verified=true;return report
+    }catch(error){report.error=error.message;throw error}
+    finally {
+      if(pending&&!report.response)report.response=await pending
+      if(leasePromise&&!report.lease)report.lease=await leasePromise
+      report.finishedAt=new Date().toISOString();save();fs.closeSync(fd)
+    }
+  }
+}
+export async function runTripRollback(config,{env=process.env}={}) {
+  if(!config.execute)return {...config,scope:'OWNED_TWO_CUSTOMER_DEPARTURE_ROLLBACK',fault:'65-second SELECT FOR UPDATE lease on the second customer pallet-lock row; first customer writes observed as uncommitted, default 50-second lock timeout rolls back the entire original request; then same original command and full normal chain',businessSqlWrites:false,deployment:false}
+  demand(env.CHONGCHONG_ENV==='test'&&['JENKINS_URL','JENKINS_USER','JENKINS_API_TOKEN','FBA_CORE_JENKINS_AGENT'].every(key=>env[key]),'EXPLICIT_TEST_EVIDENCE_PROFILE_REQUIRED')
+  const runtime=await remote(env,{},'RUNTIME');demand(runtime.rows.length===1&&runtime.rows[0].lockWaitSeconds===50,'DEFAULT_50_SECOND_LOCK_TIMEOUT_REQUIRED')
+  const sources=['scripts/fleet-trip-rollback-test-env.mjs','scripts/admin-test-env-modules/fleet-cross-customer-acceptance.mjs'].map(path=>({path,sha256:createHash('sha256').update(fs.readFileSync(path)).digest('hex')}))
+  return runCrossCustomer(config,{env,verify:async(context,fixture,options)=>{
+    fs.writeFileSync(resolve('reports',`fleet-cross-${config.runId}`,'rollback-source.json'),JSON.stringify({runtime,sources},null,2),{flag:'wx'})
+    return verifyFleetCrossCustomer(context,fixture,{...options,verifyDepartureRollback:createDepartureRollbackVerifier(env,config,runtime)})
+  }})
+}
+if(process.argv[1]&&import.meta.url===pathToFileURL(resolve(process.argv[1])).href) {
+  try {const config=parseCrossCustomerArguments(process.argv.slice(2)),result=await runTripRollback(config);console.log(JSON.stringify(config.execute?{success:result.success,error:result.error,summary:result.summary,requests:result.requestCount,goalComplete:false}:result,null,2));if(config.execute&&!result.success)process.exitCode=1}
+  catch(error){console.error(/^[A-Z_0-9]+$/.test(error.message)?error.message:'ROLLBACK_ENTRY_FAILED');process.exitCode=1}
+}

scripts/fleet-trip-rollback-test-env.test.mjs

@@ -0,0 +1,39 @@
+import test from 'node:test'
+import assert from 'node:assert/strict'
+import { validateRollbackScope, rollbackSql, buildRollbackRemoteScript, assertMidTransactionObservation, runTripRollback } from './fleet-trip-rollback-test-env.mjs'
+const scope={marker:'OWN_RUN',tripNo:'TR-OWN',firstCustomerId:'101',secondCustomerId:'102',firstDispatchId:'201',secondDispatchId:'202',firstOutboundId:'301',secondPalletId:'401',firstDispatchNo:'DS-OWN',expectedImage:'sha256:'+'a'.repeat(64)}
+const rows=()=>[{collection:'lease',connectionId:11,lockId:'99'},{collection:'wait',requester:12,blocker:11,lockId:'99'},
+  {collection:'firstDispatch',id:'201',status:'DEPARTED',shipped:1},{collection:'firstOutbound',id:'301',status:'OUTBOUNDED'},{collection:'firstReplay',count:1}]
+test('plan and import are network-free; execution rejects missing TEST profile before any network',async()=>{
+  const plan=await runTripRollback({execute:false},{env:{}})
+  assert.equal(plan.businessSqlWrites,false);assert.equal(plan.deployment,false)
+  await assert.rejects(runTripRollback({execute:true},{env:{CHONGCHONG_ENV:'prod'}}),/TEST/)
+})
+test('scope rejects reused customers, injection and unpinned runtime',()=>{
+  assert.equal(validateRollbackScope(scope),scope)
+  for(const changed of [{secondCustomerId:'101'},{secondDispatchId:'201'},{firstDispatchId:'1 OR 1=1'},{marker:"RUN';DELETE"},{expectedImage:'latest'}])assert.throws(()=>validateRollbackScope({...scope,...changed}))
+})
+test('lease contains only owned locking reads, fixed duration and rollback, without mutation or schema change',()=>{
+  const sql=rollbackSql(scope,'LEASE')
+  assert.match(sql,/dispatch_order_id=202/);assert.match(sql,/warehouse_pallet_id=401/);assert.match(sql,/FOR UPDATE;.*SLEEP\(65\)\);ROLLBACK;/)
+  assert.doesNotMatch(sql,/\b(?:INSERT|DELETE|ALTER|DROP|TRUNCATE|REPLACE|KILL|SET GLOBAL)\b/)
+  assert.equal((sql.match(/UPDATE/g)||[]).length,1)
+  assert.throws(()=>rollbackSql(scope,'CUSTOM_SQL'))
+  assert.throws(()=>buildRollbackRemoteScript(scope,'LEASE',"agent');evil()"))
+})
+test('dirty observation is separate from final consistent cleanup evidence',()=>{
+  assert.match(rollbackSql(scope,'OBSERVE'),/READ UNCOMMITTED/)
+  assert.match(rollbackSql(scope,'RELEASED'),/REPEATABLE READ/)
+  assert.match(rollbackSql(scope,'RELEASED'),/READ ONLY/)
+  assert.equal(assertMidTransactionObservation(rows()).dirtyObservationOnly,true)
+})
+for(const fault of ['foreign blocker','no wait','two waiters','not yet written','no outbound write','no replay write'])test(`refuses indirect mid-transaction evidence: ${fault}`,()=>{
+  const value=rows()
+  if(fault==='foreign blocker')value[1].blocker=77
+  if(fault==='no wait')value.splice(1,1)
+  if(fault==='two waiters')value.push({...value[1],requester:13})
+  if(fault==='not yet written')value[2].status='READY_TO_DEPART'
+  if(fault==='no outbound write')value[3].status='WAIT_OUTBOUND'
+  if(fault==='no replay write')value[4].count=0
+  assert.throws(()=>assertMidTransactionObservation(value))
+})