"""Diagnostic assembly only; immutable evidence, fixed resource gates, no resident access.""" 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] SOURCE = ROOT / '.spt/preserved/302/todlando-release/run.py' spec = importlib.util.spec_from_file_location('release302fieldbase', SOURCE) r = importlib.util.module_from_spec(spec) spec.loader.exec_module(r) g = r.g g.OUT = OUT g.TREE = ROOT / '.worktrees/302-meet-offload-field' g.TARGET = g.TREE / 'target' r.OUT = OUT r.HEAD = '0f80fef8f55c5dff5c34968a99850965b87ddd28' r.BRANCH = 'diag/302-meet-offload-field' SCOPES = { 'meet-diag': (['-p','spt-daemon','--lib'], 'test(=pairhost::meet_runtime_tests::three_subnet_rotation_keeps_lifecycle_off_broker_workers)'), 'guards': (['-p','spt-net','-p','spt-daemon','--lib','--test','netbroker','--test','netstream','--test','pairjoin'], 'package(spt-net) | (package(spt-daemon) & (test(/^nethost::tests::/) | test(/^pairhost::tests::/) | binary(/^(netbroker|netstream|pairjoin)$/)))'), '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 = [x for x in rows if x['builder'] or x['own_target']] if refused: raise RuntimeError('builder/pool user refused: '+repr([(x['pid'],x['name']) for x in refused])) g.gate = quiet_gate class FieldPhase(r.Phase): def prepare(self): if g.os.name != 'nt': raise RuntimeError('Windows only') self.receipt['source_before'] = self.source('before') source = self.receipt['source_before'] self.receipt['branch'] = self.metadata('branch','--show-current').decode().strip() if source['head'] != r.HEAD or source['diff_sha256'] != r.EMPTY_SHA256 or self.receipt['branch'] != r.BRANCH: raise RuntimeError('exact clean assembly commit required') if self.metadata('ls-files','--others','--exclude-standard').strip(): raise RuntimeError('untracked assembly source refused') g.reparse_guard(g.TARGET) if not g.TARGET.is_dir(): raise RuntimeError('relocated released warm pool required') 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 or claim['source_before']['head'] != r.HEAD or claim['source_before']['source_map_sha256'] != source['source_map_sha256']: raise RuntimeError('successful unchanged assembly pool claim required') base = Path(g.os.environ['LOCALAPPDATA'])/'Temp' g.reparse_guard(base) if g.inside(base.resolve(),ROOT): raise RuntimeError('private temp must be outside repository') private = Path(g.tempfile.mkdtemp(prefix='spt-302-field-'+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',**r.PROFILE_ENV) if self.name == 'meet-diag': overlay['SPT_RC_HITCH_DIAG'] = '1' self.env.update(overlay) self.receipt.update(environment_allowlist=overlay,private_root=str(private),profile=r.PROFILE,driver_sha256=g.digest(Path(__file__)),inherited_driver_sha256=g.digest(SOURCE),relocation_receipt_sha256=g.digest(OUT/'pool-relocation.json'),processes_before=g.census(),free_bytes_start=g.shutil.disk_usage(g.TREE).free) 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-field'],timeout=120) elif self.name in SCOPES: scope,expression = SCOPES[self.name] selected = self.inventory('inventory',scope,expression) if self.name == 'meet-diag' and len(selected) != 1: raise RuntimeError('exactly one real lifecycle regression required') self.run('tests',['cargo','nextest','run','--release','--locked',*scope,'--profile','ci-windows','--test-threads','2','--retries','0','--no-fail-fast','--success-output','immediate','-E',expression]) if self.name == 'meet-diag': self.verify_meet_stamps() elif self.name == 'checks': self.run('clippy',['cargo','clippy','--release','--locked','-p','spt-net','-p','spt-daemon','-p','spt-net-diag','--lib'],timeout=3600) self.run('trace',['traceable-reqs','check','--json'],timeout=600) elif self.name == 'version': self.run('version',[str(g.TARGET/'release/spt.exe'),'--version'],timeout=30) else: super().execute() def verify_meet_stamps(self): rows = [] for line in (self.dir/'tests.stderr').read_text(encoding='utf-8').splitlines(): if 'NET_DIAG_V1:' in line: rows.append(json.loads(line.split('NET_DIAG_V1:',1)[1])) starts = [x for x in rows if x['event']=='NET_DIAG_START'] ends = [x for x in rows if x['event']=='NET_DIAG_END'] if len(starts)!=1 or len(ends)!=1 or ends[0]['dropped_records']!=0: raise RuntimeError('complete lossless local epoch required') workers = set(starts[0]['worker_tids']) meet = [x for x in rows if x['event'].startswith(('NET_MEET_BIND_','NET_MEET_RETIRE_'))] if not meet or len(workers)!=2 or any(x['runtime']!='other' or x['tid'] in workers for x in meet): raise RuntimeError('meet boundary ran on broker worker or violated frozen runtime enum') binds = [x for x in meet if x['event']=='NET_MEET_BIND_END' and x.get('outcome')=='ok'] retired = [x for x in meet if x['event']=='NET_MEET_RETIRE_END' and x.get('outcome')=='ok'] stages = {x.get('stage') for x in retired} if len({x['subnet'] for x in binds})!=3 or not {'abort-request','task-drop','endpoint-drop'} <= stages: raise RuntimeError('three-subnet bind and complete retirement boundaries required') for stage in ('task-drop','endpoint-drop'): retired_keys = {(x['subnet'],x['step']) for x in retired if x.get('stage')==stage} if not {(x['subnet'],x['step']) for x in binds} <= retired_keys: raise RuntimeError('bound listener missing actual destruction: '+stage) self.receipt['local_boundary_proof'] = dict(interpretation='Hermetic real NetHost lifecycle; not ENLYZEAM field acceptance',worker_tids=sorted(workers),meet_tids=sorted({x['tid'] for x in meet}),binds=len(binds),retirement_stages=sorted(stages),runtime='other',run_id=starts[0]['run_id']) self.flush() r.Phase = FieldPhase if __name__ == '__main__': sys.exit(r.main())