AI Flow

#!/usr/bin/env python3
"""AI Flow silkscreen observer: persisted solver events, no provider calls or CAD edits."""
from recording_check import check as check_recording
import argparse,base64,hashlib,json,os,pathlib,subprocess,sys,time,threading,http.server,urllib.request,urllib.parse,signal
P=pathlib.Path;ROOT=P(__file__).resolve().parent;REG=P.home()/'.adom/instances/aiflow-silkscreen';REG.mkdir(parents=True,exist_ok=True)
def atomic(p,d):
 q=p.with_suffix('.tmp');q.write_text(json.dumps(d,indent=2));q.replace(p)
p=argparse.ArgumentParser();p.add_argument('action',choices=['start','serve','show','ls','state','control','stop']);p.add_argument('--run',required=True,type=P);p.add_argument('--ai-thread',required=True);p.add_argument('--port',default='auto');p.add_argument('--surface',choices=['wv','pup'],default='wv');p.add_argument('--target');p.add_argument('--url');p.add_argument('--json',default='{}');p.add_argument('--reason');a=p.parse_args();run=a.run.resolve();key=hashlib.sha256((a.ai_thread+str(run)).encode()).hexdigest()[:16];reg=REG/(key+'.json');snap=REG/(key+'-snapshot.json');folder=run/'silkscreen-dashboard';folder.mkdir(exist_ok=True);events=folder/'events.jsonl';manifest=folder/'board.json';version=hashlib.sha256((ROOT/'index.html').read_bytes()+P(__file__).read_bytes()).hexdigest()[:12]
def entry():
 try:
  e=json.loads(reg.read_text());r=json.load(urllib.request.urlopen('http://127.0.0.1:'+str(e['port'])+'/health',timeout=2));return e if r.get('key')==key else None
 except Exception:return None
def request(e,path,data=None):
 body=None if data is None else json.dumps(dict(data,aiThread=a.ai_thread)).encode();return json.load(urllib.request.urlopen(urllib.request.Request('http://127.0.0.1:'+str(e['port'])+path,data=body,headers={'Content-Type':'application/json'}),timeout=5))
if a.action=='ls':
 for f in REG.glob('*.json'):
  if f.name.endswith('-snapshot.json'):continue
  e=json.loads(f.read_text())
  try:os.kill(e['pid'],0)
  except ProcessLookupError:f.unlink();continue
  print(json.dumps(dict(e,versionDrift=e.get('version')!=version)))
 sys.exit()
e=entry()
if a.action in ['start','show']:
 if not e:
  log=open(folder/'server.log','ab');subprocess.Popen([sys.executable,str(P(__file__).resolve()),'serve','--run',str(run),'--ai-thread',a.ai_thread,'--port',a.port],stdout=log,stderr=log,start_new_session=True)
  for _ in range(80):
   time.sleep(.1);e=entry()
   if e:break
  if not e:raise SystemExit('ERROR: server did not become healthy; see '+str(folder/'server.log'))
 url=a.url or ('http://'+str(e['port'])+'.localhost:'+urllib.parse.urlparse((P.home()/'.adom/hd-proxy-url').read_text().strip()).port.__str__()+'/' if (P.home()/'.adom/hd-proxy-url').exists() else 'http://127.0.0.1:'+str(e['port'])+'/')
 if a.action=='show':
  if a.surface=='wv':opened=subprocess.run(['adom-cli','hydrogen','webview','open-or-refresh','--name','AI Flow silkscreen','--url',url],check=True,capture_output=True,text=True);print(opened.stdout);e['tabId']=json.loads(opened.stdout).get('tabId');atomic(reg,e)
  else:
   if not a.target or not a.url:raise SystemExit('ERROR: pup needs --target and a desktop-reachable --url')
   subprocess.run(['adom-bridge','pup_open_window',json.dumps({'url':url,'background':True}),'--target',a.target,'--ai-thread',a.ai_thread],check=True)
 print(json.dumps(dict(e,url=url,reused=True)));sys.exit()
if a.action in ['state','control','stop']:
 if not e:raise SystemExit('ERROR: not running; use start')
 if a.action=='stop' and not a.reason:raise SystemExit('ERROR reason_required: explain why this task-owned server should stop')
 result=request(e,'/shutdown' if a.action=='stop' else '/control' if a.action=='control' else '/state',dict(json.loads(a.json),reason=a.reason) if a.action!='state' else None)
 if a.action=='stop':
  import fcntl
  deadline=time.monotonic()+10
  with open(REG/(key+'.lock'),'a') as lease:
   while True:
    try:fcntl.flock(lease,fcntl.LOCK_EX|fcntl.LOCK_NB);fcntl.flock(lease,fcntl.LOCK_UN);break
    except BlockingIOError:
     if time.monotonic()>=deadline:raise SystemExit('ERROR: shutdown accepted but this server still holds its lease. Hint: check state before restarting; do not kill a shared process.')
     time.sleep(.1)
 print(json.dumps(result));sys.exit()
