// Explicit TEST-only row-lock fault; importing this file performs no I/O.
import fs from 'node:fs'
import { createHash } from 'node:crypto'
import { resolve } from 'node:path'
import { pathToFileURL } from 'node:url'
import { parseCrossCustomerArguments, runCrossCustomer } from './fleet-cross-customer-test-env.mjs'
import { verifyFleetCrossCustomer } from './admin-test-env-modules/fleet-cross-customer-acceptance.mjs'

const demand = (value, message) => { if (!value) throw Error(message) }
const identifier = value => typeof value === 'string' && /^[1-9][0-9]{0,19}$/.test(value)
const businessNo = value => typeof value === 'string' && /^[A-Za-z0-9_-]{1,128}$/.test(value)
const literal = value => `CONVERT(0x${Buffer.from(value).toString('hex')} USING utf8mb4) COLLATE utf8mb4_bin`
const encoded = value => '[' + Buffer.from(value).toString('base64').match(/.{1,4000}/g).map(x => `'${x}'`).join(',') + "].join('')"
const pause = ms => new Promise(done => setTimeout(done, ms))
const canonical = value => JSON.stringify(sortFields(value))
const sha = value => createHash('sha256').update(canonical(value)).digest('hex')
function sortFields(value) {
  return Array.isArray(value) ? value.map(sortFields) : value && typeof value === 'object'
    ? Object.fromEntries(Object.keys(value).sort().map(key => [key, sortFields(value[key])])) : value
}
const rr = ['SET NAMES utf8mb4 COLLATE utf8mb4_bin;', 'SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ;', 'START TRANSACTION WITH CONSISTENT SNAPSHOT, READ ONLY;']
const databaseRuntime = "SELECT JSON_OBJECT('collection','databaseRuntime','serverUuid',@@server_uuid,'databaseName',DATABASE(),'performanceSchema',@@performance_schema);"
// Every column is obtained from the actual schema; no partial projection can masquerade as a full rollback baseline.
const tables = ['ll_dispatch_order', 'll_dispatch_history', 'll_dispatch_item', 'll_dispatch_command', 'll_dispatch_pallet_lock', 'll_dispatch_physical_pallet_lock', 'll_physical_pallet_guard', 'll_dispatch_item_source',
  'll_fleet_trip', 'll_fleet_trip_dispatch', 'll_fleet_trip_history', 'll_fleet_pool_order_link', 'll_outbound_order', 'll_delivery_order',
  'll_order', 'll_loading_order_status_history', 'll_warehouse_transfer_order', 'll_dispatch_pda_task', 'll_dispatch_pda_pick',
  'll_dispatch_pda_pick_order', 'll_inventory_balance', 'll_inventory_position', 'll_inventory_ledger']
const columnsSql = `SELECT JSON_OBJECT('collection','column','table',TABLE_NAME,'column',COLUMN_NAME,'ordinal',ORDINAL_POSITION,'type',COLUMN_TYPE,'key',COLUMN_KEY) FROM information_schema.columns WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME IN (${tables.map(literal).join(',')}) ORDER BY TABLE_NAME,ORDINAL_POSITION;`
const projection = (collection, table, fields, where) => `SELECT JSON_OBJECT('collection','${collection}',${Object.entries(fields).map(([key, expression]) => `'${key}',CAST(${expression} AS CHAR)`).join(',')}) FROM ${table} WHERE ${where};`
const one = (rows, collection) => { const found = rows.filter(row => row.collection === collection); demand(found.length === 1, `EXACT_${collection.toUpperCase()}_REQUIRED`); return found[0] }

export function validateRollbackRequest(scope) {
  demand(scope && businessNo(scope.marker) && businessNo(scope.tripNo) && businessNo(scope.dispatchNo)
    && businessNo(scope.commandKey) && identifier(scope.commandVersion) && identifier(scope.warehouseId)
    && /^sha256:[a-f0-9]{64}$/.test(scope.expectedImage || ''), 'NATIVE_ROLLBACK_REQUEST_REQUIRED')
  demand(scope.members?.length === 2 && new Set(scope.members.map(m => m.customerId)).size === 2, 'TWO_DISTINCT_CUSTOMERS_REQUIRED')
  for (const member of scope.members) {
    demand(['customerId', 'dispatchItemId', 'palletId', 'positionId', 'balanceId'].every(field => identifier(member[field]))
      && ['incomingNo', 'outboundNo', 'deliveryNo'].every(field => businessNo(member[field])), 'OWNED_MEMBER_IDENTITIES_REQUIRED')
  }
  for (const field of ['incomingNo', 'dispatchItemId', 'palletId', 'positionId', 'balanceId', 'outboundNo', 'deliveryNo'])
    demand(new Set(scope.members.map(m => m[field])).size === 2, 'MEMBER_IDENTITY_REUSED')
  return scope
}

function resolutionSql(scope) {
  validateRollbackRequest(scope)
  const root = `(SELECT id FROM ll_dispatch_order WHERE dispatch_no=${literal(scope.dispatchNo)})`
  const trip = `(SELECT id FROM ll_fleet_trip WHERE trip_no=${literal(scope.tripNo)})`
  const items = `dispatch_order_id=${root}`
  const sourceOrders = `(SELECT order_id FROM ll_dispatch_item WHERE ${items})`
  return [...rr, databaseRuntime, columnsSql,
    projection('root', 'll_dispatch_order', { id: 'id', dispatchNo: 'dispatch_no', warehouseId: 'overseas_warehouse_id', version: 'record_version', status: 'status', stage: 'stage' }, `dispatch_no=${literal(scope.dispatchNo)}`),
    projection('trip', 'll_fleet_trip', { id: 'id', tripNo: 'trip_no', warehouseId: 'warehouse_id', status: 'status' }, `trip_no=${literal(scope.tripNo)}`),
    projection('binding', 'll_fleet_trip_dispatch', { dispatchId: 'dispatch_id', dispatchNo: 'dispatch_no', tripId: 'trip_id', tripNo: 'trip_no' }, `trip_id=${trip} OR dispatch_id=${root}`),
    projection('item', 'll_dispatch_item', { id: 'id', dispatchId: 'dispatch_order_id', customerId: 'customer_id', palletId: 'warehouse_pallet_id', orderId: 'order_id', sourceType: 'source_business_type', sourceNo: 'source_business_no', incomingNo: 'inbound_code', packages: 'package_count' }, items),
    projection('transfer', 'll_warehouse_transfer_order', { id: 'id', customerId: 'customer_id', incomingNo: 'transfer_no', orderId: 'lastleg_order_id', warehouseId: 'operating_warehouse_id', needRelabel: 'need_relabel', relabelNo: 'relabel_no', outboundAt: 'outbound_at' }, `lastleg_order_id IN ${sourceOrders}`),
    projection('outbound', 'll_outbound_order', { id: 'id', customerId: 'customer_id', dispatchId: 'dispatch_order_id', outboundNo: 'outbound_no', status: 'status' }, items),
    projection('delivery', 'll_delivery_order', { id: 'id', customerId: 'customer_id', dispatchId: 'dispatch_order_id', deliveryNo: 'delivery_no', outboundNo: 'outbound_no', status: 'status' }, items),
    projection('task', 'll_dispatch_pda_task', { id: 'id', dispatchId: 'dispatch_order_id', warehouseId: 'warehouse_id' }, items),
    'COMMIT;'].join('\n')
}

