"""#309 warm-pool native proof, no retries and immutable per-phase receipts.""" import importlib.util import json import os from pathlib import Path import shutil import sys import tempfile OUT=Path(__file__).resolve().parent ROOT=OUT.parents[3] SOURCE=ROOT/'.spt/preserved/310-local-update-docs/windows-validation/run.py' spec=importlib.util.spec_from_file_location('guard310',SOURCE) p=importlib.util.module_from_spec(spec);spec.loader.exec_module(p) g=p.g;g.OUT=OUT;g.TREE=ROOT/'.worktrees/309-bringup-liveness';g.TARGET=g.TREE/'target' p.AUTHORIZED += [ROOT/'.worktrees/hertz-309-er-briefing'] CELL='bound_engine_room_survives_bringup_reconcile_and_receives_native_briefing' SCOPE=['-p','spt','--test','er_briefing_bringup_reconcile_e2e'] SEAMS=['-p','spt','-p','spt-daemon','--lib','--bin','spt','--test','er_briefing_presented_e2e','--test','attach','--test','input_ack_deadlock','--test','redispatch','--test','redispatch_stall','--test','broker'] SEAM_EXPR='(package(spt) & (test(/^rc::tests::/) | binary(er_briefing_presented_e2e))) | (package(spt-daemon) & (test(/^livehost::tests::/) | binary(/^(attach|input_ack_deadlock|redispatch|redispatch_stall)$/) | test(=ack_false_input_suppresses_applied_frame) | test(=exactly_once_preserved_on_no_ack_path) | test(/^msg::tests::input_req_.*ack/) | test(=broker::tests::local_controller_wedged_scopes_to_the_brains_own_conn)))' class Phase(g.Phase): def prepare(self): if os.name!='nt':raise RuntimeError('Windows only') self.receipt['source_before']=self.source('before') self.receipt['free_bytes_start']=shutil.disk_usage(g.TREE).free self.receipt['processes_before']=g.census() g.reparse_guard(g.TARGET) foreign=self.env.get('CARGO_TARGET_DIR') if foreign and (g.TREE/foreign).resolve()!=g.TARGET.resolve():raise RuntimeError('foreign target override') base=Path(os.environ['LOCALAPPDATA'])/'Temp';g.reparse_guard(base) private=Path(tempfile.mkdtemp(prefix='spt-309-'+self.name+'-',dir=base)) home=private/'home';tmp=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['warm_pool_authorization']='SMQWACWD / TT4CRR66; moved released merged 49-267 pool, no cold pool' self.receipt['authorized_concurrent_source_trees']=list(map(str,p.AUTHORIZED)) self.receipt['driver_sha256']=g.digest(Path(__file__)) self.flush() def cell(self,control): selected=self.inventory('inventory',SCOPE,'test(='+CELL+')') if len(selected)!=1:raise RuntimeError('expected exactly one cell') env=dict(self.receipt['environment_allowlist'],SPT_TEST_EVIDENCE_DIR=str(self.dir/'evidence'),SPT_TEST_ER309_KNOB_UNSET='1' if control else '0') g.save(self.dir/'environment.json',env) args=['nextest','run','--locked',*SCOPE,'--retries','0','--test-threads','1','-E','test(='+CELL+')'] g.save(self.dir/'argv.json',args) launcher=ROOT/'.spt/preserved/309/native/launch-v2.ps1' argv=['pwsh','-NoProfile','-File',str(launcher),'-Label','todlando-309-'+self.name,'-Seconds','240','-RecordFile',str(self.dir/'native-record.json'),'-ArgsFile',str(self.dir/'argv.json'),'-Exe',shutil.which('cargo'),'-OutFile',str(self.dir/'nextest.stdout.log'),'-ErrFile',str(self.dir/'nextest.stderr.log'),'-Admission','process-tree','-Scope','step','-EnvironmentFile',str(self.dir/'environment.json')] self.run('native',argv,timeout=300) record=json.loads((self.dir/'native-record.json').read_text()) if record['subject']['native_exit']!=0 or record['coverage']!='COMPLETE' or record['termination']!='CONFIRMED_GONE' or record['job']['active_processes']!=0:raise RuntimeError('native cell exit/containment failed') self.receipt['native_record_sha256']=g.digest(self.dir/'native-record.json') self.receipt['binary_inputs']={name:g.digest(g.TARGET/'debug'/name) for name in ['spt.exe','mock-session.exe','translate_proof_fixture.exe']} def execute(self): if self.name=='pool-claim': self.run('claim',['cargo','run','--locked','-p','xtask','--','pool-claim','--pool','target','--label','309-bringup-liveness']) elif self.name=='pool-claim-direct': self.receipt['prebuilt_xtask_sha256']=g.digest(g.TARGET/'debug/xtask.exe') self.run('claim',[str(g.TARGET/'debug/xtask.exe'),'pool-claim','--pool','target','--label','309-bringup-liveness'],timeout=120) elif self.name.startswith('build'): self.run('build',['cargo','build','--locked','-p','spt','-p','mock-adapter','-p','xtask','--bin','spt','--bin','mock-session','--bin','translate_proof_fixture','--bin','xtask']) elif self.name.startswith('scheduled-'):self.cell(False) elif self.name.startswith('control-'):self.cell(True) elif self.name.startswith('seams'): self.inventory('inventory',SEAMS,SEAM_EXPR) self.run('tests',['cargo','nextest','run','--locked',*SEAMS,'--profile','ci-windows','--test-threads','2','--retries','0','--no-fail-fast','-E',SEAM_EXPR]) elif self.name.startswith('leak-'): expression='test(=livehost::tests::legacy_psyche_sweep_guard_is_id_specific_and_fail_safe)' selected=self.inventory('inventory',SEAMS,expression) if len(selected)!=1:raise RuntimeError('expected exactly one leak cell') inventory=json.loads((self.dir/'inventory.stdout').read_text()) binary=Path(inventory['rust-suites']['spt-daemon']['binary-path']) self.receipt['test_binary']={'path':str(binary),'sha256':g.digest(binary)} self.receipt['test_selector']=expression self.run('tests',['cargo','nextest','run','--locked',*SEAMS,'--profile','ci-windows','--test-threads','2','--retries','0','--no-fail-fast','-E',expression]) elif self.name=='checks': self.run('clippy',['cargo','clippy','--workspace','--locked']) self.run('trace-version',['traceable-reqs','--version'],timeout=30) self.run('trace',['traceable-reqs','check','--json'],timeout=600) elif self.name=='docs':self.run('gen',[str(g.TARGET/'debug/xtask.exe'),'gen'],timeout=300) elif self.name=='docs-check':self.run('check',[str(g.TARGET/'debug/xtask.exe'),'check'],timeout=300) elif self.name=='pool-release':self.run('release',[str(g.TARGET/'debug/xtask.exe'),'pool-release','--pool','target'],timeout=60) 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') if phase.name!='docs': for key in ('head','diff_sha256','source_map_sha256'): if phase.receipt['source_before'][key]!=phase.receipt['source_after'][key]:raise RuntimeError('source changed during phase '+key) phase.receipt['processes_after']=g.census() phase.receipt['free_bytes_end']=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())