import glob,hashlib,json,os,signal,subprocess,threading,time
from pathlib import Path
root=Path(__file__).resolve().parents[4];packet=Path(__file__).resolve().parent
head=subprocess.check_output(['git','rev-parse','HEAD'],cwd=root,text=True).strip()
assert head==(packet/'repeat-commit.txt').read_text().strip()
manifest=(packet/'candidate-manifest.sha256').read_text()
expected={p:h for h,p in (line.split('  ',1) for line in manifest.splitlines())}
def verify():
 assert subprocess.check_output(['git','rev-parse','HEAD'],cwd=root,text=True).strip()==head
 assert all(hashlib.sha256((root/p).read_bytes()).hexdigest()==h for p,h in expected.items())
verify(); base_tmp=Path((packet/'repeat-tmpdir.txt').read_text().strip())
stop=threading.Event();failed=threading.Event();lock=threading.Lock();rows=[];processes={}
start=time.time()
def save(row):
 with lock:
  rows.append(row);(packet/'repeat-results.json').write_text(json.dumps(rows,indent=2)+'\n')
 print(json.dumps(row),flush=True)
def kill(p):
 try:os.killpg(p.pid,signal.SIGKILL)
 except ProcessLookupError:pass
 p.wait()
def execute(name,files,limit,kind):
 verify();env=os.environ.copy();tmp=base_tmp/name;tmp.mkdir()
 env['TMPDIR']=str(tmp);env['MOSAIC_CONFIG']=str(tmp/'fixture-config-only.json');env['DOCKER_HOST']='unix:///nonexistent.sock'
 env['MOSAIC_SSHD_EVIDENCE']=str(packet/'attempt-logs'/name)
 path=packet/'repeat-logs'/(name+'.log');path.parent.mkdir(exist_ok=True)
 then=time.time();timed=False;aborted=False
 with path.open('wb') as output:
  p=subprocess.Popen(['node','--test','--test-reporter=tap',*files],cwd=root,env=env,stdout=output,stderr=subprocess.STDOUT,start_new_session=True)
  with lock: processes[name]=p
  while p.poll() is None:
   if failed.is_set():aborted=True;kill(p);break
   if time.time()-then>limit:timed=True;kill(p);break
   time.sleep(.2)
  with lock:processes.pop(name,None)
 text=path.read_text(errors='replace')
 import re
 counts={k:int(v) for k,v in re.findall(r'^# (tests|pass|fail|skipped|cancelled) (\d+)$',text,re.M)}
 okay=p.returncode==0 and not timed and not aborted and counts.get('tests',0)>0 and counts.get('fail')==0 and counts.get('cancelled')==0
 row={'name':name,'kind':kind,'head':head,'start':time.strftime('%Y-%m-%dT%H:%M:%SZ',time.gmtime(then)),'end':time.strftime('%Y-%m-%dT%H:%M:%SZ',time.gmtime()),'seconds':round(time.time()-then,3),'exit':p.returncode,'timeout':timed,'aborted':aborted,'counts':counts,'okay':okay,'log':str(path.relative_to(packet))}
 save(row)
 if not okay:failed.set();stop.set()
 verify();return okay
def load():
 try:
  i=0
  while not stop.is_set():
   i+=1
   if not execute(f'cli-load-{i:02}',sorted(glob.glob('packages/cli/tests/*.test.mjs',root_dir=root)),900,'cli-load'):return
 except BaseException as e:
  save({'name':'cli-load-runner','error':repr(e),'okay':False});failed.set();stop.set()
thread=threading.Thread(target=load);thread.start()
try:
 for file in ['grant-executor','host-run']:
  for i in range(1,31):
   if failed.is_set():break
   if not execute(f'{file}-{i:02}',[f'packages/bus/tests/{file}.test.mjs'],180,'repeat'):break
  if failed.is_set():break
finally:
 stop.set();thread.join()
 with lock: remaining=list(processes.values())
 for p in remaining:kill(p)
 verify()
summary={'head':head,'manifest_sha256':hashlib.sha256(manifest.encode()).hexdigest(),'start':time.strftime('%Y-%m-%dT%H:%M:%SZ',time.gmtime(start)),'end':time.strftime('%Y-%m-%dT%H:%M:%SZ',time.gmtime()),'failed':failed.is_set(),'repeat_runs':sum(r.get('kind')=='repeat' for r in rows),'load_runs':sum(r.get('kind')=='cli-load' for r in rows),'counts':{kind:{key:sum(r.get('counts',{}).get(key,0) for r in rows if r.get('kind')==kind) for key in ['tests','pass','fail','skipped','cancelled']} for kind in ['repeat','cli-load']}}
(packet/'repeat-summary.json').write_text(json.dumps(summary,indent=2)+'\n');print(json.dumps(summary),flush=True)
raise SystemExit(1 if failed.is_set() else 0)