export function resolveRollbackScope(request, rows) {
  validateRollbackRequest(request)
  const root = one(rows, 'root'), trip = one(rows, 'trip'), binding = one(rows, 'binding'), task = one(rows, 'task')
  demand(identifier(root.id) && identifier(trip.id) && identifier(task.id) && root.dispatchNo === request.dispatchNo
    && root.warehouseId === request.warehouseId && root.version === request.commandVersion && root.status === 'ACTIVE' && root.stage === 'AFTER_LOADING'
    && trip.tripNo === request.tripNo && trip.warehouseId === request.warehouseId && trip.status === 'PLANNED'
    && binding.dispatchId === root.id && binding.tripId === trip.id && binding.dispatchNo === request.dispatchNo && binding.tripNo === request.tripNo
    && task.dispatchId === root.id && task.warehouseId === request.warehouseId, 'ONE_NATIVE_ROOT_AND_FLEET_REQUIRED')
  for (const collection of ['item', 'transfer', 'outbound', 'delivery']) demand(rows.filter(r => r.collection === collection).length === 2, `TWO_${collection.toUpperCase()}_ROWS_REQUIRED`)
  const members = request.members.map(member => {
    const own = collection => one(rows.filter(r => r.customerId === member.customerId), collection)
    const item = own('item'), transfer = own('transfer'), outbound = own('outbound'), delivery = own('delivery')
    demand(item.id === member.dispatchItemId && item.dispatchId === root.id && item.palletId === member.palletId && identifier(item.orderId)
      // Native ordinary cargo leaves source_business_type/no NULL; the actual transfer/order/receipt links prove its origin.
      && item.sourceType === null && item.sourceNo === null && item.incomingNo === member.incomingNo && item.packages === '1'
      && identifier(transfer.id) && transfer.orderId === item.orderId && transfer.incomingNo === member.incomingNo
      && transfer.warehouseId === request.warehouseId && transfer.needRelabel === '0' && transfer.relabelNo === null && transfer.outboundAt === null
      && identifier(outbound.id) && outbound.dispatchId === root.id && outbound.outboundNo === member.outboundNo && outbound.status === 'WAIT_OUTBOUND'
      && identifier(delivery.id) && delivery.dispatchId === root.id && delivery.deliveryNo === member.deliveryNo
      && delivery.outboundNo === member.outboundNo && delivery.status === 'WAIT_DELIVERY', 'REAL_ORDINARY_TRANSFER_AND_CHILD_LINKS_REQUIRED')
    return { ...member, orderId: item.orderId, transferId: transfer.id, outboundId: outbound.id, deliveryId: delivery.id }
  }).sort((a, b) => a.customerId < b.customerId ? -1 : a.customerId > b.customerId ? 1 : 0) // Java String.compareTo for decimal customer IDs; never numeric/locale order.
  const scope = { ...request, dispatchId: root.id, tripId: trip.id, taskId: task.id, members,
    columns: rows.filter(row => row.collection === 'column') }
  return validateRollbackScope(scope)
}

export function validateRollbackScope(scope) {
  validateRollbackRequest(scope)
  demand(['dispatchId', 'tripId', 'taskId'].every(field => identifier(scope[field]))
    && scope.members.every(member => ['orderId', 'transferId', 'outboundId', 'deliveryId'].every(field => identifier(member[field])))
    && scope.members[0].customerId < scope.members[1].customerId, 'RESOLVED_SHARED_ROOT_AND_JAVA_CUSTOMER_ORDER_REQUIRED')
  for (const field of ['orderId', 'transferId', 'outboundId', 'deliveryId']) demand(new Set(scope.members.map(m => m[field])).size === 2, 'RESOLVED_MEMBER_ID_REUSED')
  if (scope.observedTransactions) demand(scope.observedTransactions.length === 2 && scope.observedTransactions.every(identifier)
    && new Set(scope.observedTransactions).size === 2, 'EXACT_OBSERVED_TRANSACTION_IDS_REQUIRED')
  demand(Array.isArray(scope.columns) && scope.columns.every(c => c.collection === 'column' && tables.includes(c.table)
    && /^[a-z][a-z0-9_]*$/.test(c.column) && Number.isSafeInteger(c.ordinal) && c.ordinal > 0 && typeof c.type === 'string'), 'ACTUAL_SCHEMA_COLUMNS_REQUIRED')
  for (const table of tables) {
    const columns = scope.columns.filter(c => c.table === table)
    demand(columns.length > 0 && columns.some(c => c.key === 'PRI') && new Set(columns.map(c => c.column)).size === columns.length
      && columns.every((c, i) => c.ordinal === i + 1), 'COMPLETE_ACTUAL_SCHEMA_REQUIRED')
  }
  const fields = table => new Set(scope.columns.filter(column => column.table === table).map(column => column.column))
  const guard = scope.columns.filter(column => column.table === 'll_physical_pallet_guard')
  demand(guard.length === 2 && guard.some(column => column.column === 'warehouse_pallet_id' && column.key === 'PRI' && column.type.toLowerCase() === 'bigint')
    && guard.some(column => column.column === 'created_at' && column.key !== 'PRI' && column.type.toLowerCase() === 'datetime(3)'), 'PERMANENT_GLOBAL_PHYSICAL_PALLET_GUARD_SCHEMA_REQUIRED')
  demand(fields('ll_dispatch_physical_pallet_lock').has('active_warehouse_pallet_id')
    && !fields('ll_dispatch_physical_pallet_lock').has('customer_id')
    && fields('ll_dispatch_pallet_lock').has('physical_lock_id') && !fields('ll_dispatch_pallet_lock').has('active_warehouse_pallet_id')
    && fields('ll_dispatch_item_source').has('dispatch_item_id') && fields('ll_dispatch_pda_pick_order').has('dispatch_item_source_id'),
  'CURRENT_PHYSICAL_AND_LOGICAL_SOURCE_SCHEMA_REQUIRED')
  return scope
}

