AI Flow

"""Real CLI/server lifecycle with a fake Hydrogen API (a test-owned adom-cli on PATH); no CAD or real tabs touched.

Covers: default intake, reuse, closed-tab recovery, per-run server isolation, resume, stopped-server
recovery, SSE live channel and reconnect replay, run labels and tab names, the old hex-named tab
renamed in the same pane, the tab registry, the redundant-tab challenge (same thread, other thread),
finished-tab reuse, --new-tab --why, timeline list, timeline close, headless override, visible UI failure.
"""
import json,os,pathlib,subprocess,sys,tempfile,time,urllib.request,urllib.error,signal
binary=str(pathlib.Path(sys.argv[1]).resolve())
DOT='AI Flow · '
with tempfile.TemporaryDirectory(prefix='aiflow-dashboard-test-') as root:
 p=pathlib.Path(root);bindir=p/'bin';bindir.mkdir();state=p/'hydrogen.json';calls=p/'calls.jsonl';home=p/'home'
 # two panes; the human's own tab must never be touched
 state.write_text(json.dumps({'panes':{'left':[{'name':'Visual Studio Code','url':None}],'right':[{'name':'Human notes','url':'http://human/'}]}}))
 mock=bindir/'adom-cli';mock.write_text('''#!/usr/bin/env python3
import sys,json,os,pathlib
args=sys.argv[1:];p=pathlib.Path(os.environ['TEST_HYDROGEN']);d=json.loads(p.read_text())
with open(os.environ['TEST_CALLS'],'a')as f:f.write(json.dumps(args)+'\\n')
def opt(k):return args[args.index(k)+1] if k in args else None
def find(name):
 for pane,tabs in d['panes'].items():
  for t in tabs:
   if t['name']==name:return pane,t
 return None,None
def save():p.write_text(json.dumps(d))
if args[:3]==['hydrogen','workspace','tabs']:print(json.dumps({'tabs':[{'name':t['name'],'panelId':pane,'url':t['url']} for pane,tabs in d['panes'].items() for t in tabs]}))
elif args[:3]==['hydrogen','workspace','get']:print(json.dumps({'root':{'type':'split','children':[{'type':'leaf','id':pane,'tabs':[{'panelState':{'displayName':t['name'],'url':t['url']}} for t in tabs]} for pane,tabs in d['panes'].items()]}}))
elif args[:3]==['hydrogen','webview','open-or-refresh']:
 name=opt('--name');url=opt('--url');pane,t=find(name)
 if t:t['url']=url;save();print(json.dumps({'ok':True,'action':'refreshed'}))
 else:d['panes'].setdefault(opt('--panel-id') or 'right',[]).append({'name':name,'url':url});save();print(json.dumps({'ok':True,'action':'created'}))
elif args[:3]==['hydrogen','webview','navigate']:
 pane,t=find(opt('--name'))
 if t:t['url']=args[3]
 save();print(json.dumps({'ok':True}))
elif args[:3]==['hydrogen','workspace','remove-tab']:
 pane,t=find(opt('--name'))
 if not t:print(json.dumps({'ok':False}));sys.exit(1)
 d['panes'][pane].remove(t);save();print(json.dumps({'ok':True}))
else:print('{}')
''');mock.chmod(0o755)
 env=dict(os.environ,PATH=str(bindir)+os.pathsep+os.environ['PATH'],TEST_HYDROGEN=str(state),TEST_CALLS=str(calls),AIFLOW_HOME=str(home));env.pop('AIFLOW_DASHBOARD',None);env.pop('VSCODE_PROXY_URI',None)
 runs={n:p/n for n in ['one','two','three','four']}
 def run(r,*args,thread='astra',extra=None,code=0):
  res=subprocess.run([binary,'--run',str(r),'--ai-thread',thread,*args],env=extra or env,capture_output=True,text=True,timeout=35)
  assert code is None or res.returncode==code,(args,res.returncode,res.stdout,res.stderr);return res
 def bare(*args,code=0):
  res=subprocess.run([binary,'--ai-thread','astra',*args],env=env,capture_output=True,text=True,timeout=35,cwd=root)
  assert code is None or res.returncode==code,(args,res.stdout,res.stderr);return res
 def log():return [json.loads(l) for l in calls.read_text().splitlines()]
 def opens():return sum(c[:3]==['hydrogen','webview','open-or-refresh'] for c in log())
 def tabs():return {t['name']:(pane,t['url']) for pane,ts in json.loads(state.read_text())['panes'].items() for t in ts}
 def flow_tabs():return {n:v for n,v in tabs().items() if n.startswith(DOT)}
 def cfg(r):return json.loads((r/'dashboard.json').read_text())
 def reg():return json.loads((home/'dashboards.json').read_text())['dashboards']
 def entry(r):return next(e for e in reg() if e['runDir']==str(r.resolve()))
 def edit(r,f):
  d=json.loads((r/'run.json').read_text());f(d);(r/'run.json').write_text(json.dumps(d))
 def alive(url):
  try:return json.load(urllib.request.urlopen(url+'dashboard.json',timeout=2))['runId']
  except Exception:return None
 try:
  # --- default intake: one tab named for the run, never a hash or 'adom-aiflow'
  run(runs['one'],'intake','--prompt-time','2026-10-09T10:00:00Z','--engine','codex');assert opens()==1
  a=cfg(runs['one']);assert a['name']==DOT+'one',a;assert alive(a['url'])==a['runId']
  tl=json.loads((runs['one']/'timeline'/'timeline.json').read_text())
  assert tl['title']=='one' and tl['aiThread']=='astra' and tl['runFolder']=='one' and tl['engine']=='KiCad',tl
  assert 'adom-aiflow' not in json.dumps({k:tl[k] for k in ['title','label']})
  e=entry(runs['one']);assert e['aiThread']=='astra' and e['label']=='one' and e['finished'] is False and e['port']==a['port'],e
  run(runs['one'],'status');assert opens()==1,'repeated command must reuse tab'
  # --- proxy change navigates the named tab
  s=json.loads(state.read_text());s['panes']['right'][-1]['url']='http://old-proxy/';state.write_text(json.dumps(s))
  run(runs['one'],'status');assert opens()==1 and tabs()[DOT+'one'][1]==a['url'],'proxy change must navigate the existing named tab'
  # --- closed tab reopens
  s=json.loads(state.read_text());s['panes']['right']=[t for t in s['panes']['right'] if not t['name'].startswith(DOT)];state.write_text(json.dumps(s))
  run(runs['one'],'status');assert opens()==2,'closed tab must reopen'
  # --- the label follows the ORIGINAL board's stem; the tab is renamed in its pane
  edit(runs['one'],lambda d:(d.__setitem__('sourceBoard','/boards/buck-12v5v.kicad_pcb'),d.__setitem__('boards',[str(runs['one']/'board-3.kicad_pcb')])))
  run(runs['one'],'status');ft=flow_tabs();assert list(ft)==[DOT+'buck-12v5v'] and ft[DOT+'buck-12v5v'][0]=='right',ft
  tl=json.loads((runs['one']/'timeline'/'timeline.json').read_text());assert tl['title']=='buck-12v5v' and tl['board']=='buck-12v5v.kicad_pcb',tl
  # --- migration: a 0.2.19 tab "AI Flow · <hex>" in the left pane is replaced in the same pane
  hexname=DOT+a['runId'][:8]
  s=json.loads(state.read_text());s['panes']['right']=[t for t in s['panes']['right'] if not t['name'].startswith(DOT)];s['panes']['left'].append({'name':hexname,'url':a['url']});state.write_text(json.dumps(s))
  c=cfg(runs['one']);c['name']=hexname;(runs['one']/'dashboard.json').write_text(json.dumps(c))
  r=json.loads((home/'dashboards.json').read_text());r['dashboards']=[];(home/'dashboards.json').write_text(json.dumps(r))
  n0=len(log());run(runs['one'],'status');new=log()[n0:]
  o=[c for c in new if c[:3]==['hydrogen','webview','open-or-refresh']];rm=[c for c in new if c[:3]==['hydrogen','workspace','remove-tab']]
  assert len(o)==1 and o[0][o[0].index('--panel-id')+1]=='left' and o[0][o[0].index('--name')+1]==DOT+'buck-12v5v',new
  assert rm==[['hydrogen','workspace','remove-tab','--name',hexname]],rm
  assert flow_tabs()=={DOT+'buck-12v5v':('left',a['url'])},flow_tabs()
  assert cfg(runs['one'])['name']==DOT+'buck-12v5v' and entry(runs['one'])['name']==DOT+'buck-12v5v'
  a=cfg(runs['one'])
  # --- resume activates the existing tab without reopening
  count=opens();run(runs['one'],'prompt','--text','resume');assert opens()==count
  # --- stopped server is restarted behind the same tab
  for line in subprocess.check_output(['ps','-eo','pid,args'],text=True).splitlines():
   if binary in line and str(runs['one']) in line and 'timeline serve' in line:os.kill(int(line.split()[0]),signal.SIGTERM)
  time.sleep(.2);run(runs['one'],'status');assert opens()==count,'server restart should reuse the tab'
  assert alive(a['url'])==a['runId']

  # --- CLI -> HTTP control -> SSE updates an existing view, and reconnect replays it
  stream=urllib.request.urlopen(a['url']+'events',timeout=5)
  assert stream.headers['X-Accel-Buffering']=='no'
  def event(connection):
   while True:
    line=connection.readline().decode()
    if line.startswith('data: '):return json.loads(line[6:])
  initial=event(stream);assert initial['timeline']['steps'] and initial['timeline']['aiThread']=='astra'
  count=opens();run(runs['one'],'timeline','present','--json',json.dumps({'caption':'Live test update','title':'No extra tab'}));assert opens()==count
  deadline=time.time()+5;received=event(stream)
  while not (received.get('presentation') or {}).get('caption') and time.time()<deadline:received=event(stream)
  assert received['presentation']['caption']=='Live test update'
  stream.close()
  stream=urllib.request.urlopen(a['url']+'events',timeout=5);assert event(stream)['presentation']['caption']=='Live test update';stream.close()
  req=urllib.request.Request(a['url']+'state',data=json.dumps({'presentation':{'caption':'anonymous'}}).encode(),headers={'Content-Type':'application/json'})
  try:urllib.request.urlopen(req);raise AssertionError('anonymous mutation accepted')
  except urllib.error.HTTPError as e:assert e.code==400 and 'caller_identity_required' in e.read().decode()
  assert json.load(urllib.request.urlopen(a['url']+'state'))['presentation']['caption']=='Live test update'
  assert 'timeline serve'not in (runs['one']/'run.jsonl').read_text()

  # --- same thread, second unfinished run: challenged, no new tab
  count=opens();res=run(runs['two'],'intake','--prompt-time','2026-10-09T10:00:00Z','--engine','claude')
  assert opens()==count and 'name' not in cfg(runs['two']),res.stderr
  assert 'WARNING: no new AI Flow tab for this run' in res.stderr and DOT+'buck-12v5v' in res.stderr and '--reuse-tab' in res.stderr and '--new-tab --why' in res.stderr and 'keeps the human' in res.stderr,res.stderr
  res=run(runs['two'],'status');assert opens()==count and 'WARNING: this run has no dashboard tab' in res.stderr,res.stderr
  res=run(runs['two'],'timeline','open',code=2);assert 'ERROR' in res.stdout+res.stderr and '--reuse-tab' in res.stdout+res.stderr and opens()==count
  res=run(runs['two'],'timeline','open','--new-tab',code=2);assert '--why' in res.stdout+res.stderr and opens()==count
  # --- --new-tab --why opens and records the reason
  res=run(runs['two'],'timeline','open','--new-tab','--why','the human compares both layouts side by side');assert opens()==count+1,res
  b=cfg(runs['two']);assert b['name']==DOT+'two' and b['port']!=a['port'];assert entry(runs['two'])['why']=='the human compares both layouts side by side'
  assert alive(a['url'])==a['runId'] and alive(b['url'])==b['runId'],'second run stole first server'
  # --- close --run stops only that run's verified server and tab
  res=run(runs['two'],'timeline','close');assert DOT+'two' not in tabs() and 'stopped timeline server' in res.stdout,res.stdout
  time.sleep(.3);assert alive(b['url']) is None and alive(a['url'])==a['runId']
  assert all(e['runDir']!=str(runs['two'].resolve()) for e in reg())

  # --- a finished run's tab is reused (renamed in place) by the same thread's next run
  edit(runs['one'],lambda d:d['clock'].__setitem__('finishTime','2026-10-09T12:00:00Z'))
  n0=len(log());res=run(runs['three'],'intake','--prompt-time','2026-10-09T13:00:00Z','--engine','codex');new=log()[n0:]
  assert 'reused tab' in res.stderr and 'finished' in res.stderr,res.stderr
  assert flow_tabs()=={DOT+'three':('left',cfg(runs['three'])['url'])},flow_tabs()
  o=[c for c in new if c[:3]==['hydrogen','webview','open-or-refresh']];assert o and o[0][o[0].index('--panel-id')+1]=='left',new
  assert ['hydrogen','workspace','remove-tab','--name',DOT+'buck-12v5v'] in new
  time.sleep(.3);assert alive(a['url']) is None,'the finished run that gave its tab up keeps no server'
  assert all(e['runDir']!=str(runs['one'].resolve()) for e in reg())

  # --- another thread on the same board: warning, no tab; --new-tab --why adds the thread to the name
  reqs=p/'req.json';reqs.write_text(json.dumps({'title':'three'}))
  count=opens();res=run(runs['four'],'intake','--prompt-time','2026-10-09T14:00:00Z','--engine','claude','--requirements',str(reqs),thread='nova')
  assert opens()==count and 'is already being driven by thread "astra"' in res.stderr and DOT+'three' in res.stderr and '--new-tab --why' in res.stderr,res.stderr
  run(runs['four'],'timeline','open','--new-tab','--why','second opinion on the same board',thread='nova');assert opens()==count+1
  assert cfg(runs['four'])['name']==DOT+'three · nova',cfg(runs['four'])
  assert json.loads((runs['four']/'timeline'/'timeline.json').read_text())['aiThread']=='nova'

  # --- timeline list: registry reconciled with Hydrogen; stray AI Flow tabs are unknown; human tabs never listed
  s=json.loads(state.read_text());s['panes']['right'].append({'name':DOT+'stray','url':'http://stray/'});state.write_text(json.dumps(s))
  res=bare('timeline','list');out=res.stdout
  assert DOT+'three\n' in out and DOT+'three · nova' in out and 'AI thread nova' in out and 'AI thread astra' in out and 'not finished' in out and 'server alive' in out,out
  assert DOT+'stray' in out and 'unknown' in out and 'Human notes' not in out and 'second opinion on the same board' in out,out

  # --- timeline close: refuses non-AI-Flow names; --finished closes only finished AI Flow tabs
  res=bare('timeline','close','--name','Human notes',code=2);assert 'Human notes' in tabs()
  edit(runs['three'],lambda d:d['clock'].__setitem__('deliveredTime','2026-10-09T15:00:00Z'))
  before=set(tabs());res=bare('timeline','close','--finished');after=set(tabs())
  assert before-after=={DOT+'three'},(before-after,res.stdout)
  assert {'Human notes','Visual Studio Code',DOT+'stray',DOT+'three · nova'}<=after
  removed=[c[4] for c in log() if c[:3]==['hydrogen','workspace','remove-tab']];assert 'Human notes' not in removed and 'Visual Studio Code' not in removed
  bare('timeline','close','--name',DOT+'stray');assert DOT+'stray' not in tabs() and 'Human notes' in tabs()

  # --- headless override and visible UI failure
  off=dict(env,AIFLOW_DASHBOARD='off');run(p/'headless','intake','--prompt-time','2026-10-09T10:00:00Z',extra=off);assert not (p/'headless'/'dashboard.json').exists()
  mock.write_text('#!/bin/sh\nexit 1\n');run(runs['four'],'status',thread='nova')
  res=run(runs['four'],'status',thread='nova');assert 'Dashboard unavailable'in res.stderr
  print('PASS: intake tab named for the run, reuse, navigate, closed-tab recovery, label from the original board with rename in place, hex tab migrated in its pane, resume, server restart, SSE + replay, same-thread challenge, --new-tab --why, close --run with verified server stop, finished-tab reuse, other-thread warning, name collision adds the thread, list, close --finished/--name guard, headless override, visible UI failure')
 finally:
  # Only terminate test-owned timeline processes identified by exact temporary run path.
  for line in subprocess.check_output(['ps','-eo','pid,args'],text=True).splitlines():
   if binary in line and root in line and 'timeline serve'in line:
    os.kill(int(line.split()[0]),signal.SIGTERM)