Codex

#!/usr/bin/env python3
import argparse,json,os,platform,re,shutil,socket,subprocess,sys,time,urllib.request,urllib.error,urllib.parse
from pathlib import Path
from http.server import SimpleHTTPRequestHandler,ThreadingHTTPServer
ROOT=Path(__file__).resolve().parent
VERSION=json.loads((ROOT/'package.json').read_text())['version']
def version_tuple(value):
 m=re.fullmatch(r'(\d+)\.(\d+)\.(\d+)(?:[-+].*)?',str(value).strip())
 return tuple(map(int,m.groups())) if m else (0,0,0)
def cli_state():
 """The Codex CLI on PATH, and whether it is the golden-image wrapper onto the extension's
 bundled binary (the no-duplicates rule) or a second install (npm, standalone)."""
 p=shutil.which('codex');kind='missing';target=None;version=None
 if p:
  try:head=open(p,'rb').read(2048).decode('utf8','replace')
  except Exception:head=''
  real=str(Path(p).resolve())
  if 'Adom golden image' in head or '/code-server/extensions/' in real:kind='wrapper';target='the extension\'s bundled binary'
  elif 'node' in head.split('\n',1)[0]:kind='npm';target=real
  else:kind='standalone';target=real
  try:
   import subprocess;version=subprocess.run([p,'--version'],capture_output=True,text=True,timeout=15).stdout.strip()[:60] or None
  except Exception:pass
 return {'path':p,'kind':kind,'target':target,'version':version,'duplicate':kind in('npm','standalone') and any(Path.home().glob('.local/share/code-server/extensions/openai.chatgpt-*/bin/linux-x86_64/codex'))}
def signin_state():
 """Read ~/.codex/auth.json (shared by the extension and the CLI); never a live call."""
 ch=Path(os.environ.get('CODEX_HOME',Path.home()/'.codex'));f=ch/'auth.json';email=None;ok=False
 try:
  d=json.loads(f.read_text());tok=d.get('tokens') or {};ok=bool(tok.get('access_token') or tok.get('refresh_token') or d.get('OPENAI_API_KEY'))
  idt=tok.get('id_token')
  if idt:
   import base64;seg=idt.split('.')[1];seg+='='*(-len(seg)%4);email=json.loads(base64.urlsafe_b64decode(seg)).get('email')
 except Exception:pass
 return {'signedIn':ok,'email':email,'file':str(f),'shared':'the extension and the CLI read this same file'}
def context():
 ch=Path(os.environ.get('CODEX_HOME',Path.home()/'.codex'))
 return {'app':'adom/codex','version':VERSION,'cli':cli_state(),'signin':signin_state(),'container':{'system':platform.system(),'kernel':platform.release(),'hostname':platform.node(),'cwd':os.getcwd()},'tools':{n:shutil.which(n) for n in ['codex','adom-wiki','adom-cli','adom-bridge']},'instructionsInstalled':(ch/'skills/codex-adom/SKILL.md').exists(),'hostTarget':'Not inferred; select and verify through Adom Bridge','hydrogenControlConfigured':(Path.home()/'.adom/hydrogen-control-url').exists()}
