"""Unchanged #302 source, optimized release proofs; never overwrite phase evidence.""" 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] DEBUG_OUT = ROOT / '.spt/preserved/302/todlando-windows' SOURCE = DEBUG_OUT / 'run.py' spec = importlib.util.spec_from_file_location('validated302', SOURCE) p = importlib.util.module_from_spec(spec) spec.loader.exec_module(p) g = p.g g.OUT = OUT HEAD = '3142753a8e4eedf1404f75a82eea88bcb3331a32' BRANCH = 'diag/302-net-worker-stamps' EMPTY_SHA256 = 'e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855' PROFILE = {'name': 'release', 'opt_level': '3', 'debug': 'line-tables-only', 'strip': 'none'} PROFILE_ENV = {'CARGO_PROFILE_RELEASE_OPT_LEVEL': '3', 'CARGO_PROFILE_RELEASE_DEBUG': 'line-tables-only', 'CARGO_PROFILE_RELEASE_STRIP': 'none'} SCOPE = p.SCOPE EXPR = p.EXPR class Phase(g.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() baseline = json.loads((DEBUG_OUT / 'build-3/receipt.json').read_text()) if (source['head'] != HEAD or source['diff_sha256'] != EMPTY_SHA256 or source['source_map_sha256'] != baseline['source_before']['source_map_sha256'] or self.receipt['branch'] != BRANCH): raise RuntimeError('exact clean diagnostic source and branch required') self.receipt['free_bytes_start'] = g.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') if self.name != 'pool-claim': claim = json.loads((OUT / 'pool-claim/receipt.json').read_text()) if claim['status'] != 'passed' or claim['driver_exit'] != 0 or claim['source_before']['head'] != source['head']: raise RuntimeError('successful same-HEAD release-leg pool claim required') 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-release-' + 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', **PROFILE_ENV) self.env.update(overlay) self.receipt['environment_allowlist'] = overlay self.receipt['profile'] = PROFILE self.receipt['inherited_toolchain_environment'] = {key: value for key, value in self.env.items() if key.startswith(('CARGO_', 'RUST')) or key in ('HOME', 'USERPROFILE')} self.receipt['warm_pool_authorization'] = 'Same released 302 pool; retained debug xtask is pool controller only, never a release subject' self.receipt['driver_sha256'] = g.digest(Path(__file__)) self.receipt['inherited_driver_sha256'] = g.digest(SOURCE) context = DEBUG_OUT / 'cpu-contention-context.json' self.receipt['historical_contention_context'] = {'path': str(context), 'sha256': g.digest(context), 'interpretation': 'Historical CI report only. Current census samples do not establish continuous overlap or CPU impact.'} self.flush() def inventory(self, label, scope, expression): return super().inventory(label, ['--release', '--locked', *scope], expression) 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-release'], timeout=120) elif self.name == 'build': stdout, _ = self.run('build', ['cargo', 'build', '--release', '--locked', '-p', 'spt', '--bin', 'spt', '-p', 'spt-daemon', '--example', 'net_diag_smoke', '--message-format=json-render-diagnostics']) expected = {str((g.TARGET / name).resolve()): name for name in ('release/spt.exe', 'release/examples/net_diag_smoke.exe')} artifacts = {} for line in stdout.read_text().splitlines(): row = json.loads(line) if row.get('reason') == 'compiler-artifact' and row.get('executable'): executable = str(Path(row['executable']).resolve()) if executable in expected: if row['profile']['opt_level'] != '3': raise RuntimeError('release artifact is not optimized at opt-level 3') artifacts[expected[executable]] = row if len(artifacts) != len(expected): raise RuntimeError('missing release compiler artifact identity') self.receipt['compiler_artifacts'] = artifacts self.receipt['built_subjects'] = { name: {'exe_sha256': g.digest(g.TARGET / name), 'pdb_sha256': g.digest((g.TARGET / name).with_suffix('.pdb'))} for name in expected.values() } elif self.name in ('seams', 'stack-unit'): scope, expression = (SCOPE, EXPR) if self.name == 'seams' else (['-p', 'spt-net-diag', '--lib'], 'test(stacks::windows::tests::)') self.inventory('inventory', scope, expression) self.run('tests', ['cargo', 'nextest', 'run', '--release', '--locked', *scope, '--profile', 'ci-windows', '--test-threads', '2', '--retries', '0', '--no-fail-fast', '-E', expression]) elif self.name in ('smoke', 'forced'): self.env['SPT_RC_HITCH_DIAG'] = '1' self.receipt['environment_allowlist']['SPT_RC_HITCH_DIAG'] = '1' symbols = ROOT / '.spt/preserved/302/symbols/flat' g.reparse_guard(symbols) if not symbols.is_dir(): raise RuntimeError('prepared HFENDULEAM symbol directory missing') self.env['SPT_NET_DIAG_SYMBOL_DIR'] = str(symbols) self.receipt['environment_allowlist']['SPT_NET_DIAG_SYMBOL_DIR'] = str(symbols) symbol_receipt = ROOT / '.spt/preserved/302/symbols/preparation-receipt.json' prepared = json.loads(symbol_receipt.read_text()) for module in prepared['modules']: if not module['identity_match'] or g.digest(Path(module['flat_path'])) != module['pdb_sha256']: raise RuntimeError('prepared local symbol identity/hash mismatch') self.receipt['symbol_receipt_sha256'] = g.digest(symbol_receipt) amendment = ROOT / '.spt/preserved/302/raw-copy-resume-v1' analyzer = amendment / 'step1-analyze.py' delivery = amendment / 'DELIVERY.json' if g.digest(delivery) != 'bd7f8f3e16e63b6f8c5a1bd7fa12daa89a29544d2b5b58c6b6bbcfe2aa2c0538' or g.digest(analyzer) != '6f36563c6882aed3d4d14f5b276531911d844f6c629b41963378f883a4e9ab24': raise RuntimeError('amended analyzer delivery pin mismatch') verifier = DEBUG_OUT / 'verify.py' analyzer_inputs = [analyzer, amendment / 'step1_rounds.py', delivery, ROOT / '.spt/preserved/302/STEP1-DRIVER.md', verifier] self.receipt['analyzer_inputs'] = {str(path): g.digest(path) for path in analyzer_inputs} binary = g.TARGET / 'release/examples/net_diag_smoke.exe' self.receipt['binary_sha256'] = g.digest(binary) self.receipt['pdb_sha256'] = g.digest(binary.with_suffix('.pdb')) args = ['freeze'] if self.name == 'forced' else [] self.run('native', [str(binary), *args], timeout=120) self.run('analyze', [sys.executable, str(analyzer), '--samples', str(self.dir / 'native.stdout'), '--log', str(self.dir / 'native.stderr'), '--out', str(self.dir / 'analysis')], expected=2, timeout=30) self.run('verify', [sys.executable, str(verifier), str(self.dir)], timeout=30) if any(g.digest(Path(path)) != digest for path, digest in self.receipt['analyzer_inputs'].items()): raise RuntimeError('analyzer inputs changed during native proof') if g.digest(binary) != self.receipt['binary_sha256'] or g.digest(binary.with_suffix('.pdb')) != self.receipt['pdb_sha256']: raise RuntimeError('release fixture or PDB changed during native proof') elif self.name == 'trace': self.run('trace', ['traceable-reqs', 'check', '--json'], timeout=600) elif self.name == 'checks': self.run('clippy', ['cargo', 'clippy', '--release', '--workspace', '--locked']) self.run('trace', ['traceable-reqs', 'check', '--json'], timeout=600) elif self.name == 'pool-release': self.run('release', [str(g.TARGET / 'debug/xtask.exe'), 'pool-release', '--pool', str(g.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') before = json.loads(Path(phase.receipt['source_before']['source_map']).read_text()) after = json.loads(Path(phase.receipt['source_after']['source_map']).read_text()) changed = [key for key in before.keys() | after.keys() if before.get(key) != after.get(key)] phase.receipt['source_changes'] = changed if changed or phase.receipt['source_after']['head'] != HEAD or phase.receipt['source_after']['diff_sha256'] != EMPTY_SHA256: raise RuntimeError('unexpected source changes during release 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())