#!/usr/bin/env python3
"""Explicit real-model spike: independent App Server, new test thread only."""
import asyncio,json,pathlib,sys,tempfile,os
ROOT=pathlib.Path(tempfile.mkdtemp(prefix='codex-control-spike-')); CWD=str(ROOT/'project'); pathlib.Path(CWD).mkdir()
DISCOVERY_CWD=str(pathlib.Path(sys.argv[1] if len(sys.argv)>1 else '.').resolve())
print('Proof directory:',ROOT,flush=True)
class RPC:
 async def start(self):
  self.p=await asyncio.create_subprocess_exec('codex','app-server','--listen','stdio://',stdin=-1,stdout=-1,stderr=open(ROOT/'stderr.log','a'))
  self.n=0; self.pending={}; self.events=[]; self.task=asyncio.create_task(self.read())
  await self.call('initialize',{'clientInfo':{'name':'device_control_spike','version':'0.2.2'},'capabilities':{'experimentalApi':True}})
  self.p.stdin.write(b'{"method":"initialized"}\n'); await self.p.stdin.drain(); return self
 async def read(self):
  while line:=await self.p.stdout.readline():
   obj=json.loads(line)
   if 'id' in obj and obj['id'] in self.pending: self.pending.pop(obj['id']).set_result(obj)
   elif 'method' in obj:
    self.events.append(obj)
    if 'id' in obj:
     self.p.stdin.write((json.dumps({'id':obj['id'],'result':{'decision':'decline'}})+'\n').encode()); await self.p.stdin.drain()
 async def call(self,m,p):
  self.n+=1; f=asyncio.get_running_loop().create_future(); self.pending[self.n]=f
  self.p.stdin.write((json.dumps({'id':self.n,'method':m,'params':p})+'\n').encode()); await self.p.stdin.drain()
  r=await asyncio.wait_for(f,60)
  if 'error' in r: raise RuntimeError(m+': '+json.dumps(r['error']))
  return r['result']
 async def stop(self):
  self.p.stdin.close()
  try: await asyncio.wait_for(self.p.wait(),12)
  except TimeoutError: self.p.terminate(); await self.p.wait()
  await self.task
 async def turn(self,sid,txt):
  mark=len(self.events)
  r=await self.call('turn/start',{'threadId':sid,'input':[{'type':'text','text':txt}]})
  print('input accepted',r['turn']['id'],flush=True)
  for _ in range(180):
   events=self.events[mark:]
   done=next((e for e in events if e['method']=='turn/completed'),None)
   if done:
    output=''.join(e['params'].get('delta','') for e in events if e['method']=='item/agentMessage/delta')
    print('turn',done['params']['turn']['status'],'output',output,'events',sorted({e['method'] for e in events}),flush=True)
    return {'status':done['params']['turn']['status'],'output':output,'events':sorted({e['method'] for e in events})}
   await asyncio.sleep(1)
  raise TimeoutError('turn did not complete')
async def main():
 r=await RPC().start(); result={}
 try:
  listing=await r.call('thread/list',{'limit':5}); result['static_list']=len(listing['data']); print('static threads',len(listing['data']),flush=True)
  listing=await r.call('thread/list',{'cwd':DISCOVERY_CWD,'limit':100}); result['cwd_count']=len(listing['data']); result['cwd_filter']=all(x['cwd']==DISCOVERY_CWD for x in listing['data']); print('cwd',result['cwd_count'],result['cwd_filter'],flush=True)
  created=await r.call('thread/start',{'cwd':CWD,'approvalPolicy':'on-request','sandbox':'read-only','baseInstructions':'You are a connectivity test. Do not use tools unless explicitly asked. Reply exactly as requested.'})
  sid=created['thread']['id']; result['session_id']=sid; print('test thread',sid,flush=True)
  result['first']=await r.turn(sid,'Reply exactly SESSION_READY')
  assert result['first']['status']=='completed' and result['first']['output'].strip()=='SESSION_READY'
  await r.stop(); r=await RPC().start()
  listing=await r.call('thread/list',{'cwd':CWD,'limit':100}); result['rediscovery']=any(x['id']==sid for x in listing['data'])
  resumed=await r.call('thread/resume',{'threadId':sid,'excludeTurns':True}); result['same_id']=resumed['thread']['id']==sid; print('restart rediscovery',result['rediscovery'],'same id',result['same_id'],flush=True)
  assert result['rediscovery'] and result['same_id'] and result['cwd_filter']
  result['read']=bool(await r.call('thread/read',{'threadId':sid}))
  result['settings']=await r.call('thread/settings/update',{'threadId':sid,'model':created['model'],'effort':'low','approvalPolicy':'on-request'})
  result['resumed']=await r.turn(sid,'Reply exactly SESSION_RESUMED')
  assert result['resumed']['status']=='completed' and result['resumed']['output'].strip()=='SESSION_RESUMED'
  result['turns']=len((await r.call('thread/turns/list',{'threadId':sid,'limit':10}))['data'])
  items=await r.call('thread/items/list',{'threadId':sid,'limit':20}); result['items_keys']=list(items)
 except Exception as e:
  result['error']=str(e); print('ERROR',e,flush=True)
 finally:
  await r.stop(); (ROOT/'result.json').write_text(json.dumps(result,indent=2)); print(json.dumps(result,indent=2),flush=True)
  if 'error' in result: raise SystemExit(1)
if __name__ == '__main__':
 asyncio.run(main())