function snapshotSql(scope) {
  const source = scope.members.map(m => `(customer_id=${literal(m.customerId)} AND id=${m.orderId})`).join(' OR ')
  const histories = scope.members.map(m => `(customer_id=${literal(m.customerId)} AND order_id=${m.orderId})`).join(' OR ')
  const customer = `customer_id IN (${scope.members.map(m => literal(m.customerId)).join(',')})`
  const stock = `${customer} AND overseas_warehouse_id=${literal(scope.warehouseId)}`
  const predicates = {
    ll_dispatch_order: `id=${scope.dispatchId}`, ll_dispatch_history: `dispatch_id=${scope.dispatchId}`,
    ll_dispatch_item: `dispatch_order_id=${scope.dispatchId}`, ll_dispatch_command: `dispatch_id=${scope.dispatchId}`,
    ll_dispatch_pallet_lock: `dispatch_order_id=${scope.dispatchId}`,
    ll_dispatch_physical_pallet_lock: `dispatch_order_id=${scope.dispatchId} OR warehouse_pallet_id IN (${scope.members.map(member => member.palletId).join(',')})`,
    ll_physical_pallet_guard: `warehouse_pallet_id IN (${scope.members.map(member => member.palletId).join(',')})`,
    ll_dispatch_item_source: `dispatch_order_id=${scope.dispatchId} OR dispatch_item_id IN (${scope.members.map(member => member.dispatchItemId).join(',')})`,
    ll_fleet_trip: `id=${scope.tripId}`,
    ll_fleet_trip_dispatch: `trip_id=${scope.tripId} OR dispatch_id=${scope.dispatchId}`, ll_fleet_trip_history: `trip_id=${scope.tripId}`,
    ll_fleet_pool_order_link: `dispatch_no=${literal(scope.dispatchNo)}`, ll_outbound_order: `dispatch_order_id=${scope.dispatchId}`,
    ll_delivery_order: `dispatch_order_id=${scope.dispatchId}`, ll_order: source, ll_loading_order_status_history: histories,
    ll_warehouse_transfer_order: `lastleg_order_id IN (${scope.members.map(m => m.orderId).join(',')})`,
    ll_dispatch_pda_task: `dispatch_order_id=${scope.dispatchId}`, ll_dispatch_pda_pick: `task_id=${scope.taskId}`,
    ll_dispatch_pda_pick_order: `pick_id IN (SELECT id FROM ll_dispatch_pda_pick WHERE task_id=${scope.taskId})`,
    ll_inventory_balance: stock, ll_inventory_position: stock, ll_inventory_ledger: stock,
  }
  return [columnsSql, ...tables.flatMap(table => {
    const columns = scope.columns.filter(c => c.table === table)
    const json = columns.map(c => `'${c.column}',CAST(\`${c.column}\` AS CHAR CHARACTER SET utf8mb4)`).join(',')
    return [`SELECT JSON_OBJECT('collection','coverage','table','${table}','count',COUNT(*)) FROM ${table} WHERE ${predicates[table]};`,
      `SELECT JSON_OBJECT('collection','row','table','${table}','row',JSON_OBJECT(${json})) FROM ${table} WHERE ${predicates[table]};`]
  })].join('\n')
}

function lockSql(scope) {
  const target = scope.members[1]
  const held = `l.OBJECT_SCHEMA=DATABASE() AND l.OBJECT_NAME='ll_warehouse_transfer_order' AND l.INDEX_NAME='PRIMARY' AND l.LOCK_DATA='${target.transferId}'`
  const lease = `SELECT JSON_OBJECT('collection','lease','connectionId',CAST(t.trx_mysql_thread_id AS CHAR),'transactionId',CAST(t.trx_id AS CHAR),'lockId',l.LOCK_DATA,'mode',l.LOCK_MODE,'ageSeconds',TIMESTAMPDIFF(SECOND,t.trx_started,NOW())) FROM performance_schema.data_locks l JOIN information_schema.innodb_trx t ON t.trx_id=l.ENGINE_TRANSACTION_ID WHERE ${held} AND l.LOCK_STATUS='GRANTED' AND t.trx_query LIKE ${literal('%ROLLBACK_' + scope.marker + '%')};`
  const waits = `SELECT JSON_OBJECT('collection','wait','requester',CAST(r.trx_mysql_thread_id AS CHAR),'blocker',CAST(b.trx_mysql_thread_id AS CHAR),'requestingTransactionId',CAST(r.trx_id AS CHAR),'blockingTransactionId',CAST(b.trx_id AS CHAR),'lockId',l.LOCK_DATA,'blockingLockId',bl.LOCK_DATA,'objectName',l.OBJECT_NAME,'indexName',l.INDEX_NAME,'requestStatus',l.LOCK_STATUS,'blockingStatus',bl.LOCK_STATUS,'statement',r.trx_query,'sessionLockWaitSeconds',v.VARIABLE_VALUE,'ownsRoot',EXISTS(SELECT 1 FROM performance_schema.data_locks dl WHERE dl.ENGINE_TRANSACTION_ID=r.trx_id AND dl.OBJECT_SCHEMA=DATABASE() AND dl.OBJECT_NAME='ll_dispatch_order' AND dl.INDEX_NAME='PRIMARY' AND dl.LOCK_DATA='${scope.dispatchId}' AND dl.LOCK_STATUS='GRANTED' AND dl.LOCK_MODE LIKE 'X%')) FROM performance_schema.data_lock_waits w JOIN performance_schema.data_locks l ON l.ENGINE_LOCK_ID=w.REQUESTING_ENGINE_LOCK_ID AND l.ENGINE=w.ENGINE JOIN performance_schema.data_locks bl ON bl.ENGINE_LOCK_ID=w.BLOCKING_ENGINE_LOCK_ID AND bl.ENGINE=w.ENGINE JOIN information_schema.innodb_trx r ON r.trx_id=w.REQUESTING_ENGINE_TRANSACTION_ID JOIN information_schema.innodb_trx b ON b.trx_id=w.BLOCKING_ENGINE_TRANSACTION_ID JOIN performance_schema.threads th ON th.PROCESSLIST_ID=r.trx_mysql_thread_id JOIN performance_schema.variables_by_thread v ON v.THREAD_ID=th.THREAD_ID AND v.VARIABLE_NAME='innodb_lock_wait_timeout' WHERE ${held} AND bl.OBJECT_SCHEMA=DATABASE() AND bl.OBJECT_NAME=l.OBJECT_NAME AND bl.INDEX_NAME='PRIMARY' AND bl.LOCK_DATA=l.LOCK_DATA AND b.trx_query LIKE ${literal('%ROLLBACK_' + scope.marker + '%')};`
  return [lease, waits]
}