if e:print('OK: already running');sys.exit()
settingspath=REG/'settings.json';settings=json.loads(settingspath.read_text()) if settingspath.exists() else dict(ttlHours=24,speed=4,showObstacles=True)
state=json.loads(snap.read_text()) if snap.exists() else dict(mode='live',cursor=-1,playing=False,layer='F.SilkS',view='2d');clients=set();lock=threading.RLock();last=time.time();revision=0;history=[];offset=0

def payload():return dict(state,settings=settings,events=history,aiThread=a.ai_thread,version=version)
def save():atomic(snap,state)
def broadcast():
 global revision
 revision+=1
 # Bounded state messages contain the current event; history is fetched separately.
 data=dict(state,settings=settings,event=history[state['cursor']] if 0<=state['cursor']<len(history) else None,total=len(history),revision=revision)
 for q in list(clients):
  try:q.write(('event: state\ndata: '+json.dumps(data)+'\n\n').encode());q.flush()
  except Exception:clients.discard(q)
class Handler(http.server.BaseHTTPRequestHandler):
 def log_message(self,*args):pass
 def response(self,data,code=200,ctype='application/json'):
  b=data if isinstance(data,bytes) else json.dumps(data).encode();self.send_response(code);self.send_header('Content-Type',ctype);self.send_header('Cache-Control','no-store');self.send_header('Content-Length',str(len(b)));self.end_headers();self.wfile.write(b)
 def do_GET(self):
  global last
  last=time.time();path=urllib.parse.urlparse(self.path).path
  if path=='/health':return self.response(dict(key=key,version=version))
  if path=='/state':return self.response(payload())
  if path=='/settings':return self.response(dict(settings=settings,enums={'ttlHours':'1..168','speed':'0.5..60','showObstacles':'boolean'}))
  if path=='/stream':
   self.send_response(200);self.send_header('Content-Type','text/event-stream');self.send_header('Cache-Control','no-cache');self.send_header('X-Accel-Buffering','no');self.end_headers();self.wfile.write((':'+(' '*2048)+'\n\n').encode());clients.add(self.wfile);broadcast()
   try:
    while True:time.sleep(15);self.wfile.write(b': heartbeat\n\n');self.wfile.flush()
   except Exception:clients.discard(self.wfile)
   return
  if path in ['/','/index.html']:return self.response((ROOT/'index.html').read_bytes(),ctype='text/html')
  if path=='/board.json':return self.response(manifest.read_bytes() if manifest.exists() else b'{"obstacles":[],"labels":[],"bounds":[0,0,100,100]}')
  if path.startswith('/recordings/'):
   f=(folder/path.lstrip('/')).resolve()
   if not f.is_relative_to((folder/'recordings').resolve()) or not f.is_file():return self.response({'error':'not found'},404)
   return self.response(f.read_bytes(),ctype='video/webm')
  if path.startswith('/models/'):
   f=(folder/path.lstrip('/')).resolve()
   if not f.is_relative_to((folder/'models').resolve()) or not f.is_file():return self.response({'error':'not found'},404)
   return self.response(f.read_bytes(),ctype='model/gltf-binary')
  self.response({'error':'not found'},404)
 def do_POST(self):
  global last
  last=time.time()
  try:
   if int(self.headers.get('Content-Length','0'))>(40_000_000 if self.path=='/recording' else 65536):raise ValueError('request too large')
   body=json.loads(self.rfile.read(int(self.headers.get('Content-Length','0'))))
   if not body.get('aiThread'):return self.response(dict(errorCode='caller_identity_required',hint='Pass aiThread or CLI --ai-thread.'),400)
   with lock:
    if self.path=='/recording':
     video=base64.b64decode(body['video'],validate=True)
     if len(video)<1024 or video[:4]!=bytes.fromhex('1a45dfa3'):raise ValueError('Refused empty or invalid WebM recording')
     clips=folder/'recordings';clips.mkdir(exist_ok=True);name='replay-'+str(time.time_ns())+'.webm';(clips/name).write_bytes(video)
     quality=check_recording(clips/name,body.get('elapsedSeconds',0));accepted=body.get('accepted') is not False and quality['accepted']
     state['lastRecording']={'file':str(clips/name),'url':'recordings/'+name,'kind':body.get('kind','walkthrough'),'bytes':len(video),'replay':True,'accepted':accepted,'rejectedReason':body.get('rejectedReason') or quality.get('reason'),'frameValidation':quality};state['recordRequest']=None;save();broadcast();self.response({'ok':True,'recording':state['lastRecording']});return
    if self.path=='/shutdown':
     if not body.get('reason'):raise ValueError('reason_required')
     self.response({'ok':True});threading.Thread(target=shutdown,daemon=True).start();return
    if self.path=='/settings':
     for k,v in body.items():
      if k=='aiThread':continue
      if k=='ttlHours' and isinstance(v,(int,float)) and 1<=v<=168:settings[k]=v
      elif k=='speed' and isinstance(v,(int,float)) and .5<=v<=60:settings[k]=v
      elif k=='showObstacles' and isinstance(v,bool):settings[k]=v
      else:raise ValueError('Allowed settings: ttlHours 1..168, speed .5..60, showObstacles boolean')
     atomic(settingspath,settings)
    elif self.path=='/control':
     for k,v in body.items():
      if k in ['aiThread','reason']:continue
      if k=='record' and v in ['overview','walkthrough']:state['recordRequest']={'id':time.time_ns(),'kind':v}
      elif k=='mode' and v in ['live','replay']:state[k]=v
      elif k=='layer' and v in ['F.SilkS','B.SilkS']:state[k]=v
      elif k=='view' and v in ['2d','3d']:state[k]=v
      elif k in ['playing','summary'] and isinstance(v,bool):state[k]=v
      elif k=='cursor' and isinstance(v,int) and -1<=v<len(history):state[k]=v
      else:raise ValueError('Allowed controls: mode live/replay, layer F.SilkS/B.SilkS, view 2d/3d, playing boolean, cursor within event history')
     if state['mode']=='live':state['cursor']=len(history)-1;state['playing']=False
    else:return self.response({'error':'not found'},404)
    save();broadcast();self.response({'ok':True,'state':state})
  except (ValueError,TypeError,KeyError) as ex:self.response({'error':str(ex)},400)
