"""Isolated #302 remedy proofs; immutable phase receipts and one owned warm pool.""" import importlib.util import json from pathlib import Path import sys sys.dont_write_bytecode = True OUT = Path(__file__).resolve().parent ROOT = OUT.parents[3] GUARD = ROOT / '.spt/preserved/308-registry-process-lock/windows-validation/run.py' spec = importlib.util.spec_from_file_location('guard302step2', GUARD) g = importlib.util.module_from_spec(spec) spec.loader.exec_module(g) g.OUT = OUT g.TREE = ROOT / '.worktrees/302-meet-offload' g.TARGET = g.TREE / 'target' BRANCH = 'fix/302-meet-offload' MEET = 'test(=pairhost::meet_runtime_tests::three_subnet_rotation_keeps_lifecycle_off_broker_workers)' GUARDS = 'package(spt-net) | (package(spt-daemon) & (test(/^nethost::tests::/) | test(/^pairhost::tests::/) | binary(/^(netbroker|netstream|pairjoin)$/)))' SCOPES = { 'meet': (['-p','spt-daemon','--lib'], MEET), 'guards': (['-p','spt-net','-p','spt-daemon','--lib','--test','netbroker','--test','netstream','--test','pairjoin'], GUARDS), 'starve': (['-p','spt-daemon','--test','net_worker_starve'], 'all()'), 'attach': (['-p','spt-daemon','--test','attach','-p','spt','--test','attach_wedge_e2e'], 'all()'), } def quiet_gate(rows): refused = [r for r in rows if r['own_target'] or r['builder']] if refused: raise RuntimeError('quiet proof admission refused; builder/pool user: ' + repr([ {k:r[k] for k in ('pid','birth','name','exe','cwd','cmd','ancestors')} for r in refused])) g.gate = quiet_gate class Phase(g.Phase): def prepare(self): if g.os.name != 'nt': raise RuntimeError('Windows proof driver only') self.receipt['source_before'] = self.source('before') self.receipt['branch'] = self.metadata('branch','--show-current').decode().strip() if self.receipt['branch'] != BRANCH: raise RuntimeError('wrong source branch') self.receipt['driver_sha256'] = g.digest(Path(__file__)) self.receipt['guard_sha256'] = g.digest(GUARD) self.receipt['processes_before'] = g.census() self.receipt['free_bytes_start'] = g.shutil.disk_usage(g.TREE).free g.reparse_guard(g.TARGET) if not g.TARGET.is_dir(): raise RuntimeError('released warm pool must already be relocated; no cold target') foreign = self.env.get('CARGO_TARGET_DIR') if foreign and (g.TREE/foreign).resolve() != g.TARGET.resolve(): raise RuntimeError('foreign inherited target override') if self.name != 'pool-claim': claim = json.loads((OUT/'pool-claim/receipt.json').read_bytes()) if claim['status'] != 'passed' or claim['driver_exit'] != 0: raise RuntimeError('successful remedy pool claim required') if claim['source_before']['head'] != self.receipt['source_before']['head']: raise RuntimeError('source HEAD changed since claim') base = Path(g.os.environ['LOCALAPPDATA'])/'Temp' g.reparse_guard(base) if g.inside(base.resolve(), ROOT): raise RuntimeError('private temporary storage must be outside repository') private = Path(g.tempfile.mkdtemp(prefix='spt-302-remedy-'+self.name+'-',dir=base)) home, tmp = private/'home', private/'tmp' home.mkdir(); tmp.mkdir() overlay = dict(CARGO_TARGET_DIR=str(g.TARGET), CARGO_BUILD_JOBS='2', NEXTEST_PROFILE='ci-windows', SPT_HOME=str(home), TEMP=str(tmp), TMP=str(tmp), SPT_INSTALL_NO_FIREWALL='1', SPT_TEST_EPHEMERAL_ADVISORY_PORTS='1') self.env.update(overlay) self.receipt['environment_allowlist'] = overlay self.receipt['private_root'] = str(private) self.receipt['relocation_receipt_sha256'] = g.digest(OUT/'pool-relocation.json') self.receipt['inherited_toolchain_environment'] = {k:v for k,v in self.env.items() if k.startswith(('CARGO_', 'RUST')) or k in ('HOME','USERPROFILE')} self.flush() def execute(self): if self.name == 'pool-claim': helper = g.TARGET/'debug/xtask.exe' self.receipt['prebuilt_xtask_sha256'] = g.digest(helper) self.run('claim',[str(helper),'pool-claim','--pool','target','--label','todlando-302-step2'],timeout=120) elif self.name in SCOPES: scope, expression = SCOPES[self.name] selected = self.inventory('inventory',['--locked',*scope],expression) if self.name == 'meet' and len(selected) != 1: raise RuntimeError('meet proof must select exactly its named regression') self.run('tests',['cargo','nextest','run','--locked',*scope,'--profile','ci-windows', '--test-threads','2','--retries','0','--no-fail-fast','--success-output','immediate','-E',expression]) elif self.name == 'checks': self.run('clippy',['cargo','clippy','--locked','-p','spt-net','-p','spt-daemon','--lib'],timeout=3600) self.run('trace',['traceable-reqs','check','--json'],timeout=600) self.run('docs-check',[str(g.TARGET/'debug/xtask.exe'),'check'],timeout=600) elif self.name == 'pool-release': self.run('release',[str(g.TARGET/'debug/xtask.exe'),'pool-release','--pool',str(g.TARGET)],timeout=120) else: raise RuntimeError('unknown phase') def main(): phase = Phase(sys.argv[1]) code = 1 try: phase.prepare() phase.execute() phase.receipt['status'] = 'passed' code = 0 except BaseException as error: phase.receipt.update(status='refused-or-failed',error=repr(error)) finally: try: phase.receipt['source_after'] = phase.source('after') before = json.loads(Path(phase.receipt['source_before']['source_map']).read_bytes()) after = json.loads(Path(phase.receipt['source_after']['source_map']).read_bytes()) changed = [k for k in before.keys() | after.keys() if before.get(k) != after.get(k)] phase.receipt['source_changes'] = changed if changed: raise RuntimeError('source changed during frozen native phase: '+repr(changed)) phase.receipt['processes_after'] = g.census() phase.receipt['free_bytes_end'] = g.shutil.disk_usage(g.TREE).free phase.receipt['target_bytes_end'] = g.target_bytes() except BaseException as error: phase.receipt.update(status='refused-or-failed',finalization_error=repr(error)) code = 1 phase.receipt.update(end=g.now(),driver_exit=code) phase.flush() print(phase.dir/'receipt.json') return code if __name__ == '__main__': sys.exit(main())