import os,sys,json,time,subprocess,urllib.request,sqlite3,datetime,pathlib
ROOT=pathlib.Path(sys.argv[1]); SIDE=sys.argv[2]; OUT=pathlib.Path(__file__).parent / (SIDE+'-'+datetime.datetime.now().strftime('%Y%m%d-%H%M%S')); OUT.mkdir()
procs=[]; files=[]
def cmd(args):
 p=subprocess.run(args,cwd=ROOT,text=True,capture_output=True,timeout=80)
 if p.returncode: raise RuntimeError(p.stderr[-1800:])
 return p.stdout

def post(port,path='',data={}):
 r=urllib.request.Request(f'http://127.0.0.1:{port}/{path}',json.dumps(data).encode(),{'Content-Type':'application/json'})
 try:
  with urllib.request.urlopen(r,timeout=30) as x:return json.load(x)
 except urllib.error.HTTPError as x:raise RuntimeError(x.read().decode())
def wait(fn,timeout=60):
 end=time.monotonic()+timeout
 while True:
  try:
   val=fn()
   if val:return val
  except (OSError,ValueError):pass
  if time.monotonic()>end:raise RuntimeError('Timed out')
  time.sleep(.3)
def say(s):
 print(s,flush=True)
 with (OUT/'transcript.txt').open('a') as f:f.write(s+'\n')
