import { createHash } from 'node:crypto'
import { confirmFleetLoading } from './fleet-trip-fulfillment.mjs'

const MODULE = 'fleet-cross-customer-acceptance'
const ROOT = '/api/platform/v1/last-leg'
const PDA = '/api/pda/v1/last-leg/dispatches'
const OPTIONS = { scenario: 'fleet-cross-customer-departure', deploymentBlocker: true }
const id = value => value == null ? '' : String(value)
const ok = response => response?.http === 200 && id(response.body?.code) === '200'
const encode = encodeURIComponent
const scoped = (path, customerId) => `${path}?${new URLSearchParams({ customerId })}`
const canonical = value => JSON.stringify(normalize(value))
function normalize(value) {
  if (Array.isArray(value)) return value.map(normalize)
  return value && typeof value === 'object' ? Object.fromEntries(Object.keys(value).sort().map(key => [key, normalize(value[key])])) : value
}
function integer(value) {
  if (!['number', 'string'].includes(typeof value) || !/^(0|[1-9][0-9]*)$/.test(String(value)) || !Number.isSafeInteger(Number(value))) throw new Error('Invalid inventory integer')
  return Number(value)
}
const buckets = ['availablePackages', 'holdPackages', 'allocatedPackages', 'pickedPackages', 'receivingPackages']
const balanceBuckets = [...buckets, 'inTransitPackages', 'procurementPackages']
export const fixtureContract = Object.freeze({
  stage: 'TWO_FRESH_CUSTOMERS_ONE_REVIEWED_DISPATCH_LOADING_NOT_PICKED',
  fixture: 'marker, warehouseId, dispatchNo, tripNo, tripProof, pdaToken, optional current-run loadingMeasurements; members[2]: marker/customerId/incomingNo/shared dispatchNo/dispatchItemId/balanceId/positionId/palletId/palletNo/receiptVerified',
  preparation: 'Two independently qualified customers formally received one pallet/box each in the same warehouse. One native Create with both actual source selections; automatic unique Fleet record, resources and both native reviews. No picking or delivery arrangement.',
  reader: 'readInventory(scope) returns all raw balances/positions/immutable movements including zero, exact marker/customer/warehouse and CUSTOMER_WAREHOUSE_COMPLETE coverage.',
  boundary: 'Prerequisite rejection is separate from an explicitly injected real mid-transaction rollback. Actual loading needs actual evidence; absent measurement fields block. No environment loader, SQL or implicit account lifecycle.',
})

export const afterLoadingContinuationContract = Object.freeze({
  ...fixtureContract,
  stage: 'TWO_OWNED_CUSTOMERS_ONE_DISPATCH_DELIVERING',
  preparation: 'Explicit continuation only: prior failed proof with verified gate and complete AFTER_PICK_1; original successful loadingCommand/confirmedLoad; current DELIVERING with automatically created execution documents.',
  boundary: 'Reconcile original identities, complete picked inventory and original loading receipt through current reads before any new write. Never repeat picking/loading or rewrite the prior proof.',
})

export async function verifyFleetCrossCustomer(context, fixture, { execute = false, readInventory, verifyDepartureRollback, completionPolling } = {}) {
  if (!execute) return fixtureContract
  return runCrossCustomerCore(context, fixture, { readInventory, verifyDepartureRollback, completionPolling })
}

export async function continueFleetCrossCustomerAfterLoading(context, fixture, { execute = false, readInventory, verifyDepartureRollback, priorProof, completionPolling } = {}) {
  if (!execute) return afterLoadingContinuationContract
  if (!priorProof) throw new Error('Explicit prior cross-customer proof is required for completed-loading continuation')
  return runCrossCustomerCore(context, { ...fixture, members: structuredClone(fixture?.members) },
    { readInventory, verifyDepartureRollback, priorProof, completionPolling })
}

