#!/usr/bin/env python3 """Runner-owned narrow P18 collection using the unchanged guarded phase protocol.""" import argparse import hashlib import json import os from pathlib import Path import shutil import sys import uuid sys.dont_write_bytecode = True import run_all_tests as shared ROSTER = [ 'tests/s4/test_original_relationship_model.gd', 'tests/s4/test_original_relationship_runtime.gd', 'tests/s4/test_relationship_event_service.gd', 'tests/s4/test_runtime_relationship_events.gd', 'tests/s6/test_runtime_snapshot_serialization.gd', 'tests/s6/test_runtime_save_lifecycle.gd', 'tests/ui/test_staff_management.gd', ] def main(argv=None): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--project-root', type=Path, default=Path(__file__).resolve().parent.parent) parser.add_argument('--output', type=Path) parser.add_argument('--godot', default='godot') parser.add_argument('--timeout', type=int, default=1200) parser.add_argument('--prepare-only', action='store_true') args = parser.parse_args(argv) root = args.project_root.resolve() out = (args.output or root / 'artifacts' / ('p18-relationship-' + uuid.uuid4().hex)).resolve() shared.require(args.timeout > 0, 'positive timeout') refs = shared.source_refs(root) # Refuse mixed collector/support checkouts; all actual Python source is pinned. for name in ['tools/run_all_tests.py', 'tools/run_p18_relationship_tests.py', 'tools/strict_test_markers.py']: shared.require(refs[name]['sha256'] == shared.sha(shared.HERE / Path(name).name), 'actual collector source ' + name) for file in shared.SUPPORT.iterdir(): if file.is_file(): shared.require(refs['tools/regression/' + file.name]['sha256'] == shared.sha(file), 'actual support source ' + file.name) shared.require(all(name in refs for name in ROSTER), 'complete narrow roster') shared.require(set(ROSTER).issubset(shared.roster(root)), 'normal discovery includes narrow roster') markers = shared.module('p18_exact_markers', shared.HERE / 'strict_test_markers.py') contract_path = root / 'docs/plans/original-completion-2026-10-02/execution/completion_marker_contracts.json' contract_sha = shared.sha(contract_path) contracts = markers.load_contracts(contract_path) shared.require(all(name in contracts for name in ROSTER), 'complete marker roster') policy = shared.read(shared.SUPPORT / 'suite-policy.json') shared.policy_coverage(ROSTER, refs, policy) for test in ROSTER: shared.phase_policy(test, Path(refs[test]['path']), policy['suite_policies']) for test in ROSTER[:2]: shared.require(contracts[test]['source_sha256_at_schema_audit'] == refs[test]['sha256'], 'new strict source-hashed marker') parse_paths = ['res://' + name for name in sorted(refs) if name.endswith('.gd')] source_bytes = sum(pin['bytes'] for pin in refs.values()) ancestor = out.parent while not ancestor.exists(): ancestor = ancestor.parent free = shutil.disk_usage(ancestor).free needed = source_bytes + shared.MIN_SPACE preview = dict(roster=ROSTER, source_files=len(refs), source_bytes=source_bytes, free_bytes=free, conservative_required_bytes=needed, capacity_available=free >= needed, parse_paths=len(parse_paths), future_native_phases=len(ROSTER) + 2, actual_checks=None, engines=0, stages=0, formal_acceptance=False) if args.prepare_only: print(json.dumps(preview, ensure_ascii=False)) return 0 shared.require(free >= needed, 'fresh source and phase disk admission') outer = shared.module('p18_owned_outer', shared.SUPPORT / 'owned_outer.py') shared.require(not outer.native.helper['live_engines'](), 'another Godot owns CPU') engine = shutil.which(args.godot) if not Path(args.godot).is_absolute() else args.godot shared.require(engine and Path(engine).is_file(), 'native launcher exists') engine = str(Path(engine).resolve()) out.mkdir(parents=True, exist_ok=False) stage = out / 'stage' stage.mkdir() (out / 'starters').mkdir() (out / 'bootstrap').mkdir() for name, pin in shared.read(shared.SUPPORT / 'bootstrap-scripts.json').items(): data = pin['text'].encode('utf-8') shared.require(hashlib.sha256(data).hexdigest() == pin['sha256'], 'unchanged external bootstrap bytes') (out / 'bootstrap' / name).write_bytes(data) protected = shared.default_player(root) expected = {name: {'bytes': pin['bytes'], 'sha256': pin['sha256']} for name, pin in refs.items()} selected = {'root_before': refs, 'map': {name: pin['sha256'] for name, pin in refs.items()}, 'timeout': args.timeout} for name, pin in refs.items(): target = stage / name target.parent.mkdir(parents=True, exist_ok=True) shared.copy_source(Path(pin['path']), target) shared.require(shared.inventory(stage, stage=True) == expected, 'complete private source copy') public = (stage / 'project.godot').read_bytes() shared.write(out / 'parse-paths.json', parse_paths) run_id = uuid.uuid4().hex shared.write(out / 'runner-input.json', dict(protocol='guarded-p18-v1', run_id=run_id, owner_runner_pid=os.getpid(), project_root=str(root), stage=str(stage), roster=ROSTER, selected_source_map=selected['map'], marker_contract_sha256=contract_sha, runner_sha256=shared.sha(Path(__file__)), shared_runner_sha256=shared.sha(shared.HERE / 'run_all_tests.py'), marker_module_sha256=shared.sha(shared.HERE / 'strict_test_markers.py'), engine=engine, engine_sha256=shared.sha(engine), default_player_before=protected, formal_acceptance=False, original_unit_complete=False)) shared.write(out / 'source-start.json', refs) diagnostics = shared.module('p18_diagnostics', shared.SUPPORT / 'diagnostic_gate.py') owners = shared.module('p18_owners', shared.SUPPORT / 'ordinary_proof_gate.py') audio = shared.module('p18_audio', shared.SUPPORT / 'audio_evidence_gate.py') phases, rows, previous = [], [], {} error = None try: for phase, test in [('import', None), ('parse', None)] + [(str(i).zfill(3), name) for i, name in enumerate(ROSTER)]: shared.require(shared.sha(contract_path) == contract_sha, 'marker contract unchanged') row = shared.execute_phase(engine, stage, out, phase, test, selected, expected, previous, policy, contracts, parse_paths, public, protected, root, outer, markers, diagnostics, owners, audio) phases.append(row) if test: row.update(sequence=len(rows), godot=engine) rows.append(row) (out / 'all-tests.json').write_text(json.dumps(rows, ensure_ascii=False, indent=2) + '\n') print(('PASS ' if row['passed'] else 'FAIL ') + test, flush=True) shared.require(row['passed'], 'first failed phase stops without retry: ' + phase) except BaseException as caught: error = type(caught).__name__ + ': ' + str(caught) finally: cleanup, cleanup_errors = outer.finish_captured(outer.OWNED) guard_errors = [] try: shared.require(shared.source_refs(root) == refs, 'whole source unchanged') shared.require(shared.sha(contract_path) == contract_sha, 'marker contract unchanged at end') shared.require(shared.default_player(root) == protected, 'default profile untouched') shared.require(shared.inventory(stage, stage=True) == dict(expected, **previous), 'whole stage exact') except BaseException as caught: guard_errors.append(type(caught).__name__ + ': ' + str(caught)) ended = bool(outer.OWNED) and all(row['engine_ended'] is True for row in cleanup) passed = error is None and not cleanup_errors and not guard_errors and ended and len(phases) == len(ROSTER) + 2 and all(row['passed'] for row in phases) context = shared.totals(rows, ROSTER) shared.write(out / 'runner-ended.json', dict(context, protocol='guarded-p18-v1', run_id=run_id, owner_runner_pid=os.getpid(), passed=passed, engine_ended=ended, actual_captured_PIDs=[process.pid for process in outer.OWNED], final_owned_cleanup=cleanup, cleanup_errors=cleanup_errors, guard_errors=guard_errors, error=error, phases=phases, all_tests_sha256=shared.sha(out / 'all-tests.json') if (out / 'all-tests.json').exists() else None, default_player_after=shared.default_player(root), formal_acceptance=False, original_unit_complete=False, independent_review='pending', native_Windows='pending')) print(json.dumps(dict(context, passed=passed, error=error, cleanup_errors=cleanup_errors, guard_errors=guard_errors)), flush=True) return 0 if passed else 1 if __name__ == '__main__': raise SystemExit(main())