baiyue-firstleg/scripts/test-event-fault-executor.test.mjs

import assert from 'node:assert/strict'
import test from 'node:test'
import { createHash } from 'node:crypto'
import { spawnSync } from 'node:child_process'
import { createTestFaultExecutor, createTestUnloadFaultExecutor, createTestDeliveryFaultExecutor, assertTestDeliveryFaultProof, assertTestUnloadFaultProof, renderTestFaultAction } from './test-event-fault-executor.mjs'
import { withDeliveryEventRecovery } from './admin-test-env-modules/phase5-root-fulfillment.mjs'

const scope = { marker: 'TEST-ROOT', customerId: '7', loadingPlanId: '99', containerBookingId: '66' }
const environment = { CHONGCHONG_ENV: 'test', JENKINS_URL: 'https://jenkins.example.invalid/',
  JENKINS_USER: 'offline', JENKINS_API_TOKEN: 'offline-secret', FBA_CORE_JENKINS_AGENT: 'java-test-agent' }
function requestedAction(options) {
  const script = options.body.get('script')
  const chunks = [...script.matchAll(/'([A-Za-z0-9+/=]+)'/g)].map(match => match[1]).join('')
  const groovy = Buffer.from(chunks, 'base64').toString('utf8')
  const argument = groovy.match(/'UTF-8'\),'([A-Za-z0-9+/=]+)'\]\)/)?.[1]
  return JSON.parse(Buffer.from(argument, 'base64').toString('utf8'))
}
function transport(calls, failActivate = false) {
  let recovered = false
  return async (_url, options) => {
    assert.equal(options.redirect, 'error')
    const { action } = requestedAction(options)
    calls.push(action)
    if (action === 'recover') recovered = true
    const data = action === 'proxy-ready' ? { ready: true } : action === 'health' ? { healthy: true }
      : action === 'status' ? { producer: { publishStatus: recovered ? 'PUBLISHED' : 'DEAD', failedAttempts: 12 } }
        : action === 'proof' ? { ...scope, restored: false } : {}
    const result = action === 'activate' && failActivate ? { ok: false, error: 'DOCKER_409' } : { ok: true, data }
    return { ok: true, text: async () => 'TEST_EVENT_JSON=' + JSON.stringify(result) }
  }
}
test('scope and actions are bounded and renderer never embeds caller credentials', () => {
  const rendered = renderTestFaultAction('status', scope, environment.FBA_CORE_JENKINS_AGENT)
  assert.ok(!rendered.includes(environment.JENKINS_API_TOKEN))
  for (const change of [{ ...scope, customerId: '7;DROP' }, { ...scope, marker: '../other' }]) {
    assert.throws(() => renderTestFaultAction('status', change, environment.FBA_CORE_JENKINS_AGENT))
  }
  assert.throws(() => renderTestFaultAction('arbitrary', scope, 'java-test-agent'))
  assert.throws(() => createTestFaultExecutor({ ...environment, CHONGCHONG_ENV: 'prod' }))
})