async function runCrossCustomerCore(context, fixture, { readInventory, verifyDepartureRollback, priorProof, completionPolling }) {
  const continuation = priorProof !== undefined
  const contract = continuation ? afterLoadingContinuationContract : fixtureContract
  if (!context?.marker || fixture?.marker !== context.marker || fixture.stage !== contract.stage
    || !fixture.dispatchNo || !fixture.tripNo || fixture.tripProof?.dispatchNo !== fixture.dispatchNo || fixture.tripProof.tripNo !== fixture.tripNo
    || !context.createdResources?.fleetTrips?.includes(fixture.tripProof)
    || !fixture.warehouseId || !fixture.pdaToken || typeof readInventory !== 'function' || fixture.members?.length !== 2
    || new Set(fixture.members.map(row => row.customerId)).size !== 2
    || fixture.members.some(row => row.marker !== context.marker || row.receiptVerified !== true || row.dispatchNo !== fixture.dispatchNo
      || ['customerId', 'incomingNo', 'dispatchItemId', 'balanceId', 'positionId', 'palletId', 'palletNo'].some(field => !id(row[field]))))
    throw new Error('Requires two distinct current-run customers and one reviewed native dispatch with complete raw inventory evidence')
  for (const field of ['incomingNo', 'dispatchItemId', 'positionId', 'palletId', 'balanceId'])
    if (new Set(fixture.members.map(row => id(row[field]))).size !== 2) throw new Error(`Cross-customer fixture reuses ${field}`)
  const polling = { maxAttempts: 41, intervalMs: 500, deadlineMs: 20000, ...completionPolling }
  if (!Number.isSafeInteger(polling.maxAttempts) || polling.maxAttempts < 1 || polling.maxAttempts > 121
    || !Number.isSafeInteger(polling.intervalMs) || polling.intervalMs < 1 || polling.intervalMs > 2000
    || !Number.isSafeInteger(polling.deadlineMs) || polling.deadlineMs < 1 || polling.deadlineMs > 60000)
    throw new Error('Finite completion polling bounds are required')
  context.createdResources ??= {}
  if (context.createdResources.fleetCrossCustomer) throw new Error('Proof already exists; retain failed run and reconcile explicitly')
  const proof = context.createdResources.fleetCrossCustomer = { marker: context.marker, dispatchNo: fixture.dispatchNo, tripNo: fixture.tripNo,
    warehouseId: fixture.warehouseId, members: fixture.members.map(({ marker, customerId, incomingNo, dispatchNo, dispatchItemId, balanceId, positionId, palletId, palletNo }) =>
      ({ marker, customerId, incomingNo, dispatchNo, dispatchItemId, balanceId, positionId, palletId, palletNo })),
    status: 'CHECKING', requests: [], snapshots: {}, gateProof: 'NOT_RUN', midTransactionRollback: 'NOT_TESTED', cleanup: 'Caller owns account/session lifecycle; retain all business fixtures' }
  const check = (name, passed) => { context.assert(MODULE, name, null, Boolean(passed), '', OPTIONS); if (!passed) throw new Error(name) }
  const equal = (a, b, name) => check(name, canonical(a) === canonical(b))
  const key = suffix => `${context.marker}-CROSS-${suffix}`
  const path = `${ROOT}/dispatches/${encode(fixture.dispatchNo)}`, pdaPath = `${PDA}/${encode(fixture.dispatchNo)}`
  const tripPath = `${ROOT}/fleet/trips/${encode(fixture.tripNo)}`
  const request = async (method, target, body, label, pda = false, rejection) => {
    const entry = { method, path: target, body, label, status: 'SENDING' }; proof.requests.push(entry); await context.checkpoint?.()
    let response
    try { response = await context.request(method, target, body, pda ? fixture.pdaToken : context.token, {}) }
    catch (error) { entry.status = 'UNKNOWN_RESULT'; await context.checkpoint?.(); throw error }
    Object.assign(entry, { status: 'RESPONDED', http: response?.http, code: response?.body?.code, requestId: response?.requestId, traceId: response?.traceId })
    await context.checkpoint?.()
    const denied = response?.http === 409 || (response?.http === 200 && id(response.body?.code) === 'A0409')
    const actorDenied = [200,403].includes(response?.http) && id(response.body?.code) === 'A0403'
    check(label, Boolean(response?.requestId && response?.traceId)
      && (rejection === 'ACTOR_FORBIDDEN' ? actorDenied : rejection ? denied && rejection(id(response.body?.message)) : ok(response)))
    return response.body?.data
  }
  const stock = async member => {
    const scope={marker:context.marker,customerId:member.customerId,warehouseId:fixture.warehouseId}
    const raw=await readInventory(scope)
    check('Inventory reader proves complete customer/warehouse scope',canonical(raw?.scope)===canonical({...scope,coverage:'CUSTOMER_WAREHOUSE_COMPLETE'}))
    for(const collection of ['balances','positions','ledger'])check(`Complete ${collection} rows have unique identities and customer ownership`,Array.isArray(raw[collection])
      && new Set(raw[collection].map(row=>id(row.id))).size===raw[collection].length
      && raw[collection].every(row=>id(row.id) && id(row.customerId)===id(member.customerId) && id(row.overseasWarehouseId)===id(fixture.warehouseId)))
    check('Original receipt balance and physical position are retained even at zero',raw.balances.some(row=>id(row.id)===id(member.balanceId))
      && raw.positions.some(row=>id(row.id)===id(member.positionId) && id(row.balanceId)===id(member.balanceId) && id(row.palletId)===id(member.palletId) && row.palletNo===member.palletNo))
    for(const balance of raw.balances) {
      for(const field of balanceBuckets) integer(balance[field])
      const positions=raw.positions.filter(row=>id(row.balanceId)===id(balance.id))
      for(const field of buckets) check(`Raw balance and every position conserve ${field}`,integer(balance[field])===positions.reduce((sum,row)=>sum+integer(row[field]),0))
    }
    check('No orphan stock positions',raw.positions.every(row=>raw.balances.some(balance=>id(balance.id)===id(row.balanceId))))
    return raw
  }
  const snapshot = async label => {
    const trip = await request('GET', tripPath, undefined, 'Read automatic unique Fleet resource record')
    check('One trip binds exactly this one native cross-customer dispatch', trip.trip?.tripNo === fixture.tripNo
      && id(trip.trip.warehouseId) === id(fixture.warehouseId) && trip.dispatches?.length === 1 && trip.dispatches[0].dispatchNo === fixture.dispatchNo)
    const dispatch = await request('GET', path, undefined, 'Read native whole dispatch')
    const items = await request('GET', `${path}/items`, undefined, 'Read actual per-customer cargo')
    const task = await request('GET', pdaPath, undefined, 'Read bound warehouse native PDA task', true)
    check('Native main/items/PDA retain both true customers and physical pallets', dispatch.dispatchNo === fixture.dispatchNo
      && items.dispatchNo === fixture.dispatchNo && task.dispatchNo === fixture.dispatchNo && id(task.warehouseId) === id(fixture.warehouseId)
      && id(dispatch.recordVersion) === id(items.recordVersion) && id(task.recordVersion) === id(dispatch.recordVersion)
      && items.items?.length === 2 && task.items?.length === 2 && fixture.members.every(member =>
        items.items.some(i => id(i.dispatchItemId) === member.dispatchItemId && id(i.customerId) === member.customerId
          && id(i.warehousePalletId) === member.palletId && i.palletNo === member.palletNo && integer(i.scheduledPackages) === 1)
        && task.items.some(i => id(i.dispatchItemId) === member.dispatchItemId && id(i.customerId) === member.customerId && i.palletNo === member.palletNo)))
    const members = []
    for (const member of proof.members) {
      let outbound = null, delivery = null
      if (member.outboundNo || member.deliveryNo) {
        check('Customer execution documents are both explicit', member.outboundNo && member.deliveryNo)
        outbound = await request('GET', scoped(`${ROOT}/outbound-orders/${encode(member.outboundNo)}`, member.customerId), undefined, 'Read customer outbound')
        delivery = await request('GET', scoped(`${ROOT}/delivery-orders/${encode(member.deliveryNo)}`, member.customerId), undefined, 'Read customer delivery')
        check('Execution documents keep customer, warehouse and shared main ownership', outbound.outbound?.outboundNo === member.outboundNo
          && outbound.outbound.dispatchNo === fixture.dispatchNo && id(outbound.outbound.overseasWarehouseId) === id(fixture.warehouseId)
          && outbound.deliveries?.length === 1 && outbound.deliveries[0].deliveryNo === member.deliveryNo
          && id(outbound.deliveries[0].customerId) === member.customerId && delivery.order?.deliveryNo === member.deliveryNo
          && id(delivery.order.customerId) === member.customerId && delivery.order.dispatchNo === fixture.dispatchNo && delivery.order.outboundNo === member.outboundNo)
      }
      members.push({ outbound, delivery, stock: await stock(member) })
    }
    const value = { trip, dispatch, items, task, members }; proof.snapshots[label] = value; return value
  }
  const pick = async (member, index) => {
    const before = await snapshot(`BEFORE_PICK_${index}`), item = before.task.items.find(i => id(i.dispatchItemId) === member.dispatchItemId)
    check('PDA exposes exactly this customer real current-run pallet and position', integer(item.scheduledPackages) === 1 && integer(item.pickedPackages) === 0
      && item.palletNo === member.palletNo && item.positions?.length === 1 && id(item.positions[0].positionId) === member.positionId
      && integer(item.positions[0].availablePackages) === 1)
    const body = { recordVersion: before.task.recordVersion, positionId: member.positionId, packages: 1, productQuantity: 1,
      idempotencyKey: key(`PICK-${index}`), remark: context.marker }
    const pickPath = `${pdaPath}/items/${encode(member.dispatchItemId)}/pick`
    await request('POST', pickPath, body, 'Pick this customer only', true)
    const after = await snapshot(`AFTER_PICK_${index}`)
    equal(before.members[1-index].stock, after.members[1-index].stock, 'Other customer inventory unchanged by picking')
    const expected = structuredClone(before.members[index].stock)
    for (const collection of ['balances', 'positions']) expected[collection].find(row => id(row.id) === (collection === 'balances' ? member.balanceId : member.positionId)).availablePackages = 0
    const counts = raw => Object.fromEntries(['balances','positions'].map(collection => [collection, raw[collection].map(row => ({ id: id(row.id),
      ...Object.fromEntries((collection === 'balances' ? balanceBuckets : buckets).map(field => [field, integer(row[field])])) }))]))
    equal(counts(expected), counts(after.members[index].stock), 'Only selected customer position and balance lose one box')
    const old = before.members[index].stock.ledger, newRows = after.members[index].stock.ledger.filter(row => !old.some(prior => id(prior.id) === id(row.id)))
    equal(old, after.members[index].stock.ledger.filter(row => old.some(prior => id(prior.id) === id(row.id))), 'Immutable old movements preserved')
    check('Single ALLOCATE/PICK/SHIP sequence has exact source and quantity', newRows.length === 3 && ['ALLOCATE','PICK','SHIP'].every(type => newRows.filter(row =>
      row.businessType === type && id(row.positionId) === member.positionId && row.businessOrderNo === fixture.dispatchNo && Number(row.deltaPackages) === (type === 'SHIP' ? -1 : 1)).length === 1))
    await request('POST', pickPath, body, 'Replay exact customer pick', true)
    equal(after, await snapshot(`PICK_REPLAY_${index}`), 'Pick replay changes neither customer nor any execution record')
  }
  try {
    let cargoBasis
    if (continuation) {
      const command = fixture.tripProof.loadingCommand
      const prior = priorProof.snapshots?.AFTER_PICK_1
      check('Continuation preserves original successful loading command and current-run picked inventory', priorProof.dispatchNo === fixture.dispatchNo && priorProof.marker === context.marker && priorProof.gateProof === 'PREREQUISITE_REJECTION_VERIFIED'
        && canonical(priorProof.members) === canonical(proof.members)
        && priorProof.requests.every(row => row.method === 'GET' || row.status === 'RESPONDED')
        && Number(prior?.dispatch?.counts?.pickedPackageCount) === 2 && id(command?.body?.recordVersion) === id(prior?.dispatch?.recordVersion)
        && canonical(fixture.tripProof.loadingPreview?.pallets?.map(p=>({warehousePalletId:id(p.warehousePalletId),palletSpaces:1,weightKg:p.weightKg??null,volumeCbm:p.volumeCbm??null,deliveryIndex:p.deliveryIndex}))) === canonical(command?.result?.result?.confirmedLoad?.pallets)
        && prior?.dispatch?.dispatchNo === fixture.dispatchNo && prior.dispatch.stage === 'LOADING'
        && command?.body?.idempotencyKey && command.result?.action === 'LOADING_COMPLETE' && command.result.stage === 'DELIVERING'
        && command.result.dispatchNo === fixture.dispatchNo && command.result.idempotencyKey === command.body.idempotencyKey
        && command.result.result?.confirmedLoad?.manifestHash === command.body.manifestHash)
      const current = await snapshot('CONTINUATION_INITIAL')
      check('Continuation is already delivering with the original automatic documents', current.dispatch.stage === 'DELIVERING'
        && current.task.stage === 'DELIVERING' && current.trip.trip.status === 'IN_TRANSIT'
        && id(current.dispatch.recordVersion) === id(command.result.recordVersion) && Number(current.dispatch.counts.shippedPackageCount) === 2
        && canonical(current.items.items) === canonical(prior.items.items)
        && current.dispatch.documents?.outbounds?.length === 2 && current.dispatch.documents?.deliveries?.length === 2)
      equal(prior.members.map(row => row.stock), current.members.map(row => row.stock), 'Continuation preserves complete picked inventory')
      proof.continued = true; proof.gateProof = priorProof.gateProof
      proof.continuation = { mode: 'EXPLICIT_COMPLETED_LOADING', gateProofSource:'PRIOR_PROOF', originalAfterPickSnapshot:structuredClone(prior), priorProofSHA256: createHash('sha256').update(canonical(priorProof)).digest('hex') }
      cargoBasis = prior
    } else {
      const initial = await snapshot('INITIAL')
    check('Fresh native fixture is reviewed LOADING with both customer receipts untouched', initial.trip.trip.status === 'PLANNED'
      && initial.dispatch.status === 'ACTIVE' && initial.dispatch.stage === 'LOADING' && initial.dispatch.reviewState === 'APPROVED'
      && initial.dispatch.reviews?.dispatchReview?.status === 'APPROVED' && initial.dispatch.reviews?.loadingReview?.status === 'APPROVED'
      && integer(initial.dispatch.counts.packageCount) === 2 && integer(initial.dispatch.counts.pickedPackageCount) === 0
      && initial.dispatch.documents?.deliveries?.length === 0 && initial.members.every((row,index) =>
        integer(row.stock.positions.find(p => id(p.id) === fixture.members[index].positionId).availablePackages) === 1
        && row.stock.balances.every(balance => balanceBuckets.filter(field => field !== 'availablePackages').every(field => integer(balance[field]) === 0))
        && row.stock.balances.reduce((sum,balance) => sum + integer(balance.availablePackages), 0) === 1))
    await pick(proof.members[0], 0)
    const rejectedBefore = await snapshot('GATE_BEFORE')
    check('Precisely one customer picked while main remains LOADING', integer(rejectedBefore.dispatch.counts.pickedPackageCount) === 1
      && rejectedBefore.dispatch.stage === 'LOADING' && rejectedBefore.task.pickingReconciled === false)
    const preview = await request('GET', `${pdaPath}/loading`, undefined, 'Read incomplete native loading manifest', true)
    await request('POST', `${pdaPath}/loading-complete`, { recordVersion: rejectedBefore.task.recordVersion, idempotencyKey: key('UNREADY-LOAD'),
      manifestHash: preview.manifestHash, remark: context.marker }, 'Unready second customer blocks explicit loading', true,
      message => /拣货.*核平|尚未核平/.test(message))
    equal(rejectedBefore, await snapshot('GATE_AFTER'), 'Rejected departure leaves both complete inventories/outbounds/deliveries/trip unchanged')
    proof.gateProof = 'PREREQUISITE_REJECTION_VERIFIED'
    await pick(proof.members[1], 1)
      cargoBasis = proof.snapshots.AFTER_PICK_1
      await confirmFleetLoading(context, MODULE, fixture.tripProof, fixture.pdaToken, fixture.loadingMeasurements, async command => {
        if (verifyDepartureRollback === undefined) return
        check('Explicit rollback verifier is callable', typeof verifyDepartureRollback === 'function')
        const ready = await snapshot('READY')
        const rollback = await verifyDepartureRollback({ context, fixture, ready, departPath: command.path, departBody: command.body, 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'), 'Rollback restores both complete inventory snapshots before loading completion')
        proof.rollbackEvidence = rollback; proof.midTransactionRollback = 'VERIFIED'
      })
    }
    const detail = await request('GET', path, undefined, 'Read documents created by the one loading-complete command')
    const documents = detail.documents
    check('Automatic loading creates both customer-isolated outbounds and deliveries', detail.dispatchNo === fixture.dispatchNo && detail.stage === 'DELIVERING'
      && documents?.deliveries?.length === 2 && documents.outbounds?.length === 2
      && new Set(documents.deliveries.map(d => d.deliveryNo)).size === 2 && new Set(documents.outbounds.map(o => o.outboundNo)).size === 2)
    for (const [index, member] of proof.members.entries()) {
      const deliveries = documents.deliveries.filter(d => id(d.customerId) === member.customerId)
      check('Each customer has its own source destination and quantity', deliveries.length === 1 && deliveries[0].status === 'DELIVERING'
        && integer(deliveries[0].packageCount) === 1 && deliveries[0].destination?.key === cargoBasis.items.items.find(i => id(i.customerId) === member.customerId)?.destination?.key
        && documents.outbounds.some(o => id(o.customerId) === member.customerId && o.status === 'OUTBOUNDED' && o.outboundNo === deliveries[0].outboundNo && integer(o.packageCount) === 1))
      Object.assign(member, { outboundNo: deliveries[0].outboundNo, deliveryNo: deliveries[0].deliveryNo })
      Object.assign(fixture.members[index], { outboundNo: member.outboundNo, deliveryNo: member.deliveryNo })
    }
    const command = fixture.tripProof.loadingCommand, departed = command.result
    proof.departure = departed
    const after = await snapshot('DEPARTED')
    check('Both customer outbounds completed and main delivering', after.trip.trip.status === 'IN_TRANSIT' && after.dispatch.stage === 'DELIVERING'
      && after.members.every(row => row.outbound.outbound.status === 'OUTBOUNDED' && row.delivery.order.status === 'DELIVERING'))
    equal(cargoBasis.members.map(row => row.stock), after.members.map(row => row.stock), 'Loading completion does not deduct either inventory again')
    const replay = await request('POST', command.path, command.body, 'Original PDA actor exact replay of loading completion', true)
    check('Loading replay returns the original complete result', replay.replayed === true && id(replay.recordVersion) === id(departed.recordVersion)
      && canonical(replay.result) === canonical(departed.result))
    const replayed = await snapshot('DEPART_REPLAY')
    equal(after, replayed, 'Loading replay preserves both complete inventories and execution documents')
    await request('POST', `${path}/loading-complete`, command.body, 'Different platform actor cannot reuse the PDA loading key', false, 'ACTOR_FORBIDDEN')
    equal(replayed, await snapshot('FOREIGN_ACTOR_DEPART_REJECTED'), 'Actor rejection preserves both complete inventories and execution facts')
    for (const [index, member] of proof.members.entries()) {
      const before = await snapshot(`BEFORE_SIGN_${index}`), documentPath = scoped(`${ROOT}/delivery-orders/${encode(member.deliveryNo)}/documents`, member.customerId)
      const docs = await request('GET', documentPath, undefined, 'Read this customer delivery document version')
      const uploadPath = scoped(`${ROOT}/delivery-orders/${encode(member.deliveryNo)}/documents/RECEIPT/files`, member.customerId)
      const uploadEntry = { method: 'POST', path: uploadPath, label: 'Upload own synthetic receipt', status: 'SENDING' }; proof.requests.push(uploadEntry); await context.checkpoint?.()
      let uploaded
      try { uploaded = await context.requestMultipart('POST', uploadPath, { name: `${context.marker}-${index}-TEST-receipt.txt`, type: 'text/plain',
        content: new TextEncoder().encode(`${context.marker} synthetic TEST receipt ${member.deliveryNo}`) }) }
      catch (error) { uploadEntry.status = 'UNKNOWN_RESULT'; await context.checkpoint?.(); throw error }
      Object.assign(uploadEntry, { status: 'RESPONDED', http: uploaded?.http, code: uploaded?.body?.code, requestId: uploaded?.requestId, traceId: uploaded?.traceId }); await context.checkpoint?.()
      check('Receipt uploaded through correct customer endpoint', ok(uploaded) && uploaded.requestId && uploaded.traceId && id(uploaded.body?.data?.fileId))
      const documents = (docs.categories || []).filter(row => row.category !== 'RECEIPT').map(row => ({ category: row.category, fileIds: (row.files || []).map(file => file.fileId) }))
      documents.push({ category: 'RECEIPT', fileIds: [id(uploaded.body.data.fileId)] })
      await request('PUT', documentPath, { expectedVersion: docs.version, idempotencyKey: key(`SIGN-${index}`), bolNo: docs.bolNo || null, documents }, 'Attach receipt to independently complete this delivery')
      const signed = await snapshot(`SIGNED_${index}`)
      check('This customer independently signed', signed.members[index].delivery.order.status === 'DELIVERED')
      equal(before.members[1-index].delivery, signed.members[1-index].delivery, 'Other customer delivery is not signed or changed implicitly')
      equal(before.members.map(row => row.stock), signed.members.map(row => row.stock), 'Signing does not mutate either customer inventory')
    }
    const started = Date.now()
    proof.completionPolling = { ...polling, status: 'WAITING', attempts: [] }
    for (let attempt = 1; attempt <= polling.maxAttempts; attempt++) {
      if (Date.now() - started >= polling.deadlineMs) break
      const entry = { attempt, elapsedMs: Date.now() - started, status: 'READING' }
      proof.completionPolling.attempts.push(entry)
      const current = await request('GET', path, undefined, 'Read bounded automatic delivery completion progress')
      check('Completion progress retains the exact normal main, counts and both execution-document owners', current.dispatchNo === fixture.dispatchNo
        && current.status === 'ACTIVE' && ['DELIVERING','DELIVERED'].includes(current.stage)
        && /^(0|[1-9]\d*)$/.test(id(current.recordVersion)) && integer(current.counts?.packageCount) === 2
        && integer(current.counts?.pickedPackageCount) === 2 && integer(current.counts?.shippedPackageCount) === 2
        && current.documents?.outbounds?.length === 2 && current.documents?.deliveries?.length === 2
        && proof.members.every(member => current.documents.outbounds.some(row => id(row.customerId) === member.customerId && row.outboundNo === member.outboundNo)
          && current.documents.deliveries.some(row => id(row.customerId) === member.customerId && row.deliveryNo === member.deliveryNo && row.outboundNo === member.outboundNo)))
      Object.assign(entry, { status: 'RESPONDED', stage: current.stage, recordVersion: current.recordVersion, elapsedMs: Date.now() - started })
      if (entry.elapsedMs >= polling.deadlineMs) break
      if (current.stage === 'DELIVERED') { proof.completionPolling.status = 'COMPLETED'; break }
      await context.checkpoint?.()
      const remaining = polling.deadlineMs - (Date.now() - started)
      if (attempt < polling.maxAttempts && remaining > 0) await new Promise(resolve => setTimeout(resolve, Math.min(polling.intervalMs, remaining)))
    }
    proof.completionPolling.elapsedMs = Date.now() - started
    if (proof.completionPolling.status !== 'COMPLETED') {
      proof.completionPolling.status = 'TIMED_OUT'
      await context.checkpoint?.()
      throw new Error('Automatic delivery completion did not reach DELIVERED within the explicit read-only polling bounds')
    }
    const final = await snapshot('FINAL')
    check('All normal customer signatures aggregate native delivery completion', final.dispatch.status === 'ACTIVE' && final.dispatch.stage === 'DELIVERED')
    proof.status = 'VERIFIED'; return proof
  } catch (error) {
    proof.status = 'FAILED'; proof.error = error.message
    if (proof.completionPolling?.status === 'WAITING') {
      proof.completionPolling.status = 'FAILED'; proof.completionPolling.error = error.message
      const attempt = proof.completionPolling.attempts.at(-1)
      if (attempt?.status === 'READING') attempt.status = 'FAILED'
    }
    throw error
  }
}