Engine v0.23.0 becomes unresponsive during concurrent trigger-provider reconnects (serial control passes)

python reproduce.py /absolute/path/to/iii /tmp/iii-reconnect-repro-new

import asyncio,json,subprocess,time,socket,uuid,os,hashlib,sys
from pathlib import Path
import websockets
ENGINE=str(Path(sys.argv[1]).resolve());OUT=Path(sys.argv[2]).resolve();OUT.mkdir();receipt={'schema':'engine-trigger-reconnect-diagnostic/v1','synthetic':True,'not_live_causal_proof':True,'cases':[]}
class Peer:
 def __init__(self,name):self.name=name;self.pending={};self.received=0;self.errors=[];self.ws=None;self.task=None
 async def open(self):
  self.ws=await websockets.connect('ws://127.0.0.1:49134',open_timeout=3,close_timeout=.5,ping_interval=None,max_size=1048576)
  self.task=asyncio.create_task(self.listen());self.registration=await self.call('engine::workers::register',{'name':self.name,'runtime':'python','version':'diagnostic','namespace':'default','pid':os.getpid()});assert self.registration=={'success':True},self.registration
 async def send(self,v):await self.ws.send(json.dumps(v))
 async def call(self,fn,data):
  id=str(uuid.uuid4());f=asyncio.get_running_loop().create_future();self.pending[id]=f
  await self.send({'type':'invokefunction','invocation_id':id,'function_id':fn,'data':data});return await asyncio.wait_for(f,3)
 async def listen(self):
  try:
   async for s in self.ws:
    v=json.loads(s);self.received+=1;t=v.get('type')
    if t=='invocationresult':
     f=self.pending.pop(v['invocation_id'],None)
     if f and not f.done():
      if v.get('error'):f.set_exception(RuntimeError(str(v['error'])[:180]))
      else:f.set_result(v.get('result'))
    elif t=='registertrigger':await self.send({'type':'triggerregistrationresult','id':v['id'],'trigger_type':v['trigger_type'],'error':None})
    elif t=='invokefunction' and v.get('invocation_id'):await self.send({'type':'invocationresult','invocation_id':v['invocation_id'],'function_id':v['function_id'],'result':{'fixture':True}})
  except Exception as e:self.errors.append(type(e).__name__+':'+str(e)[:160])
 async def close(self):
  if self.ws:await self.ws.close()
  if self.task:
   self.task.cancel();await asyncio.gather(self.task,return_exceptions=True)
async def scenario(result):
 peers=[Peer('reconnect-diagnostic-'+str(i)) for i in range(8)];audit=Peer('reconnect-diagnostic-audit')
 try:
  await audit.open();result['initial_roster']=len((await audit.call('engine::workers::list',{}))['workers'])
  await asyncio.gather(*(p.open() for p in peers));await audit.send({'type':'registerfunction','id':'fixture::noop','request_format':{},'response_format':{}})
  for j,p in enumerate(peers):
   for k in range(6):await p.send({'type':'registertriggertype','id':f'fixture-{j}-{k}','description':'synthetic reconnect liveness'})
  for j in range(8):
   for k in range(6):
    for n in range(12):await audit.send({'type':'registertrigger','id':f'fixture-bind-{j}-{k}-{n}','trigger_type':f'fixture-{j}-{k}','function_id':'fixture::noop','config':{}})
  result['registered_bindings']=576;result['roster_before']=(await audit.call('engine::workers::list',{}))['workers']
  expected={p.name for p in peers}|{audit.name};assert expected=={w['name'] for w in result['roster_before'] if w['runtime']=='python'};result['metadata_registration_verified']=True;result['metadata_successes']=9;result['registered_worker_ids']=[w['id'] for w in result['roster_before'] if w['runtime']=='python']
  result['rounds']=[]
  for round in range(4):
   async def types(j,p):
    for k in range(6):await p.send({'type':'registertriggertype','id':f'fixture-{j}-{k}','description':'synthetic reconnect liveness'})
   if result['mode']=='serial-reconnect':
    for j in range(8):
     await peers[j].close();peers[j]=Peer('reconnect-diagnostic-'+str(j));await peers[j].open();await types(j,peers[j]);await audit.call('engine::workers::list',{})
   else:
    await asyncio.gather(*(p.close() for p in peers));peers=[Peer('reconnect-diagnostic-'+str(i)) for i in range(8)]
    await asyncio.gather(*(p.open() for p in peers));await asyncio.gather(*(types(j,p) for j,p in enumerate(peers)))
   t=time.monotonic();roster=await audit.call('engine::workers::list',{});result['rounds'].append({'round':round,'latency':time.monotonic()-t,'workers':len(roster['workers']),'named_peers':sorted(w['name'] for w in roster['workers'] if w['runtime']=='python')});assert expected=={w['name'] for w in roster['workers'] if w['runtime']=='python'}
   (OUT/'progress.json').write_text(json.dumps(receipt))
  check=Peer('fresh-handshake-final');await check.open();await check.call('engine::workers::list',{});await check.close();result['final_handshake']=True
 finally:
  await asyncio.gather(*(p.close() for p in peers),audit.close(),return_exceptions=True)
receipt['engine_binary_sha256']=hashlib.sha256(Path(ENGINE).read_bytes()).hexdigest()
for mode in ['serial-reconnect','concurrent-reconnect']:
 res={'mode':mode,'passed':False};receipt['cases'].append(res);work=OUT/mode;work.mkdir();conf={'workers':[{'name':'iii-worker-manager','config':{'host':'127.0.0.1','port':49134}}]}
 cfg=work/'engine.json';cfg.write_text(json.dumps(conf));log=(OUT/(mode+'-engine.log')).open('w');proc=subprocess.Popen([ENGINE,'--no-update-check','--config',str(cfg)],cwd=work,stdout=log,stderr=log);t=time.monotonic()
 try:
  deadline=time.monotonic()+10
  while True:
   try:
    with socket.create_connection(('127.0.0.1',49134),timeout=.2):break
   except OSError:
    if time.monotonic()>deadline or proc.poll() is not None:raise RuntimeError('engine not ready')
    time.sleep(.1)
  asyncio.run(asyncio.wait_for(scenario(res),25));res['passed']=True
 except BaseException as e:res['error']=type(e).__name__+':'+str(e)[:700]
 finally:
  res['seconds']=time.monotonic()-t;proc.terminate()
  try:proc.wait(timeout=4);res['engine_normal_exit']=proc.returncode
  except subprocess.TimeoutExpired:res['engine_normal_exit']=None;proc.kill();proc.wait();res['fixture_forced_cleanup']=True
  log.close();res['log_bytes']=(OUT/(mode+'-engine.log')).stat().st_size;(OUT/'result.json').write_text(json.dumps(receipt,indent=2))
receipt['passed']=all(r['passed'] for r in receipt['cases']);(OUT/'result.json').write_text(json.dumps(receipt,indent=2));print(json.dumps(receipt))