research_session.py (8064B)
1 """Explicit portable research preparation and Haskell-only guarded execution.""" 2 import hashlib 3 import json 4 import os 5 from pathlib import Path 6 import re 7 import signal 8 import subprocess 9 import time 10 from native_library_migration import digest 11 from process_rss import ChildRSS 12 13 ROOT = Path(__file__).resolve().parent.parent 14 15 16 class ResearchSession: 17 def __init__(self, tricu, directory, prepared, compiler_bundle): 18 self.tricu, self.root = str(Path(tricu).resolve()), Path(directory).resolve() 19 self.root.mkdir(parents=True, exist_ok=True) 20 self.store = self.root / 'store' 21 self.workspace = self.root / 'workspace' 22 self.workspace.mkdir(exist_ok=True) 23 self.events = [] 24 self.deadline = time.monotonic() + 1200 25 directories = [Path(path).resolve() for path in prepared] 26 self.records = {} 27 for directory in directories: 28 for record in json.loads((directory / 'migration.json').read_text()): 29 if not record['complete']: 30 raise ValueError('incomplete research preparation') 31 self.records[(ROOT / record['path']).resolve()] = record 32 for source in (directory / 'application/objects').glob('*/*'): 33 target = self.store / 'objects' / source.parent.name / source.name 34 target.parent.mkdir(parents=True, exist_ok=True) 35 if not target.exists(): 36 os.link(source, target) 37 for source in (directory / 'workspace').glob('*.tri'): 38 (self.workspace / source.name).write_bytes(source.read_bytes()) 39 for path, record in self.records.items(): 40 if hashlib.sha256(path.read_bytes()).hexdigest() != record['after_sha256']: 41 raise ValueError('stale research preparation: ' + str(path)) 42 for record in self.records.values(): 43 payload = self.object('arboricx.module-manifest.v1', record['manifest']) 44 receipt = self.object('arboricx.build-receipt.v1', record['receipt']).decode().splitlines() 45 if ('source\t' + digest('arboricx.source.v1', (ROOT / record['path']).read_bytes()) not in receipt or 46 'output\t' + record['manifest'] not in receipt): 47 raise ValueError('research receipt mismatch') 48 (self.workspace / (record['alias'] + '.tri')).write_text('!export * from ' + record['manifest'] + '\n') 49 output, _ = self.run([self.tricu, 'arboricx', 'import', '--file', str(Path(compiler_bundle).resolve()), 50 '--store', str(self.store), '--indexed']) 51 self.compiler = re.search(rb'Created module alias .* -> ([a-f0-9]{64})', output)[1].decode() 52 self.names = {row.split('\t')[2]: row.split('\t')[1] for row in 53 (ROOT / 'bootstrap/compiler-source-map.v1').read_text().splitlines()[1:]} 54 55 def object(self, domain, pin): 56 payload = (self.store / 'objects' / pin[:3] / pin).read_bytes() 57 if digest(domain, payload) != pin: 58 raise ValueError('corrupt research object') 59 return payload 60 61 def run(self, command, seconds=1200): 62 if not 0 < seconds <= 1200: 63 raise ValueError('job must be bounded to 1200 seconds') 64 memory_mib = getattr(self, 'memory_mib', 49152) 65 if not 0 < memory_mib <= 65536: 66 raise ValueError('RSS/heap budget must be within 64 GiB') 67 command = [*map(str, command), '+RTS', '-N1', f'-M{memory_mib}m', '-RTS'] 68 seconds = min(seconds, self.deadline - time.monotonic()) 69 if seconds <= 0: 70 raise subprocess.TimeoutExpired(command, 1200) 71 start, peak, reason = time.monotonic(), 0, 'completed' 72 number = len(self.events) 73 with (self.root / f'job-{number}.stdout').open('wb') as out, (self.root / f'job-{number}.stderr').open('wb') as err: 74 proc = subprocess.Popen(command, stdout=out, stderr=err, start_new_session=True) 75 try: 76 monitor = ChildRSS(proc) 77 while proc.poll() is None: 78 rss = monitor.sample() 79 if rss is not None: 80 peak = max(peak, rss) 81 if peak > memory_mib * 1024: 82 reason = 'RSS guard'; raise RuntimeError(reason) 83 if time.monotonic() - start > seconds: 84 reason = 'time guard'; raise subprocess.TimeoutExpired(command, seconds) 85 time.sleep(.05) 86 except (OSError, RuntimeError) as error: 87 if reason == 'completed': 88 reason = 'monitoring failure: ' + str(error) 89 raise 90 except BaseException: 91 if reason == 'completed': 92 reason = 'interrupted' 93 raise 94 finally: 95 if proc.poll() is None: 96 os.killpg(proc.pid, signal.SIGTERM) 97 try: proc.wait(timeout=5) 98 except subprocess.TimeoutExpired: 99 os.killpg(proc.pid, signal.SIGKILL); proc.wait() 100 self.events.append(dict(command=command, seconds=time.monotonic()-start, 101 peak_kib=peak, reason=reason, code=proc.returncode)) 102 (self.root / 'events.json').write_text(json.dumps(self.events, indent=2)+'\n') 103 stdout = (self.root / f'job-{number}.stdout').read_bytes().strip() 104 stderr = (self.root / f'job-{number}.stderr').read_bytes() 105 if proc.returncode: 106 raise AssertionError((command, stdout, stderr)) 107 return stdout, time.monotonic()-start 108 109 def compile(self, source, names, output, seconds=1200): 110 source = Path(source).resolve() 111 if source in self.records: 112 record = self.records[source] 113 if hashlib.sha256(source.read_bytes()).hexdigest() != record['after_sha256']: 114 raise ValueError('prepared research source changed during session') 115 rows = self.object('arboricx.module-manifest.v1', record['manifest']).decode().splitlines()[1:] 116 exports = {row.split('\t')[1]: row.split('\t')[3] for row in rows} 117 command = [self.tricu, 'arboricx', 'export', '--store', str(self.store)] 118 for name in names: 119 command += ['--name', name, '--target', exports[name]] 120 _, elapsed = self.run([*command, '--output', str(output)], seconds) 121 return elapsed # packing prepared trees, not fresh source compilation 122 raw = source.read_text() 123 body, imports = [], [] 124 for line in raw.splitlines(keepends=True): 125 if line.startswith('!import "'): 126 match = re.fullmatch(r'!import "([^"\n]+)"\s+(\S+)\n', line) 127 if not match: raise ValueError('unsupported research fixture header') 128 name, namespace = match.groups() 129 reference = name if re.fullmatch('[a-f0-9]{64}', name) else self.names[name] 130 imports.append((reference, namespace)) 131 else: body.append(line) 132 if imports: 133 header = ''.join('!import '+ref+(' *' if ns == '!Local' else ' as '+ns)+'\n' for ref, ns in imports) 134 raw = header + '!export *\n' + ''.join(body) 135 if '!export ' not in raw: 136 raw = '!export *\n' + raw 137 # Fixture-only import facades expose exactly the requested inherited names. 138 if not ''.join(body).strip() and imports: 139 raw += ''.join('!export * from '+ref+'\n' for ref, _ in imports) 140 staged = self.workspace / (source.stem.replace('-', '_') + '.tri') 141 staged.write_text(raw) 142 _, elapsed = self.run([self.tricu, 'arboricx', 'compile', '--file', str(staged), 143 '--toolchain', self.compiler, '--tool-store', str(self.store), '--store', str(self.store), 144 '--compile-timeout', str(seconds), '--output', str(output), 145 *[arg for name in names for arg in ['--name', name]]], seconds) 146 return elapsed