import { openFixedActor, requireFixedActors, COST_ACTOR_SLOTS } from './admin-test-env-modules/fixed-test-actors.mjs'
import { mkdirSync, writeFileSync, openSync, closeSync, unlinkSync, readFileSync, fsyncSync } from 'node:fs'
import { resolve, dirname, basename, relative, isAbsolute, join } from 'node:path'
import { fileURLToPath, pathToFileURL } from 'node:url'
import { createHash, randomUUID } from 'node:crypto'
import { AsyncLocalStorage } from 'node:async_hooks'
import { AcceptanceRun, parseArguments, validateExecutionOptions, buildAcceptanceSummary } from './admin-test-env-acceptance.mjs'
import { redactEvidence } from './fba-core-regression-test-env.mjs'
import { createTestEvidenceReaders, buildEvidenceRemoteScript, validateEvidenceConfig } from './fba-core-regression-evidence.mjs'
import { prepareFleetTrip, finishFleetTrip } from './admin-test-env-modules/fleet-trip-fulfillment.mjs'
import { createBody, candidatePage, markerCandidates, dispatchIo, readDispatch, version as dispatchVersion, replay as replayDispatch } from './admin-test-env-modules/grouped-dispatch-acceptance.mjs'
import { verifyFleetBookingReservation } from './admin-test-env-modules/fleet-booking-reservation-acceptance.mjs'
import { verifyFleetFeeCoverage } from './admin-test-env-modules/fleet-fee-coverage-acceptance.mjs'
import { verifyFleetExternalReceipt } from './admin-test-env-modules/fleet-external-receipt-acceptance.mjs'
import { verifyFleetExternalDispatch } from './admin-test-env-modules/fleet-external-dispatch-acceptance.mjs'
import { verifyFleetHandover, cleanupFleetHandoverResources } from './admin-test-env-modules/fleet-handover-acceptance.mjs'
import { verifyFleetTripCosts } from './admin-test-env-modules/fleet-trip-cost-acceptance.mjs'
import { verifyFleetReturnRedispatch, waitForRedispatchSignatureAudit, selectReturnSources } from './admin-test-env-modules/fleet-return-redispatch-acceptance.mjs'
import { resolvePickerAccess, createPicker, disablePicker, restoreFixedPicker } from './admin-test-env-modules/phase4-overseas-pda.mjs'
import { saveReceiptAndComplete } from './admin-test-env-modules/phase5-overseas-fulfillment.mjs'

const ROOT = resolve(dirname(fileURLToPath(import.meta.url)), '..')
const MODULE = 'fleet-specialized-acceptance'
const OPTIONS = { scenario: MODULE, deploymentBlocker: true }
const id = value => value == null ? '' : String(value)
const success = response => response?.http === 200 && id(response?.body?.code) === '200'
const key = () => `fleet-${randomUUID()}`
const NATIVE_PROTOCOL = 'NATIVE_DISPATCH_V1'
const nativeDeparture = (value, dispatchNo) => value?.action === 'LOADING_COMPLETE' && value.dispatchNo === dispatchNo
  && value.status === 'ACTIVE' && value.stage === 'DELIVERING' && value.result?.loadingCompletionHash
  && Number.isFinite(Date.parse(value.result.loadingCompletedAt)) && value.result.confirmedLoad?.manifestHash
const instant = value => typeof value === 'string' && /^\d{4}-\d\d-\d\dT\d\d:\d\d(?::\d\d(?:\.\d+)?)?$/.test(value) ? `${value}Z` : value
const sameInstant = (a, b) => Number.isFinite(Date.parse(instant(a))) && Date.parse(instant(a)) === Date.parse(instant(b))
const deliveryFacts = rows => JSON.stringify(rows.map(row => [id(row.customerId), row.deliveryNo, row.outboundNo, Number(row.packageCount)]).sort())

export function nativeFleetDeliveries(detail) {
  if (!detail?.dispatchNo || !Array.isArray(detail.documents?.deliveries) || !detail.documents.deliveries.length) throw Error('Native delivery references missing')
  const rows = new Map(), owners = new Map()
  for (const row of detail.documents.deliveries) {
    if (!row.deliveryNo || !row.outboundNo || !id(row.customerId) || !Number.isSafeInteger(Number(row.packageCount)) || Number(row.packageCount) <= 0)
      throw Error('Native delivery customer, business identity or quantity missing')
    if (owners.has(row.deliveryNo) && owners.get(row.deliveryNo) !== id(row.customerId)) throw Error('Delivery cannot belong to two customers')
    const identity = `${row.customerId}:${row.deliveryNo}`, prior = rows.get(identity)
    if (prior && (prior.outboundNo !== row.outboundNo || Number(prior.packageCount) !== Number(row.packageCount) || prior.groupKey !== row.groupKey))
      throw Error('Duplicate native delivery references disagree')
    owners.set(row.deliveryNo, id(row.customerId)); rows.set(identity, row)
  }
  return [...rows.values()]
}

export function selectNativeFleetFullSource(context, original, candidates) {
  const source = context.createdResources?.phase4OverseasDispatch
  if (source?.marker !== context.marker || source.status !== 'VERIFIED' || original.data.dispatchNo !== source.dispatchNo
    || original.data.status !== 'ACTIVE' || original.data.stage !== 'BEFORE_LOADING' || Number(original.data.counts.pickedPackageCount) !== 0
    || Number(original.data.counts.shippedPackageCount) !== 0 || id(original.data.recordVersion) !== id(source.recordVersion)
    || original.items.items.length < 2 || new Set(original.items.items.map(i => i.destination?.key)).size !== 2)
    throw Error('FULL/PARTIAL require the unchanged current-run two-destination native main')
  const history = source.routeLifecycle?.stages?.find(s => s.label === '新增同目的地第二托')?.snapshot?.items
  if (!history?.length) throw Error('Current-run released duplicate source evidence missing')
  const own = original.items.items
  const eligible = markerCandidates(context, candidates).filter(c => id(c.customerId) === id(source.customerId)
    && id(c.warehouseId) === id(original.warehouseId) && history.some(i => i.sourceKey === c.sourceKey
      && id(i.warehousePalletId) === id(c.selection.warehousePalletId))
    && own.every(i => i.sourceKey !== c.sourceKey && id(i.warehousePalletId) !== id(c.selection.warehousePalletId)))
  if (eligible.length !== 1) throw Error('Need exactly one current-run released, free, non-overlapping FULL pallet; never borrow historical inventory')
  return eligible[0]
}

/** Install before login/prerequisites. A caught unknown write must not permit another request. */
export function guardFleetWrites(context, checkpoint, { guardFetch = false } = {}) {
  let stopped = null, sequence = 0
  const stop = entry => {
    context.createdResources ??= {}
    context.createdResources.fleetWriteBoundary = { status: 'STOPPED', method: entry.method, path: entry.path,
      outcome: entry.status, requestId: entry.requestId, traceId: entry.traceId }
  }
  const scope = new AsyncLocalStorage(), originalFetch = globalThis.fetch
  const direct = { async requestWithBody(method, path, body, token, headers, request) {
    const nativeResponse = await originalFetch(request), copy = nativeResponse.clone()
    const bytes = new Uint8Array(await copy.arrayBuffer()), contentType = nativeResponse.headers.get('content-type') || ''
    const content = /json|text|xml|html/i.test(contentType) ? new TextDecoder().decode(bytes) : ''
    let payload = null
    try { payload = content ? JSON.parse(content) : null } catch { /* Keep bounded diagnostic only. */ }
    return { nativeResponse, http: nativeResponse.status, body: payload, bytes, contentType, raw: payload || !content ? null : content.slice(0, 500),
      requestId: payload?.requestId || nativeResponse.headers.get('x-request-id'),
      traceId: payload?.traceId || nativeResponse.headers.get('x-trace-id') }
  } }
  const targets = ['request', 'requestWithBody', 'requestBinary'].map(name => [context, name])
  if (guardFetch) targets.push([direct, 'requestWithBody'])
  for (const [owner, name] of targets) {
    const original = owner[name]
    if (typeof original !== 'function') continue
    owner[name] = async function (method, path, body, token, headers) {
      if (stopped) throw stopped
      if (['GET', 'HEAD', 'OPTIONS'].includes(String(method).toUpperCase()))
        return scope.run(true, () => original.apply(this, arguments))
      const entry = { sequence: ++sequence, method, path, status: 'SENDING', startedAt: new Date().toISOString(),
        headers, body }
      let session = owner === direct ? String(headers?.authorization || '').replace(/^Bearer\s+/i, '')
        : token === undefined ? this.token : token
      if (owner !== direct && this.pdaToken && /\/pda-loading\/tasks\/\d+/.test(String(path))) session = this.pdaToken
      entry.actor = session && session === context.pdaToken ? { kind: 'PDA', source: 'CONTEXT_PDA_SESSION' }
        : session && session === context.token && context.options?.username ? { kind: 'PLATFORM', source: 'CONTEXT_ADMIN_SESSION' }
          : { kind: 'UNKNOWN', source: session ? 'UNRESOLVED_SESSION' : 'NO_SESSION' }
      // Multipart delegates to requestWithBody, so it is journaled exactly once, after construction.
      if (body instanceof FormData) {
        entry.body = { multipart: [] }
        for (const [field, value] of body.entries()) entry.body.multipart.push(typeof value === 'string'
          ? { field, value } : { field, name: value.name, type: value.type,
            base64: Buffer.from(await value.arrayBuffer()).toString('base64') })
      }
      try { await checkpoint(entry) } catch (error) { entry.status = 'NOT_SENT'; stopped = error; stop(entry); throw error }
      try {
        const response = await scope.run(true, () => original.apply(this, arguments))
        const http = response?.http, payload = response?.body, code = id(payload?.code)
        const expectsPdf = name === 'requestBinary' || owner === direct && /(?:^|,)\s*application\/pdf(?:\s*[;,]|$)/i.test(headers?.accept || '')
        const binary = (name === 'requestBinary' || owner === direct) && response?.bytes instanceof Uint8Array
        const pdf = expectsPdf && binary && http === 200 && /^application\/pdf(?:\s*;|$)/i.test(response.contentType || '')
          && response.bytes.length > 4 && Buffer.from(response.bytes.subarray(0, 5)).toString('ascii') === '%PDF-'
        const definite = Number.isInteger(http) && http >= 200 && http < 500
          && response.requestId && response.traceId
          && (pdf || payload && typeof payload === 'object' && !Array.isArray(payload)
            && (!expectsPdf && http === 200 && code === '200' || /^[A-Z]\d{4}$/.test(code) && (http === 200 || http >= 400)))
        Object.assign(entry, { status: definite ? 'RESPONDED' : 'UNKNOWN_RESULT', http, code,
          requestId: response?.requestId, traceId: response?.traceId, raw: boundedFleetDiagnostic(response?.raw),
          responseBody: redactFleetWrite(response?.body), finishedAt: new Date().toISOString() })
        if (binary) entry.binary = { contentType: response.contentType, bytes: response.bytes.length,
          sha256: createHash('sha256').update(response.bytes).digest('hex'), validPdf: Boolean(pdf) }
        if (!definite) { stopped = Error(`Unknown write outcome: ${method} ${path}; preserve original request, do not resend`); stop(entry) }
        await checkpoint(entry)
        if (stopped) throw stopped
        return response
      } catch (error) {
        stopped ||= Error(boundedFleetDiagnostic(error instanceof Error ? error.message : String(error)) || 'Fleet write failed')
        if (entry.status === 'SENDING') {
          entry.status = 'UNKNOWN_RESULT'; entry.finishedAt = new Date().toISOString()
          entry.raw = boundedFleetDiagnostic(error instanceof Error ? error.message : String(error))
          stop(entry); await checkpoint(entry)
        }
        stop(entry)
        throw stopped
      }
    }
  }
  if (guardFetch) globalThis.fetch = async (input, init) => {
    if (scope.getStore()) return originalFetch(input, init)
    if (stopped) throw stopped
    const request = new Request(input, init), url = new URL(request.url)
    if (url.origin !== new URL(context.options.gateway).origin) return originalFetch(input, init)
    if (['GET', 'HEAD', 'OPTIONS'].includes(request.method)) return originalFetch(input, init)
    const type = request.headers.get('content-type') || ''
    let body
    if (request.body) body = type.includes('multipart/form-data') ? await request.clone().formData()
      : type.includes('application/json') ? await request.clone().json() : await request.clone().text()
    const result = await direct.requestWithBody(request.method, url.pathname + url.search, body, undefined,
      Object.fromEntries(request.headers), request)
    return result.nativeResponse
  }
  return () => { if (guardFetch) globalThis.fetch = originalFetch }
}

