import os,sys,json,time,re,pathlib,subprocess,shlex,hashlib,signal
REPO=pathlib.Path('/home/jwoltje/.t3/worktrees/mosaic-stack/gander-row196')
ROOT=pathlib.Path(__file__).parent; OUT=ROOT/'records'; OUT.mkdir(exist_ok=True)
CASE='packages/cli/tests/case1-accept.test.mjs'
HELPER='packages/cli/tests/case-p10-timing-fixture.mjs'
HEAD='ffa93b054f28852a75017928d8478bdf9046e717'
DIGEST='277ace0b4274970d5bd4c05cf8fe99aa1e7ebc469a719e4cefcb334a75d3865d'
FILES=sorted(str(p.relative_to(REPO)) for p in (REPO/'packages/cli/tests').glob('*.test.mjs'))
PLAN=json.loads((REPO/'agents/gander/work/queue-299/MUTANT-PLAN.json').read_text())['mutants']
STATE={'grant':'sage-3268','m4_ruling':'sage-3273 / mos-1426','head':HEAD,'manifest':DIGEST,'status':'PREPARED','cli_file_list':FILES,'cli_file_count':len(FILES),'stages':[],'restores':[]}
def utc():return time.strftime('%Y-%m-%dT%H:%M:%SZ',time.gmtime())
def git(*args):return subprocess.check_output(['git',*args],cwd=REPO,text=True).strip()
def save(): (OUT/'results.json').write_text(json.dumps(STATE,indent=2)+'\n')
def stop(reason):
 STATE['status']='STOP';STATE['reason']=reason;STATE['stopped_utc']=utc();save();print('STOP '+reason,flush=True);raise SystemExit(1)
def load(): return pathlib.Path('/proc/loadavg').read_text().strip()
def busy(label):
 r=subprocess.run(['bash','/mnt/storage/scratch/sage-busy-now.sh'],capture_output=True,text=True)
 (OUT/(label+'.busy.log')).write_text(r.stdout+r.stderr)
 if r.returncode or r.stdout.strip().splitlines()[-1:]!=['CLEAR']:stop('busy refusal before '+label)
def parse_metrics(text):
 result=[]
 for line in text.splitlines():
  m=re.match(r'^\s*#\s*(\{.*\})\s*$',line)
  if m:
   try:
    item=json.loads(m[1])
    if 'probeSeconds' in item:result.append(item)
   except json.JSONDecodeError:pass
 return result
def execute(label,args,limit,expected_red=False,syntax=False):
 busy(label)
 tmp=subprocess.check_output(['mktemp','-d','/mnt/storage/scratch/tmp/gXXXXXX'],text=True).strip()
 env=os.environ.copy();env['TMPDIR']=tmp
 argv=['timeout','-k','10',str(limit),*args]
 meta={'label':label,'argv':argv,'limit_seconds':limit,'tmpdir':tmp,'started_utc':utc(),'start_load':load(),'log':str(OUT/(label+'.log')),'expected_red':expected_red}
 (OUT/(label+'.command.json')).write_text(json.dumps(meta,indent=2)+'\n')
 STATE['active']=meta;save();print('START '+json.dumps(meta),flush=True)
 shell='set -o pipefail\n'+shlex.join(argv)+' 2>&1 | tee '+shlex.quote(meta['log'])
 start=time.monotonic();samples=[];p=subprocess.Popen(['bash','-c',shell],cwd=REPO,env=env,start_new_session=True)
 while True:
  samples.append({'wall_seconds':round(time.monotonic()-start,3),'utc':utc(),'load':load()})
  try:rc=p.wait(timeout=5);break
  except subprocess.TimeoutExpired:
   if time.monotonic()-start>limit+30:
    os.killpg(p.pid,signal.SIGKILL);p.wait();stop('outer controller timeout '+label)
 text=pathlib.Path(meta['log']).read_text();seconds=round(time.monotonic()-start,3)
 counts={}
 for key in ['tests','pass','fail','cancelled','skipped','todo']:
  matches=re.findall(r'^# '+key+r' (\d+)\s*$',text,re.M)
  if matches:counts[key]=int(matches[-1])
 row={**meta,'rc':rc,'wall_seconds':seconds,'finished_utc':utc(),'end_load':load(),'counts':counts,'metrics':parse_metrics(text),'skip_lines':[s for s in text.splitlines() if '# SKIP' in s],'external_timeout':rc in (124,137)}
 (OUT/(label+'.load.json')).write_text(json.dumps(samples,indent=2)+'\n')
 STATE['stages'].append(row);STATE.pop('active',None);save();print('END '+json.dumps(row),flush=True)
 if row['external_timeout']:stop('timeout '+label)
 if not expected_red and rc!=0:stop('unexpected red '+label)
 if not expected_red and not syntax:
  if len(row['metrics'])!=4:stop('missing direct P10 diagnostics '+label)
  for m in row['metrics']:
   lo,hi=(1.25,2.75) if m['fault']=='inner' else (3.25,4.75)
   if not lo<=m['probeSeconds']<hi:stop('band overshoot '+label)
 return row,text