const unloadScope = { marker: 'TEST-ROOT', customerId: '7', containerBookingId: '66' }
const originalImageId = 'sha256:' + 'a'.repeat(64), originalFingerprint = 'b'.repeat(64)
function unloadProof(nanoseconds = false) {
  const payload = { actualEmptyReportedAt: null, containerNo: 'OFFLINE66', sourceContainerBookingId: '66',
    unloadingCompletedAt: nanoseconds ? '2026-10-02T01:00:00.136827515Z' : '2026-10-02T01:00:00Z', unloadingPhotoUploadedAt: '2026-10-02T00:59:00Z' }
  const payloadHash = 'sha256:' + createHash('sha256').update(JSON.stringify(payload)).digest('hex')
  const eventId = '00000000-0000-4000-8000-000000000001', entryId = '123'
  const envelope = { eventId, eventType: 'OverseasContainerUnloaded', eventVersion: 1, tenantId: '0',
    aggregateId: '66', aggregateType: 'CONTAINER_BOOKING', aggregateVersion: 3, payloadHash, payload }
  const row = { entryId, eventId, payloadHash, envelope, eventType: envelope.eventType, customerId: '0',
    aggregateId: '66', aggregateType: 'CONTAINER_BOOKING', eventKey: 'OverseasContainerUnloaded:66', publishStatus: 'PUBLISHED' }
  const container = { id: '66', containerNo: 'OFFLINE66', version: '9', emptyReportSourceVersion: '3',
    unloadingCompletedAt: nanoseconds ? '2026-10-02 01:00:00.137' : '2026-10-02 01:00:00', unloadingPhotoUploadedAt: '2026-10-02 00:59:00',
    estimatedEmptyReportedAt: nanoseconds ? '2026-10-05 01:00:00.137' : '2026-10-05 01:00:00', actualEmptyReportedAt: null }
  return JSON.parse(JSON.stringify({ ...unloadScope, readOnly: true, producer: row,
    beforeRetry: { ...row, publishStatus: 'DEAD', failedAttempts: 12, container }, afterRetry: { ...row, container },
    consumer: { count: 1, status: 'PROCESSED', eventId, payloadHash, eventType: envelope.eventType, aggregateId: '66', payload },
    attempts: Array.from({ length: 13 }, (_, i) => ({ attemptId: `attempt-${i}`, entryId, eventId, serviceName: 'lastleg', outcome: i < 11 ? 'RETRY' : i === 11 ? 'DEAD' : 'DUPLICATE' })),
    proxyRecords: Array.from({ length: 13 }, (_, i) => ({ eventId, payloadHash, attempt: i + 1, matched: true,
      outcome: i ? 'DUPLICATE' : 'APPLIED', errorCode: i < 12 ? 'ACK_LOSS_INJECTED' : null })),
    retryRequestId: '00000000-0000-4000-8000-000000000002', retryAuditCount: 1,
    retryAudit: { requestId: '00000000-0000-4000-8000-000000000002', entryId, eventId, payloadHash, previousFailures: 12 } }))
}
function unloadTransport(calls, proof = unloadProof(), finish = {}) {
  let recovered = false
  return async (_url, options) => {
    const config = requestedAction(options), { action } = config
    assert.equal(config.direction, 'UNLOAD')
    assert.deepEqual(config.scope, unloadScope)
    calls.push(action)
    if (action === 'recover') recovered = true
    const data = action === 'prepare' ? { originalImageId, originalFingerprint }
      : action === 'proxy-ready' ? { ready: true } : action === 'health' ? { healthy: true }
        : action === 'status' ? { producer: { publishStatus: recovered ? 'PUBLISHED' : 'DEAD', failedAttempts: 12 } }
          : action === 'proof' ? proof : action === 'finish' ? { restored: true, healthy: true, proxyStopped: true,
            producerName: 'firstleg-test-lastleg', originalImageId, currentImageId: originalImageId,
            originalFingerprint, currentFingerprint: originalFingerprint, ...finish } : {}
    return { ok: true, text: async () => 'TEST_EVENT_JSON=' + JSON.stringify({ ok: true, data }) }
  }
}
test('unload factory pins exactly three scope fields and verifies restoration before success', async () => {
  const calls = [], executor = createTestUnloadFaultExecutor(environment, unloadTransport(calls))
  await assert.rejects(executor.prepare({ ...unloadScope, eventType: 'Other' }), /Invalid/)
  await executor.prepare(unloadScope)
  await assert.rejects(executor.prepare({ ...unloadScope, containerBookingId: '67' }), /scope drift/)
  await assert.rejects(executor.readProof({ ...unloadScope, eventType: 'Other' }), /scope drift/)
  const proof = await executor.readProof(unloadScope)
  assert.equal(proof.restored, true)
  assert.equal(proof.restoration.currentImageId, originalImageId)
  await executor.cleanup()
  assert.deepEqual(calls, ['prepare', 'proxy-ready', 'activate', 'health', 'status', 'recover', 'status', 'proof', 'restore', 'health', 'finish'])
})
test('unload proof rejects changed original event, fabricated ACK, nonunique audit and changed business facts', () => {
  assertTestUnloadFaultProof(unloadProof(), unloadScope)
  for (const mutate of [
    p => { p.producer.customerId = '7' }, p => { p.producer.eventKey = 'other' },
    p => { p.producer.envelope.tenantId = '7' }, p => { p.producer.envelope.payload.containerNo = 'CHANGED' },
    p => { p.beforeRetry.entryId = '124' }, p => { p.beforeRetry.envelope.aggregateVersion = 4 },
    p => { p.beforeRetry.failedAttempts = 11 }, p => { p.afterRetry.eventId = 'OTHER' },
    p => { p.consumer.count = 2 }, p => { p.consumer.status = 'RECEIVED' }, p => { p.consumer.payload = {} },
    p => { p.attempts.pop() }, p => { p.attempts[0].serviceName = 'firstleg' },
    p => { p.proxyRecords[0].outcome = 'DUPLICATE' }, p => { p.proxyRecords[0].errorCode = null },
    p => { p.proxyRecords[12].attempt = 1 }, p => { p.retryAuditCount = 2 },
    p => { p.retryAudit.requestId = 'OTHER' }, p => { p.afterRetry.container.version = '10' },
    p => { p.beforeRetry.container.estimatedEmptyReportedAt = p.afterRetry.container.estimatedEmptyReportedAt = '2026-10-04 01:00:00' },
  ]) {
    const proof = unloadProof(); mutate(proof)
    assert.throws(() => assertTestUnloadFaultProof(proof, unloadScope), /Invalid unload fault proof/)
  }
})