export function rollbackSql(scope, mode) {
  if (mode === 'RESOLVE') return resolutionSql(scope)
  validateRollbackScope(scope)
  const second = scope.members[1]
  if (mode === 'LEASE') return `SET NAMES utf8mb4 COLLATE utf8mb4_bin;${databaseRuntime}SET SESSION innodb_lock_wait_timeout=5;START TRANSACTION;SELECT JSON_OBJECT('collection','held','lockId',CAST(id AS CHAR),'customerId',customer_id,'connectionId',CAST(CONNECTION_ID() AS CHAR)) FROM ll_warehouse_transfer_order WHERE id=${second.transferId} AND customer_id=${literal(second.customerId)} AND lastleg_order_id=${second.orderId} AND transfer_no=${literal(second.incomingNo)} AND need_relabel=0 AND relabel_no IS NULL AND outbound_at IS NULL FOR UPDATE;SELECT JSON_OBJECT('marker','ROLLBACK_${scope.marker}','elapsed',SLEEP(65));ROLLBACK;SELECT JSON_OBJECT('released',JSON_EXTRACT('true','$'));`
  if (mode === 'LOCKS' || mode === 'RELEASED') return [...rr, databaseRuntime, ...lockSql(scope),
    ...(mode === 'RELEASED' ? [(() => {
      demand(scope.observedTransactions, 'OBSERVED_TRANSACTIONS_REQUIRED_FOR_RELEASE_PROOF')
      return `SELECT JSON_OBJECT('collection','remainingTransaction','transactionId',CAST(trx_id AS CHAR)) FROM information_schema.innodb_trx WHERE trx_id IN (${scope.observedTransactions.map(literal).join(',')});`
    })()] : []), 'COMMIT;'].join('\n')
  if (['BASELINE', 'RESTORED', 'OBSERVE'].includes(mode)) return [
    ...(mode === 'OBSERVE' ? ['SET NAMES utf8mb4 COLLATE utf8mb4_bin;', 'SET SESSION TRANSACTION ISOLATION LEVEL READ UNCOMMITTED;', 'START TRANSACTION READ ONLY;'] : rr),
    databaseRuntime, ...(mode === 'OBSERVE' ? lockSql(scope) : []), snapshotSql(scope), 'COMMIT;'].join('\n')
  throw Error('ROLLBACK_MODE_INVALID')
}

export function buildRollbackRemoteScript(scope, mode, agent) {
  demand(['RUNTIME', 'RESOLVE', 'LEASE', 'LOCKS', 'OBSERVE', 'BASELINE', 'RESTORED', 'RELEASED'].includes(mode)
    && /^[A-Za-z0-9_.-]{1,128}$/.test(agent || ''), 'ROLLBACK_REMOTE_MODE_REQUIRED')
  const sql = mode === 'RUNTIME' ? [databaseRuntime,
    ...['data_locks', 'data_lock_waits', 'threads', 'variables_by_thread'].map(table => `SELECT JSON_OBJECT('collection','privilege','table','${table}','count',COUNT(*)) FROM performance_schema.${table} WHERE 1=0;`),
    "SELECT JSON_OBJECT('collection','privilege','table','innodb_trx','count',COUNT(*)) FROM information_schema.innodb_trx WHERE 1=0;"].join('\n') : rollbackSql(scope, mode)
  const imageGuard = mode === 'RUNTIME' ? '' : ` or c['Image']!='${scope.expectedImage}'`
  const python = `import json,os,subprocess,urllib.parse,re
try:
 c=json.loads(subprocess.check_output(['docker','inspect','firstleg-test-lastleg'],text=True,stderr=subprocess.DEVNULL))[0]
 e=dict(x.split('=',1) for x in c['Config']['Env'])
 if c['Name']!='/firstleg-test-lastleg' or c['Config'].get('Labels',{}).get('com.docker.compose.project')!='firstleg-test' or e.get('DEPLOY_ENV')!='test' or c['State'].get('Health',{}).get('Status')!='healthy'${imageGuard}:raise RuntimeError('TEST_RUNTIME_CHANGED')
 u=urllib.parse.urlsplit(e['LASTLEG_DB_URL'].removeprefix('jdbc:'));host='127.0.0.1' if u.hostname=='host.docker.internal' else u.hostname
 q=subprocess.run(['mysql','--host='+host,'--port='+str(u.port or 3306),'--user='+e['LASTLEG_DB_USERNAME'],'--batch','--raw','--skip-column-names',u.path.lstrip('/')],input=${JSON.stringify(sql)},capture_output=True,text=True,env=dict(os.environ,MYSQL_PWD=e['LASTLEG_DB_PASSWORD']),timeout=${mode === 'LEASE' ? 75 : 25})
 if q.returncode:
  code=re.search(r'ERROR ([0-9]{3,5})',q.stderr);print('TRIP_ROLLBACK_JSON='+json.dumps({'ok':False,'error':'MYSQL_'+(code.group(1) if code else 'UNKNOWN')}))
 else:print('TRIP_ROLLBACK_JSON='+json.dumps({'ok':True,'imageId':c['Image'],'rows':[json.loads(x) for x in q.stdout.splitlines() if x.strip()]}))
except Exception as error:
 print('TRIP_ROLLBACK_JSON='+json.dumps({'ok':False,'error':str(error) if re.fullmatch('[A-Z_]{1,64}',str(error)) else 'REMOTE_ROLLBACK_FAILED'}))
`
  const groovy = `def p=new ProcessBuilder(['python3','-c',new String(java.util.Base64.decoder.decode(${encoded(python)}),'UTF-8')]).redirectErrorStream(true).start();def out=p.inputStream.getText('UTF-8');p.waitFor();print(out)`
  return `def c=jenkins.model.Jenkins.instance.getComputer('${agent}');if(c==null||c.channel==null)throw new RuntimeException('TEST agent unavailable');print(hudson.util.RemotingDiagnostics.executeGroovy(new String(java.util.Base64.decoder.decode(${encoded(groovy)}),'UTF-8'),c.channel))`
}