http.server.ThreadingHTTPServer.allow_reuse_address=True
import fcntl
lease=open(REG/(key+'.lock'),'w')
try:fcntl.flock(lease,fcntl.LOCK_EX|fcntl.LOCK_NB)
except BlockingIOError:raise SystemExit('OK: another instance for this run/thread is starting or running')
for attempt in range(30):
 try:
  server=http.server.ThreadingHTTPServer(('127.0.0.1',0 if a.port=='auto' else int(a.port)),Handler);break
 except OSError:
  if attempt==29:raise
  time.sleep(.1)
server.daemon_threads=True
atomic(reg,dict(pid=os.getpid(),port=server.server_port,thread=a.ai_thread,run=str(run),version=version,startedAt=time.time()))
def shutdown():
 save();current=json.loads(reg.read_text()) if reg.exists() else {};reg.unlink(missing_ok=True)
 if current.get('tabId'):
  subprocess.run(['adom-cli','hydrogen','workspace','remove-tab','--tab-id',current['tabId']],capture_output=True)
 server.shutdown()
def watch():
 global offset,history
 previous=time.time()
 while True:
  time.sleep(.05)
  with lock:
   if events.exists():
    with events.open() as f:
     f.seek(offset)
     while True:
      pos=f.tell();line=f.readline()
      if not line or not line.endswith('\n'):break
      try:history.append(json.loads(line))
      except json.JSONDecodeError:pass
      offset=f.tell()
    if state['mode']=='live' and state['cursor']!=len(history)-1:state['cursor']=len(history)-1;broadcast()
   if state['mode']=='replay' and state['playing']:
    start=max([i for i,e in enumerate(history) if e.get('kind')=='start'] or [0])
    eligible=[i for i in range(start,len(history)) if (not history[i].get('layer') or history[i].get('layer')==state.get('layer')) and (not state.get('summary') or history[i].get('kind') in ['placement','unresolved'])]
    speed=max(1,len(eligible)/5) if state.get('summary') else settings['speed']
    elapsed=time.time()-previous
    if elapsed>=1/speed:
     remaining=[i for i in eligible if i>state['cursor']];count=max(1,int(elapsed*speed));previous+=count/speed
     if remaining:state['cursor']=remaining[min(count,len(remaining))-1]
     if len(remaining)<=count:state['cursor']=len(history)-1;state['playing']=False
     broadcast()
   else:previous=time.time()
   if time.time()-last>settings['ttlHours']*3600:shutdown();return
threading.Thread(target=watch,daemon=True).start();signal.signal(signal.SIGTERM,lambda *_:threading.Thread(target=shutdown,daemon=True).start());print(json.dumps({'port':server.server_port}),flush=True);server.serve_forever();save();reg.unlink(missing_ok=True)