test('unload timestamps follow DATETIME(3) rounding rather than allowing time drift', () => {
  const proof=unloadProof(true)
  assertTestUnloadFaultProof(proof,unloadScope)
  for(const row of [proof.beforeRetry,proof.afterRetry]) {
    row.container.unloadingCompletedAt='2026-10-02 01:00:00.138'
    row.container.estimatedEmptyReportedAt='2026-10-05 01:00:00.138'
  }
  assert.throws(()=>assertTestUnloadFaultProof(proof,unloadScope),/CONTAINER_FACTS/)
})
test('invalid unload proof still restores and incomplete restoration never becomes success', async () => {
  const invalid = unloadProof(); invalid.consumer.count = 0
  const calls = [], executor = createTestUnloadFaultExecutor(environment, unloadTransport(calls, invalid))
  await executor.prepare(unloadScope)
  await assert.rejects(executor.readProof(unloadScope), /INBOX/)
  assert.deepEqual(calls.slice(-3), ['restore', 'health', 'finish'])
  for (const change of [{ restored: false }, { healthy: false }, { proxyStopped: false },
    { currentImageId: 'sha256:' + 'c'.repeat(64) }, { currentFingerprint: 'c'.repeat(64) }]) {
    const failed = createTestUnloadFaultExecutor(environment, unloadTransport([], unloadProof(), change))
    await failed.prepare(unloadScope)
    await assert.rejects(failed.readProof(unloadScope), /restoration evidence mismatch/)
  }
})
test('Python reverse SQL routing, scope isolation and full configuration recovery fail closed offline', () => {
  const result = spawnSync(process.env.PYTHON || 'python', ['-c', String.raw`
import ast, copy, importlib.util, io, pathlib, sys
from unittest.mock import patch
sys.dont_write_bytecode = True
path = pathlib.Path('scripts/test-event-fault-runtime.py')
ast.parse(path.read_text(encoding='utf-8'))
spec = importlib.util.spec_from_file_location('fault_runtime', path)
m = importlib.util.module_from_spec(spec); spec.loader.exec_module(m)
scope = {'marker':'TEST-ROOT','customerId':'7','containerBookingId':'66'}
assert m.target_setting({'direction':'UNLOAD'}) == ('FIRSTLEG_SERVICE_URL','http://firstleg:9202')
assert m.scope_directory(scope, 'UNLOAD') != m.scope_directory(dict(scope, loadingPlanId='99'))
try: m.scope_directory(dict(scope, eventType='Other'), 'UNLOAD'); assert False
except RuntimeError as error: assert str(error) == 'INVALID_SCOPE'
queries=[]
with patch.object(m, 'inspect', side_effect=lambda name: name), patch.object(m, 'read_sql', side_effect=lambda name, sql: queries.append((name, sql)) or []):
    assert m.producer(scope, 'UNLOAD') is None
assert queries[0][0] == 'firstleg-test-lastleg'
assert "customer_id='0'" in queries[0][1] and "event_key='OverseasContainerUnloaded:66'" in queries[0][1]
assert "JSON_EXTRACT(payload_json,'$')" in queries[0][1] and "publishStatus',status" in queries[0][1]
row={'entryId':'123','eventId':'00000000-0000-4000-8000-000000000001','payloadHash':'sha256:'+'a'*64,'publishStatus':'DEAD','failedAttempts':12,'envelope':{'eventType':'OverseasContainerUnloaded'}}
retry_state={'direction':'UNLOAD','operationsRequestId':'00000000-0000-4000-8000-000000000002'}
retry_container={'Config':{'Env':['EVENT_OPERATIONS_TOKEN=offline']},'NetworkSettings':{'Networks':{'firstleg-test-network':{'IPAddress':'172.20.0.3'}}}}
with patch.object(m,'producer',return_value=row),patch.object(m,'control',return_value=row),patch.object(m,'container_facts',return_value={'version':'9'}),patch.object(m,'business_ids',side_effect=AssertionError('reverse must not query inventory or orders')),patch.object(m,'private_write'),patch.object(m,'inspect',return_value=retry_container),patch.object(m.urllib.request,'urlopen',return_value=io.BytesIO(b'{"code":"200"}')) as http:
    assert m.recover(scope,pathlib.Path('/offline'),retry_state)['retrySubmitted']
    assert http.call_args.args[0].full_url == 'http://172.20.0.3:9303/internal/lastleg/event-operations/123/retry'
    assert retry_state['beforeRetry']['container']=={'version':'9'}
    assert retry_state['beforeRetry']['envelope']==row['envelope']
row['publishStatus']='PUBLISHED'
queries=[]
with patch.object(m,'producer',return_value=row),patch.object(m,'control',return_value={'records':[]}),patch.object(m,'container_facts',return_value={'version':'9'}),patch.object(m,'business_ids',side_effect=AssertionError('reverse must not query inventory or orders')),patch.object(m,'inspect',side_effect=lambda name:name),patch.object(m,'read_sql',side_effect=lambda name,sql: queries.append((name,sql)) or []),patch.object(m,'private_write') as save:
    proof=m.proof(scope,retry_state)
    save.assert_called_once_with(m.scope_directory(scope,'UNLOAD')/'proof.json',proof)
assert proof['retryAuditCount']==0 and proof['retryRequestId']==retry_state['operationsRequestId']
assert [item[0] for item in queries]==['firstleg-test-lastleg','firstleg-test-lastleg','firstleg-test-business']
assert "WHERE entry_id='123' AND event_type='OverseasContainerUnloaded'" in queries[1][1]
assert 'fl_container_event_inbox' in queries[2][1]
container={'Name':'/firstleg-test-lastleg','Id':'container123456789','Image':'sha256:'+'a'*64,
 'Config':{'Image':'lastleg:test','Env':['FIRSTLEG_SERVICE_URL=http://firstleg:9202','A=1'],'Labels':{'com.docker.compose.project':'firstleg-test'}},
 'HostConfig':{'Binds':['source:/data:rw'],'PortBindings':{'9303/tcp':[{'HostPort':'9303'}]}},
 'Mounts':[{'Type':'bind','Source':'/source','Destination':'/data','RW':True}],
 'NetworkSettings':{'Networks':{'firstleg-test-network':{'IPAddress':'172.20.0.3','Aliases':['lastleg-service','container123']}}},
 'State':{'Health':{'Status':'healthy'}}}
fingerprint=m.fingerprint(container)
multi=copy.deepcopy(container)
multi['NetworkSettings']['Networks']['firstleg-test-network']['MacAddress']='02:42:ac:15:00:04'
multi['NetworkSettings']['Networks']['second-test-network']={'IPAddress':'172.22.0.7','MacAddress':'02:42:ac:16:00:07','Aliases':['lastleg']}
legacy=copy.deepcopy(multi); legacy['Config']['MacAddress']='02:42:ac:16:00:07'
assert m.fingerprint(legacy)==m.fingerprint(multi)
changed=copy.deepcopy(legacy); changed['Config']['MacAddress']='02:42:ac:99:00:01'
assert m.fingerprint(changed)!=m.fingerprint(multi)
changed=copy.deepcopy(legacy); changed['NetworkSettings']['Networks']['second-test-network']['MacAddress']='02:42:ac:16:00:08'
assert m.fingerprint(changed)!=m.fingerprint(multi)
state={'direction':'UNLOAD','restored':False,'image':container['Config']['Image'],'originalImageId':container['Image'],
 'originalFingerprint':fingerprint,'originalBody':m.recreation(container,m.env_of(container)),'proxyPid':123,'controlToken':'offline'}
for mutate in [lambda c:c['HostConfig'].update(Binds=['other:/data:rw']),lambda c:c['NetworkSettings']['Networks']['firstleg-test-network'].update(IPAddress='172.20.0.4'),lambda c:c.update(Image='sha256:'+'b'*64),lambda c:c['State']['Health'].update(Status='unhealthy')]:
    changed=copy.deepcopy(container); mutate(changed)
    with patch.object(m,'inspect',return_value=changed), patch.object(m,'private_write') as write:
        try: m.finish(pathlib.Path('/offline'),copy.deepcopy(state)); assert False
        except RuntimeError as error: assert str(error).startswith('RESTORED_')
        write.assert_not_called()
with patch.object(m,'inspect',return_value=container),patch.object(m.Path,'exists',return_value=False),patch.object(m,'private_write'):
    restored=m.finish(pathlib.Path('/offline'),copy.deepcopy(state))
assert restored['proxyStopped'] and restored['currentFingerprint']==fingerprint and restored['originalImageId']==restored['currentImageId']
with patch.object(m,'inspect',return_value=container),patch.object(m.Path,'exists',return_value=True),patch.object(m.Path,'read_bytes',return_value=b'python3\x00'+str(pathlib.Path('/offline')/'proxy.py').encode()+b'\x00'),patch.object(m.os,'kill'),patch.object(m.time,'sleep'),patch.object(m,'private_write') as write:
    try: m.finish(pathlib.Path('/offline'),copy.deepcopy(state)); assert False
    except RuntimeError as error: assert str(error)=='PROXY_NOT_STOPPED'
    write.assert_not_called()
`], { encoding: 'utf8', cwd: new URL('..', import.meta.url), timeout: 10000 })
  assert.equal(result.status, 0, result.stderr || result.error?.message)
})
test('only original scoped event is recovered and restoration precedes a successful proof', async () => {
  const calls = [], executor = createTestFaultExecutor(environment, transport(calls))
  await executor.prepare(scope)
  await executor.prepare({ ...scope, loadingPlanId: '100' })
  const proof = await executor.readProof({ containerBookingId: '66', loadingPlanId: '99', customerId: '7', marker: 'TEST-ROOT' })
  assert.equal(proof.restored, true)
  await executor.cleanup()
  assert.deepEqual(calls, ['prepare', 'proxy-ready', 'activate', 'health', 'status', 'recover',
    'status', 'proof', 'restore', 'health', 'finish'])
})
test('failed activation retains task ownership for explicit restoration, never retries activation', async () => {
  const calls = [], executor = createTestFaultExecutor(environment, transport(calls, true))
  await assert.rejects(executor.prepare(scope), /DOCKER_409/)
  await executor.cleanup()
  assert.equal(calls.filter(name => name === 'activate').length, 1)
  assert.deepEqual(calls.slice(-3), ['restore', 'health', 'finish'])
})