class Handler(SimpleHTTPRequestHandler):
 def __init__(self,*a,**kw):super().__init__(*a,directory=str(ROOT),**kw)
 def log_message(self,*a):pass
 def end_headers(self):
  origin=self.headers.get('Origin')
  if self.command=='GET' and self.path=='/api/status' and origin in {'http://tauri.localhost','https://tauri.localhost','tauri://localhost'}:
   self.send_header('Access-Control-Allow-Origin',origin);self.send_header('Vary','Origin')
  super().end_headers()
 def json(self,obj,status=200):
  b=json.dumps(obj).encode();self.send_response(status);self.send_header('Content-Type','application/json');self.send_header('Content-Length',str(len(b)));self.end_headers();self.wfile.write(b)
 def proxy(self):
  path=Path.home()/'.adom/hydrogen-control-url'
  if not path.exists():return self.json({'ok':False,'error':'Hydrogen control API unavailable','hint':'Open Adom Hydrogen in this workspace; its dynamic control URL is missing.'},503)
  base=path.read_text().strip()
  data=self.rfile.read(int(self.headers.get('Content-Length','0'))) if self.command=='POST' else None
  upstream=self.path.removeprefix('/codex') if self.path.startswith('/codex/') else self.path
  if self.command=='POST' and self.path=='/codex/ai-threads/run-prompt':
   try:
    status=json.load(urllib.request.urlopen(base.rstrip('/')+'/updater/status',timeout=10))
    current=status.get('updater',{}).get('current_version') or 'unknown'
    if version_tuple(current)<(1,0,233):return self.json({'ok':False,'error':f'Send to AI requires Hydrogen 1.0.233 or newer (found {current}). Update Hydrogen, or copy the prompt into Codex. No prompt was sent.'},409)
    payload=json.loads(data);payload.pop('type',None);payload['prefer']='codex';payload['background']=True;data=json.dumps(payload).encode()
   except (ValueError,TypeError,AttributeError):return self.json({'error':'Valid JSON object required'},400)
   except (OSError,urllib.error.URLError):return self.json({'error':'Cannot verify Hydrogen version; no prompt was sent.'},503)
  req=urllib.request.Request(base.rstrip('/')+upstream,data=data,method=self.command,headers={'Content-Type':'application/json'})
  try:r=urllib.request.urlopen(req,timeout=120)
  except urllib.error.HTTPError as e:r=e
  except Exception as e:return self.json({'ok':False,'error':'Hydrogen control API unavailable','detail':str(e)},503)
  b=r.read();self.send_response(r.status);self.send_header('Content-Type',r.headers.get('Content-Type','application/json'));self.send_header('Content-Length',str(len(b)));self.end_headers();self.wfile.write(b)
 def do_GET(self):
  if self.path in ['/api/status','/state']:return self.json(context())
  if self.path=='/events':
   b=('data: '+json.dumps(context())+'\n\n').encode();self.send_response(200);self.send_header('Content-Type','text/event-stream');self.send_header('Content-Length',str(len(b)));self.end_headers();self.wfile.write(b);return
  if self.path in ['/updater/status','/codex/updater/status'] or self.path.startswith(('/ai-threads/run-prompt/status?','/codex/ai-threads/run-prompt/status?')):return self.proxy()
  if self.path=='/':self.path='/dashboard/index.html'
  # Serve only package-owned public assets, never arbitrary host files.
  if not self.path.startswith(('/dashboard/','/videos/','/docs/')):return self.send_error(404)
  return super().do_GET()
 def do_OPTIONS(self):
  if self.path not in ['/ai-threads/run-prompt','/codex/ai-threads/run-prompt']:return self.send_error(404)
  self.send_response(204);self.send_header('Allow','POST, OPTIONS');self.send_header('Content-Length','0');self.end_headers()
 def do_POST(self):
  origin=self.headers.get('Origin')
  if origin and urllib.parse.urlsplit(origin).netloc!=self.headers.get('Host'):return self.json({'error':'Origin refused'},403)
  if self.headers.get('Content-Type','').split(';')[0]!='application/json':return self.json({'error':'JSON required'},415)
  if self.path in ['/ai-threads/run-prompt','/codex/ai-threads/run-prompt']:return self.proxy()
  return self.send_error(404)
def main():
 p=argparse.ArgumentParser();p.add_argument('command',choices=['context','serve','_serve']);p.add_argument('--print-url',action='store_true');p.add_argument('--port',type=int,default=47863);a=p.parse_args()
 if a.command=='context':print(json.dumps(context(),indent=2));return
 if a.command=='_serve':ThreadingHTTPServer(('127.0.0.1',a.port),Handler).serve_forever();return
 state=Path.home()/'.local/state/adom/codex';state.mkdir(parents=True,exist_ok=True)
 urlfile=state/'dashboard-url'
 urls=([urlfile.read_text().strip()] if urlfile.exists() else [])+[f'http://127.0.0.1:{a.port}/']
 for candidate in urls:
  try:
   r=json.load(urllib.request.urlopen(candidate+'api/status',timeout=1))
   if r.get('app')=='adom/codex' and r.get('version')==VERSION:print(candidate,flush=True);return
  except Exception:pass
 with socket.socket() as sock:
  try:sock.bind(('127.0.0.1',a.port))
  except OSError:sock.bind(('127.0.0.1',0))
  a.port=sock.getsockname()[1]
 url=f'http://127.0.0.1:{a.port}/'
 try:
  r=json.load(urllib.request.urlopen(url+'api/status',timeout=1));assert r.get('app')=='adom/codex' and r['version']==VERSION
 except Exception:
  log=Path.home()/'.local/state/adom/codex';log.mkdir(parents=True,exist_ok=True)
  with (log/'dashboard.log').open('ab') as out:subprocess.Popen([sys.executable,str(Path(__file__).resolve()),'_serve','--port',str(a.port)],stdout=out,stderr=out,start_new_session=True)
  for _ in range(50):
   try:
    ready=json.load(urllib.request.urlopen(url+'api/status',timeout=.5))
    if ready.get('app')!='adom/codex' or ready.get('version')!=VERSION:raise RuntimeError('Port is occupied by a different service')
    break
   except Exception:time.sleep(.1)
  else:raise RuntimeError('Dashboard failed to start; inspect '+str(log/'dashboard.log'))
 urlfile.write_text(url);print(url,flush=True)
if __name__=='__main__':main()