tricu

compiler_candidate.py (4725B)

---

 """Shared opt-in candidate workspace, with explicitly authenticated checkpoint reuse.
 
 No recovery compilation, generation promotion, or writes through source hardlinks.
 Only immutable CAS objects are linked; mutable checkpoint files are copied.
 """
 import hashlib
 import json
 import os
 from pathlib import Path
 import shutil
 import time
 from bootstrap_generations import source_map
 from collections_batch import ProbeRunner, audit_workspace
 from native_library_migration import digest
 ROOT=Path(__file__).resolve().parent.parent
 
 def sha(raw):return hashlib.sha256(raw).hexdigest()
 
 class Candidate:
     def __init__(self,args,closure):
         if not 0 < args.memory_mib <= 65536: raise ValueError('memory budget must be within 64 GiB')
         self.root=args.directory.resolve();self.root.mkdir(parents=True,exist_ok=args.resume)
         self.workspace=self.root/'workspace';self.workspace.mkdir(exist_ok=args.resume)
         self.runner=ProbeRunner(args.tricu,self.root,1500,memory_mib=args.memory_mib)
         self.report=dict(complete=False,diagnostic_reuse=args.resume,compiler_sha256=sha(args.compiler.read_bytes()),runtime_sha256=sha(args.tricu.read_bytes()),limits=dict(memory_mib=args.memory_mib,wall_seconds=1500,compile_seconds=900))
         self.tool=self.runner.imported(args.compiler,'tool');raw=(self.runner.store/'objects'/self.tool[:3]/self.tool).read_bytes()
         assert digest('arboricx.module-manifest.v1',raw)==self.tool
         self.producer=next(row.split('\t')[3] for row in raw.decode().splitlines()[1:] if row.split('\t')[1]=='sourceBuildGraph')
         if args.seed and not args.resume:
             seed=args.seed.resolve();previous=json.loads((seed/'report.json').read_text())
             assert previous['complete'] and previous['compiler_sha256']==self.report['compiler_sha256']
             assert audit_workspace(seed/'workspace',seed/'store',self.producer)==previous['modules']
             for source in (seed/'store/objects').glob('*/*'):
                 target=self.runner.store/'objects'/source.parent.name/source.name
                 target.parent.mkdir(parents=True,exist_ok=True)
                 if not target.exists():os.link(source,target)
             shutil.copytree(seed/'store/cache',self.runner.store/'cache',dirs_exist_ok=True)
             self.report['seed_report_sha256']=sha((seed/'report.json').read_bytes())
             (self.root/'seed.json').write_text(json.dumps(dict(report_sha256=self.report['seed_report_sha256'],compiler_sha256=previous['compiler_sha256']))+'\n')
         elif (self.root/'seed.json').exists():self.report['seed']=json.loads((self.root/'seed.json').read_text())
         _,self.entries,self.needed=source_map(ROOT/'bootstrap/compiler-source-map.v1',ROOT,closure)
         for alias in self.needed:(self.workspace/(alias+'.tri')).write_bytes((ROOT/self.entries[alias][1]).read_bytes())
     def build(self,alias,names=()):
         output=self.root/(alias+'.arboricx');argv=['compile','--file',self.workspace/(alias+'.tri'),'--toolchain',self.tool,'--tool-store',self.runner.store,'--store',self.runner.store,'--output',output,'--compile-timeout','900']
         for name in names:argv+=['--name',name]
         self.runner.run('compile-'+alias,argv,930);return output
     def audit(self):
         self.report['modules']=audit_workspace(self.workspace,self.runner.store,self.producer)
         for alias in self.needed:assert (self.workspace/(alias+'.tri')).read_bytes()==(ROOT/self.entries[alias][1]).read_bytes()
         return self.report['modules']
     def export(self,alias,names):
         output=self.root/(alias+'-selected.arboricx');argv=['export','--store',self.runner.store,'--output',output]
         for name in names:argv+=['--target',self.report['modules'][alias]['export_hashes'][name],'--name',name]
         self.runner.run('export-'+alias,argv,120);return output
     def invoke(self,pin,name,arguments,label=None,prefix=()):
         argv=list(prefix)
         for i,argument in enumerate(arguments):
             path=self.root/f'argument-{i}.tree';path.write_bytes(argument);argv+=['--arg-tree-file',path]
         return self.runner.invoke(label or name,pin,name,argv)
     def save(self):
         self.report['seconds']=time.monotonic()-self.runner.started
         (self.root/'report.json').write_text(json.dumps(self.report,indent=2)+'\n')
 
 def options(description):
     import argparse
     p=argparse.ArgumentParser(description=description);p.add_argument('tricu',type=Path)
     p.add_argument('--compiler',type=Path,required=True);p.add_argument('--directory',type=Path,required=True)
     p.add_argument('--memory-mib',type=int,default=8192)
     p.add_argument('--seed',type=Path);p.add_argument('--resume',action='store_true');return p