AI Flow

"""Exercise early component inventory, optional marking and review-note preservation."""
import os
os.environ["AIFLOW_DASHBOARD"] = "off"  # Tests must never open the user workspace.
import pathlib,subprocess,tempfile,sys,json
binary,board,spec=map(lambda p:str(pathlib.Path(p).resolve()),sys.argv[1:])
with tempfile.TemporaryDirectory(prefix='aiflow-components-') as d:
 base=[binary,'--ai-thread','ESC AI Flow Astra','--run',d]
 subprocess.run(base+['start','--board',board,'--spec',spec,'--engine','component-workflow-test'],check=True,capture_output=True,text=True)
 text=subprocess.run(base+['plan'],check=True,capture_output=True,text=True).stdout
 assert text.index('step components') < text.index('step libraries') < text.index('step models') < text.index('step library-tour') < text.index('step placement')
 assert text.index('Search the wiki FIRST')<text.index('use adom-parts-search next')
 for phrase in ['ALL ADOM USERS','private reference-only','MPN marking is ON by default','Cache keys','per-file provenance mandatory','datasheet page/figure/table','Unknown provenance stays unknown']:
  assert phrase in text,phrase
 result=subprocess.run(base+['components'],check=True,capture_output=True,text=True)
 p=pathlib.Path(d)/'components.json';v=json.load(open(p));assert len(v['components'])==149;assert v['mpnMarking']=='on'
 assert all('assetProvenance' in r['review'] for r in v['components'])
 refs=[r['reference'] for r in v['components']];assert len(set(refs))==149
 v['components'][0]['review']['notes']='Preserve this independent review';p.write_text(json.dumps(v))
 subprocess.run(base+['components','--etch','on'],check=True,capture_output=True,text=True)
 v=json.load(open(p));assert v['mpnMarking']=='on';assert v['components'][0]['review']['notes']=='Preserve this independent review'
 subprocess.run(base+['components'],check=True,capture_output=True,text=True);assert json.load(open(p))['mpnMarking']=='on'
 subprocess.run(base+['components','--etch','off'],check=True,capture_output=True,text=True);assert json.load(open(p))['mpnMarking']=='off'
 v=json.load(open(p));v['components'][0]['properties']['Value']='deliberately stale source value';v['components'][0]['review']['status']='pass';p.write_text(json.dumps(v))
 subprocess.run(base+['components'],check=True,capture_output=True,text=True)
 v=json.load(open(p));assert v['components'][0]['review']['status']=='stale';assert v['components'][0]['review']['notes']=='Preserve this independent review'
 p.write_text('{invalid');r=subprocess.run(base+['components'],capture_output=True,text=True);assert r.returncode!=0;assert p.read_text()=='{invalid'
 print(result.stdout)
 print('PASS: early plan order, 149 references, explicit marking preference, preserved review notes, stale source invalidation, malformed register untouched.')