save()
if len(sys.argv)<2 or sys.argv[1]!='run':sys.exit(0)
if git('rev-parse','HEAD')!=HEAD or git('status','--porcelain'):stop('candidate/tree mismatch')
manifest=(REPO/'agents/gander/work/queue-299/MANIFEST.sha256').read_bytes()
if hashlib.sha256(manifest).hexdigest()!=DIGEST:stop('manifest mismatch')
for m in PLAN:
 data=(REPO/m['path']).read_text()
 if hashlib.sha256(data.encode()).hexdigest()!=m['source_sha256']:stop('source sha mismatch '+m['id'])
 if m.get('scope_begin'):data=data[data.index(m['scope_begin']):data.index(m['scope_end'])]
 if data.count(m['old'])!=1:stop('nonunique target '+m['id'])
STATE['status']='RUNNING';STATE['started_utc']=utc();save()
print('PREAMBLE '+json.dumps({'head':HEAD,'manifest':DIGEST,'source':'e6751ecab','case_limit':1200,'cli_limit':5400,'cli_file_count':len(FILES),'cli_files':FILES,'ambient_load_is_target_only':True}),flush=True)
for i,path in enumerate([HELPER,CASE]):execute('syntax-'+str(i+1),['node','--check',path],120,syntax=True)
for i in range(1,4):execute('case1-whole-'+str(i),['node','--test','--test-reporter=tap',CASE],1200)
execute('cli-whole',['node','--test','--test-reporter=tap',*FILES],5400)
for m in PLAN:
 original=(REPO/m['path']).read_bytes(); before=hashlib.sha256(original).hexdigest();text=original.decode()
 if m.get('scope_begin'):
  start=text.index(m['scope_begin']);end=text.index(m['scope_end']);part=text[start:end];changed=text[:start]+part.replace(m['old'],m['new'],1)+text[end:]
 else:changed=text.replace(m['old'],m['new'],1)
 row=None;output=''
 try:
  (REPO/m['path']).write_text(changed)
  ext=m['path'].rsplit('.',1)[-1];syntax=['bash','-n',m['path']] if ext=='sh' else ['node','--check',m['path']]
  execute(m['id']+'-syntax',syntax,120,syntax=True)
  row,output=execute(m['id'],['node','--test','--test-reporter=tap','--test-name-pattern','^'+m['killer']+'$',CASE],1200,expected_red=True)
 finally:
  (REPO/m['path']).write_bytes(original)
  restored=hashlib.sha256((REPO/m['path']).read_bytes()).hexdigest();status=git('status','--porcelain')
  STATE['restores'].append({'id':m['id'],'path':m['path'],'before_sha256':before,'restored_sha256':restored,'git_status':status,'identical':before==restored});save()
  if before!=restored or status:stop('restore drift '+m['id'])
 if row is None:stop('no result '+m['id'])
 if row['rc']==0:stop('survivor '+m['id'])
 named=bool(re.search(r'^not ok \d+ - '+re.escape(m['killer'])+r'\s*$',output,re.M))
 assertion='ERR_ASSERTION' in output
 if not named or not assertion:stop('wrong failure '+m['id'])
 if m['id'] in ['M1','M2','M3'] and not ('direct probe ' in output and row['metrics']):stop('wrong timing kill '+m['id'])
 if m['id']=='M4':
  if 'probe-outer-timeout' not in output or 'probe-inner-command' not in output:stop('wrong outer-reason failure M4')
  row['guard_return_receipt']={'ruling':'sage-3273 / mos-1426','configured_ms':7000,'named_command_returned':True,'rc':row['rc'],'wall_seconds':row['wall_seconds'],'actual_refusal':'probe-inner-command','external_timeout':False,'evidence':'named outer-reason ERR_ASSERTION in raw log after the unchanged safety guard returns; guard itself has no log channel'}
 row['killed']=True;row['kill_test']=m['killer'];row['mutated_sha256']=hashlib.sha256(changed.encode()).hexdigest();save()
STATE['status']='GREEN';STATE['finished_utc']=utc();save();print('GREEN all whole stages and M1-M6',flush=True)