"""NO4JJOEL exact-commit cohorts; Main owns admission, checkout and cleanup. Execute only with Doyle's Windows END authority and a BOX_CLEAR receipt. Workflow: prove; list --selector CELL --proof RECEIPT; cohort --selector CELL --proof RECEIPT --inventory RECEIPT. Every invocation requires a fresh --label. No claims, checkouts, target allocation, artifact adoption or automatic reaping. Adapted from the preserved B2FGTAFA S2 triage-run.py; original rig unchanged. """ import argparse from contextlib import ExitStack from datetime import datetime, timezone import hashlib import json import os from pathlib import Path import re import shutil import subprocess import tempfile import time ROOT = Path(r'C:\Users\decid\Documents\projects\spt-core') HERE = Path(__file__).resolve().parent TREE = ROOT / '.worktrees/hertz-304-phase-b' TARGET = TREE / 'target' LAUNCHER = ROOT / '.spt/preserved/hertz-fp-driver-review/d2/field-rig-r5-S6ESSJ3N/support/fp-bin/launch-v2.ps1' CANDIDATE = '7890ead39bb7f14ed44aaae44b0951f098ffe9ac' CELLS = { 'attachment': ('spt', 'webserve_attachment_e2e', 'an_attachment_is_snapshot_served_fetched_back_and_named_by_its_message'), 'g1': ('spt-daemon', 'inject_control_wedge', 'g1_choreography_happy_path_payload_reaches_pty_and_controller_keeps_control'), 'g7': ('spt-daemon', 'inject_control_wedge', 'g7_native_injects_mid_active_bypassing_the_idle_gate'), } GIB = 1024 ** 3 GOLDEN_ENV = { 'SPT_TEST_EPHEMERAL_ADVISORY_PORTS': '1', 'SPT_ATTACH_GATE_WATCHDOG_MS': '120000', 'SPT_ATTACH_IPC_DEADLINE_MS': '30000', } def utc(): return datetime.now(timezone.utc).isoformat().replace('+00:00', 'Z') def require(condition, message): if not condition: raise RuntimeError(message) def save(path, value): with path.open('x', encoding='utf-8', newline='\n') as stream: json.dump(value, stream, indent=2) stream.write('\n') stream.flush() os.fsync(stream.fileno()) def load(path): return json.loads(Path(path).read_text(encoding='utf-8-sig')) def sha256(path): digest = hashlib.sha256() with Path(path).open('rb') as stream: for chunk in iter(lambda: stream.read(1024 * 1024), b''): digest.update(chunk) return digest.hexdigest() def evidence(path): return {'path': str(Path(path).resolve()), 'sha256': sha256(path)} def verify_evidence(item): require(sha256(item['path']) == item['sha256'], 'EVIDENCE_HASH_MISMATCH') return Path(item['path']) def selector_contract(selector, arm): package, binary, test = CELLS[selector] return { 'selector': selector, 'environment_arm': arm, 'spt_environment': GOLDEN_ENV if arm == 'golden' else {}, 'scope': ['-p', package, '--test', binary], 'filter': f'package(={package}) & kind(test) & binary(={binary}) & test(={test})', } def scrubbed_environment(private_temp, arm): env = dict(os.environ) removed = [] for name in list(env): upper = name.upper() if (upper.startswith(('OWL_', 'SPT_', 'NEXTEST_')) or upper in ('CARGO_TARGET_DIR', 'RUSTFLAGS', 'CARGO_ENCODED_RUSTFLAGS', 'CARGO_BUILD_JOBS', 'CARGO_INCREMENTAL', 'TEMP', 'TMP', 'TMPDIR')): removed.append(name) del env[name] overlay = { 'CARGO_TARGET_DIR': str(TARGET), 'CARGO_BUILD_JOBS': '2', 'CARGO_INCREMENTAL': '0', 'TEMP': str(private_temp), 'TMP': str(private_temp), 'TMPDIR': str(private_temp), } if arm == 'golden': overlay.update(GOLDEN_ENV) env.update(overlay) record = {'utc': utc(), 'removed_names': sorted(removed), 'remaining_names': sorted(env), 'overlay': overlay, 'parent_environment_unchanged': True} return env, record def final_native(record, label): subject = record.get('subject', {}) job = record.get('job', {}) require(record.get('version') == 2 and record.get('label') == label and record.get('scope') == 'run' and record.get('coverage') == 'COMPLETE' and record.get('termination') == 'CONFIRMED_GONE' and record.get('completion_reason') == 'stop_requested' and record.get('error', 'missing') is None and record.get('ended_utc') and job.get('assigned') is True and job.get('membership_read') is True and job.get('kill_on_close_read') is True and job.get('limit_flags') == 8192 and job.get('active_processes') == 0 and type(subject.get('native_exit')) is int, 'INFRASTRUCTURE_NATIVE_CONTAINMENT_OR_TIMEOUT') class Invocation: def __init__(self, args): self.args = args require(os.name == 'nt', 'WINDOWS_REQUIRED') require(re.fullmatch(r'[A-Za-z0-9][A-Za-z0-9_-]{0,63}', args.label), 'UNSAFE_LABEL') require(args.selector != 'attachment' or args.arm == 'consumer', 'ATTACHMENT_REQUIRES_CONSUMER_SHAPE') authority = load(args.authority) require(authority.get('candidate') == CANDIDATE and isinstance(authority.get('windows_end_message'), str) and authority['windows_end_message'].strip() and isinstance(authority.get('windows_end_utc'), str) and authority['windows_end_utc'].strip(), 'WINDOWS_END_AUTHORITY_REQUIRED') ended = datetime.fromisoformat(authority['windows_end_utc'].replace('Z', '+00:00')) require(ended.utcoffset() is not None and ended <= datetime.now(timezone.utc), 'INVALID_WINDOWS_END_UTC') admission = load(args.admission) self.authority = evidence(args.authority) self.admission = evidence(args.admission) require(admission.get('candidate') == CANDIDATE and admission.get('verdict') == 'BOX_CLEAR' and admission.get('authority_sha256') == self.authority['sha256'] and isinstance(admission.get('roots'), list) and admission['roots'] and Path(admission['roots'][0]).resolve() == TARGET.resolve(), 'BOX_CLEAR_ADMISSION_REQUIRED') require(TREE.is_dir() and TARGET.is_dir(), 'MAIN_OWNED_TREE_AND_TARGET_REQUIRED') self.rig = [evidence(Path(__file__)), evidence(LAUNCHER), evidence(LAUNCHER.with_suffix('.cs'))] self.proof = None self.inventory = None if args.action != 'prove': require(args.proof is not None, 'PROVE_RECEIPT_REQUIRED') self.proof = self.receipt(args.proof, 'prove', 'PROVED') require(len(self.proof.get('controls', [])) == 2 and [row['classification'] for row in self.proof['controls']] == ['control_exit_7', 'expected_injected_resource_refusal'], 'INVALID_PROVE_CONTROLS') if args.action == 'cohort': require(args.inventory is not None, 'INVENTORY_RECEIPT_REQUIRED') self.inventory = self.receipt(args.inventory, 'list', 'INVENTORIED') require(self.inventory.get('contract') == selector_contract(args.selector, args.arm) and self.inventory.get('proof') == evidence(args.proof), 'INVENTORY_CELL_OR_PROOF_MISMATCH') raw = verify_evidence(self.inventory['inventory']) require(inventory_executables(raw, args.selector) == self.inventory['executables'], 'INVENTORY_EXECUTABLES_CHANGED') # Reserve the invocation before creating any executable child or private temp. self.dest = HERE / args.label self.dest.mkdir() temp_parent = (Path(os.environ['LOCALAPPDATA']) / 'Temp').resolve() require(temp_parent.is_dir() and not temp_parent.is_relative_to(ROOT.resolve()), 'USER_TEMP_OUTSIDE_GIT_REQUIRED') self.private_temp = Path(tempfile.mkdtemp(prefix='hertz-NO4JJOEL-', dir=temp_parent)) self.env, env_record = scrubbed_environment(self.private_temp, args.arm) save(self.dest / 'environment-names.json', env_record) self.base = {'candidate': CANDIDATE, 'action': args.action, 'label': args.label, 'environment_arm': args.arm, 'spt_environment': GOLDEN_ENV if args.arm == 'golden' else {}, 'load': {'execution': 'ten serial invocations; one selected cell per invocation', 'profile': 'ci-windows', 'nextest_threads': 'profile default, one selected cell', 'full_239_cell_heavy_parallel_load': 'UNMEASURED', 'cotenant_producers': 'none at cited admission sample; sample is not a lease', 'admission': self.admission}, 'started_utc': utc(), 'authority': self.authority, 'admission': self.admission, 'tree': str(TREE), 'target': str(TARGET), 'private_temp': str(self.private_temp), 'temp_cleanup_owner': 'Main', 'rig': self.rig} save(self.dest / 'invocation.json', self.base) self.pwsh = shutil.which('pwsh', path=self.env.get('PATH')) self.cargo = shutil.which('cargo', path=self.env.get('PATH')) git = shutil.which('git', path=self.env.get('PATH')) require(self.pwsh and git and (args.action == 'prove' or self.cargo), 'REQUIRED_EXECUTABLE_ABSENT') # Execution-time check only: this module performs no subprocess work at import. checked = subprocess.run([git, '-C', str(TREE), 'rev-parse', 'HEAD'], env=self.env, stdin=subprocess.DEVNULL, capture_output=True, timeout=30) save(self.dest / 'candidate-check.json', { 'utc': utc(), 'exit': checked.returncode, 'head': checked.stdout.decode('utf-8', errors='replace').strip(), }) require(checked.returncode == 0 and checked.stdout.decode().strip() == CANDIDATE, 'CANDIDATE_HEAD_MISMATCH') self.disk_roots = [ROOT] if self.private_temp.anchor.lower() != ROOT.anchor.lower(): self.disk_roots.append(self.private_temp) self.disk_start = {str(path): shutil.disk_usage(path).free for path in self.disk_roots} self.disk_minimum = dict(self.disk_start) self.capacity() def receipt(self, path, action, verdict): receipt = load(path) require(receipt.get('candidate') == CANDIDATE and receipt.get('action') == action and receipt.get('verdict') == verdict and receipt.get('rig') == self.rig and receipt.get('authority') == self.authority and receipt.get('tree') == str(TREE) and receipt.get('target') == str(TARGET), 'PREREQUISITE_RECEIPT_MISMATCH') for item in receipt.get('artifacts', []): verify_evidence(item) require(receipt.get('artifacts'), 'PREREQUISITE_EVIDENCE_REQUIRED') return receipt def pressure(self): readings = {} refused = False for path in self.disk_roots: key = str(path) free = shutil.disk_usage(path).free self.disk_minimum[key] = min(self.disk_minimum[key], free) growth = self.disk_start[key] - free readings[key] = {'free_bytes': free, 'growth_bytes': growth} refused |= free <= 32 * GIB or growth >= 64 * GIB return refused, readings def capacity(self): refused, readings = self.pressure() if refused: save(self.dest / ('capacity-refusal-' + str(time.monotonic_ns()) + '.json'), {'utc': utc(), 'reason': 'resource_refusal', 'injected': False, 'disks': readings}) raise RuntimeError('INFRASTRUCTURE_REAL_CAPACITY_REFUSAL') def run(self, suffix, exe, argv, seconds, control=None): self.capacity() label = self.args.label + '-' + suffix dest = self.dest / suffix dest.mkdir() child_temp = self.private_temp / suffix child_temp.mkdir() env, names = scrubbed_environment(child_temp, self.args.arm) save(dest / 'environment-names.json', names) save(dest / 'environment.json', names['overlay']) save(dest / 'argv.json', argv) save(dest / 'executable.json', evidence(exe)) command = [self.pwsh, '-NoProfile', '-NonInteractive', '-File', str(LAUNCHER), '-Label', label, '-Seconds', str(seconds), '-Scope', 'run', '-Admission', 'process-tree', '-RecordFile', str(dest / 'native.json'), '-ArgsFile', str(dest / 'argv.json'), '-Exe', exe, '-OutFile', str(dest / 'stdout.log'), '-ErrFile', str(dest / 'stderr.log'), '-EnvironmentFile', str(dest / 'environment.json'), '-StopFile', str(dest / 'stop')] save(dest / 'command.json', {'utc': utc(), 'argv': command, 'cwd': str(TREE)}) started = time.monotonic() stop = None monitor_error = None forced_launcher_exit = False journal = None journal_tail = b'' native = {} with (dest / 'launcher.log').open('xb') as log, ExitStack() as readers: proc = subprocess.Popen(command, cwd=TREE, env=env, stdin=subprocess.DEVNULL, stdout=log, stderr=subprocess.STDOUT) while proc.poll() is None: if stop is None: try: refused, disks = self.pressure() # Native publishes complete journal rows before replacing its # status file. Read that append-only stream, never contend with # Windows rename/delete sharing on a live native.json. journal_path = dest / 'native.json.events.jsonl' if journal is None and journal_path.exists(): journal = readers.enter_context(journal_path.open('rb')) if journal is not None: journal_tail += journal.read() rows = journal_tail.split(b'\n') journal_tail = rows.pop() for row in rows: if row: native = json.loads(row)['record'] if native: require(native.get('version') == 2 and native.get('label') == label and native.get('scope') == 'run', 'NATIVE_POLL_IDENTITY_MISMATCH') subject = native.get('subject', {}) marker = False if control == 'resource' and (dest / 'stdout.log').exists(): text = (dest / 'stdout.log').read_text(encoding='utf-8-sig', errors='replace') marker = re.search(r'(?m)^CHILD_STARTED [1-9][0-9]*\r?$', text) is not None reason = None injected = False # Actual capacity refusal always wins over an injected control or exit. if refused: reason = 'resource_refusal' elif subject.get('state') == 'EXITED' and type(subject.get('native_exit')) is int: reason = 'subject_exit_observed' elif (control == 'resource' and marker and subject.get('state') == 'RUNNING' and subject.get('native_exit') is None): reason = 'resource_refusal' injected = True if reason: stop = {'reason': reason, 'utc': utc(), 'injected': injected, 'child_started_marker': marker, 'subject_before_stop': subject, 'disks': disks} save(dest / 'stop', stop) except Exception as error: monitor_error = type(error).__name__ + ': ' + str(error) if stop is None: stop = {'reason': 'monitoring_failure', 'utc': utc(), 'injected': False, 'error': monitor_error} try: save(dest / 'stop', stop) except Exception as stop_error: monitor_error += '; stop write: ' + str(stop_error) if time.monotonic() - started > seconds + 30: # Exceptional watchdog only; never claim this establishes containment. forced_launcher_exit = True proc.kill() break time.sleep(0.1) launcher_exit = proc.wait(timeout=30) save(dest / 'launcher-exit.json', { 'utc': utc(), 'launcher_exit': launcher_exit, 'forced_launcher_exit': forced_launcher_exit, 'monitor_error': monitor_error, }) native = load(dest / 'native.json') result = {'label': label, 'ended_utc': utc(), 'launcher_exit': launcher_exit, 'native_exit': native.get('subject', {}).get('native_exit'), 'stop': stop, 'monitor_error': monitor_error, 'forced_launcher_exit': forced_launcher_exit, 'elapsed_seconds': time.monotonic() - started, 'native': native, 'disk_start': self.disk_start, 'disk_minimum': dict(self.disk_minimum), 'classification': 'infrastructure_failure'} try: final_native(native, label) require(launcher_exit == 0 and not monitor_error and not forced_launcher_exit and stop, 'INFRASTRUCTURE_LAUNCHER_OR_MONITOR_FAILURE') require(stop['reason'] != 'resource_refusal' or stop['injected'], 'INFRASTRUCTURE_REAL_CAPACITY_REFUSAL') if control == 'resource': require(stop['reason'] == 'resource_refusal' and stop['injected'] and stop['child_started_marker'] and stop['subject_before_stop']['state'] == 'RUNNING' and stop['subject_before_stop']['native_exit'] is None and native['subject']['state'] == 'TERMINATED', 'RESOURCE_CONTROL_NOT_EXERCISED') result['classification'] = 'expected_injected_resource_refusal' else: require(stop['reason'] == 'subject_exit_observed' and native['subject']['state'] == 'EXITED', 'DIRECT_EXIT_NOT_OBSERVED') if control == 'exit': require(result['native_exit'] == 7, 'EXIT_CONTROL_NOT_SEVEN') result['classification'] = 'control_exit_7' else: require(result['native_exit'] in (0, 100), 'INFRASTRUCTURE_UNEXPECTED_NATIVE_EXIT') result['classification'] = 'passed' if result['native_exit'] == 0 else 'test_red' finally: save(dest / 'result.json', result) print(json.dumps({'label': label, 'launcher_exit': launcher_exit, 'native_exit': result['native_exit'], 'classification': result['classification']}), flush=True) return result, dest def finish(self, verdict, details): artifacts = [evidence(path) for path in sorted(self.dest.rglob('*')) if path.is_file()] receipt = dict(self.base, ended_utc=utc(), verdict=verdict, artifacts=artifacts, **details) path = HERE / (self.args.label + '-' + self.args.action + '-receipt.json') save(path, receipt) print(json.dumps({'receipt': str(path), 'verdict': verdict}), flush=True) def inventory_executables(path, selector): inventory = load(path) package, binary, test = CELLS[selector] require(Path(inventory['rust-build-meta']['target-directory']).resolve() == TARGET.resolve(), 'INVENTORY_TARGET_MISMATCH') included = [] for suite in inventory['rust-suites'].values(): for name, case in suite.get('testcases', {}).items(): status = case.get('filter-match', {}).get('status') require(status in ('matches', 'mismatch'), 'UNKNOWN_INVENTORY_FILTER_STATUS') if status == 'matches': included.append((suite, name, case)) require(len(included) == 1, 'EXACTLY_ONE_INCLUDED_TEST_REQUIRED') suite, name, case = included[0] require(suite.get('package-name') == package and suite.get('binary-name') == binary and suite.get('kind') == 'test' and suite.get('status') == 'listed' and name == test and case.get('kind') == 'test' and case.get('ignored') is False, 'EXACT_INCLUDED_CELL_REQUIRED') paths = {Path(suite['binary-path']).resolve()} for binaries in inventory['rust-build-meta'].get('non-test-binaries', {}).values(): for item in binaries: if item.get('kind') == 'bin-exe': paths.add((TARGET / item['path']).resolve()) require(all(path.is_relative_to(TARGET.resolve()) for path in paths), 'INVENTORY_EXECUTABLE_OUTSIDE_OWN_TARGET') return [evidence(path) for path in sorted(paths)] def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('action', choices=['prove', 'list', 'cohort']) parser.add_argument('--authority', type=Path, required=True) parser.add_argument('--admission', type=Path, required=True) parser.add_argument('--label', required=True) parser.add_argument('--selector', choices=CELLS, default='attachment') parser.add_argument('--arm', choices=['consumer', 'golden'], default='consumer') parser.add_argument('--proof', type=Path) parser.add_argument('--inventory', type=Path) args = parser.parse_args() invocation = Invocation(args) results = [] try: if args.action == 'prove': result, _ = invocation.run('direct-exit', invocation.pwsh, ['-NoProfile', '-NonInteractive', '-Command', 'exit 7'], 60, 'exit') results.append(result) script = ("$ErrorActionPreference='Stop'; " "$child=Start-Process -FilePath (Get-Process -Id $PID).Path " "-ArgumentList @('-NoProfile','-NonInteractive','-Command','Start-Sleep -Seconds 600') " "-PassThru -NoNewWindow; $child.Refresh(); " "if ($child.HasExited) { throw 'CHILD_EXITED_BEFORE_MARKER' }; " "[Console]::Out.WriteLine('CHILD_STARTED ' + $child.Id); [Console]::Out.Flush(); " "Start-Sleep -Seconds 600") result, _ = invocation.run('resource-control', invocation.pwsh, ['-NoProfile', '-NonInteractive', '-Command', script], 60, 'resource') results.append(result) invocation.finish('PROVED', {'controls': results}) return 0 contract = selector_contract(args.selector, args.arm) common = contract['scope'] + ['--profile', 'ci-windows', '-E', contract['filter']] if args.action == 'list': result, dest = invocation.run('inventory', invocation.cargo, ['nextest', 'list'] + common + ['--message-format', 'json'], 2400) require(result['native_exit'] == 0, 'INFRASTRUCTURE_INVENTORY_FAILED') executables = inventory_executables(dest / 'stdout.log', args.selector) invocation.finish('INVENTORIED', {'contract': contract, 'proof': evidence(args.proof), 'inventory': evidence(dest / 'stdout.log'), 'executables': executables, 'result': result}) return 0 argv = ['nextest', 'run'] + common + [ '--retries', '0', '--no-fail-fast', '--success-output', 'immediate', '--failure-output', 'immediate', '--color', 'never'] for number in range(1, 11): for item in invocation.inventory['executables']: verify_evidence(item) result, _ = invocation.run(f'attempt-{number:02}', invocation.cargo, argv, 180) results.append(result) for item in invocation.inventory['executables']: verify_evidence(item) invocation.finish('COHORT_COMPLETE', {'contract': contract, 'proof': evidence(args.proof), 'inventory_receipt': evidence(args.inventory), 'attempt_count': 10, 'results': results}) return 0 if all(row['native_exit'] == 0 for row in results) else 100 except Exception as error: invocation.finish('INFRASTRUCTURE_FAILURE', {'error': type(error).__name__ + ': ' + str(error), 'completed_results': results}) return 125 if __name__ == '__main__': raise SystemExit(main())