try:
 for i,label in enumerate(['Parent','Remote']):
  home=OUT/label; (home/'userdata').mkdir(parents=True)
  bindir=home/'bin';bindir.mkdir();shim=bindir/'hostnamectl';shim.write_text('#!/bin/sh\necho '+label+'\n');shim.chmod(0o755)
  settings={'linkedEnvironmentsEnabled':True,'defaultModelSelection':{'instanceId':'devReplay','model':'multi-turn'},'providerInstances':{k:{'driver':k,'enabled':False} for k in ['codex','claudeAgent','cursor','grok','muse','pi','opencode','antigravity']}}
  (home/'userdata/settings.json').write_text(json.dumps(settings))
  log=(OUT/(label+'-server.log')).open('w');files.append(log)
  env={k:os.environ[k] for k in ['PATH','HOME','SHELL'] if k in os.environ};env['PATH']=str(bindir)+':'+env['PATH'];env.update(T3CODE_DEV_SEED='0',T3CODE_AUTO_BOOTSTRAP_PROJECT_FROM_CWD='0',PROOF_CONTROL_PORT=str(19400+i),PROOF_REPLAY_DELAY_MS='12000')
  p=subprocess.Popen(['node','scripts/dev-runner.ts','dev:server','--home-dir',str(home),'--host','127.0.0.1','--port',str(19500+i),'--no-seed'],cwd=ROOT,env=env,stdout=log,stderr=log);procs.append(p)
  wait(lambda:post(19400+i) if p.poll() is None else (_ for _ in ()).throw(RuntimeError('Server exited at startup')),90)
  workspace=home/'workspace';workspace.mkdir();(workspace/'README.md').write_text('Synthetic follower proof\n')
  cmd(['git','init','--quiet',str(workspace)])
  cmd(['git','-C',str(workspace),'add','.'])
  cmd(['git','-C',str(workspace),'-c','user.name=Proof','-c','user.email=proof@example.invalid','commit','-qm','Synthetic data'])
  cmd(['node','apps/server/src/bin.ts','project','add',str(workspace),'--title',label,'--base-dir',str(home)])
 say('Surface: two real T3 development servers; replay provider; fresh synthetic databases; loopback HTTP/MCP link.')
 say('Commit: '+cmd(['git','rev-parse','HEAD']).strip())
 pairing=json.loads(cmd(['node','apps/server/src/bin.ts','auth','pairing','create','--json','--base-dir',str(OUT/'Remote')]))
 (OUT/'pairing-private.json').write_text(json.dumps(pairing))
 token=pairing.get('token') or pairing.get('credential') or pairing.get('pairingCode')
 if not token:raise RuntimeError('Pairing response keys '+str(list(pairing)))
 link=post(19400,'link',{'url':'http://127.0.0.1:19501','pairingCode':token,'access':'full-access','expectedLabel':'Remote'})
 say('Linked remote environment '+link['environmentId']+' with full-access pairing.')
 db=sqlite3.connect(f'file:{OUT}/Remote/userdata/statev2.sqlite?mode=ro',uri=True)
 tables=db.execute("select name from sqlite_master where type='table' and name like '%project%'").fetchall()
 (OUT/'tables-private.json').write_text(json.dumps(tables))
 projectId=db.execute('select project_id from projection_projects limit 1').fetchone()[0]
 for faults,label in [(1,'transient'),(3,'repeated')]:
  launched=json.loads(cmd(['node','apps/server/src/bin.ts','thread','start',str(OUT/'Parent/workspace'),'Wait for remote delegated work.','--title',label+' follower defect','--provider','devReplay','--model','multi-turn','--json','--base-dir',str(OUT/'Parent')]))
  tid=launched['threadId']
  state=wait(lambda: (s if any(r['status'] in ['running','starting'] for r in s['runs']) else None) if (s:=post(19400,'state',{'threadId':tid})) else None)
  task=post(19400,'delegate',{'threadId':tid,'faults':faults,'task':{'task':'Complete the synthetic remote task.','title':label+' remote work','target':{'environmentId':link['environmentId'],'projectId':projectId,'providerInstanceId':'devReplay','model':'multi-turn'}}})
  say('\nCASE '+label+': injected '+str(faults)+' consecutive defect(s) at follower entry; task '+task['taskId'])
  for seconds in [1,9,22,35]:
   time.sleep(seconds if seconds==1 else seconds-prev)
   prev=seconds
   s=post(19400,'state',{'threadId':tid});t=next(t for t in s['subagents'] if t['id']==task['taskId'])
   remote=post(19401,'state',{'threadId':t['remoteChild']['threadId']})
   summary={'atSeconds':seconds,'taskStatus':t['status'],'taskResult':t['result'],'remoteThreadId':t['remoteChild']['threadId'],'remoteRuns':[{'id':r['id'],'status':r['status']} for r in remote['runs']],'parentRuns':[{'id':r['id'],'status':r['status']} for r in s['runs']],'follower':s['proofFaults'].get(task['taskId'])}
   say('  +'+str(seconds)+'s: local task='+t['status']+'; remote runs='+','.join(r['status'] for r in remote['runs'])+'; follower attempts='+str(summary['follower']['attempts']))
   say('    Parent runs: '+', '.join(r['status'] for r in s['runs']))
   say('    Remote thread: '+t['remoteChild']['threadId'])
   say('    Result: '+str(t['result']))
   say('    Internal exception leaked in result: '+str('PRIVATE_PROOF_EXCEPTION' in (t['result'] or '')))
   if seconds == 35:
    wakes=[m for m in s['messages'] if m.get('role')=='user' and m.get('text','').startswith('Delegated task ')]
    say('    Persisted parent completion wake messages: '+str(len(wakes)))
    say('    Completed parent runs: '+str(sum(r['status']=='completed' for r in s['runs'])))
   with (OUT/(label+'-observations.jsonl')).open('a') as f:f.write(json.dumps(summary)+'\n')
   (OUT/(label+'-state-'+str(seconds)+'-private.json')).write_text(json.dumps(s,default=str))
 say('\nRun finished.')
finally:
 for p in procs:
  p.terminate()
 for p in procs:
  try:p.wait(20)
  except subprocess.TimeoutExpired:p.kill();p.wait()
 for f in files:f.close()
 print('Servers stopped.',flush=True)
 if os.environ.get('DISPLAY'):time.sleep(5)