function boundedFleetDiagnostic(value) {
  return typeof value === 'string' ? value.replace(/Bearer\s+[^\s"'<>]+/gi, 'Bearer [REDACTED]')
    .replace(/((?:password|accessToken|refreshToken|authorization)["']?\s*[=:]\s*["']?)[^\s,;"'<>]+/gi, '$1[REDACTED]')
    .replace(/https?:\/\/[^\s"'<>]+/gi, '[REDACTED_URL]').slice(0, 500) : null
}

export function redactFleetWrite(value, secrets = []) {
  const sensitive = name => name !== 'draftToken' && /password|authorization|cookie|privateKey|api.?key|secret|token$|^(?:loginId|username)$/i.test(name)
  if (Array.isArray(value)) return value.map(item => redactFleetWrite(item, secrets))
  if (value && typeof value === 'object' && typeof value.field === 'string' && sensitive(value.field))
    return { field: value.field, value: '[REDACTED]' }
  if (value && typeof value === 'object') return Object.fromEntries(Object.entries(value).map(([name, item]) => [name,
    sensitive(name)
      ? '[REDACTED]' : redactFleetWrite(item, secrets)]))
  if (typeof value === 'string') return secrets.filter(secret => typeof secret === 'string' && secret.length)
    .reduce((text, secret) => text.split(secret).join('[REDACTED]'), value)
  return value
}

export function parseFleetArguments(args) {
  const config = { execute: false, environment: 'TEST', runId: 'plan', scenario: 'handover' }
  let action = null
  const seen = new Set()
  for (let index = 0; index < args.length; index++) {
    const name = args[index]
    if (['--plan', '--execute'].includes(name)) {
      if (action) throw Error('Choose exactly one plan or execute action')
      action = name; config.execute = name === '--execute'; continue
    }
    if (!['--environment', '--run-id', '--output', '--resume-report', '--resume-stage', '--resume-sha256', '--scenario'].includes(name) || !args[index + 1] || args[index + 1].startsWith('--')) throw Error('Unsupported or missing argument')
    if (seen.has(name)) throw Error('Duplicate argument')
    seen.add(name)
    const value = args[++index]
    if (name === '--environment') config.environment = value
    else if (name === '--run-id') config.runId = value
    else if (name === '--output') config.outputDir = value
    else if (name === '--resume-report') config.resumeReport = value
    else if (name === '--resume-sha256') config.resumeSha256 = value
    else if (name === '--scenario') config.scenario = value
    else config.resumeStage = value
  }
  if (config.environment !== 'TEST' || !/^[A-Za-z0-9][A-Za-z0-9_-]{0,47}$/.test(config.runId)
    || (config.execute && (!args.includes('--environment') || config.runId === 'plan'))) throw Error('Explicit TEST and fresh run-id required')
  if (Boolean(config.resumeReport) !== Boolean(config.resumeStage) || (config.resumeStage && !['FULL_COVERAGE', 'FULL_COVERAGE_PERSISTED', 'EXTERNAL_RECEIPT_READY', 'EXTERNAL_PALLETIZED', 'EXTERNAL_RECEIPT_CONFIRMED', 'TERMINAL_READONLY', 'DEPARTED_BEFORE_COSTS', 'DEPARTED_AFTER_COSTS_BEFORE_HANDOVER', 'DEPARTED_VEHICLE_CREATED', 'RETURN_SOURCE_SELECTION_BEFORE_RETURN'].includes(config.resumeStage))) throw Error('Explicit reviewed checkpoint required')
  if (config.resumeStage?.startsWith('DEPARTED_') || config.resumeStage === 'RETURN_SOURCE_SELECTION_BEFORE_RETURN' ? !/^[a-f0-9]{64}$/.test(config.resumeSha256 || '') : config.resumeSha256)
    throw Error('Departed recovery requires the exact reviewed report SHA256')
  if (config.outputDir && !/^reports\/[A-Za-z0-9][A-Za-z0-9_-]{0,95}$/.test(config.outputDir)) throw Error('Output must be a fresh named directory directly under reports')
  if (!['handover', 'return-redispatch'].includes(config.scenario) || (config.scenario === 'return-redispatch' && config.resumeReport && config.resumeStage !== 'RETURN_SOURCE_SELECTION_BEFORE_RETURN')
    || (config.resumeStage === 'RETURN_SOURCE_SELECTION_BEFORE_RETURN' && config.scenario !== 'return-redispatch'))
    throw Error('Return redispatch only accepts a fresh run, never a reviewed old checkpoint')
  return config
}
export function fleetExecutionPlan(config) {
  const afterCosts = ['DEPARTED_AFTER_COSTS_BEFORE_HANDOVER', 'DEPARTED_VEHICLE_CREATED'].includes(config.resumeStage)
  return { environment: 'TEST', execution: config.execute ? 'EXECUTE' : 'PLAN', runId: config.runId, scenario: config.scenario || 'handover',
    returnRedispatch: config.scenario === 'return-redispatch' ? { required: 'Two current-run domestic sources; A >=3 boxes; current departure covers A and B in full',
      stages: ['A actual return 2', 'original remaining signatures; A incomplete and B complete', 'RETURN_RECEIPT draft lock and pre-arrange', 'new PDA picks and DEPART 2', 'redispatch signature; A complete', 'source event/ACK read-only evidence remains required'] } : null,
    outputDir: config.outputDir || `reports/fleet-${config.runId}`, producer: config.resumeReport ? `Reviewed ${config.resumeStage} checkpoint in the same original fixture` : 'phase4-overseas-dispatch with its normal fresh prerequisites',
    resumeReport: config.resumeReport || null, resumeStage: config.resumeStage || null, resumeSha256: config.resumeSha256 || null,
    stages: config.resumeStage === 'RETURN_SOURCE_SELECTION_BEFORE_RETURN'
      ? ['Verify exact source-selection-failure report SHA256 and three definitive native delivery reads', 'Read-only original DEPART/handover/cost/cargo/resources and DELIVERING versions',
        'Recheck repaired A/B return candidates and complete source quantities before any write', 'Create fresh bound PDA session',
        'Original sample actual return and redispatch; no repeated original departure/handover/cost/start-delivery',
        'Wait for coordinator pre-signature A0/B1 event evidence gate', 'New signature and source ACK evidence; close new and original resources']
      : config.resumeStage?.startsWith('DEPARTED_')
      ? ['Verify reviewed report SHA256', afterCosts
        ? 'Read-only exact trip/version/DEPART/cargo, unchanged verified costs, empty handovers and unsigned deliveries'
        : 'Read-only exact original trip/version/DEPART/cargo and empty costs/handovers/unsigned deliveries',
        ...(config.resumeStage === 'DEPARTED_VEHICLE_CREATED' ? ['Verify exact successful vehicle creation, complete current vehicle snapshot and zero resource assignments', 'Explicitly reuse the created vehicle; skip its creation POST and continue from new driver'] : []),
        afterCosts ? 'Create fresh PDA session; retain verified costs' : 'Create fresh bounded cost/PDA sessions',
        afterCosts ? 'Match original verified cost snapshot; no cost commands' : 'Original trip operating costs with actual allocation availability',
        'Retain original DEPART; no picking or departure commands',
        'In-transit handover and PDA readback', 'Original delivery signatures and resource release', 'Close newly created test sessions']
      : ['two disjoint current-run native sources', 'independent FULL loading/coverage/departure/POD/resource closure', 'PARTIAL original main and reservation lifecycle',
      'external actual receipt', 'external dispatch and agreed fee', 'Finance consumer persistence', 'one native main PDA picking and explicit actual loading',
      'native current-version whole-main departure and cross-end exact replay', 'actual in-transit handover and PDA readback', 'same-fixture optional H5 checkpoint',
      'per-destination receipt and replacement-resource release', 'explicit handover-fixture cleanup'],
    boundaries: 'TEST only; serial fresh fixtures; no SQL writes, deployment or configuration changes. Failure retains original requests/resources. No automatic rerun/resume.',
    limits: 'Finance pool proves consumer persistence; producer ACK requires separate event evidence. UI and remaining fleet product scope are separate.' }
}

export function reviewedFullCheckpoint(report, stage = 'FULL_COVERAGE') {
  const resources = report?.createdResources, proof = resources?.fleetAcceptance, full = resources?.fleetFeeCoverageCases?.[0]
  const trip = resources?.fleetTrips?.find(row => row.tripNo === proof?.tripNo)
  if (proof?.protocol !== NATIVE_PROTOCOL || !proof.dispatchNo || !trip || trip.dispatchNo !== proof.dispatchNo
    || JSON.stringify(proof.dispatchNos) !== JSON.stringify([proof.dispatchNo]) || resources.dispatchWriteBoundary?.outcomeUnknown
    || resources.fleetWriteBoundary?.status === 'STOPPED'
    || (proof.departure && !nativeDeparture(proof.departure, proof.dispatchNo)))
    throw Error('Only current native single-dispatch reports may resume; legacy or unknown outcomes require explicit reconciliation')
  if (stage === 'RETURN_SOURCE_SELECTION_BEFORE_RETURN') {
    const attempt = trip?.returnRedispatch, originals = attempt?.originalDeliveries, writes = attempt?.requests?.filter(row => row.method !== 'GET')
    const failures = report?.steps?.filter(row => row.passed === false)
    if (report?.success !== false || proof?.stage !== 'RETURN_REDISPATCH' || proof.status !== 'CHECKING'
      || proof.marker !== attempt?.marker || trip?.costAcceptance?.status !== 'VERIFIED' || trip.handover?.status !== 'VERIFIED'
      || !nativeDeparture(proof.departure, proof.dispatchNo)
      || trip.closure || proof.closure || proof.receipts?.length || attempt.status !== 'FAILED'
      || attempt.failure !== 'Fresh redispatch fixture requires source A with at least 3 boxes and a separate domestic source B'
      || attempt.exceptionNo || attempt.sources || attempt.newTripNo || attempt.originalSignatures
      || originals?.length !== 3 || new Set(originals.map(row => row.deliveryNo)).size !== 3 || writes?.length !== 0
      || originals.some(row => id(row.customerId) !== id(proof.customerId) || id(row.overseasWarehouseId) !== id(proof.warehouseId)
        || row.dispatchNo !== proof.dispatchNo || row.status !== 'DELIVERING' || row.signedAt || Number(row.signedPackages || 0))
      || attempt.requests.some(row => row.status !== 'RESPONDED' || row.http !== 200 || id(row.code) !== '200' || !row.requestId || !row.traceId)
      || attempt.requests.length !== 3 || attempt.requests.some(row => {
        const original = originals.find(item => item.deliveryNo === row.result?.order?.deliveryNo)
        return row.method !== 'GET' || Object.hasOwn(row.result || {}, 'replayed') || !original
          || row.path !== `/api/platform/v1/last-leg/delivery-orders/${original.deliveryNo}?customerId=${proof.customerId}`
          || JSON.stringify(row.result.order) !== JSON.stringify(original)
      }) || new Set(attempt.requests.map(row => row.result?.order?.deliveryNo)).size !== 3
      || resources.fleetCostActors?.status !== 'SESSIONS_CLOSED' || failures?.length !== 1 || failures[0].name !== '模块执行'
      || failures[0].evidence !== attempt.failure)
      throw Error('Return source recovery requires the exact native candidate failure, three already-delivering documents and no return/signature writes')
    return { resources, marker: proof.marker, returnSourceSelection: true }
  }
  if (stage === 'DEPARTED_BEFORE_COSTS' || stage === 'DEPARTED_AFTER_COSTS_BEFORE_HANDOVER' || stage === 'DEPARTED_VEHICLE_CREATED') {
    const vehicleCreated = stage === 'DEPARTED_VEHICLE_CREATED'
    const afterCosts = stage !== 'DEPARTED_BEFORE_COSTS'
    const costs = trip?.costAcceptance, failures = report?.steps?.filter(row => row.passed === false)
    const failureName = vehicleCreated ? '接替车辆返回真实身份及车牌' : afterCosts ? '每张排车已发车且货量与客户起运仓一致' : '独立预期覆盖全部实物且没有重叠托盘'
    const handover = trip?.handover
    const creation = vehicleCreated ? handover?.requests?.filter(row => row.method !== 'GET') : null
    if (vehicleCreated && (creation?.length !== 1 || creation[0].method !== 'POST'
      || creation[0].path !== '/api/platform/v1/last-leg/master-data/vehicles' || creation[0].status !== 'RESPONDED'
      || creation[0].http !== 200 || id(creation[0].code) !== '200' || !creation[0].requestId || !creation[0].traceId
      || !/^[1-9]\d*$/.test(id(creation[0].result?.id)) || id(creation[0].result.recordVersion) !== '0'
      || creation[0].result.plateNumber !== creation[0].body?.plateNumber?.toUpperCase()
      || creation[0].result.status !== 'IN_SERVICE' || handover.replacement?.driverId))
      throw Error('Created vehicle recovery requires the exact definitive successful creation and no later handover write')
    if (report?.success !== false || proof?.stage !== (afterCosts ? 'IN_TRANSIT_HANDOVER' : 'TRIP_OPERATING_COSTS') || proof.status !== 'CHECKING'
      || proof.marker !== costs?.marker || !nativeDeparture(proof.departure, proof.dispatchNo)
      || !proof.departureRequest?.idempotencyKey || !costs?.departure?.snapshot?.confirmedLoad?.manifestHash
      || costs.status !== (afterCosts ? 'VERIFIED' : 'FAILED') || !Array.isArray(costs.costs) || costs.costs.length !== (afterCosts ? 2 : 0) || !costs.requests?.length
      || costs.requests.some(row => row.status !== 'RESPONDED' || !row.requestId || !row.traceId
        || (afterCosts ? ![200, 403].includes(row.http) || !['200', 'A0403', 'A0409'].includes(id(row.code)) : row.method !== 'GET' || row.http !== 200 || id(row.code) !== '200'))
      || (afterCosts ? handover?.status !== 'FAILED' || !handover.requests?.length || handover.confirmRequest || handover.selection
        || handover.requests.some(row => (row.method !== 'GET' && row !== creation?.[0]) || row.status !== 'RESPONDED' || row.http !== 200 || id(row.code) !== '200')
        || !costs.final?.page || costs.final.page.records?.length !== 2 : trip.handover)
      || trip.closure || proof.receipts || proof.closure || trip.returnRedispatch
      || resources.fleetCostActors?.status !== 'SESSIONS_CLOSED' || !proof.picker?.membershipId
      || failures?.length !== 2 || !failures.some(row => row.name === failureName)
      || failures.some(row => ![failureName, '模块执行'].includes(row.name)))
      throw Error('Departed checkpoint must be the reviewed read-only cost assertion failure, with no later business writes')
    return { resources, marker: proof.marker, departed: true, costsAlreadyVerified: afterCosts,
      createdVehicle: vehicleCreated ? { snapshot: creation[0].result, creationRequest: creation[0] } : null }
  }
  if (stage === 'TERMINAL_READONLY') {
    const failures = report?.steps?.filter(row => row.passed === false)
    if (report?.success !== false || proof?.stage !== 'RECEIPTS_AND_RESOURCE_RELEASE'
      || trip?.closure?.status !== 'CARGO_COMPLETED_AND_RESOURCES_RELEASED'
      || proof.closure?.status !== trip.closure.status || trip.externalReceipt?.status !== 'VERIFIED'
      || trip.externalDispatch?.status !== 'VERIFIED' || trip.externalDispatch.dispatchNo !== proof.dispatchNo || trip.reassignment?.status !== 'REJECTED_WITHOUT_MUTATION'
      || !proof.receipts?.length || proof.finance?.status !== 'CONSUMER_PERSISTED'
      || !failures?.length || failures.some(row => !['本次车次货物结案及车辆司机释放完成', '模块执行'].includes(row.name)))
      throw Error('Terminal reconciliation requires completed original business proofs and only the known aggregate assertion failure')
    return { resources, marker: proof.marker, terminal: true }
  }
  if (stage === 'EXTERNAL_PALLETIZED' || stage === 'EXTERNAL_RECEIPT_CONFIRMED') {
    const confirmed = stage === 'EXTERNAL_RECEIPT_CONFIRMED'
    const partial = resources?.fleetFeeCoverageCases?.find(row => row.mode === 'PARTIAL'), receipt = trip?.externalReceipt
    if (report?.success !== false || proof?.stage !== 'EXTERNAL_RECEIPT' || proof.status !== 'CHECKING'
      || full?.proof?.status !== 'VERIFIED' || proof.fullExecution?.closure?.status !== 'CARGO_COMPLETED_AND_RESOURCES_RELEASED' || partial?.proof || partial?.dispatchNo !== trip?.dispatchNo
      || trip.reservation?.status !== 'VERIFIED' || receipt?.status !== 'CHECKING' || trip.externalDispatch
      || (confirmed ? (!receipt.receiptRequest?.idempotencyKey || !receipt.requests?.some(row => row.label === '运营按实际1托位和新单价30确认实收' && row.method === 'POST' && row.http === 200 && String(row.code) === '200')) : receipt.receiptRequest)
      || !receipt.palletizeRequest?.idempotencyKey || !receipt.warehousePalletId || !proof.picker?.membershipId || !proof.pickerMarker
      || !receipt.requests?.some(row => row.label === 'PDA正式完成外接收货及库存入账' && row.method === 'POST' && row.http === 200 && String(row.code) === '200'))
      throw Error('Palletized checkpoint requires the exact successful PDA command and no subsequent receipt confirmation')
    return { resources, marker: proof.marker, receiptReady: true, palletized: !confirmed, receiptConfirmed: confirmed }
  }
  if (stage === 'EXTERNAL_RECEIPT_READY') {
    const partial = resources?.fleetFeeCoverageCases?.find(row => row.mode === 'PARTIAL')
    if (report?.success !== false || proof?.stage !== 'RESERVATIONS' || proof.status !== 'CHECKING'
      || full?.proof?.status !== 'VERIFIED' || proof.fullExecution?.closure?.status !== 'CARGO_COMPLETED_AND_RESOURCES_RELEASED' || partial?.proof || partial?.dispatchNo !== trip?.dispatchNo
      || trip.reservation?.status !== 'VERIFIED' || trip.externalReceipt || trip.externalDispatch || proof.picker
      || !report.steps?.some(row => row.name === '创建海外PDA仓库操作员71' && row.code === 'A0409' && row.message === '手机号已存在'))
      throw Error('Receipt-ready checkpoint requires verified coverage/reservations and an explicitly rejected operator creation with no receipt writes')
    return { resources, marker: proof.marker, receiptReady: true }
  }
  if (report?.success !== false || proof?.stage !== 'FULL_COVERAGE' || proof.status !== 'CHECKING' || full?.mode !== 'FULL'
    || full.marker !== proof.marker || full.dispatchNo === trip?.dispatchNo || !resources.fleetTrips.some(t => t.dispatchNo === full.dispatchNo && t.tripNo === proof.fullTripNo)
    || !['CHECKING','UNPROVEN'].includes(full.proof?.status) || proof.fullExecution?.departureRequest
    || !full.proof.requests?.length
    || full.proof.requests.some(row => row.status !== 'RESPONDED' || !row.requestId || !row.traceId)
    || resources.fleetFeeCoverageCases.length !== 2 || !proof.originalDispatch?.detail || proof.originalDispatch.detail.dispatchNo !== proof.dispatchNo)
    throw Error('Only a reviewed read-only FULL_COVERAGE failure can resume; mutated/unknown outcomes require case-specific reconciliation')
  const persisted = stage === 'FULL_COVERAGE_PERSISTED'
  const fullTrip = resources.fleetTrips.find(row => row.dispatchNo === full.dispatchNo)
  if (persisted ? (!full.proof.outboundNo || !full.proof.deliveryNos?.length || !fullTrip?.loadingCommand?.body?.idempotencyKey
    || !fullTrip.loadingCommand.body.measurements?.length || fullTrip.loadingCommand.body.recordVersion == null
    || !nativeDeparture(fullTrip.loadingCommand.result, full.dispatchNo)
    || !fullTrip.loadingCommand.afterReplay?.main?.documents?.outbounds?.some(o => o.outboundNo === full.proof.outboundNo)
    || JSON.stringify(fullTrip.loadingCommand.afterReplay.main.documents.deliveries.map(d => d.deliveryNo).sort()) !== JSON.stringify([...full.proof.deliveryNos].sort()))
    : (full.proof.requests.some(row => row.method !== 'GET') || full.proof.outboundNo))
    throw Error('Mutated/unknown outcomes require the exact successful command and persisted document reconciliation')
  return { resources, marker: proof.marker, persisted }
}

export async function verifyFleetTerminalState(context) {
  const proof = context.createdResources.fleetAcceptance
  const trip = context.createdResources.fleetTrips.find(row => row.tripNo === proof.tripNo)
  const evidence = proof.terminalReadback = { status: 'CHECKING', requests: [], deliveries: [], dispatches: [] }
  const check = (label,passed) => { context.assert(MODULE,label,null,Boolean(passed),'',OPTIONS);if(!passed)throw Error(label) }
  async function get(path) { const r=await context.request('GET',path);evidence.requests.push({path,http:r.http,code:r.body?.code,requestId:r.requestId,traceId:r.traceId});check('终态只读回查成功',success(r)&&r.requestId&&r.traceId);return r.body.data }
  check('原生结案助手已完成货物与资源核对', proof.protocol===NATIVE_PROTOCOL && trip?.dispatchNo===proof.dispatchNo && trip.closure?.status==='CARGO_COMPLETED_AND_RESOURCES_RELEASED')
  const current=await get(`/api/platform/v1/last-leg/fleet/trips/${encodeURIComponent(trip.tripNo)}`)
  check('当前车次已完成且两项资源结束事实与原证明一致',current.trip?.status==='COMPLETED'
    && current.trip.completedAt===trip.closure.finalTrip.completedAt && id(current.trip.recordVersion)===id(trip.closure.finalTrip.recordVersion)
    && JSON.stringify(current.trip.vehicleRelease)===JSON.stringify(trip.closure.finalTrip.vehicleRelease)
    && JSON.stringify(current.trip.driverRelease)===JSON.stringify(trip.closure.finalTrip.driverRelease)
    && current.trip.vehicleRelease&&current.trip.driverRelease&&current.dispatches?.length===1&&current.dispatches[0].dispatchNo===proof.dispatchNo)
  evidence.trip=current.trip
  const detail=await get(`/api/platform/v1/last-leg/dispatches/${encodeURIComponent(proof.dispatchNo)}`)
  check('原生主单全部正常派送完成且数量守恒',detail.dispatchNo===proof.dispatchNo&&detail.status==='ACTIVE'&&detail.stage==='DELIVERED'
    && Number(detail.counts?.packageCount)===Number(detail.counts?.shippedPackageCount)&&Number(detail.counts?.pickedPackageCount)===Number(detail.counts?.shippedPackageCount)
    && Number(detail.counts?.shippedPackageCount)===Number(proof.departureSnapshot.detail.counts.shippedPackageCount))
  evidence.dispatches.push(detail)
  const refs=nativeFleetDeliveries(detail)
  check('签收证明恰好覆盖主单每客户派送一次',JSON.stringify(proof.receipts.map(r=>`${r.customerId}:${r.deliveryNo}`).sort())===JSON.stringify(refs.map(r=>`${r.customerId}:${r.deliveryNo}`).sort()))
  for(const ref of refs){const delivery=await get(`/api/platform/v1/last-leg/delivery-orders/${encodeURIComponent(ref.deliveryNo)}?customerId=${encodeURIComponent(id(ref.customerId))}`)
    check('每张原派送保留客户和真实签收时间',delivery.order?.deliveryNo===ref.deliveryNo&&id(delivery.order.customerId)===id(ref.customerId)&&delivery.order.status==='DELIVERED'
      && delivery.order.signedAt===proof.receipts.find(r=>r.deliveryNo===ref.deliveryNo&&r.customerId===id(ref.customerId)).signedAt);evidence.deliveries.push(delivery.order)}
  const probe=trip.closure.nextDispatch
  check('原生空排车占用保护证明完整',trip.closure.subsequentTripClaimProtection==='VERIFIED_NATIVE_EMPTY_DISPATCH'&&probe?.dispatchNo&&probe.finalStatus==='CANCELLED')
  const cancelled=await get(`/api/platform/v1/last-leg/dispatches/${encodeURIComponent(probe.dispatchNo)}`)
  check('本轮资源核对原生空排车已取消',cancelled.dispatchNo===probe.dispatchNo&&cancelled.status==='CANCELLED'&&Number(cancelled.counts?.packageCount)===0&&Number(cancelled.counts?.pickedPackageCount)===0)
  const customerId=id(proof.finance.customerId),pool=await get(`/api/platform/finance/v1/fee-pool?customerId=${encodeURIComponent(customerId)}&direction=RECEIVABLE`)
  const fees=pool.filter(row=>id(row.sourceId)===id(proof.finance.source.sourceFeeId)&&id(row.sourceVersion)===id(proof.finance.source.sourceVersion))
  check('原外接应收仍唯一且金额正确',fees.length===1&&Number(fees[0].amount)===30&&fees[0].currency==='CNY'&&id(fees[0].customerId)===customerId&&fees[0].sourceType==='OVERSEAS_DELIVERY_FEE')
  evidence.fees=fees;evidence.status='VERIFIED';proof.status='HTTP_CHAIN_VERIFIED_WITH_ACK_GAP';proof.stage='FINISHED';return proof
}
/** All business checks happen before creating replacement test sessions or issuing a business command. */
export async function verifyDepartedRecovery(context, { costsAlreadyVerified = false, createdVehicle, returnSourceReady = false } = {}) {
  const proof = context.createdResources.fleetAcceptance
  const trip = context.createdResources.fleetTrips.find(row => row.tripNo === proof.tripNo)
  const original = trip.costAcceptance.departure, snapshot = original.snapshot, load = snapshot.confirmedLoad
  const expectedTrip = returnSourceReady ? trip.handover.finalTrip : snapshot.trip
  const base = `/api/platform/v1/last-leg/fleet/trips/${encodeURIComponent(trip.tripNo)}`
  const evidence = { status: 'CHECKING', tripNo: trip.tripNo, requests: [], dispatchDetails: [], deliveries: [] }
  context.createdResources.fleetDepartureRecoveryVerification = evidence
  const check = (ok, message) => { if (!ok) throw Error(`Departed recovery refused: ${message}`) }
  const same = (a, b) => JSON.stringify(a) === JSON.stringify(b)
  async function get(path) {
    const r = await context.request('GET', path)
    evidence.requests.push({ path, http: r.http, code: r.body?.code, requestId: r.requestId, traceId: r.traceId })
    check(success(r) && r.requestId && r.traceId, 'read-only response not definitive')
    return r.body.data
  }
  const current = await get(base)
  check(current.trip?.status === 'IN_TRANSIT' && id(current.trip.recordVersion) === id(expectedTrip.recordVersion)
    && sameInstant(current.trip.departedAt, proof.departureSnapshot.detail.dispatchInfo.departedAt) && id(current.trip.warehouseId) === id(proof.warehouseId)
    && id(current.trip.vehicleId) === id(expectedTrip.vehicleId) && id(current.trip.driverId) === id(expectedTrip.driverId)
    && !current.trip.vehicleRelease && !current.trip.driverRelease, 'trip version, resources or departure changed')
  const history = await get(`${base}/history?page=1&size=100`)
  const departures = history.records?.filter(row => row.action === 'DEPART')
  check(Array.isArray(history.records) && Number(history.total) === history.records.length
    && departures?.length === 1 && same(departures[0], original)
    && history.records.every(row => BigInt(row.recordVersion) <= BigInt(expectedTrip.recordVersion)), 'immutable DEPART/history changed')
  if (returnSourceReady) {
    const expectedHistory = trip.handover.requests.filter(row => row.method === 'GET' && row.path.startsWith(`${base}/history?`)).at(-1)?.result
    check(expectedHistory && same(history.records, expectedHistory.records), 'verified handover history changed')
    check(typeof context.context?.readRecoveryTripResources === 'function', 'current resource occupancy reader missing')
    const bindings = await context.context.readRecoveryTripResources(trip.tripNo)
    check(bindings.length === 2 && bindings.every(row => row.tripNo === trip.tripNo)
      && bindings.some(row => row.resourceKind === 'VEHICLE' && id(row.resourceId) === id(expectedTrip.vehicleId))
      && bindings.some(row => row.resourceKind === 'DRIVER' && id(row.resourceId) === id(expectedTrip.driverId)), 'current resources no longer belong to the original trip')
    evidence.resourceBindings = bindings
    const vehicleSnapshot = trip.handover.requests.find(row => row.method === 'POST'
      && row.path === '/api/platform/v1/last-leg/master-data/vehicles' && id(row.result?.id) === id(expectedTrip.vehicleId))?.result
    const driverSnapshot = trip.handover.requests.find(row => row.method === 'POST'
      && row.path === `/api/platform/v1/last-leg/fleet/drivers/${expectedTrip.driverId}/review`)?.result
    check(vehicleSnapshot && driverSnapshot, 'verified current resource snapshots missing')
    const vehicle = await get(`/api/platform/v1/last-leg/master-data/vehicles/${expectedTrip.vehicleId}`)
    const driver = await get(`/api/platform/v1/last-leg/fleet/drivers/${expectedTrip.driverId}`)
    check(same(vehicle, vehicleSnapshot), 'current vehicle snapshot changed')
    check(['id', 'recordVersion', 'active', 'reviewStatus', 'licenseFileId', 'details'].every(field => same(driver[field], driverSnapshot[field])), 'current driver identity/version/qualification facts changed')
    evidence.currentVehicle = vehicle; evidence.currentDriver = driver
  }
  const costs = await get(`${base}/costs?page=1&size=100`), handovers = await get(`${base}/handovers`)
  check((costsAlreadyVerified ? same(costs, trip.costAcceptance.final)
    : costs.page?.records?.length === 0 && Number(costs.page.total) === 0 && costs.summary?.totals?.length === 0)
    && costs.summary.manifestHash === load.manifestHash && id(costs.summary.departureVersion) === id(original.recordVersion), 'costs already written or cargo basis changed')
  if (returnSourceReady) {
    const expected = trip.handover.requests.filter(row => row.method === 'GET' && row.path.startsWith(`${base}/handovers?`)).at(-1)?.result
    check(expected && Number(handovers.total) === 1 && same(handovers.records, expected.records), 'original verified handover changed')
  } else check(handovers.records?.length === 0 && Number(handovers.total) === 0, 'handover already exists')
  check(proof.protocol === NATIVE_PROTOCOL && nativeDeparture(proof.departure,proof.dispatchNo)
    && trip.dispatchNo === proof.dispatchNo && current.dispatches?.length === 1 && current.dispatches[0].dispatchNo === proof.dispatchNo
    && JSON.stringify(proof.dispatchNos) === JSON.stringify([proof.dispatchNo]), 'trip dispatch binding changed')
  const detail=await get(`/api/platform/v1/last-leg/dispatches/${encodeURIComponent(proof.dispatchNo)}`)
  const items=await get(`/api/platform/v1/last-leg/dispatches/${encodeURIComponent(proof.dispatchNo)}/items`)
  const cargo=load.cargo
  const facts=rows=>{
    const totals=new Map()
    for(const row of rows){const identity=JSON.stringify([id(row.dispatchItemId),id(row.warehousePalletId),id(row.customerId)]);totals.set(identity,(totals.get(identity)||0)+Number(row.scheduledPackages??row.packages))}
    return [...totals].sort(([a],[b])=>a.localeCompare(b))
  }
  check(detail.dispatchNo===proof.dispatchNo&&detail.status==='ACTIVE'&&detail.stage==='DELIVERING'&&items.dispatchNo===proof.dispatchNo
    && id(items.recordVersion)===id(detail.recordVersion)&&Number(detail.counts?.packageCount)===Number(detail.counts?.shippedPackageCount)
    && Number(detail.counts?.pickedPackageCount)===Number(detail.counts?.shippedPackageCount)&&cargo.every(row=>row.dispatchNo===proof.dispatchNo)
    && same(facts(items.items),facts(cargo))&&proof.departureSnapshot?.detail?.dispatchNo===proof.dispatchNo
    && same(items,proof.departureSnapshot.items),'dispatch shipped cargo changed')
  evidence.dispatchDetails.push({detail,items})
  const refs=nativeFleetDeliveries(detail)
  check(new Set(refs.map(d=>d.deliveryNo)).size===refs.length
    && deliveryFacts(refs)===deliveryFacts(nativeFleetDeliveries(proof.departureSnapshot.detail)),'duplicate or changed delivery identities')
  for(const ref of refs){
    const response=await get(`/api/platform/v1/last-leg/delivery-orders/${encodeURIComponent(ref.deliveryNo)}?customerId=${encodeURIComponent(id(ref.customerId))}`),order=response.order
    const expected=(returnSourceReady?trip.returnRedispatch.originalDeliveries:proof.departureDeliveries)?.find(d=>d.deliveryNo===ref.deliveryNo&&id(d.customerId)===id(ref.customerId))
    check(expected&&order?.deliveryNo===ref.deliveryNo&&id(order.customerId)===id(ref.customerId)&&order.dispatchNo===proof.dispatchNo
      && order.outboundNo===ref.outboundNo&&order.status==='DELIVERING'&&id(order.version)===id(expected.version)
      && order.outboundNo===expected.outboundNo&&Number(order.packageCount)===Number(ref.packageCount)&&Number(order.packageCount)===Number(expected.packageCount)
      && (!returnSourceReady||order.vehiclePlate===expectedTrip.vehiclePlate&&order.driverName===expectedTrip.driverName&&order.driverPhone===expectedTrip.driverPhone)
      && !order.signedAt&&!Number(order.signedPackages||0)&&!order.completedAt&&order.documentSummary?.categories?.every(r=>Number(r.fileCount)===0),'delivery has started, changed or been signed')
    evidence.deliveries.push(order)
  }
  check(evidence.deliveries.reduce((n,row)=>n+Number(row.packageCount),0)===Number(proof.departureSnapshot.detail.counts.shippedPackageCount),'delivery quantity mismatch')
  if (createdVehicle) {
    const vehicle = await get(`/api/platform/v1/last-leg/master-data/vehicles/${encodeURIComponent(createdVehicle.snapshot.id)}`)
    const canonical = value => value && typeof value === 'object' && !Array.isArray(value)
      ? Object.fromEntries(Object.keys(value).sort().map(key => [key, canonical(value[key])])) : value
    check(same(canonical(vehicle), canonical(createdVehicle.snapshot)), 'created vehicle snapshot changed')
    check(typeof context.context?.readReviewedVehicleOccupation === 'function', 'vehicle occupancy reader missing')
    const occupation = await context.context.readReviewedVehicleOccupation(vehicle.id)
    check(id(occupation?.vehicleId) === id(vehicle.id) && occupation.assignmentCount === 0, 'created vehicle is occupied')
    evidence.reviewedVehicle = { ...createdVehicle, snapshot: vehicle, occupation }
  }
  Object.assign(evidence, { status: 'VERIFIED', tripVersion: current.trip.recordVersion, manifestHash: load.manifestHash,
    departure: departures[0], costSummary: costs.summary })
  await context.checkpoint()
  return { trip, dispatchDetails: evidence.dispatchDetails, reviewedVehicle: evidence.reviewedVehicle }
}

export async function readReviewedVehicleOccupation(env, vehicleId, fetcher = globalThis.fetch) {
  validateEvidenceConfig(env)
  if (!/^[1-9]\d*$/.test(id(vehicleId))) throw Error('Exact numeric reviewed vehicle ID required')
  const sql = ['SET NAMES utf8mb4 COLLATE utf8mb4_bin;', 'SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ;',
    'START TRANSACTION WITH CONSISTENT SNAPSHOT, READ ONLY;',
    `SELECT JSON_OBJECT('vehicleId','${vehicleId}','assignmentCount',COUNT(*)) FROM ll_fleet_resource_assignment WHERE resource_kind='VEHICLE' AND resource_id=${vehicleId};`, 'COMMIT;'].join('\n')
  const script = buildEvidenceRemoteScript('lastleg', sql, env.FBA_CORE_JENKINS_AGENT)
  const response = await fetcher(new URL('scriptText', env.JENKINS_URL.replace(/\/?$/, '/')), { method: 'POST',
    headers: { Authorization: 'Basic ' + Buffer.from(env.JENKINS_USER + ':' + env.JENKINS_API_TOKEN).toString('base64'), 'Content-Type': 'application/x-www-form-urlencoded' },
    body: new URLSearchParams({ script }), signal: AbortSignal.timeout(55000) })
  if (!response.ok) throw Error('Reviewed vehicle occupancy HTTP failure')
  const lines = (await response.text()).split(/\r?\n/).filter(line => line.startsWith('FBA_CORE_JSON='))
  if (lines.length !== 1) throw Error('Reviewed vehicle occupancy response ambiguous')
  const result = JSON.parse(lines[0].slice('FBA_CORE_JSON='.length))
  if (result.ok !== true || result.rows?.length !== 1) throw Error('Reviewed vehicle occupancy query failed')
  return result.rows[0]
}

export async function readRecoveryTripResources(env, tripNo, fetcher = globalThis.fetch) {
  validateEvidenceConfig(env)
  if (!/^[A-Za-z0-9_-]{1,128}$/.test(tripNo)) throw Error('Exact recovery trip number required')
  const sql = ['SET NAMES utf8mb4 COLLATE utf8mb4_bin;', 'SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ;',
    'START TRANSACTION WITH CONSISTENT SNAPSHOT, READ ONLY;',
    `SELECT JSON_OBJECT('resourceKind',resource_kind,'resourceId',CAST(resource_id AS CHAR),'tripNo',trip_no) FROM ll_fleet_resource_assignment WHERE trip_no='${tripNo}' ORDER BY resource_kind;`, 'COMMIT;'].join('\n')
  const script = buildEvidenceRemoteScript('lastleg', sql, env.FBA_CORE_JENKINS_AGENT)
  const response = await fetcher(new URL('scriptText', env.JENKINS_URL.replace(/\/?$/, '/')), { method: 'POST',
    headers: { Authorization: 'Basic ' + Buffer.from(env.JENKINS_USER + ':' + env.JENKINS_API_TOKEN).toString('base64'), 'Content-Type': 'application/x-www-form-urlencoded' },
    body: new URLSearchParams({ script }), signal: AbortSignal.timeout(55000) })
  if (!response.ok) throw Error('Recovery resources readonly HTTP failure')
  const lines = (await response.text()).split(/\r?\n/).filter(line => line.startsWith('FBA_CORE_JSON='))
  if (lines.length !== 1) throw Error('Recovery resource evidence ambiguous')
  const result = JSON.parse(lines[0].slice('FBA_CORE_JSON='.length))
  if (result.ok !== true || !Array.isArray(result.rows)) throw Error('Recovery resource query failed')
  return result.rows
}

export async function verifyReturnSourceRecovery(context, { verifyOriginal = verifyDepartedRecovery } = {}) {
  const original = await verifyOriginal(context, { costsAlreadyVerified: true, returnSourceReady: true })
  const run = context.createdResources.fleetAcceptance, rows = []
  const previous = original.trip.returnRedispatch
  for (const delivery of previous.originalDeliveries) {
    const response = await context.request('GET', `/api/platform/v1/last-leg/delivery-orders/${delivery.deliveryNo}/return-inbound-options?customerId=${run.customerId}`)
    if (!success(response) || !response.requestId || !response.traceId || !Array.isArray(response.body.data)) throw Error('Repaired return candidates must be definitive')
    for (const relation of response.body.data) {
      const tail = await context.request('GET', `/api/platform/v1/last-leg/orders/${relation.lastMileOrderId}?customerId=${run.customerId}`)
      if (!success(tail) || !tail.requestId || !tail.traceId || relation.deliveryNo !== delivery.deliveryNo
        || relation.outboundNo !== delivery.outboundNo || id(tail.body.data?.order?.customerId) !== id(run.customerId)
        || id(tail.body.data.order.overseasWarehouseId) !== id(run.warehouseId)
        || id(tail.body.data.order.id) !== id(relation.lastMileOrderId)) throw Error('Repaired source owner or delivery identity mismatch')
      rows.push({ relation, tail: tail.body.data.order, delivery })
    }
  }
  const selected = selectReturnSources(rows)
  context.createdResources.fleetDepartureRecoveryVerification.returnSourceSelection = selected
  await context.checkpoint()
  return original
}

export async function continueReviewedReturnSources(context, { verify = verifyReturnSourceRecovery,
  resolveAccess = resolvePickerAccess, makePicker = createPicker, closePicker = disablePicker,
  redispatch = verifyFleetReturnRedispatch, closeOriginal = finishFleetTrip, cleanup = cleanupFleetHandoverResources } = {}) {
  const verified = await verify(context), trip = verified.trip, run = context.createdResources.fleetAcceptance
  trip.previousSourceSelectionAttempts = [trip.returnRedispatch]; delete trip.returnRedispatch
  run.previousPicker = run.picker; delete run.picker
  run.pickerMarker = `${context.marker}-RETURN-RECOVERY-${randomUUID().slice(0, 8)}`
  await context.checkpoint()
  let picker
  try {
    const child = Object.create(context); child.marker = run.pickerMarker
    const access = await resolveAccess(context)
    picker = access ? await makePicker(child, 71, run.warehouseId, access) : null
    if (!picker?.token || !picker.membershipId) throw Error('Source recovery requires a fresh bound PDA session')
    run.picker = { membershipId: picker.membershipId, username: picker.username }; await context.checkpoint()
    const partial = context.createdResources.fleetFeeCoverageCases.find(row => row.mode === 'PARTIAL')
    const result = await redispatch(context, MODULE, trip, { customerId: run.customerId, warehouseId: run.warehouseId,
      pickerToken: picker.token, dispatchNos: run.dispatchNos, deliveryNos: nativeFleetDeliveries(verified.dispatchDetails[0].detail).map(ref => ref.deliveryNo), sourceSelectionRecovery: true })
    run.receipts = result.originalSignatures.map(row => ({ customerId: id(row.customerId || run.customerId), deliveryNo: row.deliveryNo, signedAt: row.signedAt }))
    context.context.saveFleetReturnRedispatchFixture(result)
    run.closure = await closeOriginal(context, MODULE, trip)
    if (run.closure?.status !== 'CARGO_COMPLETED_AND_RESOURCES_RELEASED') throw Error('Recovered original trip did not close')
    await cleanup(context, MODULE, trip)
    run.stage = 'FINISHED'; run.status = 'HTTP_CHAIN_VERIFIED_WITH_ACK_GAP'; await context.checkpoint()
    return run
  } finally { await closePicker(context, picker, 71) }
}

export async function runFleetHandoverStage(context, trip, pickerToken, dispatchDetails, { verify = verifyFleetHandover, onReady, reviewedVehicle } = {}) {
  const state = await verify(context, MODULE, trip, pickerToken, dispatchDetails, { reviewedVehicle })
  const run = context.createdResources.fleetAcceptance
  if (state?.status !== 'VERIFIED' || state.marker !== context.marker || run?.tripNo !== trip.tripNo
    || !nativeDeparture(run.departure, trip.dispatchNo) || !state.originalDeparture?.snapshot?.confirmedLoad || !state.browser
    || trip.costAcceptance?.status !== 'VERIFIED' || trip.costAcceptance.marker !== context.marker)
    throw Error('Only the current verified DEPART and handover can publish a shared fixture')
  const fixture = { marker: context.marker, runId: context.context?.fleetRunId, tripNo: trip.tripNo, warehouseId: trip.warehouseId, dispatchNos: state.dispatchNos,
    departure: run.departure, immutableDeparture: state.originalDeparture, handoverNo: state.handoverNo,
    originalResources: state.previous, replacementResources: state.replacement, browser: state.browser,
    costs: trip.costAcceptance,
    allocationStatus: trip.costAcceptance.expected?.allocationStatus,
    allocationBlockReason: trip.costAcceptance.expected?.allocationBlockReason,
    status: 'DEPART_AND_HANDOVER_VERIFIED', requestCount: state.requestCount, pdaReadCount: state.pdaReadCount }
  context.context?.saveFleetHandoverFixture?.(fixture)
  if (onReady) await onReady(fixture)
  return fixture
}

export function fleetCostShares(trip, dispatchDetails) {
  const external = trip.externalReceipt, externalDispatch = trip.externalDispatch
  if (external?.status !== 'VERIFIED' || !external.bookingNo || !external.warehousePalletId
    || !trip.dispatchNo || externalDispatch?.dispatchNo !== trip.dispatchNo)
    throw Error('Cost expectations require this run\'s verified external receipt')
  const bundle = dispatchDetails?.length === 1 ? dispatchDetails[0] : null
  const detail = bundle?.detail, items = bundle?.items
  if (detail?.dispatchNo !== trip.dispatchNo || items?.dispatchNo !== trip.dispatchNo
    || id(detail.recordVersion) !== id(items.recordVersion) || !items.items?.length)
    throw Error('Cost expectations require the same native dispatch Detail and Items version')
  const pallets = [...new Set(items.items.map(item => id(item.warehousePalletId)))].sort()
  const measured = trip.confirmedLoad?.pallets
  if (pallets.some(value => !value) || !trip.confirmedLoad?.manifestHash || !Array.isArray(measured)
    || measured.length !== pallets.length || new Set(measured.map(row => id(row.warehousePalletId))).size !== measured.length
    || measured.some(row => !pallets.includes(id(row.warehousePalletId)) || !Number.isSafeInteger(row.palletSpaces) || row.palletSpaces < 1))
    throw Error('Cost expectations require exact actual loading measurements for every distinct pallet')
  const externalId = id(external.warehousePalletId), own = pallets.filter(value => value !== externalId)
  if (!pallets.includes(externalId) || !own.length) throw Error('External receipt and own cargo require distinct actual pallets')
  const share = (sourceType, businessNo, ids) => ({ businessNo, sourceType, dispatchNo: trip.dispatchNo,
    warehousePalletIds: ids, palletSpaces: measured.filter(row => ids.includes(id(row.warehousePalletId)))
      .reduce((sum, row) => sum + row.palletSpaces, 0) })
  return [share('OWN', trip.dispatchNo, own), share('EXTERNAL', external.bookingNo, [externalId])]
}

export function fleetCostActorJournalData(value) {
  const clean = redactEvidence(value)
  function removeCredentials(item) {
    if (Array.isArray(item)) return item.map(removeCredentials)
    if (!item || typeof item !== 'object') return item
    const result = {}, forbidden = /^(?:password|newPassword|temporaryPassword|accessToken|refreshToken|authorization)$/i
    let removed = false
    for (const [name, field] of Object.entries(item)) {
      if (forbidden.test(name)) removed = true
      else result[name] = removeCredentials(field)
    }
    if (removed) result.credentialFieldsRemoved = true
    return result
  }
  return removeCredentials(clean)
}

export async function prepareFleetCostActors(context) {
  if (!context.marker || context.createdResources.fleetCostActors || typeof context.checkpoint !== 'function')
    throw Error('Cost actors require fresh current-run ownership and durable checkpoints')
  requireFixedActors(context.options, COST_ACTOR_SLOTS)
  const proof = context.createdResources.fleetCostActors = { marker: context.marker, status: 'CHECKING', actors: [], requests: [] }
  const sessions = []
  const io = Object.create(context)
  io.request = async (...args) => {
    const entry = { method: args[0], path: args[1], body: fleetCostActorJournalData(args[2]), status: 'SENDING' }
    proof.requests.push(entry); await context.checkpoint()
    try {
      const response = await context.request(...args)
      Object.assign(entry, { status: 'RESPONDED', http: response?.http, code: response?.body?.code, requestId: response?.requestId, traceId: response?.traceId })
      await context.checkpoint()
      return response
    } catch (error) {
      entry.status = 'UNKNOWN_RESULT'; await context.checkpoint(); throw error
    }
  }
  async function close() {
    const result = await Promise.allSettled(sessions.map(actor => actor.close()))
    proof.status = result.some(row => row.status === 'rejected') ? 'CLEANUP_FAILED' : 'SESSIONS_CLOSED'
    await context.checkpoint()
    if (proof.status === 'CLEANUP_FAILED') throw Error('固定车队权限账号会话退出未全部确认')
  }
  try {
    for (const kind of ['reader', 'denied']) {
      const actor = await openFixedActor(io, `fleet-cost.${kind}`, kind === 'reader' ? ['platform:fleet:trip:list'] : [])
      sessions.push(actor)
      proof.actors.push({ kind, username: actor.username, membershipId: actor.membershipId, roleId: actor.roleId, reused: true })
    }
    if (sessions[0].token === sessions[1].token) throw Error('固定车队权限账号必须有独立会话')
    proof.status = 'READY'; await context.checkpoint()
    return { marker: context.marker, listOnlyToken: sessions[0].token, deniedToken: sessions[1].token, close }
  } catch (error) {
    proof.status = 'FAILED'; await context.checkpoint()
    try { await close() } catch (cleanupError) { throw new AggregateError([error, cleanupError], '固定账号核验和退出均失败') }
    throw error
  }
}

export async function findFleetActorPhone(context, actorSuffix) {
  if (!context.marker || !/^[a-f0-9]{16}$/.test(actorSuffix)) throw Error('Fresh actor phone scope required')
  for (let attempt = 0; attempt < 100; attempt++) {
    const number = createHash('sha256').update(`${context.marker}:${actorSuffix}:${attempt}`).digest().readUInt32BE(0)
    const phone = `18${String(number).padStart(9, '0').slice(-9)}`
    const users = await context.request('GET', `/api/platform/v1/iam/users?${new URLSearchParams({ page: '1', size: '20', keyword: phone })}`)
    if (!success(users) || !users.requestId || !users.traceId || !Array.isArray(users.body.data?.records)
      || !Number.isSafeInteger(Number(users.body.data.total)) || Number(users.body.data.total) !== users.body.data.records.length)
      throw Error('Cannot verify platform actor phone availability')
    if (users.body.data.records.some(row => row.phone === phone)) continue
    const customer = await context.request('GET', `/api/platform/v1/iam/tenants/administrator-phone-availability?phone=${encodeURIComponent(phone)}`)
    if (!success(customer) || !customer.requestId || !customer.traceId) throw Error('Cannot verify customer administrator phone availability')
    if (customer.body.data === true) return phone
  }
  throw Error('No available phone in the fresh actor scope; do not replay user creation')
}

async function nativeRequest(context, method, path, body, token) {
  const result = await context.request(method, path, body, token)
  if (!success(result) || !result.requestId || !result.traceId) throw Error(`Native fleet request failed: ${method} ${path}`)
  return result.body.data
}
export async function readNativeFleetBundle(context, dispatchNo) {
  const base = `/api/platform/v1/last-leg/dispatches/${encodeURIComponent(dispatchNo)}`
  const detail = await nativeRequest(context, 'GET', base)
  const items = await nativeRequest(context, 'GET', `${base}/items`)
  if (detail.dispatchNo !== dispatchNo || items.dispatchNo !== dispatchNo || id(dispatchVersion(detail.recordVersion)) !== id(dispatchVersion(items.recordVersion))) throw Error('Native Detail/Items version mismatch')
  return { detail, items }
}

// 只读取本轮生产者和运营已确认实测，不从计划托位、计费重量或常数生成实际量。
export async function currentFleetMeasurements(context, trip) {
  const path = `/api/platform/v1/last-leg/dispatches/${encodeURIComponent(trip.dispatchNo)}/loading`
  const preview = await nativeRequest(context, 'GET', path)
  if (preview.dispatchNo !== trip.dispatchNo || preview.status !== 'ACTIVE' || !preview.pallets?.length) throw Error('Current native physical loading preview missing')
  const producer = context.createdResources.phase4OverseasDispatch?.loadingMeasurements
  if (producer && (producer.marker !== context.marker || producer.environment !== 'TEST' || producer.dispatchNo !== trip.dispatchNo || !Array.isArray(producer.pallets)))
    throw Error('Physical measurements must be the exact current-run producer receipt for this dispatch')
  const actual = (producer?.pallets || []).map(row => structuredClone(row))
  const receipt = trip.externalReceipt?.finalBooking?.receipt
  for (const measured of receipt?.pallets || []) {
    const source = receipt.source?.pallets?.find(p => id(p.warehousePalletId) === id(measured.warehousePalletId))
    if (!source || !Number.isFinite(Date.parse(receipt.confirmedAt)) || !receipt.operatorId || !receipt.source?.incomingNo) throw Error('External actual measurement ownership missing')
    actual.push({ warehousePalletId: id(measured.warehousePalletId), palletSpaces: measured.palletSpaces,
      weightKg: measured.weightKg ?? source.weightKg, volumeCbm: measured.volumeCbm ?? source.volumeCbm,
      deliveryIndex: trip.stops.findIndex(s => s.destinationKey === trip.externalReceipt.finalBooking.cargo.destinationKey),
      evidence: ['palletSpaces','weightKg','volumeCbm'].map(field => ({ field, value: field === 'palletSpaces' ? measured[field] : measured[field] ?? source[field],
        actual: true, sourceType: 'FLEET_RECEIPT_CONFIRMATION', sourceBusinessNo: receipt.source.incomingNo, observedAt: receipt.confirmedAt })) })
  }
  if (new Set(actual.map(p => id(p.warehousePalletId))).size !== actual.length) throw Error('Duplicate current-run physical measurement receipts')
  for (const pallet of preview.pallets) {
    const supplied = actual.find(p => id(p.warehousePalletId) === id(pallet.warehousePalletId))
    for (const field of ['palletSpaces','weightKg','volumeCbm']) {
      const source = pallet[field] == null ? supplied : pallet, value = source?.[field]
      if (!(Number(value) > 0) || !source?.evidence?.some(e => e.field === field && e.actual === true && e.sourceBusinessNo && e.sourceType
        && Number.isFinite(Date.parse(e.observedAt)) && Number(e.value) === Number(value)))
        throw Error(`Missing current-run actual ${field} for ${pallet.palletNo || 'unassociated pallet'}; no planning constants or external receipt injection`)
    }
    const index = pallet.deliveryIndex ?? supplied?.deliveryIndex
    if (!Number.isSafeInteger(index) || index < 1 || !trip.stops?.[index]?.destinationKey)
      throw Error('Current-run physical pallet has no verified unloading stop')
  }
  if (actual.some(p => !preview.pallets.some(row => id(row.warehousePalletId) === id(p.warehousePalletId))
    && !receipt?.source?.pallets?.some(row => id(row.warehousePalletId) === id(p.warehousePalletId)))) throw Error('Physical receipts contain a pallet outside this dispatch and its confirmed incoming')
  return { environment: 'TEST', marker: context.marker, dispatchNo: trip.dispatchNo, pallets: actual }
}

export async function departNativeFleet(context, trip, pickerToken, run) {
  if (!pickerToken || !run || run.dispatchNo !== trip.dispatchNo || run.tripNo !== trip.tripNo || run.departureRequest)
    throw Error('Native departure requires owned main and a new original command')
  const path = `/api/platform/v1/last-leg/dispatches/${encodeURIComponent(trip.dispatchNo)}`
  const after = await readNativeFleetBundle(context, trip.dispatchNo)
  const refs = nativeFleetDeliveries(after.detail), original = trip.loadingCommand
  if (!nativeDeparture(original?.result, trip.dispatchNo) || !original.body?.idempotencyKey
    || after.detail.status !== 'ACTIVE' || after.detail.stage !== 'DELIVERING'
    || id(after.detail.recordVersion) !== id(original.result.recordVersion)
    || Number(after.detail.counts?.shippedPackageCount) !== Number(after.detail.counts?.packageCount)
    || refs.reduce((sum, d) => sum + Number(d.packageCount), 0) !== Number(after.detail.counts.packageCount))
    throw Error('Automatic loading completion must have an original success receipt and complete real delivery documents')
  const departed = await nativeRequest(context, 'GET', `/api/pda/v1/last-leg/dispatches/${encodeURIComponent(trip.dispatchNo)}/commands/${encodeURIComponent(original.body.idempotencyKey)}`, undefined, pickerToken)
  if (!nativeDeparture(departed, trip.dispatchNo) || id(departed.recordVersion) !== id(original.result.recordVersion)
    || JSON.stringify(departed.result) !== JSON.stringify(original.result.result)) throw Error('Original loading command readback changed its outcome')
  run.departureRequest = original.body; run.departure = departed; run.departureReplay = original.replay
  run.departureSnapshot = after; run.dispatchNos = [trip.dispatchNo]
  run.departureDeliveries = []
  for (const ref of nativeFleetDeliveries(after.detail)) {
    const value = await nativeRequest(context, 'GET', `/api/platform/v1/last-leg/delivery-orders/${encodeURIComponent(ref.deliveryNo)}?customerId=${encodeURIComponent(id(ref.customerId))}`)
    if (value.order?.deliveryNo !== ref.deliveryNo || id(value.order.customerId) !== id(ref.customerId) || value.order.dispatchNo !== trip.dispatchNo
      || value.order.status !== 'DELIVERING' || value.order.version == null || Number(value.order.packageCount) !== Number(ref.packageCount)) throw Error('Native departure delivery versions or owners changed')
    run.departureDeliveries.push(value.order)
  }
  await context.checkpoint?.(); return { departed, dispatchDetails: [after] }
}

export async function runFleetScenario(context, { wait = ms => new Promise(resolve => setTimeout(resolve, ms)), onHandoverReady,
  costActors, resumeFull = false, resumePersisted = false, resumeReceiptReady = false, resumePalletized = false, resumeReceiptConfirmed = false, terminalReadOnly = false } = {}) {
  if (terminalReadOnly) return verifyFleetTerminalState(context)
  if (costActors?.marker !== context.marker || !costActors.listOnlyToken || !costActors.deniedToken
    || costActors.listOnlyToken === costActors.deniedToken || [costActors.listOnlyToken, costActors.deniedToken].includes(context.token))
    throw Error('Prepare two independent current-run platform cost permission sessions before starting the fleet scenario')
  const resumed = resumeFull || resumeReceiptReady
  const proof = resumed ? context.createdResources.fleetAcceptance : context.createdResources.fleetAcceptance = {
    protocol: NATIVE_PROTOCOL, marker: context.marker, status: 'CHECKING', stage: 'SOURCES', checkpoints: [] }
  if (proof.protocol !== NATIVE_PROTOCOL) throw Error('Legacy multi-dispatch reports cannot enter the native main chain')
  function stage(value) { proof.stage = value; proof.checkpoints.push({ stage: value, at: new Date().toISOString() }); console.log(`FLEET_STAGE=${value}`) }
  const source = context.createdResources.phase4OverseasDispatch, receiving = context.createdResources.phase4OverseasPda
  if (source?.marker !== context.marker || source.status !== 'VERIFIED' || !source.dispatchNo || receiving?.marker !== context.marker
    || receiving.finalStatus !== 'COMPLETED' || !receiving.customerId || !receiving.overseasWarehouseId) throw Error('Current-run native dispatch and real received warehouse evidence required')
  const customerId = id(receiving.customerId), warehouseId = id(receiving.overseasWarehouseId)
  Object.assign(proof, { customerId, warehouseId, dispatchNo: source.dispatchNo, dispatchNos: [source.dispatchNo] })
  const journal = dispatchIo(context, MODULE, proof, 'FLEET-NATIVE')
  let full, partial, trip
  const template = { plannedDeliveryAt: new Date(Date.now()+21600000).toISOString(), deliveryMethod: 'TRUCK_PALLET', pickupMethod: 'WAREHOUSE_PICKUP',
    loadingType: 'PALLET', vehicleType: 'TRUCK', vehicleSize: '本轮实际装载', appointmentAccount: 'SELF', remark: `${context.marker} TEST原费用覆盖`,
    fees: [{ feeType: 'DELIVERY_FEE', feeName: '未覆盖原入仓派送费', direction: 'RECEIVABLE', currency: 'CNY', amount: 10 }] }
  if (!resumed) {
    const original = await readDispatch(journal, source.dispatchNo)
    const candidate = selectNativeFleetFullSource(context, original, await candidatePage(journal, warehouseId))
    proof.originalDispatch = { detail: original.data, items: original.items, route: original.route }
    proof.fullCreateRequest = createBody(context.marker, 'FLEET-FULL-INDEPENDENT', [candidate])
    const created = await journal.command('POST', '/api/platform/v1/last-leg/dispatches', proof.fullCreateRequest, '释放的本轮真实第三托创建独立FULL原生排车')
    await replayDispatch(journal, '/api/platform/v1/last-leg/dispatches', proof.fullCreateRequest, created, '独立FULL创建原键恢复')
    if (!created.dispatchNo || created.dispatchNo === source.dispatchNo) throw Error('FULL must be a distinct newly created native dispatch')
    const fullTrip = await prepareFleetTrip(context, MODULE, created.dispatchNo, warehouseId)
    trip = await prepareFleetTrip(context, MODULE, source.dispatchNo, warehouseId)
    Object.assign(proof, { tripNo: trip.tripNo, fullTripNo: fullTrip.tripNo })
    full = { marker: context.marker, mode: 'FULL', dispatchNo: created.dispatchNo, customerId, transferBody: structuredClone(template) }
    partial = { marker: context.marker, mode: 'PARTIAL', dispatchNo: source.dispatchNo, customerId, transferBody: structuredClone(template) }
    context.createdResources.fleetFeeCoverageCases = [full, partial]
    proof.fullExecution = { protocol: NATIVE_PROTOCOL, marker: context.marker, dispatchNo: full.dispatchNo, tripNo: fullTrip.tripNo }
  } else {
    full = context.createdResources.fleetFeeCoverageCases.find(row => row.mode === 'FULL')
    partial = context.createdResources.fleetFeeCoverageCases.find(row => row.mode === 'PARTIAL')
    trip = context.createdResources.fleetTrips.find(row => row.dispatchNo === proof.dispatchNo)
    if (!full || !partial || partial.dispatchNo !== trip?.dispatchNo || full.dispatchNo === partial.dispatchNo) throw Error('Reviewed native FULL/PARTIAL original identities missing')
    if (resumeFull && !resumePersisted && JSON.stringify(await readNativeFleetBundle(context, full.dispatchNo)) !== JSON.stringify(proof.fullBeforeAttempt))
      throw Error('Read-only FULL checkpoint changed physical state; review the original commands before any new session or write')
  }
  const access = await resolvePickerAccess(context)
  let picker
  try {
    if (!resumeReceiptReady) {
      const fullTrip = context.createdResources.fleetTrips.find(t => t.dispatchNo === full.dispatchNo)
      stage('FULL_COVERAGE')
      if (!resumed) proof.fullBeforeAttempt = await readNativeFleetBundle(context, full.dispatchNo)
      let fullPicker
      try {
        const child = Object.create(context)
        proof.fullExecution.pickerMarker = `${context.marker}-FULL-PDA-${randomUUID().slice(0,8)}`
        child.marker = proof.fullExecution.pickerMarker
        fullPicker = access ? await createPicker(child, 161, warehouseId, access) : null
        if (!fullPicker?.token || !fullPicker.membershipId) throw Error('Independent FULL requires its own bound-warehouse PDA')
        proof.fullExecution.picker = { membershipId: fullPicker.membershipId, username: fullPicker.username }
        await context.checkpoint?.()
        if (!resumePersisted && full.proof) { full.previousReadOnlyAttempts ??= []; full.previousReadOnlyAttempts.push(full.proof); delete full.proof }
        await verifyFleetFeeCoverage(context, MODULE, full, { reconcilePersisted: resumePersisted, pickerToken: fullPicker.token,
          actualMeasurements: resumePersisted ? undefined : await currentFleetMeasurements(context, fullTrip) })
        const fullDeparted = await departNativeFleet(context, fullTrip, fullPicker.token, proof.fullExecution)
        proof.fullExecution.receipts = []
        for (const [index, ref] of nativeFleetDeliveries(fullDeparted.dispatchDetails[0].detail).entries()) {
          const detail = await nativeRequest(context, 'GET', `/api/platform/v1/last-leg/delivery-orders/${encodeURIComponent(ref.deliveryNo)}?customerId=${encodeURIComponent(id(ref.customerId))}`)
          const signed = await saveReceiptAndComplete(context, detail.order, id(ref.customerId), 161+index)
          if (signed?.order?.status !== 'DELIVERED' || !signed.order.signedAt) throw Error('FULL source requires actual customer POD completion')
          proof.fullExecution.receipts.push({customerId:id(ref.customerId),deliveryNo:ref.deliveryNo,signedAt:signed.order.signedAt})
        }
        proof.fullExecution.closure = await finishFleetTrip(context, MODULE, fullTrip)
        if (proof.fullExecution.closure.status !== 'CARGO_COMPLETED_AND_RESOURCES_RELEASED') throw Error('FULL independent resources did not finish')
      } finally { await disablePicker(context, fullPicker, 161) }
      stage('RESERVATIONS'); await verifyFleetBookingReservation(context, MODULE, trip)
    }
    if (full.proof?.status !== 'VERIFIED' || proof.fullExecution?.closure?.status !== 'CARGO_COMPLETED_AND_RESOURCES_RELEASED'
      || trip.reservation?.status !== 'VERIFIED' || partial.proof) throw Error('PARTIAL must retain its unpicked real main after independent FULL closure and reservation')
    if (resumePalletized || resumeReceiptConfirmed) {
      picker = access ? await restoreFixedPicker(context, proof.picker, 71, warehouseId, access) : null
    } else {
      const child = Object.create(context); proof.pickerMarker = `${context.marker}-FPDA-${randomUUID().slice(0,8)}`; child.marker = proof.pickerMarker
      picker = access ? await createPicker(child, 71, warehouseId, access) : null
    }
    if (!picker?.token || !picker.membershipId) throw Error('Fresh bound-warehouse fleet PDA required')
    proof.picker = {membershipId:picker.membershipId,username:picker.username}; await context.checkpoint?.()
    stage('EXTERNAL_RECEIPT')
    await verifyFleetExternalReceipt(context, MODULE, trip, picker.token, {reconcilePalletized:resumePalletized,reconcileConfirmed:resumeReceiptConfirmed})
    stage('EXTERNAL_DISPATCH')
    await verifyFleetExternalDispatch(context, MODULE, trip, {pickerToken:picker.token,
      actualMeasurements: await currentFleetMeasurements(context, trip), coverageCase:partial})
    const external = trip.externalDispatch
    stage('FINANCE_CONSUMER')
    const externalCustomer = id(external.customerId)
    const sources = await nativeRequest(context, 'GET', `/api/platform/v1/last-leg/finance-fee-sources?${new URLSearchParams({customerId:externalCustomer,businessType:'DELIVERY_ORDER',businessNo:external.deliveryNo})}`)
    const target = sources.filter(row => row.feeItemCode === 'DELIVERY_FEE' && row.direction === 'RECEIVABLE' && row.sourceStatus === 'ACTIVE'
      && row.inboundCode === trip.externalReceipt.incomingNo && id(row.customerId) === externalCustomer)
    if (target.length !== 1 || Number(target[0].originalAmount) !== 30 || target[0].originalCurrency !== 'CNY') throw Error('Exact current-run external fee source missing')
    let matched = []
    for (let attempt=0;attempt<30;attempt++) {
      const pool = await nativeRequest(context, 'GET', `/api/platform/finance/v1/fee-pool?customerId=${encodeURIComponent(externalCustomer)}&direction=RECEIVABLE`)
      if (!Array.isArray(pool) || pool.some(row => id(row.customerId) !== externalCustomer || row.direction !== 'RECEIVABLE')) throw Error('Finance consumer scope mismatch')
      matched = pool.filter(row => id(row.sourceId) === id(target[0].sourceFeeId) && id(row.sourceVersion) === id(target[0].sourceVersion))
      if (matched.length) break
      if (attempt<29) await wait(1000)
    }
    if (matched.length !== 1 || matched[0].sourceType !== 'OVERSEAS_DELIVERY_FEE' || Number(matched[0].amount) !== 30 || matched[0].currency !== 'CNY') throw Error('Finance did not persist exactly this external fee')
    proof.finance = {status:'CONSUMER_PERSISTED',customerId:externalCustomer,source:target[0],pool:matched,producerAck:'NOT_VERIFIED'}
    external.financeAcknowledgement = 'CONSUMER_PERSISTED_PRODUCER_ACK_NOT_VERIFIED'
    stage('WHOLE_TRIP_DEPARTURE')
    const departure = await departNativeFleet(context, trip, picker.token, proof)
    await completeDepartedFleetScenario(context, trip, picker, departure.dispatchDetails, {costActors,onHandoverReady})
  } finally { await disablePicker(context, picker, 71) }
  return proof
}
export async function completeDepartedFleetScenario(context, trip, picker, dispatchDetails, { costActors, onHandoverReady,
  costsAlreadyVerified = false, reviewedVehicle,
  verifyCosts = verifyFleetTripCosts, handover = runFleetHandoverStage, completeDelivery = saveReceiptAndComplete,
  closeTrip = finishFleetTrip, cleanup = cleanupFleetHandoverResources } = {}) {
  const proof=context.createdResources.fleetAcceptance,customerId=proof.customerId,warehouseId=proof.warehouseId
  if(proof.protocol!==NATIVE_PROTOCOL||!nativeDeparture(proof.departure,trip.dispatchNo)||proof.dispatchNo!==trip.dispatchNo
    ||dispatchDetails?.length!==1||dispatchDetails[0].detail?.dispatchNo!==trip.dispatchNo)throw Error('Post-departure continuation requires one native main bundle and its actual DEPART command')
  const stage=value=>{proof.stage=value;proof.checkpoints.push({stage:value,at:new Date().toISOString()});console.log('FLEET_STAGE='+value)}
  if(!costsAlreadyVerified){stage('TRIP_OPERATING_COSTS');await verifyCosts(context,MODULE,trip,{...costActors,shares:fleetCostShares(trip,dispatchDetails)})}
  else if(trip.costAcceptance?.status!=='VERIFIED')throw Error('Recovery cost proof must remain verified')
  stage('IN_TRANSIT_HANDOVER')
  await handover(context,trip,picker.token,dispatchDetails.map(bundle=>bundle.detail),{reviewedVehicle,
    onReady:onHandoverReady?async fixture=>{stage('PDA_UI_CHECKPOINT');await onHandoverReady({fixture,picker})}:undefined})
  stage('RECEIPTS_AND_RESOURCE_RELEASE');proof.receipts=[]
  const current=await readNativeFleetBundle(context,trip.dispatchNo),refs=nativeFleetDeliveries(current.detail)
  if(context.context.fleetScenario==='return-redispatch'){
    stage('RETURN_REDISPATCH')
    if(refs.some(ref=>id(ref.customerId)!==id(customerId)))throw Error('Explicit return-redispatch source selection requires the one actual source customer')
    const returned=await verifyFleetReturnRedispatch(context,MODULE,trip,{customerId,warehouseId,pickerToken:picker.token,
      dispatchNos:[trip.dispatchNo],deliveryNos:refs.map(ref=>ref.deliveryNo)})
    proof.receipts=returned.originalSignatures.map(row=>({customerId:id(row.customerId||customerId),deliveryNo:row.deliveryNo,signedAt:row.signedAt}))
    context.context.saveFleetReturnRedispatchFixture(returned)
  }else{
    for(const[index,ref]of refs.entries()){
      const delivery=await nativeRequest(context,'GET',`/api/platform/v1/last-leg/delivery-orders/${encodeURIComponent(ref.deliveryNo)}?customerId=${encodeURIComponent(id(ref.customerId))}`)
      if(delivery.order?.deliveryNo!==ref.deliveryNo||id(delivery.order.customerId)!==id(ref.customerId)||delivery.order.dispatchNo!==trip.dispatchNo)throw Error('POD source customer or dispatch drift')
      const completed=await completeDelivery(context,delivery.order,id(ref.customerId),index+71)
      if(completed?.order?.status!=='DELIVERED'||!completed.order.signedAt)throw Error('每个目的地真实保存签收资料后完成')
      proof.receipts.push({customerId:id(ref.customerId),deliveryNo:ref.deliveryNo,signedAt:completed.order.signedAt})
    }
  }
  proof.closure=await closeTrip(context,MODULE,trip)
  if(proof.closure?.status!=='CARGO_COMPLETED_AND_RESOURCES_RELEASED')throw Error('本次车次货物结案及车辆司机释放完成')
  stage('HANDOVER_RESOURCE_CLEANUP');await cleanup(context,MODULE,trip)
  proof.status='HTTP_CHAIN_VERIFIED_WITH_ACK_GAP';stage('FINISHED');return proof
}
export async function continueReviewedDeparture(context, { verify = verifyDepartedRecovery, prepareActors = prepareFleetCostActors,
  costsAlreadyVerified = false, createdVehicle,
  resolveAccess = resolvePickerAccess, makePicker = createPicker, closePicker = disablePicker,
  complete = completeDepartedFleetScenario } = {}) {
  const verified = await verify(context, { costsAlreadyVerified, createdVehicle })
  const proof = context.createdResources.fleetAcceptance, trip = verified.trip
  if (!costsAlreadyVerified) {
    context.createdResources.previousFleetCostActors = context.createdResources.fleetCostActors
    trip.previousReadOnlyCostAttempts = [trip.costAcceptance]
    delete context.createdResources.fleetCostActors
    delete trip.costAcceptance
  } else {
    trip.previousReadOnlyHandoverAttempts = [trip.handover]
    delete trip.handover
  }
  proof.previousPicker = proof.picker
  delete proof.picker
  proof.pickerMarker = `${context.marker}-RECOVERY-PDA-${randomUUID().slice(0, 8)}`
  await context.checkpoint()
  let actors, picker
  try {
    if (!costsAlreadyVerified) {
      actors = await prepareActors(context)
      if (actors?.marker !== context.marker || !actors.listOnlyToken || !actors.deniedToken
      || actors.listOnlyToken === actors.deniedToken || [actors.listOnlyToken, actors.deniedToken].includes(context.token))
        throw Error('Recovery requires fresh independent cost sessions')
    }
    const pickerContext = Object.create(context); pickerContext.marker = proof.pickerMarker
    const access = await resolveAccess(context)
    picker = access ? await makePicker(pickerContext, 71, proof.warehouseId, access) : null
    if (!picker?.token || !picker.membershipId) throw Error('Recovery requires a new bound PDA session')
    proof.picker = { membershipId: picker.membershipId, username: picker.username }
    await context.checkpoint()
    return await complete(context, trip, picker, verified.dispatchDetails, { costActors: actors, costsAlreadyVerified, reviewedVehicle: verified.reviewedVehicle })
  } finally {
    try { await closePicker(context, picker, 71) } finally { await actors?.close() }
  }
}

export async function runFleetAcceptance(config, { env = process.env, readers, createContext, scenario = runFleetScenario, prepareCostActors = prepareFleetCostActors, root = ROOT } = {}) {
  if (!config.execute || config.environment !== 'TEST' || env.CHONGCHONG_ENV?.toLowerCase() !== 'test'
    || basename(root) !== 'baiyue-firstleg' || basename(dirname(root)) !== 'chongchong-global') throw Error('Explicit TEST workspace and configuration required')
  if (scenario === runFleetScenario && config.resumeStage !== 'TERMINAL_READONLY' && typeof prepareCostActors !== 'function')
    throw Error('Current-run cost actor preparation and lifecycle must be provided before any environment requests')
  const base = new URL(env.CHONGCHONG_ADMIN_API_BASE_URL)
  if (!['http:', 'https:'].includes(base.protocol) || base.username || base.password || base.search || base.hash
    || /(?:^|[.-])(?:dev|prod|production)(?:[.-]|$)/i.test(base.hostname)) throw Error('Invalid TEST gateway')
  for (const name of ['CHONGCHONG_ADMIN_USERNAME', 'CHONGCHONG_ADMIN_PASSWORD', 'ADMIN_PDA_USERNAME', 'ADMIN_PDA_PASSWORD']) if (!env[name]) throw Error(`Missing ${name}`)
  if (config.outputDir && !/^reports\/[A-Za-z0-9][A-Za-z0-9_-]{0,95}$/.test(config.outputDir)) throw Error('Report scope invalid')
  const output = resolve(root, config.outputDir || `reports/fleet-${config.runId}`), rel = relative(resolve(root, 'reports'), output)
  if (!rel || rel.startsWith('..') || isAbsolute(rel)) throw Error('Report scope invalid')
  mkdirSync(resolve(root, 'reports'), { recursive: true })
  const lockPath = resolve(root, 'reports/.fleet-acceptance.lock'), lock = openSync(lockPath, 'wx')
  let context, initialized = false, report, restoreFetch
  const secrets = Object.entries(env).filter(([name]) => /PASSWORD|TOKEN|USERNAME|JENKINS_USER/.test(name)).map(([,value]) => value)
  const save = (name, value) => writeFileSync(join(output, name), JSON.stringify(redactEvidence(value, [...secrets,
    context?.token, context?.pdaToken, context?.refreshToken, context?.context?.merchantQualificationFixture?.password]), null, 2) + '\n')
  try {
    writeFileSync(lock, JSON.stringify({ pid: process.pid, runId: config.runId, startedAt: new Date().toISOString() }))
    mkdirSync(output, { recursive: false }); initialized = true; mkdirSync(join(output, 'evidence'))
    save('manifest.json', fleetExecutionPlan(config))
    let resumed
    if (config.resumeReport) {
      const previous = resolve(root, config.resumeReport), previousRelative = relative(resolve(root, 'reports'), previous)
      if (!previousRelative || previousRelative.startsWith('..') || isAbsolute(previousRelative)) throw Error('Resume report must be inside this TEST repository reports')
      const bytes = readFileSync(previous)
      if ((config.resumeStage?.startsWith('DEPARTED_') || config.resumeStage === 'RETURN_SOURCE_SELECTION_BEFORE_RETURN') && createHash('sha256').update(bytes).digest('hex') !== config.resumeSha256)
        throw Error('Reviewed departure report SHA256 mismatch; no environment request sent')
      resumed = reviewedFullCheckpoint(JSON.parse(bytes.toString('utf8')), config.resumeStage)
      save('reviewed-checkpoint.json', { source: config.resumeReport, sha256: createHash('sha256').update(bytes).digest('hex'), stage: config.resumeStage })
    } else {
      readers ||= createTestEvidenceReaders(env)
      await readers.preflight()
    }
    const options = validateExecutionOptions({ ...parseArguments(['--modules', 'phase4-overseas-dispatch', '--quiet', 'true']),
      gateway: base.origin, username: env.CHONGCHONG_ADMIN_USERNAME, password: env.CHONGCHONG_ADMIN_PASSWORD,
      pdaUsername: env.ADMIN_PDA_USERNAME, pdaPassword: env.ADMIN_PDA_PASSWORD, merchantUsername: '', merchantPassword: '',
      markerNamespace: createHash('sha256').update(config.runId).digest('hex').slice(0, 12) })
    context = createContext ? createContext(options) : new AcceptanceRun(options)
    let writeCheckpoint = 0
    restoreFetch = guardFleetWrites(context, entry => {
      const payload = JSON.stringify(redactFleetWrite({ marker: context.marker, runId: config.runId, request: entry },
        [...secrets, context.token, context.pdaToken, context.refreshToken]), null, 2) + '\n'
      const descriptor = openSync(join(output, 'evidence', `write-${String(++writeCheckpoint).padStart(6, '0')}.json`), 'wx')
      try { writeFileSync(descriptor, payload); fsyncSync(descriptor) }
      finally { closeSync(descriptor) }
    }, { guardFetch: true })
    context.context.fleetRunId = config.runId
    context.context.fleetScenario = config.scenario || 'handover'
    context.context.readReviewedVehicleOccupation = vehicleId => readReviewedVehicleOccupation(env, vehicleId)
    context.context.readRecoveryTripResources = tripNo => readRecoveryTripResources(env, tripNo)
    context.context.onBeforeRedispatchSignature = proof => waitForRedispatchSignatureAudit(context, proof,
      { root, output, secrets: [...secrets, context.token, context.pdaToken, context.refreshToken] })
    context.context.saveFleetHandoverFixture = fixture => save('handover-fixture.json', fixture)
    context.context.saveFleetReturnRedispatchFixture = fixture => save('return-redispatch-fixture.json', fixture)
    let handoverCheckpoint = 0
    context.context.checkpointFleetHandover = proof => {
      const payload = JSON.stringify(redactEvidence({ marker: context.marker, runId: config.runId,
        proof, createdResources: context.createdResources }, [...secrets, context.token, context.pdaToken, context.refreshToken]), null, 2) + '\n'
      // Never truncate the prior checkpoint: interruption or disk failure leaves the earlier request recoverable.
      const descriptor = openSync(join(output, 'evidence', `handover-${String(++handoverCheckpoint).padStart(6, '0')}.json`), 'wx')
      try { writeFileSync(descriptor, payload); fsyncSync(descriptor) }
      finally { closeSync(descriptor) }
    }
    context.checkpoint = () => context.context.checkpointFleetHandover(context.createdResources.fleetAcceptance)
    const fleetScenario = async (ctx, options = {}) => {
      if (options.returnSourceSelection) return continueReviewedReturnSources(ctx)
      if (options.resumeDeparted) return continueReviewedDeparture(ctx, { prepareActors: prepareCostActors,
        costsAlreadyVerified: options.costsAlreadyVerified, createdVehicle: options.createdVehicle })
      const costActors = options.terminalReadOnly || scenario !== runFleetScenario ? undefined : await prepareCostActors?.(ctx)
      try { return await scenario(ctx, { ...options, costActors }) }
      finally { await costActors?.close() }
    }
    let evidenceNumber = 0
    context.context.phase4ReceivingEvidenceReader = async scope => {
      const evidence = await readers.readReceiving(scope)
      save(`evidence/receiving-${++evidenceNumber}.json`, { scope, evidence }); return evidence
    }
    const executeModule = context.executeModule.bind(context)
    context.executeModule = async (name, handler) => {
      await executeModule(name, handler)
      save('progress.json', { stage: name, steps: context.steps, modules: context.modules, createdResources: context.createdResources })
      if (context.steps.some(step => step.passed === false) || context.modules.at(-1)?.success === false) throw Error(`Producer stopped: ${name}`)
      if (name === 'phase4-overseas-dispatch') {
        await executeModule(MODULE, () => fleetScenario(context))
        if (context.modules.at(-1)?.success === false) throw Error('Fleet case stopped at original fixture')
      }
    }
    if (resumed) {
      context.createdResources = resumed.resources; context.marker = resumed.marker
      const login = await context.request('POST', '/api/auth/login', { loginId: options.username, password: options.password }, '')
      if (!success(login) || !login.body?.data?.accessToken) throw Error('Resume administrator login failed')
      context.token = login.body.data.accessToken; context.refreshToken = login.body.data.refreshToken || ''
      await context.executeModule(MODULE, () => fleetScenario(context, { resumeFull: !resumed.receiptReady,
        resumeDeparted: resumed.departed,
        returnSourceSelection: resumed.returnSourceSelection,
        costsAlreadyVerified: resumed.costsAlreadyVerified,
        createdVehicle: resumed.createdVehicle,
        resumePersisted: resumed.persisted, resumeReceiptReady: resumed.receiptReady, resumePalletized: resumed.palletized, resumeReceiptConfirmed: resumed.receiptConfirmed, terminalReadOnly: resumed.terminal }))
      const summary = buildAcceptanceSummary(context.steps, context.modules, context.coverageGaps)
      report = { success: summary.failed === 0, marker: context.marker, summary, steps: context.steps, modules: context.modules,
        createdResources: context.createdResources, resumedFrom: config.resumeReport }
    } else report = await context.run()
    report.fleetExecution = fleetExecutionPlan(config)
    // Keep the unresolved producer ACK explicit; a green consumer row cannot close the complete fleet goal.
    report.fleetGoalComplete = false
  } catch (error) {
    report = { success: false, fatal: error instanceof Error ? error.message : 'FLEET_ACCEPTANCE_FAILED',
      summary: context ? buildAcceptanceSummary(context.steps, context.modules, context.coverageGaps) : { failed: 1 },
      steps: context?.steps || [], modules: context?.modules || [], createdResources: context?.createdResources || {} }
  } finally {
    restoreFetch?.()
    if (initialized) save('report.json', report || { success: false, fatal: 'RUN_INTERRUPTED' })
    closeSync(lock); unlinkSync(lockPath)
  }
  return { success: report?.success === true, report: join(output, 'report.json'), summary: report?.summary,
    fleet: report?.createdResources?.fleetAcceptance?.status, fatal: report?.fatal }
}

if (process.argv[1] && import.meta.url === pathToFileURL(resolve(process.argv[1])).href) {
  try { const config = parseFleetArguments(process.argv.slice(2)); const result = config.execute ? await runFleetAcceptance(config) : fleetExecutionPlan(config)
    console.log(JSON.stringify(result, null, 2)); if (config.execute && !result.success) process.exitCode = 1
  } catch (error) { console.error(error instanceof Error ? error.message : 'Fleet acceptance failed'); process.exitCode = 1 }
}