export function fullRollbackSnapshot(rows, scope) {
  validateRollbackScope(scope)
  demand(canonical(rows.filter(r => r.collection === 'column')) === canonical(scope.columns), 'ACTUAL_SCHEMA_CHANGED')
  const result = {}
  for (const table of tables) {
    const coverage = one(rows.filter(r => r.table === table), 'coverage')
    const values = rows.filter(r => r.collection === 'row' && r.table === table).map(r => r.row)
    const columns = scope.columns.filter(c => c.table === table).map(c => c.column).sort()
    demand(Number.isSafeInteger(coverage.count) && coverage.count === values.length
      && values.every(r => canonical(Object.keys(r).sort()) === canonical(columns) && Object.values(r).every(v => v === null || typeof v === 'string'))
      && new Set(values.map(canonical)).size === values.length, 'COMPLETE_FULL_FIELD_SNAPSHOT_REQUIRED')
    result[table] = values.sort((a, b) => canonical(a) < canonical(b) ? -1 : canonical(a) > canonical(b) ? 1 : 0)
  }
  return result
}
const exactRow = (snapshot, table, id) => {
  const found = snapshot[table].filter(row => row.id === id)
  demand(found.length === 1, `EXACT_OWNED_ROW_REQUIRED_${table}`); return found[0]
}
const immutableTables = ['ll_physical_pallet_guard', 'll_dispatch_physical_pallet_lock', 'll_dispatch_item_source', 'll_dispatch_pda_task', 'll_dispatch_pda_pick', 'll_dispatch_pda_pick_order', 'll_inventory_balance', 'll_inventory_position', 'll_inventory_ledger']
export function assertRollbackBaseline(snapshot, scope) {
  const root = exactRow(snapshot, 'll_dispatch_order', scope.dispatchId), trip = exactRow(snapshot, 'll_fleet_trip', scope.tripId)
  demand(snapshot.ll_dispatch_order.length === 1 && snapshot.ll_fleet_trip.length === 1 && snapshot.ll_fleet_trip_dispatch.length === 1
    && root.status === 'ACTIVE' && root.stage === 'AFTER_LOADING' && root.record_version === scope.commandVersion && root.departed_at === null
    && root.shipped_package_count === '0' && trip.status === 'PLANNED' && trip.departed_at === null, 'READY_NATIVE_BASELINE_REQUIRED')
  demand(snapshot.ll_fleet_pool_order_link.length === 0, 'ORDINARY_TRANSFER_FIXTURE_CANNOT_HAVE_POOL_LINKS')
  demand(snapshot.ll_dispatch_item_source.length === 0, 'ORDINARY_TRANSFER_FIXTURE_CANNOT_HAVE_RELABEL_CONTRIBUTIONS')
  for (const table of ['ll_dispatch_item', 'll_outbound_order', 'll_delivery_order', 'll_order', 'll_warehouse_transfer_order', 'll_dispatch_pallet_lock', 'll_dispatch_physical_pallet_lock', 'll_physical_pallet_guard', 'll_dispatch_pda_pick', 'll_dispatch_pda_pick_order'])
    demand(snapshot[table].length === 2, 'EXACT_TWO_CUSTOMER_BASELINE_REQUIRED')
  demand(snapshot.ll_dispatch_pda_task.length === 1 && snapshot.ll_dispatch_pda_task[0].id === scope.taskId, 'ORIGINAL_TASK_REQUIRED')
  demand(!snapshot.ll_dispatch_command.some(r => r.action === 'DEPART' || r.idempotency_key === scope.commandKey)
    && !snapshot.ll_dispatch_history.some(r => r.action === 'DEPART') && !snapshot.ll_fleet_trip_history.some(r => r.action === 'DEPART')
    && !snapshot.ll_loading_order_status_history.some(r => r.event_id === 'DISPATCH_DEPART:' + scope.dispatchNo), 'FRESH_DEPARTURE_REQUIRED')
  for (const member of scope.members) {
    const guards = snapshot.ll_physical_pallet_guard.filter(row => row.warehouse_pallet_id === member.palletId)
    demand(guards.length === 1 && /^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}(?:\.\d{1,3})?$/.test(guards[0].created_at || ''), 'OWNED_PERMANENT_PHYSICAL_PALLET_GUARD_REQUIRED')
    const order = exactRow(snapshot, 'll_order', member.orderId), transfer = exactRow(snapshot, 'll_warehouse_transfer_order', member.transferId)
    const outbound = exactRow(snapshot, 'll_outbound_order', member.outboundId), delivery = exactRow(snapshot, 'll_delivery_order', member.deliveryId)
    demand([order, transfer, outbound, delivery].every(r => r.customer_id === member.customerId)
      && order.picked_package_count === '1' && order.shipped_package_count === '0'
      && transfer.lastleg_order_id === member.orderId && transfer.transfer_no === member.incomingNo && transfer.outbound_at === null
      && outbound.status === 'WAIT_OUTBOUND' && delivery.status === 'WAIT_DELIVERY' && delivery.dispatched_at === null, 'OWNED_SOURCE_AND_CHILD_BASELINE_REQUIRED')
    const picks = snapshot.ll_dispatch_pda_pick.filter(r => r.customer_id === member.customerId)
    const attribution = snapshot.ll_dispatch_pda_pick_order.filter(r => r.customer_id === member.customerId)
    demand(picks.length === 1 && picks[0].task_id === scope.taskId && picks[0].dispatch_item_id === member.dispatchItemId
      && picks[0].inventory_position_id === member.positionId && picks[0].packages === '1' && picks[0].return_carton_id === null
      && attribution.length === 1 && attribution[0].pick_id === picks[0].id && attribution[0].order_id === member.orderId
      && identifier(attribution[0].receipt_id) && attribution[0].packages === '1' && attribution[0].return_receipt_id === null
      && attribution[0].return_carton_id === null && attribution[0].dispatch_item_source_id === null, 'COMMITTED_REAL_PICK_ATTRIBUTION_REQUIRED')
    const ship = snapshot.ll_inventory_ledger.filter(r => r.customer_id === member.customerId && r.business_type === 'SHIP')
    demand(ship.length === 1 && ship[0].business_order_no === scope.dispatchNo && ship[0].position_id === member.positionId
      && ship[0].balance_id === member.balanceId && ship[0].delta_packages === '-1', 'PRIOR_COMMITTED_SHIP_REQUIRED')
    const claims = snapshot.ll_dispatch_pallet_lock.filter(r => r.customer_id === member.customerId)
    demand(claims.length === 1 && claims[0].dispatch_order_id === scope.dispatchId && claims[0].warehouse_pallet_id === member.palletId
      && claims[0].status === 'ACTIVE' && identifier(claims[0].physical_lock_id), 'ACTIVE_OWNED_CLAIM_REQUIRED')
    const physical = exactRow(snapshot, 'll_dispatch_physical_pallet_lock', claims[0].physical_lock_id)
    demand(physical.warehouse_id === scope.warehouseId && physical.warehouse_pallet_id === member.palletId
      && physical.active_warehouse_pallet_id === member.palletId && physical.dispatch_order_id === scope.dispatchId
      && physical.status === 'ACTIVE' && physical.draft_token === null && physical.draft_container_id === null
      && physical.expires_at === null && physical.released_at === null, 'ACTIVE_NATIVE_PHYSICAL_PALLET_OWNER_REQUIRED')
  }
  return snapshot
}

