#!/usr/bin/env python3 """Root-only native collection. Does not run regression or claim acceptance.""" import argparse import hashlib import importlib.util import json import os from pathlib import Path import shutil import subprocess import sys import time import uuid import re sys.dont_write_bytecode = True HERE = Path(__file__).resolve().parent CONTRACT = 'docs/plans/original-completion-2026-10-02/execution/completion_marker_contracts.json' CONDITIONS = ('support_connection', 'invalid_position_reason', 'demolish_rebuild', 'pan_zoom_world_identity', 'separate_pid_resume') OS_CHILD_TIMEOUT_SECONDS = 2400 RECORDER = 'docs/plans/original-completion-2026-10-02/execution/p08-03/passive_os_placement_recorder.gd' def collect_os(r, root, out, stage, engine, commit, refs, protected, outer): """P08-02 collect_native_os protocol; deliberately not execute_phase/headless. Root supplies CUA ack/receipt files in the requested directories. Neither controller nor passive endpoints generate input or change permissions. """ r.require(not outer.native.helper['live_engines'](), 'external engine present') before = r.inventory(stage, stage=True) public = (stage/'project.godot').read_bytes() pair = out/'os-pair' pair.mkdir() r.copy_source(out/'pair/diagnostic-budget.json', pair/'diagnostic-budget.json') phases = [] for role in ('writer', 'reader'): r.require(not outer.native.helper['live_engines'](), 'external engine present') r.require(r.source_refs(root) == refs and r.default_player(root) == protected, 'OS before guards') r.require(r.inventory(stage, stage=True) == before, 'OS stage before guard') r.require(shutil.disk_usage(out).free >= 500*1024*1024, 'OS disk admission') folder = out/'os-runs'/role folder.mkdir(parents=True) name = 'TheTree-PCK-tests-'+uuid.uuid4().hex+'-p08-placement-'+role profile = r.profile_path(name) r.require(not profile.exists(), 'fresh OS private profile') formal_source = out/('pair/initial-formal' if role == 'writer' else 'os-pair/formal') formal_before = r.inventory(formal_source) r.require(set(formal_before) == {'world.bin','manifest.json','thumbnail.png','journal.log'}, 'all formal inputs') formal = profile/'saves/manual_ui' formal.mkdir(parents=True) for member, pin in formal_before.items(): r.copy_source(formal_source/member, formal/member) r.require(r.sha(formal/member) == pin['sha256'], 'formal fixture copy SHA') recorder = folder/'passive_os.gd' r.copy_source(root/RECORDER, recorder) proof = folder/'teardown' (proof/'owned-audio').mkdir(parents=True) env = r.fresh_environment(dict(os.environ)) env.update(P08_PLACEMENT_ROOT=str(pair), P08_PLACEMENT_ROLE=role, THE_TREE_AUDIO_TEARDOWN_PROOF_DIR=str(proof/'owned-audio'), THE_TREE_AUDIO_FIXTURE_CONTROL_PROOF=str(proof/'control.json')) command = [engine, '--path', str(stage), '--script', str(recorder)] base = {'protocol':'p08-placement-native-os-v1', 'role':role, 'commit':commit, 'source_map':{n:v['sha256'] for n,v in refs.items()}, 'stage':str(stage), 'stage_before':before, 'profile':str(profile), 'profile_name':name, 'profile_before':{'existed':False, 'files':{}}, 'formal_input':formal_before, 'formal_source':str(formal_source), 'owner_pid':os.getpid(), 'default_before':protected, 'command':command, 'engine_sha256':r.sha(engine), 'controller_sha256':r.sha(root/'tools/run_p08_normal_placement.py'), 'recorder_sha256':r.sha(recorder), 'formal_acceptance':False, 'timeout_seconds':OS_CHILD_TIMEOUT_SECONDS, 'environment':{k:v for k,v in env.items() if k.startswith(('P08_', 'THE_TREE_'))}, 'environment_sha256':hashlib.sha256(json.dumps(env,sort_keys=True).encode()).hexdigest()} r.write(folder/'started.json', base) error = None; children = []; guards = []; exit_code = None; owner_proof = None try: (stage/'project.godot').write_bytes(r.private_project(public, name)) with (folder/'engine-native.log').open('x') as log: process = subprocess.Popen(command, cwd=stage, env=env, stdout=log, stderr=subprocess.STDOUT) children.append(process) r.write(folder/'process-started.json', dict(base, actual_pid=process.pid, monotonic_ns=time.monotonic_ns(), state='waiting_for_root_cua_requests')) print(json.dumps({'actual_pid':process.pid, 'request_directory':str(pair), 'state':'waiting_for_root_cua', 'role':role}), flush=True) exit_code = process.wait(timeout=base['timeout_seconds']) r.require(exit_code == 0, 'ordinary native OS exit0') raw = (folder/'engine-native.log').read_text() r.require('ERROR:' not in raw and 'SCRIPT ERROR:' not in raw, 'OS native diagnostics') matches = re.findall(r'(?m)^P08 passive OS placement: ([0-9]+) checks, 0 failures$', raw) r.require(len(matches) == 1 and int(matches[0]) > 0, 'passive observer completion') observed = r.read(pair/(role+'-profile.json')) r.require(observed['pid'] == process.pid and Path(observed['user_data_dir']).resolve() == profile.resolve() and observed['display'] != 'headless', 'actual OS private profile/display') owners = r.module('p08_os_owner_gate', r.SUPPORT/'ordinary_proof_gate.py') owner_proof = owners.ordinary_gate(proof, process.pid, folder, ['single']) except BaseException as caught: error = type(caught).__name__+': '+str(caught) finally: try: cleanup, cleanup_errors = outer.finish_captured(children) finally: (stage/'project.godot').write_bytes(public) try: r.require(r.source_refs(root) == refs, 'OS source unchanged') r.require(r.inventory(stage, stage=True) == before, 'OS stage unchanged') r.require(r.default_player(root) == protected, 'OS default player unchanged') r.require(r.inventory(formal_source) == formal_before, 'formal input unchanged') r.require(r.sha(recorder) == base['recorder_sha256'], 'recorder unchanged') except BaseException as caught: guards.append(type(caught).__name__+': '+str(caught)) passed = not (error or guards or cleanup_errors) and len(children) == 1 and exit_code == 0 r.write(folder/'ended.json', dict(base, actual_pid=children[0].pid if children else None, exit_code=exit_code, engine_ended=bool(cleanup) and all(v['engine_ended'] for v in cleanup), error=error, guard_errors=guards, cleanup_errors=cleanup_errors, owned_cleanup=cleanup, ordinary_owner_proof=owner_proof, stage_after=r.inventory(stage,stage=True), default_after=r.default_player(root), source_after=r.source_refs(root), profile_after=r.inventory(profile), passed=passed)) phases.append('os-runs/'+role+'/ended.json') r.require(passed, 'native OS collection failed: '+str(error or guards or cleanup_errors)) return phases def module(name, path): spec = importlib.util.spec_from_file_location(name, path) obj = importlib.util.module_from_spec(spec) spec.loader.exec_module(obj) return obj def absolute(value): path = Path(value) if not path.is_absolute(): raise argparse.ArgumentTypeError('absolute path required') return path.resolve() def collect(args): root, out = args.project_root, args.output if out.is_relative_to(root): raise ValueError('evidence output must be outside the frozen source checkout') r = module('p08_standard_runner', root/'tools/run_all_tests.py') commit = subprocess.check_output(['git', '-C', str(root), 'rev-parse', 'HEAD'], text=True).strip() r.require(commit == args.commit, 'actual final submitted commit required') r.require(not subprocess.check_output(['git','-C',str(root),'status','--porcelain'], text=True), 'frozen clean worktree required') r.require(not out.exists(), 'fresh output; never overwrite earlier proof') refs = r.source_refs(root) checker = module('p08_initial_inputs', root/'tools/check_p08_normal_placement_evidence.py') inputs_before = checker.input_map(root) outer = r.module('p08_owned_outer', r.SUPPORT/'owned_outer.py') r.require(not outer.native.helper['live_engines'](), 'one native engine') engine = shutil.which(args.godot) r.require(engine is not None, 'native engine exists') engine = str(Path(engine).resolve()) out.mkdir(parents=True) stage = out/'stage'; stage.mkdir() (out/'starters').mkdir(); (out/'bootstrap').mkdir() for name, pin in r.read(r.SUPPORT/'bootstrap-scripts.json').items(): data = pin['text'].encode() r.require(hashlib.sha256(data).hexdigest() == pin['sha256'], 'bootstrap pin') (out/'bootstrap'/name).write_bytes(data) for name, pin in refs.items(): target = stage/name; target.parent.mkdir(parents=True, exist_ok=True) r.copy_source(Path(pin['path']), target) r.require(r.sha(target) == pin['sha256'], 'whole copied source '+name) expected = {n: {'bytes':v['bytes'], 'sha256':v['sha256']} for n,v in refs.items()} selected = {'root_before':refs, 'map':{n:v['sha256'] for n,v in refs.items()}, 'timeout':180} r.require(r.inventory(stage, stage=True) == expected, 'complete private source') protected = r.default_player(root); public = (stage/'project.godot').read_bytes() paths = ['res://'+n for n in sorted(refs) if n.endswith('.gd')] r.write(out/'parse-paths.json', paths) markers = r.module('p08_strict_markers', root/'tools/strict_test_markers.py') contracts = markers.load_contracts(root/CONTRACT) policy = r.read(r.SUPPORT/'suite-policy.json') diagnostics = r.module('p08_diagnostics', r.SUPPORT/'diagnostic_gate.py') owners = r.module('p08_owners', r.SUPPORT/'ordinary_proof_gate.py') audio = r.module('p08_audio', r.SUPPORT/'audio_evidence_gate.py') r.write(out/'runner-input.json', {'protocol':'p08-normal-placement-v1', 'commit':commit, 'selected_source_map':selected['map'], 'engine':engine, 'engine_sha256':r.sha(engine), 'owner_pid':os.getpid(), 'default_player':protected, 'diagnostic_seed':42533, 'inputs':inputs_before}) phases = []; previous = {}; cases = []; error = None; os_proof = None deadline = time.monotonic()+(7200 if args.input_mode == 'native_os' else 900) saved_env = {k:os.environ.get(k) for k in ('P08_PLACEMENT_ROOT','P08_PLACEMENT_ROLE','P08_PLACEMENT_INPUT')} def phase(label, test): remaining = int(deadline-time.monotonic()) r.require(remaining >= (180 if label == 'import' else 1), '900s collection deadline') selected['timeout'] = min(180, remaining) phase_out = out/'runs'/label phase_out.mkdir(parents=True) (phase_out/'starters').mkdir(); (phase_out/'bootstrap').mkdir() for original in (out/'bootstrap').iterdir(): r.copy_source(original,phase_out/'bootstrap'/original.name) r.require(r.sha(original)==r.sha(phase_out/'bootstrap'/original.name),'bootstrap full copy') r.write(phase_out/'parse-paths.json',paths) row = r.execute_phase(engine, stage, phase_out, label, test, selected, expected, previous, policy, contracts, paths, public, protected, root, outer, markers, diagnostics, owners, audio) phases.append('runs/'+label+'/'+label+'-receipt.json') r.require(not row['failed'], 'strict phase failed: '+label) try: phase('import', None); phase('parse', None) pair = out/'pair' preserved = {} for role in ('writer', 'reader'): os.environ.update(P08_PLACEMENT_ROOT=str(pair), P08_PLACEMENT_ROLE=role, P08_PLACEMENT_INPUT='handler') phase(role, 'tests/ui/test_p08_normal_placement_ops.gd') actual = r.inventory(pair) r.require(all(actual.get(n) == pin for n, pin in preserved.items()), 'reader preserved every writer and formal byte') preserved = actual cases.append({'directory':'pair', 'phases':phases[-2:], 'files':preserved}) if args.input_mode == 'native_os': os_phases = collect_os(r, root, out, stage, engine, commit, refs, protected, outer) checker = module('p08_os_input_map', root/'tools/check_p08_normal_placement_evidence.py') os_proof = 'os-proof.json' r.write(out/os_proof, {'source_map':selected['map'], 'inputs':checker.input_map(root), 'input_method':'native_os', 'observer_mutations':[], 'phases':os_phases, 'directory':'os-pair'}) except BaseException as caught: error = type(caught).__name__+': '+str(caught) finally: cleanup, cleanup_errors = outer.finish_captured(outer.OWNED) for key,value in saved_env.items(): if value is None: os.environ.pop(key, None) else: os.environ[key] = value if r.source_refs(root) != refs or r.default_player(root) != protected: error = (error+'; ' if error else '')+'protected root/profile changed' if checker.input_map(root) != inputs_before: error = (error+'; ' if error else '')+'spec/fixture/controller inputs changed' r.write(out/'runner-ended.json', {'error':error, 'cleanup':cleanup, 'cleanup_errors':cleanup_errors, 'phases':phases, 'formal_acceptance':False, 'native_Windows':'pending'}) checker = module('p08_placement_checker', root/'tools/check_p08_normal_placement_evidence.py') r.write(out/'index.json', {'schema':'p08-normal-placement-v1', 'commit':commit, 'source_map':selected['map'], 'inputs':inputs_before, 'marker_contract_sha256':r.sha(root/CONTRACT), 'spec_sha256':hashlib.sha256((root/'docs/spec/acceptance/P08-03.json').read_text().strip().encode()).hexdigest(), 'cases':cases, 'phases':phases, 'conditions':list(CONDITIONS), 'artifacts':{str(p.relative_to(out)):{'bytes':p.stat().st_size,'sha256':r.sha(p)} for p in out.rglob('*') if p.is_file() and 'stage' not in p.relative_to(out).parts}, 'os_proof':os_proof, 'regression':None, 'acceptance':'pending', 'collection_error':error}) return 1 if error or cleanup_errors else 0 def main(argv=None): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--project-root', type=absolute, required=True) parser.add_argument('--output', type=absolute, required=True) parser.add_argument('--commit', required=True) parser.add_argument('--godot', default='godot') parser.add_argument('--input-mode', choices=('handler', 'native_os'), default='handler') parser.add_argument('--collect', action='store_true', required=True) return collect(parser.parse_args(argv)) if __name__ == '__main__': raise SystemExit(main())