const deliveryScope={marker:'TEST-DELIVERY',customerId:'7',deliveryId:'90',sourceInboundId:'200',sourceOrderId:'100',rootOrderId:'100'}
function deliveryProof(mode='ROOT_PENDING') {
  const scope={...deliveryScope,...(mode==='CHILD_FIRST'?{sourceOrderId:'101',sourceInboundId:'201'}:{})}
  const eventId='00000000-0000-4000-8000-000000000001', entryId='123'
  const payload={deliveryOrderId:'90',deliveryNo:'DL-OFFLINE',lastlegOrderId:'300',lastlegOrderNo:'LL-OFFLINE',
    sourceInboundId:scope.sourceInboundId,inboundCode:'IN-OFFLINE',sourceOrderId:scope.sourceOrderId,
    sourceOrderNo:'FL-OFFLINE',rootOrderId:'100',fbaWarehouseCode:'ABE4',completedAt:'2026-10-03T01:00:00Z'}
  const payloadHash='sha256:'+createHash('sha256').update(JSON.stringify(payload,Object.keys(payload).sort())).digest('hex')
  const envelope={eventId,eventType:'OverseasDeliveryCompleted',eventVersion:1,tenantId:'7',aggregateType:'INBOUND_ORDER',aggregateId:scope.sourceInboundId,aggregateVersion:5,payloadHash,payload}
  const producer={entryId,eventId,eventType:envelope.eventType,customerId:'7',aggregateId:scope.sourceInboundId,aggregateType:'INBOUND_ORDER',eventKey:`OverseasDeliveryCompleted:DL-OFFLINE:${scope.sourceInboundId}`,envelope,payloadHash,publishStatus:'PUBLISHED'}
  const ownRoot=mode!=='CHILD_FIRST', childComplete=mode!=='ROOT_PENDING', rootComplete=mode==='ALL_COMPLETE'
  const orders=['100','101'].map(id=>({id,rootOrderId:'100',customerId:'7',orderRole:id==='100'?'MASTER':'CHILD',
    orderNo:id===scope.sourceOrderId?'FL-OFFLINE':'OTHER',inboundCode:id===scope.sourceOrderId?'IN-OFFLINE':'OTHER',
    status:(id==='100'?rootComplete:childComplete)?'COMPLETED':'APPROVED',version:3,
    completedAt:(id==='100'?rootComplete:childComplete)?'2026-10-03 01:00:00':null}))
  const consumer={eventId,eventType:envelope.eventType,aggregateId:scope.sourceInboundId,payloadHash,payload,status:'PROCESSED'}
  const business={inbound:[{id:scope.sourceInboundId,customerId:'7',orderId:scope.sourceOrderId,orderNo:'FL-OFFLINE',inboundCode:'IN-OFFLINE',processStatus:'DELIVERED',orderStatusSnapshot:orders.find(o=>o.id===scope.sourceOrderId).status,version:2}],
    inboundHistory:[{id:'41',inboundId:scope.sourceInboundId,action:'COMPLETE_DELIVERY',fromStatus:'LOADED',toStatus:'DELIVERED',occurredAt:'2026-10-03 01:00:00'}],
    inbox:[{...consumer,processedAt:'2026-10-03 01:00:01'}],order:orders,
    orderHistory:[...(ownRoot?[{id:'51',orderId:'100',action:'OVERSEAS_DELIVERED',occurredAt:'2026-10-03 01:00:00'}]:[]),...(childComplete?[{id:'52',orderId:'101',action:'OVERSEAS_DELIVERED',occurredAt:'2026-10-03 01:00:00'}]:[]),...(rootComplete?[{id:'53',orderId:'100',action:'FULFILLMENT_COMPLETED'}]:[])],
    completionEvent:orders.filter(o=>o.status==='COMPLETED').map((o,i)=>({id:String(60+i),orderId:o.id,eventType:'FIRST_LEG_ORDER_FULFILLMENT_COMPLETED',status:'PUBLISHED',payload:{orderId:o.id}}))}
  const old=unloadProof()
  return JSON.parse(JSON.stringify({...scope,readOnly:true,producer,consumer:{...consumer,count:1},
    beforeRetry:{...producer,publishStatus:'DEAD',failedAttempts:12,business},afterRetry:{...producer,business},
    attempts:old.attempts,proxyRecords:old.proxyRecords.map(p=>({...p,payloadHash})),retryAudit:{...old.retryAudit,payloadHash},retryAuditCount:1,retryRequestId:old.retryRequestId}))
}
function deliveryTransport(calls, proof=deliveryProof(), finish={}) {
  let recovered=false
  return async(_url,options)=>{
    const {action,direction,scope}=requestedAction(options)
    assert.equal(direction,'DELIVERY');assert.deepEqual(scope,deliveryScope);calls.push(action)
    if(action==='recover')recovered=true
    const data=action==='prepare'?{originalImageId,originalFingerprint}:action==='proxy-ready'?{ready:true}:action==='health'?{healthy:true}
      :action==='status'?{producer:{publishStatus:recovered?'PUBLISHED':'DEAD',failedAttempts:12}}:action==='proof'?proof
        :action==='finish'?{restored:true,healthy:true,proxyStopped:true,producerName:'firstleg-test-lastleg',originalImageId,currentImageId:originalImageId,originalFingerprint,currentFingerprint:originalFingerprint,...finish}:{}
    return {ok:true,text:async()=> 'TEST_EVENT_JSON='+JSON.stringify({ok:true,data})}
  }
}
test('DELIVERY has a separate strict scope and restores before accepting original event proof',async()=>{
  const calls=[],executor=createTestDeliveryFaultExecutor(environment,deliveryTransport(calls))
  await assert.rejects(executor.prepare({...deliveryScope,eventType:'OverseasContainerUnloaded'}),/Invalid/)
  await executor.prepare(deliveryScope)
  await assert.rejects(executor.prepare({...deliveryScope,deliveryId:'91'}),/scope drift/)
  const proof=await executor.readProof(deliveryScope)
  assert.equal(proof.restored,true);assert.equal(proof.restoration.currentImageId,originalImageId)
  assert.deepEqual(calls,['prepare','proxy-ready','activate','health','status','recover','status','proof','restore','health','finish'])
})
test('DELIVERY proof distinguishes root pending, child first and all completed with immutable original evidence',()=>{
  for(const mode of ['ROOT_PENDING','CHILD_FIRST','ALL_COMPLETE']) {
    const p=deliveryProof(mode),scope={...deliveryScope,sourceOrderId:p.sourceOrderId,sourceInboundId:p.sourceInboundId}
    assertTestDeliveryFaultProof(p,scope)
  }
  for(const mutate of [p=>p.producer.envelope.payload.deliveryId='90',p=>delete p.producer.envelope.payload.deliveryOrderId,
    p=>p.producer.customerId='0',p=>p.producer.eventKey='wrong',p=>p.beforeRetry.entryId='124',
    p=>p.consumer.count=2,p=>p.attempts.pop(),p=>p.proxyRecords[0].outcome='DUPLICATE',
    p=>p.retryAuditCount=2,p=>p.afterRetry.business.inbound[0].version++,p=>p.afterRetry.business.order[0].version++,
    p=>p.beforeRetry.business.inbox[0].status='RECEIVED',p=>p.beforeRetry.business.inboundHistory.push({...p.beforeRetry.business.inboundHistory[0]}),
    p=>{p.beforeRetry.business.order[0].status=p.afterRetry.business.order[0].status='COMPLETED'},
    p=>p.beforeRetry.business.completionEvent.push({orderId:'100',status:'PUBLISHED'})]) {
    const proof=deliveryProof();mutate(proof);assert.throws(()=>assertTestDeliveryFaultProof(proof,deliveryScope),/Invalid delivery/)
  }
})
test('DELIVERY evidence failure still restores and restoration drift is a blocking failure',async()=>{
  const p=deliveryProof();p.consumer.count=0
  const calls=[],executor=createTestDeliveryFaultExecutor(environment,deliveryTransport(calls,p))
  await executor.prepare(deliveryScope);await assert.rejects(executor.readProof(deliveryScope),/INBOX/)
  assert.deepEqual(calls.slice(-3),['restore','health','finish'])
  const broken=createTestDeliveryFaultExecutor(environment,deliveryTransport([],deliveryProof(),{proxyStopped:false}))
  await broken.prepare(deliveryScope);await assert.rejects(broken.readProof(deliveryScope),/Delivery restoration/)
})
test('F40 hook faults only first signature, captures whole inventory and always cleans failed signing',async()=>{
  const calls=[],restoration={restored:true,healthy:true,proxyStopped:true}
  const inventory={scope:{...deliveryScope,warehouseId:'9',coverage:'CUSTOMER_WAREHOUSE_COMPLETE'},balances:[],positions:[],ledger:[]}
  const parentProof={}
  const context={context:{f40RootFulfillmentProof:parentProof,f40DeliveryEventFaultExecutor:{prepare:async s=>{assert.deepEqual(s,deliveryScope);calls.push('prepare')},
    readProof:async()=>{calls.push('proof');return {...deliveryProof(),restored:true,restoration}},cleanup:async()=>{calls.push('cleanup');return restoration}},
    f40ReadOnlyInventoryEvidence:async s=>{assert.equal(s.warehouseId,'9');calls.push('inventory');return structuredClone(inventory)}},createdResources:{phase3OrderSplitting:{}}}
  assert.equal(await withDeliveryEventRecovery(context,deliveryScope,'9',0,async()=>{calls.push('sign');return 'signed'}),'signed')
  assert.deepEqual(calls,['inventory','prepare','sign','proof','inventory','cleanup'])
  assert.equal(parentProof.deliveryEventRecovery.status,'VERIFIED')
  assert.equal(context.createdResources.phase5RootFulfillment,undefined)
  await withDeliveryEventRecovery(context,deliveryScope,'9',1,async()=>calls.push('second'))
  assert.equal(calls.at(-1),'second')
  context.context.f40RootFulfillmentProof={}
  await assert.rejects(withDeliveryEventRecovery(context,deliveryScope,'9',0,async()=>{throw Error('sign failed')}),/sign failed/)
  assert.equal(calls.at(-1),'cleanup')
  assert.equal(context.context.f40RootFulfillmentProof.deliveryEventRecovery.restored,true)
})
test('Python DELIVERY SQL uses actual payload, original Lastleg retry, and Order database without inventory queries',()=>{
  const result=spawnSync(process.env.PYTHON || 'python',['-c',String.raw`
import importlib.util,sys,ast,pathlib,copy
from unittest.mock import patch,MagicMock,mock_open
sys.dont_write_bytecode=True
p=pathlib.Path('scripts/test-event-fault-runtime.py');ast.parse(p.read_text(encoding='utf-8'))
spec=importlib.util.spec_from_file_location('runtime',p);m=importlib.util.module_from_spec(spec);spec.loader.exec_module(m)
s={'marker':'TEST-DELIVERY','customerId':'7','deliveryId':'90','sourceInboundId':'200','sourceOrderId':'100','rootOrderId':'100'}
assert m.scope_directory(s,'DELIVERY')!=m.scope_directory({'marker':s['marker'],'customerId':'7','containerBookingId':'200'},'UNLOAD')
assert m.producer_name({'direction':'DELIVERY'})=='firstleg-test-lastleg'
assert m.target_setting({'direction':'DELIVERY'})==('FIRSTLEG_SERVICE_URL','http://firstleg:9202')
queries=[]
with patch.object(m,'inspect',side_effect=lambda name:{'Name':'/'+name}),patch.object(m,'require_test'),patch.object(m,'read_sql',side_effect=lambda container,sql:queries.append((container['Name'],sql)) or []):
 assert m.producer(s,'DELIVERY') is None
 m.delivery_facts(s,'00000000-0000-4000-8000-000000000001')
assert [q[0] for q in queries]==['/firstleg-test-lastleg','/firstleg-test-business','/firstleg-test-order']
assert "$.payload.deliveryOrderId" in queries[0][1] and "customer_id='7'" in queries[0][1]
assert 'fl_inbound_status_history' in queries[1][1] and 'fl_container_event_inbox' in queries[1][1]
assert 'ord_order_status_history' in queries[2][1] and 'FIRST_LEG_ORDER_FULFILLMENT_COMPLETED' in queries[2][1]
assert all('inventory' not in q[1] for q in queries)
original={'Name':'/firstleg-test-lastleg','Id':'original-container','Image':'sha256:'+'a'*64,
 'Config':{'Image':'lastleg:test','Env':['FIRSTLEG_SERVICE_URL=http://firstleg:9202'],'Labels':{'com.docker.compose.project':'firstleg-test'}},
 'HostConfig':{},'Mounts':[],'NetworkSettings':{'Networks':{'firstleg-test-network':{'IPAddress':'172.20.0.3','Aliases':['lastleg-service']}}},
 'State':{'Running':True,'Health':{'Status':'healthy'}}}
consumer=copy.deepcopy(original);consumer['Name']='/firstleg-test-business'
with patch.object(m,'producer',return_value=None),patch.object(m,'inspect',side_effect=lambda name:original if name=='firstleg-test-lastleg' else consumer),patch.object(m.Path,'mkdir'),patch.object(m.Path,'write_bytes'),patch('builtins.open',mock_open()),patch.object(m.socket,'socket') as sock,patch.object(m.subprocess,'Popen') as process,patch.object(m,'private_write'):
 sock.return_value.getsockname.return_value=('127.0.0.1',18001)
 m.prepare(s,pathlib.Path('/offline'),{'direction':'DELIVERY','proxySource':''})
 args=process.call_args.args[0]
 assert args[args.index('--delivery-id')+1]=='90' and args[args.index('--event-type')+1]=='OverseasDeliveryCompleted'
 assert '--container-booking-id' not in args and args[args.index('--upstream')+1].endswith(':9202/internal/firstleg/v1/container-events')
state={'direction':'DELIVERY','restored':False,'prepared':True,'image':'lastleg:test','originalId':'original-container',
 'originalImageId':original['Image'],'originalFingerprint':m.fingerprint(original),'originalBody':m.recreation(original,m.env_of(original),full=True),'proxyPort':18001}
restored=copy.deepcopy(original);restored['Id']='restored-new-id'
with patch.object(m,'control',side_effect=OSError('proxy already stopped')),patch.object(m,'inspect',return_value=restored),patch.object(m,'private_write'),patch.object(m,'replace') as replace,patch.object(m,'docker') as docker:
 assert m.restore(pathlib.Path('/offline'),state)=={'restoreStarted':True}
 assert state['prepared'] is False
 replace.assert_not_called();docker.assert_not_called()
state['prepared']=True
restored['Config']['Env']=['FIRSTLEG_SERVICE_URL=http://firstleg:9202','UNEXPECTED=1']
with patch.object(m,'control'),patch.object(m,'inspect',return_value=restored),patch.object(m,'replace') as replace:
 try: m.restore(pathlib.Path('/offline'),state);assert False
 except RuntimeError as error: assert str(error)=='CONCURRENT_PRODUCER_CHANGE'
 replace.assert_not_called()
`],{encoding:'utf8',cwd:new URL('..',import.meta.url),timeout:10000})
  assert.equal(result.status,0,result.stderr||result.error?.message)
})