export function assertTargetTransferUpdate(statement, scope) {
  demand(typeof statement === 'string', 'TARGET_TRANSFER_UPDATE_NOT_PROVED')
  const sql = statement.replace(/`/g, '').replace(/\s+/g, ' ').replace(/\s*([=,])\s*/g, '$1').trim()
  const value = "(?:\\?|'[^']*')"
  const match = sql.match(new RegExp(`^UPDATE ll_warehouse_transfer_order SET status='OUTBOUNDED',\\s*outbound_at=${value},\\s*updated_at=${value} WHERE customer_id=(\\?|'[0-9]+') AND lastleg_order_id=(\\?|[0-9]+) AND outbound_at IS NULL$`, 'i'))
  const second = scope.members[1]
  demand(match && (match[1] === '?' || match[1] === `'${second.customerId}'`) && (match[2] === '?' || match[2] === second.orderId), 'TARGET_TRANSFER_UPDATE_NOT_PROVED')
  return true
}

export function assertMidTransactionObservation(rows, scope, baseline) {
  const lease = one(rows, 'lease'), wait = one(rows, 'wait'), second = scope.members[1]
  demand(identifier(lease.connectionId) && identifier(wait.requester) && wait.blocker === lease.connectionId && wait.requester !== wait.blocker
    && identifier(lease.transactionId) && identifier(wait.requestingTransactionId) && wait.blockingTransactionId === lease.transactionId
    && wait.requestingTransactionId !== lease.transactionId
    && lease.lockId === second.transferId && wait.lockId === lease.lockId && wait.blockingLockId === lease.lockId
    && /^X(?:,|$)/.test(lease.mode) && wait.objectName === 'll_warehouse_transfer_order' && wait.indexName === 'PRIMARY'
    && wait.requestStatus === 'WAITING' && wait.blockingStatus === 'GRANTED' && wait.ownsRoot === 1, 'OWNED_LATE_ROW_WAIT_NOT_PROVED')
  assertTargetTransferUpdate(wait.statement, scope)
  demand(wait.sessionLockWaitSeconds === '50', 'REQUESTING_SESSION_50_SECOND_LOCK_TIMEOUT_NOT_PROVED')
  const dirty = fullRollbackSnapshot(rows, scope)
  for (const member of scope.members) {
    const before = exactRow(baseline, 'll_order', member.orderId), order = exactRow(dirty, 'll_order', member.orderId)
    const outbound = exactRow(dirty, 'll_outbound_order', member.outboundId), delivery = exactRow(dirty, 'll_delivery_order', member.deliveryId)
    demand(outbound.status === 'OUTBOUNDED' && delivery.status === 'DELIVERING' && delivery.dispatched_at !== null
      && BigInt(delivery.version) === BigInt(exactRow(baseline, 'll_delivery_order', member.deliveryId).version) + 1n
      && BigInt(order.picked_package_count) === BigInt(before.picked_package_count) - 1n
      && BigInt(order.shipped_package_count) === BigInt(before.shipped_package_count) + 1n
      && order.outbound_order_no === member.outboundNo, 'CUSTOMER_UNCOMMITTED_CHILD_AND_SOURCE_WRITE_NOT_PROVED')
    const history = dirty.ll_loading_order_status_history.filter(r => r.customer_id === member.customerId && r.order_id === member.orderId && r.event_id === 'DISPATCH_DEPART:' + scope.dispatchNo)
    demand(history.length === 1 && !baseline.ll_loading_order_status_history.some(r => r.id === history[0].id), 'UNCOMMITTED_SOURCE_HISTORY_NOT_PROVED')
  }
  const firstTransfer = exactRow(dirty, 'll_warehouse_transfer_order', scope.members[0].transferId)
  demand(firstTransfer.status === 'OUTBOUNDED' && firstTransfer.outbound_at !== null
    && canonical(exactRow(dirty, 'll_warehouse_transfer_order', second.transferId)) === canonical(exactRow(baseline, 'll_warehouse_transfer_order', second.transferId)), 'FIRST_TRANSFER_WRITTEN_SECOND_TRANSFER_WAITING_NOT_PROVED')
  for (const table of ['ll_dispatch_order', 'll_fleet_trip', 'll_dispatch_command', 'll_dispatch_history', 'll_fleet_trip_history', 'll_dispatch_pallet_lock', ...immutableTables])
    demand(canonical(dirty[table]) === canonical(baseline[table]), 'LATE_PHASE_OR_COMMITTED_SCAN_FACTS_CHANGED')
  return { leaseConnectionId: lease.connectionId, requestingConnectionId: wait.requester, leaseTransactionId: lease.transactionId,
    requestingTransactionId: wait.requestingTransactionId, lockId: lease.lockId,
    sessionLockWaitSeconds: wait.sessionLockWaitSeconds, statement: wait.statement, dirtyObservationOnly: true, firstCustomerId: scope.members[0].customerId, secondCustomerId: second.customerId }
}

export function assertRestoredSnapshot(baseline, restored) {
  demand(canonical(baseline) === canonical(restored), 'FULL_FIELD_COMMITTED_ROLLBACK_SNAPSHOT_CHANGED')
  return { beforeSha256: sha(baseline), afterSha256: sha(restored), fullFields: true, committedRead: true }
}
export function requestRollbackDeparture(context, path, body, token) {
  // Existing 60-second, single-attempt channel: an unknown HTTP outcome is never retried here.
  return context.requestWithBody('POST', path, body, token)
}
export function assertDatabaseLockConflict(response) {
  demand(response?.http === 200 && String(response.body?.code) === 'A0409' && response.body?.message === '资源正在处理中，请稍后重试'
    && response.requestId && response.traceId, 'EXPECTED_DATABASE_LOCK_CONFLICT_REQUIRED')
}
async function remote(env, scope, mode) {
  const script = buildRollbackRemoteScript(scope, mode, env.FBA_CORE_JENKINS_AGENT)
  const response = await fetch(new URL('scriptText', env.JENKINS_URL.replace(/\/?$/, '/')), { method: 'POST', redirect: 'error',
    signal: AbortSignal.timeout(mode === 'LEASE' ? 95000 : 45000), headers: { Authorization: 'Basic ' + Buffer.from(env.JENKINS_USER + ':' + env.JENKINS_API_TOKEN).toString('base64'),
      'Content-Type': 'application/x-www-form-urlencoded' }, body: new URLSearchParams({ script }) })
  demand(response.ok, 'ROLLBACK_EVIDENCE_HTTP_FAILED')
  const lines = (await response.text()).split(/\r?\n/).filter(x => x.startsWith('TRIP_ROLLBACK_JSON='))
  demand(lines.length === 1, 'ROLLBACK_REMOTE_RESULT_AMBIGUOUS')
  const result = JSON.parse(lines[0].slice('TRIP_ROLLBACK_JSON='.length))
  demand(result.ok === true, `ROLLBACK_${mode}_EVIDENCE_BLOCKED_${/^[A-Z_0-9]+$/.test(result.error || '') ? result.error : 'REMOTE_FAILED'}`)
  const identity = one(result.rows, 'databaseRuntime')
  demand(identity.performanceSchema === 1 && identity.serverUuid && identity.databaseName && /^sha256:[a-f0-9]{64}$/.test(result.imageId || ''), 'OBSERVABLE_DATABASE_RUNTIME_REQUIRED')
  if (scope.runtime) demand(result.imageId === scope.expectedImage && canonical(identity) === canonical(scope.runtime), 'PINNED_DATABASE_RUNTIME_CHANGED')
  return result
}

export function createDepartureRollbackVerifier(env, config, runtime, { read = remote, wait = pause } = {}) {
  return async ({ context, fixture, ready, departPath, departBody, snapshot }) => {
    const output = resolve('reports', `fleet-cross-${config.runId}`, 'rollback.json'), fd = fs.openSync(output, 'wx', 0o600)
    const originalCommand = canonical({ path: departPath, body: departBody })
    const report = { verified: false, scope: 'OWNED_TWO_CUSTOMER_DEPARTURE_ROLLBACK', startedAt: new Date().toISOString(), samples: [],
      command: JSON.parse(originalCommand), runtime }
    const save = () => { fs.ftruncateSync(fd, 0); fs.writeSync(fd, JSON.stringify(report, null, 2), 0, 'utf8'); fs.fsyncSync(fd) }
    let leasePromise, pending
    try {
      demand(ready.trip.dispatches?.length === 1 && ready.trip.dispatches[0].dispatchNo === fixture.dispatchNo && fixture.members.length === 2
        && ready.dispatch.dispatchNo === fixture.dispatchNo && String(ready.task.recordVersion) === String(departBody.recordVersion), 'ONE_NATIVE_READY_SCOPE_REQUIRED')
      const request = validateRollbackRequest({ marker: context.marker, tripNo: fixture.tripNo, dispatchNo: fixture.dispatchNo, warehouseId: fixture.warehouseId,
        expectedImage: runtime.imageId, runtime: one(runtime.rows, 'databaseRuntime'), commandKey: departBody.idempotencyKey, commandVersion: String(departBody.recordVersion),
        members: fixture.members.map(({ customerId, incomingNo, dispatchItemId, palletId, positionId, balanceId, outboundNo, deliveryNo }) =>
          ({ customerId, incomingNo, dispatchItemId, palletId, positionId, balanceId, outboundNo, deliveryNo })) })
      const resolved = await read(env, request, 'RESOLVE')
      const scope = report.target = resolveRollbackScope(request, resolved.rows)
      const baseline = report.baseline = assertRollbackBaseline(fullRollbackSnapshot((await read(env, scope, 'BASELINE')).rows, scope), scope)
      save()
      leasePromise = read(env, scope, 'LEASE').then(value => ({ value }), error => ({ error: error.message }))
      let lease
      for (let i = 0; i < 12; i++) {
        const rows = (await read(env, scope, 'LOCKS')).rows; report.samples.push({ phase: 'LEASE', rows }); save()
        if (rows.filter(x => x.collection === 'lease').length === 1) {
          lease = one(rows, 'lease')
          demand(lease.lockId === scope.members[1].transferId && identifier(lease.connectionId) && /^X(?:,|$)/.test(lease.mode)
            && Number.isSafeInteger(lease.ageSeconds) && lease.ageSeconds >= 0 && lease.ageSeconds <= 5, 'FRESH_OWNED_65_SECOND_LEASE_REQUIRED')
          break
        }
        demand(i !== 11, 'LIVE_LEASE_NOT_OBSERVED'); await wait(250)
      }
      pending = requestRollbackDeparture(context, departPath, departBody, fixture.pdaToken).then(value => ({ value }), error => ({ error: error.message }))
      for (let i = 0; i < 20; i++) {
        const rows = (await read(env, scope, 'OBSERVE')).rows; report.samples.push({ phase: 'MID_TRANSACTION', rows }); save()
        if (rows.some(x => x.collection === 'wait')) {
          report.observation = assertMidTransactionObservation(rows, scope, baseline)
          demand(report.observation.leaseConnectionId === lease.connectionId && report.observation.leaseTransactionId === lease.transactionId, 'LEASE_CONNECTION_CHANGED')
          scope.observedTransactions = [report.observation.leaseTransactionId, report.observation.requestingTransactionId]
          save(); break
        }
        demand(i !== 19, 'MID_TRANSACTION_NOT_OBSERVED'); await wait(250)
      }
      report.response = await pending; save()
      report.lease = await leasePromise; save()
      demand(!report.lease.error && report.lease.value.rows.at(-1)?.released === true, 'LEASE_RELEASE_NOT_PROVED')
      const held = one(report.lease.value.rows, 'held')
      demand(held.lockId === lease.lockId && held.connectionId === lease.connectionId && held.customerId === scope.members[1].customerId, 'ORIGINAL_LEASE_RESULT_CHANGED')
      const released = report.released = (await read(env, scope, 'RELEASED')).rows
      demand(!released.some(r => ['lease', 'wait', 'remainingTransaction'].includes(r.collection)), 'OWNED_ROW_LOCK_OR_TRANSACTION_REMAINS')
      // A new RR consistent snapshot after the server outcome and lease completion, never the dirty observer transaction.
      report.restored = fullRollbackSnapshot((await read(env, scope, 'RESTORED')).rows, scope)
      report.consistentSnapshot = assertRestoredSnapshot(baseline, report.restored)
      demand(!report.response.error && report.response.value?.http !== 0, 'DEPARTURE_OUTCOME_UNKNOWN_RETAIN_ORIGINAL_COMMAND')
      assertDatabaseLockConflict(report.response.value)
      demand(canonical(ready) === canonical(await snapshot('ROLLBACK_AFTER_FAULT')), 'PUBLIC_AND_COMPLETE_INVENTORY_SNAPSHOT_CHANGED')
      demand(canonical({ path: departPath, body: departBody }) === originalCommand, 'ORIGINAL_DEPART_COMMAND_CHANGED')
      // The existing caller now sends this identical original command once, then verifies replay and both customer receipts.
      report.verified = true; return report
    } catch (error) { report.error = error.message; throw error }
    finally {
      if (pending && !report.response) report.response = await pending
      if (leasePromise && !report.lease) report.lease = await leasePromise
      report.finishedAt = new Date().toISOString(); save(); fs.closeSync(fd)
    }
  }
}

export async function runTripRollback(config, { env = process.env } = {}) {
  if (!config.execute) return { ...config, scope: 'OWNED_TWO_CUSTOMER_DEPARTURE_ROLLBACK', fault: 'One native dispatch; two ordinary WAREHOUSE_TRANSFER customers in Java string order; 65-second read-only row lease on the second transfer; exact UPDATE waiter and its own 50-second session timeout; uncommitted first customer writes; fresh full-field committed rollback baseline; same original command and normal signing chain', businessSqlWrites: false, deployment: false }
  demand(config.environment === 'TEST' && env.CHONGCHONG_ENV === 'test' && ['JENKINS_URL', 'JENKINS_USER', 'JENKINS_API_TOKEN', 'FBA_CORE_JENKINS_AGENT'].every(key => env[key]), 'EXPLICIT_TEST_EVIDENCE_PROFILE_REQUIRED')
  const runtime = await remote(env, {}, 'RUNTIME')
  demand(runtime.rows.filter(row => row.collection === 'privilege').length === 5, 'LOCK_AND_SESSION_OBSERVABILITY_REQUIRED')
  const sources = ['scripts/fleet-trip-rollback-test-env.mjs', 'scripts/admin-test-env-modules/fleet-cross-customer-acceptance.mjs'].map(path => ({ path, sha256: createHash('sha256').update(fs.readFileSync(path)).digest('hex') }))
  return runCrossCustomer(config, { env, verify: async (context, fixture, options) => {
    fs.writeFileSync(resolve('reports', `fleet-cross-${config.runId}`, 'rollback-source.json'), JSON.stringify({ runtime, sources }, null, 2), { flag: 'wx' })
    return verifyFleetCrossCustomer(context, fixture, { ...options, verifyDepartureRollback: createDepartureRollbackVerifier(env, config, runtime) })
  } })
}
if (process.argv[1] && import.meta.url === pathToFileURL(resolve(process.argv[1])).href) {
  try { const config = parseCrossCustomerArguments(process.argv.slice(2)), result = await runTripRollback(config); console.log(JSON.stringify(config.execute ? { success: result.success, error: result.error, summary: result.summary, requests: result.requestCount, goalComplete: false } : result, null, 2)); if (config.execute && !result.success) process.exitCode = 1 }
  catch (error) { console.error(/^[A-Z_0-9]+$/.test(error.message) ? error.message : 'ROLLBACK_ENTRY_FAILED'); process.exitCode = 1 }
}