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' 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 } } /** Owned departure only; reviewed source-selection recovery preserves prior starts. Requests are checkpointed before dispatch. */ 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, sizeBytes: 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, onBeforeRedispatchSignature: context.context?.onBeforeRedispatchSignature, 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)) if (input.sourceSelectionRecovery) check('Reviewed source recovery must retain the already started delivery; never start it again', current.order.status === 'DELIVERING') 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 with a stored quantity', signed?.order?.signedAt && Number.isSafeInteger(signed.order.signedPackages) && signed.order.signedPackages >= 0) 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) 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() const newSignature = await deps.saveReceiptAndComplete(io, newDelivery.order, customerId, 271) check('Redispatched two boxes actually signed', newSignature?.order?.signedAt && 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 && current.signedPackages === 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 } } 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 }