2026-10-03 22:13:15 +08:00
import { existsSync , openSync , writeFileSync , fsyncSync , closeSync , readFileSync } from 'node:fs'
import { join , resolve } from 'node:path'
import { redactEvidence } from '../fba-core-regression-test-env.mjs'
2026-10-03 16:11:52 +08:00
import { randomUUID , createHash } from 'node:crypto'
import { prepareFleetTrip , confirmFleetLoading , finishFleetTrip } from './fleet-trip-fulfillment.mjs'
import { pickAll , saveReceiptAndComplete } from './phase5-overseas-fulfillment.mjs'
const BASE = '/api/platform/v1/last-leg'
const id = value => value == null ? '' : String ( value )
const key = () => `redispatch- ${ randomUUID () } `
const scoped = ( path , customerId ) => ` ${ path } ? ${ new URLSearchParams ({ customerId } )}`
const ok = response => response ? . http === 200 && String ( response ? . body ? . code ) === '200'
const copy = value => value === undefined ? undefined : structuredClone ( value )
function definiteResponse ( response ) {
const http = response ? . http , body = response ? . body
if ( ! Number . isInteger ( http ) || http < 200 || http >= 500 || ! body || typeof body !== 'object' || Array . isArray ( body )
|| ! response . requestId || ! response . traceId ) return false
const code = id ( body . code )
return http === 200 && code === '200' || /^[A-Z]\d{4}$/ . test ( code ) && ( http === 200 || http >= 400 )
}
export function selectReturnSources ( rows ) {
const domestic = rows . filter ( row => row . tail ? . orderType === 'FIRST_LEG' && row . tail . sourceInboundId && row . tail . inboundCode === row . relation . inboundCode )
const a = domestic . find ( row => Number ( row . tail . packageCount ) > 2 && row . relation . returnSources ? . some ( source => Number ( source . maxReturnPackages ) >= 2 ))
const b = domestic . find ( row => a && row . tail . sourceInboundId !== a . tail . sourceInboundId )
if ( ! a || ! b ) throw Error ( 'Fresh redispatch fixture requires source A with at least 3 boxes and a separate domestic source B' )
for ( const selected of [ a , b ]) {
const packages = domestic . filter ( row => row . tail . sourceInboundId === selected . tail . sourceInboundId )
. reduce (( sum , row ) => sum + ( row . relation . returnSources || []). reduce (( n , source ) => n + Number ( source . maxReturnPackages ), 0 ), 0 )
if ( packages !== Number ( selected . tail . packageCount )) throw Error ( 'Current departure must cover each selected source completely; no old shipments may fill the fixture' )
}
return { a , b }
}
2026-10-03 22:13:15 +08:00
/** Owned departure only; reviewed source-selection recovery preserves prior starts. Requests are checkpointed before dispatch. */
2026-10-03 16:11:52 +08:00
export async function verifyFleetReturnRedispatch ( context , module , trip , input , dependencies = {}) {
const run = context . createdResources ? . fleetAcceptance
if ( ! context . marker || ! context . createdResources ? . fleetTrips ? . includes ( trip ) || run ? . marker !== context . marker
|| run . tripNo !== trip . tripNo || run . departure ? . status !== 'IN_TRANSIT' || ! input ? . pickerToken
|| id ( run . customerId ) !== id ( input . customerId ) || id ( run . warehouseId ) !== id ( input . warehouseId )
|| JSON . stringify (( run . departure . dispatches || []). map ( row => row . dispatchNo ). sort ()) !== JSON . stringify ([...( input . dispatchNos || [])]. sort ())
|| ! input . deliveryNos ? . length || trip . returnRedispatch || typeof context . checkpoint !== 'function' )
throw Error ( 'Redispatch requires a new owned departure, current PDA session and durable checkpoints; no automatic restart' )
const proof = trip . returnRedispatch = { marker : context . marker , status : 'CHECKING' , requests : [], returnPackages : 2 ,
readOnlyEvidenceRequired : [ 'Before redispatch signature: source A completion outbox count=0 and domestic inbound not DELIVERED; source B completion outbox count=1' ,
'After redispatch signature and exact replay: A and B each have exactly one OVERSEAS_DELIVERY_COMPLETED outbox event by customer+sourceInboundId; inspect delivery ACK and domestic inbound completion separately' ,
'Join return receipt ledger_id and source_outbound_no to immutable RETURN_STOCK_IN; join new picks by return_receipt_id; original shipment and signed facts remain unchanged' ] }
const check = ( label , condition ) => { if ( ! condition ) throw Error ( label ) }
const io = Object . create ( context )
io . assert = (... args ) => { context . assert (... args ); if ( ! args [ 3 ]) throw Error ( args [ 1 ]) }
async function invoke ( method , path , body , token , file , options ) {
const entry = { method , path , body : copy ( body ), actor : token ? 'PDA' : 'PLATFORM' , status : 'SENDING' }
if ( file ) entry . file = { name : file . name , type : file . type , bytes : file . content . byteLength , sha256 : createHash ( 'sha256' ). update ( file . content ). digest ( 'hex' ) }
proof . requests . push ( entry ); await context . checkpoint ()
try {
const response = file ? await context . requestMultipart ( method , path , file ) : await context . request ( method , path , body , token , options )
if ( ! definiteResponse ( response )) {
Object . assign ( entry , { status : 'UNKNOWN_RESULT' , http : response ? . http , code : response ? . body ? . code ,
requestId : response ? . requestId , traceId : response ? . traceId })
await context . checkpoint ()
throw Error ( `HTTP transport outcome unknown: ${ method } ${ path } ; preserve the original request, do not resend` )
}
Object . assign ( entry , { status : 'RESPONDED' , http : response ? . http , code : response ? . body ? . code ,
requestId : response ? . requestId , traceId : response ? . traceId , result : copy ( response ? . body ? . data ) })
await context . checkpoint ()
check ( `Missing HTTP correlation: ${ method } ${ path } ` , response ? . requestId && response ? . traceId )
return response
} catch ( error ) { if ( entry . status === 'SENDING' ) { entry . status = 'UNKNOWN_RESULT' ; await context . checkpoint () } throw error }
}
io . request = ( method , path , body , token , options ) => invoke ( method , path , body , token , null , options )
io . requestMultipart = ( method , path , file ) => invoke ( method , path , undefined , undefined , file )
async function request ( method , path , body , token ) {
const response = await io . request ( method , path , body , token ); check ( `HTTP failed: ${ method } ${ path } ` , ok ( response )); return response . body . data
}
const deps = { prepareFleetTrip , confirmFleetLoading , finishFleetTrip , pickAll , saveReceiptAndComplete ,
2026-10-03 22:13:15 +08:00
onBeforeRedispatchSignature : context . context ? . onBeforeRedispatchSignature ,
2026-10-03 16:11:52 +08:00
wait : ms => new Promise ( resolve => setTimeout ( resolve , ms )), ... dependencies }
const customerId = id ( input . customerId ), warehouseId = id ( input . warehouseId )
const tail = relation => request ( 'GET' , scoped ( ` ${ BASE } /orders/ ${ relation . lastMileOrderId } ` , customerId ))
const delivery = number => request ( 'GET' , scoped ( ` ${ BASE } /delivery-orders/ ${ encodeURIComponent ( number ) } ` , customerId ))
async function domestic ( selected ) {
const page = await request ( 'GET' , `/api/platform/inbounds? ${ new URLSearchParams ({ tenantId : customerId , inboundCode : selected . relation . inboundCode , page : '1' , size : '200' } )}` )
const matches = ( page . records || []). filter ( row => row . inboundCode === selected . relation . inboundCode )
check ( 'Domestic source must resolve uniquely through its current inbound code' , matches . length === 1 && matches [ 0 ]. processStatus )
return matches [ 0 ]
}
async function domesticBoundary ( a , b , allSigned ) {
const snapshots = []
for ( let attempt = 0 ; attempt < 30 ; attempt ++ ) {
const A = await domestic ( a ), B = await domestic ( b ); snapshots . push ({ A , B })
if ( ! allSigned ) check ( 'Incomplete A must never be marked delivered domestically' , A . processStatus !== 'DELIVERED' )
if ( B . processStatus === 'DELIVERED' && ( ! allSigned || A . processStatus === 'DELIVERED' )) return snapshots
if ( attempt < 29 ) await deps . wait ( 1000 )
}
throw Error ( allSigned ? 'Domestic source completion has not converged after redispatch signature' : 'Independent source B domestic completion did not converge' )
}
try {
const rows = []
proof . originalDeliveries = []
for ( const number of [... new Set ( input . deliveryNos )]) {
const current = await delivery ( number )
check ( 'Fixture deliveries must be unsigned and from the supplied current dispatches' , ! current . order . signedAt && input . dispatchNos . includes ( current . order . dispatchNo ))
proof . originalDeliveries . push ( copy ( current . order ))
2026-10-03 22:13:15 +08:00
if ( input . sourceSelectionRecovery ) check ( 'Reviewed source recovery must retain the already started delivery; never start it again' , current . order . status === 'DELIVERING' )
2026-10-03 16:11:52 +08:00
if ( current . order . status === 'WAIT_DELIVERY' ) {
check ( 'Delivery status mutation requires the current version' , Number . isSafeInteger ( Number ( current . order . version ))
&& current . order . version != null && String ( current . order . version ). trim () !== '' && Number ( current . order . version ) >= 0 )
const started = await request ( 'PUT' , scoped ( ` ${ BASE } /delivery-orders/ ${ number } /status` , customerId ), {
status : 'DELIVERING' , expectedVersion : Number ( current . order . version ), remark : ` ${ context . marker } current-run delivery underway` ,
})
check ( 'Return candidates are opened only after formal delivery start' , started . deliveryStatus === 'DELIVERING' )
}
const relations = await request ( 'GET' , scoped ( ` ${ BASE } /delivery-orders/ ${ encodeURIComponent ( number ) } /return-inbound-options` , customerId ))
for ( const relation of relations ) {
check ( 'Return source relation belongs to the original delivery' , relation . deliveryNo === number && relation . outboundNo === current . order . outboundNo )
const detail = await tail ( relation )
check ( 'Tail source customer and warehouse match this run' , id ( detail . order . customerId ) === customerId && id ( detail . order . overseasWarehouseId ) === warehouseId )
rows . push ({ relation , tail : detail . order , delivery : current . order })
}
}
const { a , b } = selectReturnSources ( rows )
proof . domesticBefore = { A : await domestic ( a ), B : await domestic ( b ) }
check ( 'Both source inbounds must be unfinished in this fresh sample' , proof . domesticBefore . A . processStatus !== 'DELIVERED' && proof . domesticBefore . B . processStatus !== 'DELIVERED' )
proof . sources = { A : copy ( a ), B : copy ( b ) }; await context . checkpoint ()
const source = a . relation . returnSources . find ( row => Number ( row . maxReturnPackages ) >= 2 )
const marked = await request ( 'POST' , scoped ( ` ${ BASE } /delivery-orders/ ${ a . delivery . deliveryNo } /mark-return` , customerId ), {
inboundOrderId : a . relation . inboundOrderId , reason : ` ${ context . marker } two boxes refused before signature` , returnDirection : 'OVERSEAS_WAREHOUSE' ,
returnLogisticsNo : ` ${ context . marker } -RET` , items : [{ outboundPalletItemId : source . outboundPalletItemId , inventoryPositionId : source . inventoryPositionId , packageCount : 2 }], idempotencyKey : key (),
})
proof . exceptionNo = marked . exceptionNo
const exceptionPath = ` ${ BASE } /exception-orders/ ${ encodeURIComponent ( marked . exceptionNo ) } `
const resource = `/api/platform/v1/overseas-warehouse-resources/ ${ warehouseId } `
const palletCode = `RET- ${ randomUUID () } `
const palletId = id ( await request ( 'POST' , ` ${ resource } /pallets` , { palletCode , palletType : '木托盘' , capacityCbm : 2 , unitPrice : 0 , status : 'IDLE' , remark : context . marker }))
const pallet = await request ( 'GET' , ` ${ resource } /pallets/ ${ palletId } ` )
const locationPage = await request ( 'GET' , ` ${ resource } /locations?page=1&size=200` )
const location = ( Array . isArray ( locationPage ) ? locationPage : locationPage . records || []). find ( row => id ( row . id ) === id ( pallet . locationId ))
check ( 'New return pallet has a real same-warehouse location' , id ( pallet . warehouseId ) === warehouseId && location ? . locationCode && pallet . palletCode === palletCode )
const started = await request ( 'POST' , scoped ( ` ${ exceptionPath } /start-processing` , customerId ), { expectedVersion : marked . version , idempotencyKey : key (), handlingAction : '实际退回2箱' })
const uploaded = await io . requestMultipart ( 'POST' , scoped ( ` ${ exceptionPath } /evidence` , customerId ), {
name : ` ${ context . marker } -return.txt` , type : 'text/plain' , content : new TextEncoder (). encode ( ` ${ context . marker } synthetic evidence of 2 returned boxes` ),
})
check ( 'Return evidence uploaded through owned exception API' , ok ( uploaded ) && uploaded . body . data ? . fileId )
const receipt = { expectedVersion : started . version , idempotencyKey : key (), handlingResult : '原出库2箱实际退回入仓' , evidenceFileIds : [ uploaded . body . data . fileId ],
receipts : [{ outboundPalletItemId : source . outboundPalletItemId , inventoryPositionId : source . inventoryPositionId , receivedPackages : 2 ,
targetPalletId : palletId , targetPalletNo : palletCode , targetLocationId : location . id , targetLocationCode : location . locationCode }] }
const restored = await request ( 'POST' , scoped ( ` ${ exceptionPath } /complete` , customerId ), receipt )
check ( 'Exactly two boxes restored' , restored . status === 'COMPLETED' && Number ( restored . restoredPackages ) === 2 )
const replay = await request ( 'POST' , scoped ( ` ${ exceptionPath } /complete` , customerId ), receipt )
check ( 'Same return key replays the same adjustment' , replay . replayed === true && id ( replay . inventoryAdjustmentId ) === id ( restored . inventoryAdjustmentId ))
proof . returned = { palletId , palletCode , adjustmentId : restored . inventoryAdjustmentId }
proof . originalSignatures = []
for ( const [ index , number ] of input . deliveryNos . entries ()) {
const current = await delivery ( number )
const signed = await deps . saveReceiptAndComplete ( io , current . order , customerId , 171 + index )
check ( 'Original remaining cargo actually signed' , signed ? . order ? . signedAt )
proof . originalSignatures . push ( copy ( signed . order ))
}
const beforeA = await tail ( a . relation ), beforeB = await tail ( b . relation )
check ( 'A remains incomplete while independently complete B is delivered' , ! beforeA . order . subOrderStatus ? . includes ( 'DELIVERY:DELIVERED' ) && beforeB . order . subOrderStatus ? . includes ( 'DELIVERY:DELIVERED' ))
proof . beforeRedispatch = { A : beforeA , B : beforeB }
proof . domesticBeforeRedispatch = await domesticBoundary ( a , b , false )
const candidates = await request ( 'GET' , ` ${ BASE } /dispatches/candidates? ${ new URLSearchParams ({ sourceStage : 'RECEIVED' , orderNo : a . relation . inboundCode } )}` )
const matches = candidates . filter ( row => row . sourceType === 'RETURN_RECEIPT' && id ( row . warehousePalletId ) === palletId && row . returnReceiptNo && id ( row . orderId ) === id ( a . relation . lastMileOrderId ))
check ( 'Exactly one fresh receipt candidate contains only the returned two boxes' , matches . length === 1 && Number ( matches [ 0 ]. packageCount ) === 2 && ! matches [ 0 ]. planPalletId )
const selected = matches [ 0 ], draftToken = key ()
const container = await request ( 'POST' , ` ${ BASE } /dispatches/drafts/containers` , { draftToken , capacity : 30 })
await request ( 'POST' , ` ${ BASE } /dispatches/draft-locks/lock` , { sourceType : 'RETURN_RECEIPT' , returnReceiptId : selected . returnReceiptId , draftToken , draftContainerId : container . id })
const arrange = { planPalletIds : [], returnReceiptIds : [ selected . returnReceiptId ], draftToken }
const arranged = await request ( 'POST' , ` ${ BASE } /dispatches/pre-arrange` , arrange )
check ( 'Redispatch is a new execution document containing only two boxes' , arranged . dispatchNo && ! input . dispatchNos . includes ( arranged . dispatchNo ) && Number ( arranged . packageCount ) === 2 )
const arrangedReplay = await request ( 'POST' , ` ${ BASE } /dispatches/pre-arrange` , arrange )
check ( 'Draft replay keeps the same redispatch' , arrangedReplay . dispatchNo === arranged . dispatchNo )
const newTrip = await deps . prepareFleetTrip ( io , module , arranged . dispatchNo , warehouseId )
proof . newTripNo = newTrip . tripNo ; proof . returnReceiptNo = selected . returnReceiptNo ; proof . redispatchNo = arranged . dispatchNo
const tripDetail = await request ( 'GET' , ` ${ BASE } /fleet/trips/ ${ newTrip . tripNo } ` )
const transfer = await request ( 'POST' , ` ${ BASE } /dispatches/ ${ arranged . dispatchNo } /transfer-delivery` , {
idempotencyKey : key (), plannedDeliveryAt : new Date ( Date . now () + 21600000 ). toISOString (). slice ( 0 , 19 ), deliveryMethod : 'TRUCK_PALLET' ,
pickupMethod : 'WAREHOUSE_PICKUP' , loadingType : 'PALLET' , vehicleType : 'TRUCK' , vehicleSize : '30托' ,
vehiclePlate : tripDetail . trip . vehiclePlate , driverName : tripDetail . trip . driverName , driverPhone : tripDetail . trip . driverPhone , appointmentAccount : 'SELF' ,
fees : [{ feeType : 'DELIVERY_FEE' , feeName : '退回重派验收费用' , direction : 'RECEIVABLE' , currency : 'CNY' , amount : 1 }], remark : context . marker ,
})
check ( 'Redispatch produces a new single destination delivery' , transfer . deliveries ? . length === 1 && ! input . deliveryNos . includes ( transfer . deliveries [ 0 ]. deliveryNo ))
check ( 'Fresh return stock can be picked via PDA' , await deps . pickAll ( io , input . pickerToken , arranged . dispatchNo ))
const recordVersion = await deps . confirmFleetLoading ( io , module , newTrip )
const departBody = { recordVersion , idempotencyKey : key (), remark : '退回2箱重新实际发车' }
const departPath = `/api/pda/v1/last-leg/dispatches/ ${ arranged . dispatchNo } /depart`
const departed = await request ( 'POST' , departPath , departBody , input . pickerToken )
check ( 'New DEPART ships exactly two returned boxes' , departed . tripNo === newTrip . tripNo && Number ( departed . shippedPackages ) === 2 )
const departedAgain = await request ( 'POST' , departPath , departBody , input . pickerToken )
check ( 'Departure replay does not duplicate shipping' , departedAgain . replayed === true && Number ( departedAgain . shippedPackages ) === 2 )
const newDelivery = await delivery ( transfer . deliveries [ 0 ]. deliveryNo )
2026-10-03 22:13:15 +08:00
proof . preSignatureGate = { status : 'WAITING' , deliveryNo : newDelivery . order . deliveryNo , newTripNo : newTrip . tripNo ,
departed , originalSignatureCount : proof . originalSignatures . length }
await context . checkpoint ()
check ( 'Readonly source event audit gate is required before redispatch signature' , typeof deps . onBeforeRedispatchSignature === 'function' )
const audit = await deps . onBeforeRedispatchSignature ( proof )
check ( 'Pre-signature audit proves A has zero completion events and B has one' , audit ? . sourceAEventCount === 0
&& audit . sourceBEventCount === 1 && /^[a-f0-9]{64}$/ . test ( audit . auditEvidenceSha256 || '' ))
Object . assign ( proof . preSignatureGate , { status : 'VERIFIED' , audit })
await context . checkpoint ()
2026-10-03 16:11:52 +08:00
const newSignature = await deps . saveReceiptAndComplete ( io , newDelivery . order , customerId , 271 )
check ( 'Redispatched two boxes actually signed' , newSignature ? . order ? . signedAt && Number ( newSignature . order . signedPackages ) === 2 )
const afterA = await tail ( a . relation ), afterB = await tail ( b . relation )
check ( 'A completes only after redispatch; B remains complete' , afterA . order . subOrderStatus ? . includes ( 'DELIVERY:DELIVERED' ) && afterB . order . subOrderStatus === beforeB . order . subOrderStatus )
for ( const original of proof . originalSignatures ) {
const current = ( await delivery ( original . deliveryNo )). order
check ( 'Original signed facts stay immutable' , current . signedAt === original . signedAt && Number ( current . signedPackages ) === Number ( original . signedPackages ) && Number ( current . packageCount ) === Number ( original . packageCount ))
}
proof . afterRedispatch = { A : afterA , B : afterB , newSignature , departed }
proof . domesticAfterRedispatch = await domesticBoundary ( a , b , true )
proof . closure = await deps . finishFleetTrip ( io , module , newTrip )
check ( 'New trip cargo and resource lifecycle completes' , proof . closure ? . status === 'CARGO_COMPLETED_AND_RESOURCES_RELEASED' )
proof . status = 'HTTP_CHAIN_VERIFIED_SOURCE_EVENT_EVIDENCE_REQUIRED' ; proof . sourceEventsVerified = false
await context . checkpoint (); return proof
} catch ( error ) { proof . status = 'FAILED' ; proof . failure = error . message ; await context . checkpoint (); throw error }
}
2026-10-03 22:13:15 +08:00
export async function waitForRedispatchSignatureAudit ( context , proof , { root , output , secrets = [],
wait = ms => new Promise ( resolve => setTimeout ( resolve , ms )), timeoutMs = 20 * 60 * 1000 } = {}) {
const gateNonce = randomUUID (), approvalPath = join ( output , 'approve-before-redispatch-signature.json' )
const checkpointPath = join ( output , 'before-redispatch-signature.json' )
const checkpoint = { gateNonce , marker : context . marker , runId : context . context . fleetRunId ,
originalTripNo : context . createdResources . fleetAcceptance . tripNo , newTripNo : proof . newTripNo ,
sourceAInboundId : proof . sources . A . tail . sourceInboundId , sourceBInboundId : proof . sources . B . tail . sourceInboundId ,
customerId : context . createdResources . fleetAcceptance . customerId , proof , createdResources : context . createdResources }
const descriptor = openSync ( checkpointPath , 'wx' )
try { writeFileSync ( descriptor , JSON . stringify ( redactEvidence ( checkpoint , secrets ), null , 2 ) + '\n' ); fsyncSync ( descriptor ) }
finally { closeSync ( descriptor ) }
const progressDescriptor = openSync ( join ( output , 'progress.json' ), 'w' )
try {
writeFileSync ( progressDescriptor , JSON . stringify ( redactEvidence ({ stage : 'RETURN_REDISPATCH_PRE_SIGNATURE' , marker : context . marker ,
steps : context . steps || [], modules : context . modules || [], createdResources : context . createdResources }, secrets ), null , 2 ) + '\n' )
fsyncSync ( progressDescriptor )
} finally { closeSync ( progressDescriptor ) }
console . log ( 'FLEET_SIGNATURE_AUDIT_WAITING=' + JSON . stringify ({ checkpointPath , approvalPath , gateNonce , marker : context . marker }))
const started = Date . now ()
while ( ! existsSync ( approvalPath )) {
if ( Date . now () - started >= timeoutMs ) throw Error ( 'Pre-signature audit gate timed out; do not sign or resume automatically' )
await wait ( 1000 )
}
const approval = JSON . parse ( readFileSync ( approvalPath , 'utf8' ))
if ( approval . gateNonce !== gateNonce || approval . marker !== checkpoint . marker || approval . newTripNo !== checkpoint . newTripNo
|| approval . sourceAEventCount !== 0 || approval . sourceBEventCount !== 1 || ! /^[a-f0-9]{64}$/ . test ( approval . auditEvidenceSha256 || '' )
|| ! /^reports\/[A-Za-z0-9_-]+(?:\/[A-Za-z0-9_.-]+)?\.json$/ . test ( approval . auditEvidencePath || '' ))
throw Error ( 'Pre-signature approval does not match this exact gate and source counts' )
const evidencePath = resolve ( root , approval . auditEvidencePath )
if ( ! existsSync ( evidencePath ) || createHash ( 'sha256' ). update ( readFileSync ( evidencePath )). digest ( 'hex' ) !== approval . auditEvidenceSha256 )
throw Error ( 'Pre-signature audit evidence SHA256 mismatch' )
return approval
}
