"""Real-server long-thread benchmark. Usage: python3 prove.py CHECKOUT OUTPUT_DIR PORT [turns=200] [toolsPerTurn=11] [user].
Fresh synthetic state only. Does not read any installed T3 home. Run identically on each revision.
Requires node and installed workspace dependencies. OUTPUT_DIR must be a new directory.
"""
import os,sys,json,time,subprocess,pathlib,urllib.request,statistics,hashlib,signal
repo=pathlib.Path(sys.argv[1]).resolve(); out=pathlib.Path(sys.argv[2]).resolve(); port=int(sys.argv[3]); turns=int(sys.argv[4]) if len(sys.argv)>4 else 200
tools_per_turn=int(sys.argv[5]) if len(sys.argv)>5 else 11
human=len(sys.argv)>6 and sys.argv[6]=='user'
out.mkdir(parents=True,exist_ok=False)
home=out/'home'; (home/'userdata').mkdir(parents=True)
env={**os.environ,'T3CODE_DEV_SEED':'0'}
cli=['node',str(repo/'apps/server/src/bin.ts')]
def run(args):
 p=subprocess.run(cli+args+['--base-dir',str(home)],cwd=repo,env=env,text=True,capture_output=True)
 if p.returncode: raise RuntimeError(f'CLI {args[:2]} failed: {p.stderr[-1500:]} {p.stdout[-1500:]}')
 return p.stdout
# Same synthetic replay on both sides. Zero simulated delay affects seeding only.
fixture=repo/'apps/server/scripts/dev-replay/command_output_transcript.ndjson'
original=fixture.read_bytes()
frames=[]
for turn in range(20):
 frames.append({'type':'expect_outbound','frame':{'kind':'request','method':'session/prompt','params':'<any>'}})
 for i in range(tools_per_turn):
  update={'sessionUpdate':'tool_call','toolCallId':f'synthetic-{i}','kind':'other','title':f'fixture_tool_{i}','status':'completed','rawInput':{'index':i},'rawOutput':{'text':'x'*40000}}
  if turn==19 and i==5: update['rawOutput']['htmlRender']={'html':'<p>Synthetic inline output</p>'}
  frames.append({'type':'emit_inbound','afterMs':0,'frame':{'kind':'notification','method':'session/update','params':{'sessionId':'fixture','update':update}}})
 frames.append({'type':'emit_inbound','afterMs':0,'frame':{'kind':'notification','method':'session/update','params':{'sessionId':'fixture','update':{'sessionUpdate':'agent_message_chunk','content':{'type':'text','text':'Complete.'}}}}})
 frames.append({'type':'emit_inbound','frame':{'kind':'response','method':'session/prompt','result':{'stopReason':'end_turn'}}})
fixture.write_text('\n'.join(map(json.dumps,frames))+'\n')
settings={'defaultModelSelection':{'instanceId':'devReplay','model':'command-output'},'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))
workspace=out/'synthetic-project'; workspace.mkdir(); (workspace/'README.md').write_text('Synthetic benchmark data only.\n')
subprocess.run(['git','init','-q',str(workspace)],check=True)
subprocess.run(['git','-C',str(workspace),'add','.'],check=True)
subprocess.run(['git','-C',str(workspace),'-c','user.name=Fixture','-c','user.email=fixture@example.invalid','commit','-qm','Synthetic fixture'],check=True)
log=open(out/'private-server.log','w')
server=subprocess.Popen(cli+['serve','--host','127.0.0.1','--port',str(port),'--dev-url',f'http://127.0.0.1:{port+1}','--base-dir',str(home)],cwd=repo,env=env,stdout=log,stderr=log,start_new_session=True)
base=f'http://127.0.0.1:{port}'
try:
 for i in range(120):
  if server.poll() is not None: raise RuntimeError('server exited; inspect private-server.log')
  try:
   urllib.request.urlopen(base+'/.well-known/t3/environment',timeout=1).read();break
  except Exception: time.sleep(1)
 else: raise RuntimeError('startup timed out')
 run(['project','add',str(workspace),'--title','Synthetic performance fixture'])
 if human:
  import sqlite3,uuid
  con=sqlite3.connect('file:'+str(home/'userdata/statev2.sqlite')+'?mode=ro',uri=True)
  pid=con.execute('select project_id from projection_projects limit 1').fetchone()[0];con.close()
  token=run(['auth','session','issue','--ttl','1h','--token-only']).strip()
  def dispatch(command):
   command.update(commandId=str(uuid.uuid4()),createdBy='user',creationSource='web')
   p=subprocess.run(['node',str(pathlib.Path(__file__).with_name('command.mjs'))],input=json.dumps({'base':base,'token':token,'command':command}),text=True,capture_output=True,timeout=30)
   if p.returncode:raise RuntimeError(p.stderr)
  tid=str(uuid.uuid4())
  dispatch({'type':'thread.create','threadId':tid,'projectId':pid,'title':'Long synthetic history','modelSelection':{'instanceId':'devReplay','model':'command-output'},'runtimeMode':'full-access','interactionMode':'default','branch':None,'worktreePath':None})
  for i in range(turns):
   dispatch({'type':'message.dispatch','threadId':tid,'messageId':str(uuid.uuid4()),'text':f'Synthetic turn {i+1}','attachments':[],'dispatchMode':{'type':'start_immediately'}})
   for attempt in range(600):
    req=urllib.request.Request(base+'/api/orchestration/threads/'+tid,headers={'Authorization':'Bearer '+token,'x-t3-orchestration-protocol':'2'})
    with urllib.request.urlopen(req,timeout=60) as response: snapshot=json.load(response)
    runs=snapshot['projection']['runs']
    if len(runs)==i+1 and runs[-1]['status'] in ['completed','failed','cancelled']:
     if runs[-1]['status']!='completed':raise RuntimeError('Synthetic run failed')
     break
    time.sleep(.1)
   else:raise RuntimeError('Seed turn timed out')
   if (i+1)%10==0:print('seed',i+1,flush=True)
 else:
  started=json.loads(run(['thread','start',str(workspace),'Synthetic turn 1','--title','Long synthetic history','--provider','devReplay','--model','command-output','--wait','--json']))
  tid=started['threadId']; print('seed 1',flush=True)
  for i in range(1,turns):
   run(['thread','send',tid,f'Synthetic turn {i+1}','--wait','--json'])
   if (i+1)%10==0: print('seed',i+1,flush=True)
 token=run(['auth','session','issue','--scope','orchestration:read','--ttl','1h','--token-only']).strip()
 headers={'Authorization':'Bearer '+token,'x-t3-orchestration-protocol':'2'}
 def get(path):
  req=urllib.request.Request(base+path,headers=headers)
  with urllib.request.urlopen(req,timeout=60) as r: return json.loads(r.read())
 path='/api/orchestration/threads/'+tid
 snap=get(path+'/bounded?deferWork=1'); (out/'private-shape.json').write_text(json.dumps(snap))
 print('bounded keys',list(snap),flush=True)
 cursor=snap.get('historyCursor')
 if not cursor: raise RuntimeError('No older history cursor: '+str(list(snap)))
 olderpath=path+'/history?deferWork=1&cursor='+urllib.parse.quote(cursor)
 older=get(olderpath)
 # Compare semantic content with nondeterministic ids/timestamps removed. Retain order,
 # item types, tool names, payload text, statuses, and deferred-work counts/boundaries.
 def normalize(x):
  if isinstance(x,list): return [normalize(v) for v in x]
  if not isinstance(x,dict): return x
  return {k:normalize(v) for k,v in x.items() if not (k.lower().endswith('id') or k.lower().endswith('ids') or k.lower().endswith('at') or k in ['snapshotSequence','position','ordinal','historyCursor','nextCursor','cursor','sourceItemId','entryId','nativeItemRef','lineage'])}
 semantics={'snapshotItems':normalize(snap['projection']['visibleTurnItems']),'older':normalize(older),'hasMoreHistory':snap['hasMoreHistory']}
 (out/'semantics.json').write_text(json.dumps(semantics,sort_keys=True))
 def timed(path):
  t=time.perf_counter(); value=get(path); return (time.perf_counter()-t)*1000,value
 rounds=[]
 from concurrent.futures import ThreadPoolExecutor
 for r in range(5):
  for _ in range(3): get(path+'/bounded?deferWork=1');get(olderpath)
  opens=[];pages=[]
  for _ in range(40):
   t,_=timed(path+'/bounded?deferWork=1');opens.append(t)
   t,_=timed(olderpath);pages.append(t)
  concurrent=[]
  with ThreadPoolExecutor(max_workers=6) as pool:
   for _ in range(15):
    concurrent.extend(pool.map(lambda _:timed(path+'/bounded?deferWork=1')[0],range(6)))
  result={'round':r+1,'openMedianMs':statistics.median(opens),'olderMedianMs':statistics.median(pages),'combinedMedianMs':statistics.median([a+b for a,b in zip(opens,pages)]),'sixConcurrentP95Ms':sorted(concurrent)[int(.95*(len(concurrent)-1))],'rawOpenMs':opens,'rawOlderMs':pages,'rawConcurrentMs':concurrent}
  rounds.append(result);print(json.dumps({k:v for k,v in result.items() if not k.startswith('raw')}),flush=True)
 sha=subprocess.check_output(['git','rev-parse','HEAD'],cwd=repo,text=True).strip()
 result={'commit':sha,'environment':'Linux x86_64, Node '+subprocess.check_output(['node','-v'],text=True).strip()+', loopback HTTP, real T3 server, fresh synthetic SQLite database, same VM','seed':{'authorship':'user/web' if human else 'agent/mcp','turns':turns,'toolsPerTurn':tools_per_turn,'outputBytesPerTool':40000,'inlineToolsEvery20Turns':1},'semanticSha256':hashlib.sha256(json.dumps(semantics,sort_keys=True).encode()).hexdigest(),'rounds':rounds}
 (out/'measurement.json').write_text(json.dumps(result,indent=2))
 print('COMPLETE',sha,flush=True)
finally:
 os.killpg(server.pid,signal.SIGTERM)
 try: server.wait(timeout=10)
 except subprocess.TimeoutExpired: os.killpg(server.pid,signal.SIGKILL);server.wait()
 fixture.write_bytes(original);log.close()