#!/usr/bin/env python3 """Synthetic evidence ONLY: disposable end-to-end check() positives and corruptions. No Godot/OS execution, no actual acceptance proof. Mutations update hashes so the semantic gate, not merely a stale digest, must reject invalid state. """ import copy import hashlib import json from pathlib import Path import tempfile import unittest import check_p08_migration_evidence as c ROOT = Path(__file__).resolve().parent.parent BANNER = 'Godot Engine v4.7.2.stable.official.ed1daf0bf - https://godotengine.org\n' R04 = json.loads((ROOT/'docs/plans/original-completion-2026-10-02/execution/p08-02/ROOT_REPAIR_INPUT-r04.json').read_text()) ACTUAL = R04['actual_readonly_diagnostics'] REQUESTS = ACTUAL['CARGO_ENDPOINT_READONLY_DIAGNOSTIC.json'][0]['requests'] REVISIONS = ACTUAL['ACTUAL_WORLD_REVISION_DELTAS.json'][0] def typed(value): if isinstance(value,tuple): return {'type':'Vector3i','text':'Vector3i('+', '.join(map(str,value))+')'} if isinstance(value,dict): return {'type':'Dictionary','entries':[[typed(k),typed(v)] for k,v in value.items()]} if isinstance(value,list): return {'type':'Array','element_type':0,'element_class':'','items':[typed(x) for x in value]} kind = 'Nil' if value is None else 'bool' if type(value) is bool else 'int' if type(value) is int else 'String' return {'type':kind,'text':json.dumps(value)} class Proof: def __init__(self, path): self.path = path runner = c.module('p08_synthetic_source',ROOT/'tools/run_all_tests.py') self.runner = runner source = {n:p['sha256'] for n,p in runner.source_refs(ROOT).items()} self.index = {'schema':'p08-evidence-v1','commit':'e196578c1a4ad2fdf84642565bd102e524be386f', 'source_map':source,'spec_sha256':c.SPEC_SHA,'marker_contract_sha256':c.sha(ROOT/c.CONTRACT), 'inputs':{str(p.relative_to(ROOT)):c.sha(p) for p in (ROOT/'docs/spec/inputs/p08-02').iterdir() if p.is_file()}, 'cases':[],'conditions':{k:[x for x in c.CASES if k!='unsolvable_preservation' or x.endswith('refusal')] for k in c.CONDITIONS}, 'artifacts':{},'acceptance':'pending','collection_error':None,'phases':[], 'os_proof':'os.json','regression':'regression.json'} for number,name in enumerate(c.CASES): self.case(name,number) self.os(); self.regression(); self.save() def put(self,name,value): path = self.path/name; path.parent.mkdir(parents=True,exist_ok=True) path.write_bytes(value if isinstance(value,bytes) else json.dumps(value).encode()) self.index['artifacts'][name] = {'bytes':path.stat().st_size,'sha256':c.sha(path)} def variant(self,prefix,name,value): node = typed(value) self.put(prefix+name+'.json',node) # Deliberately not a native Variant: this is an isolated Python proof fixture. self.put(prefix+name+'.var',json.dumps(node).encode()) def save(self): (self.path/'index.json').write_text(json.dumps(self.index)) def state(self,operations): jobs = [{'job_id':j,'state':'PLANNED' if j=='j2' else 'BUILDING','progress':0 if j=='j2' else 1266,'reserved_cells':[(x,2,0) for x in range(offset,offset+6)],'construction_cost':{'gold':1}} for j,offset in [('j1',0),('j2',10)]] facilities = [{'instance_id':'dorm','occupied_cells':[[30,0,1]]},{'instance_id':'clinic','occupied_cells':[[31,0,1]]}] facilities += [{'instance_id':'facility_00000005','occupied_cells':[(32,0,1)]}, {'instance_id':'facility_00000006','occupied_cells':[(33,0,1)]}] guests = [{'guest_id':g} for g in ['p08.walker']+['g'+str(n) for n in range(5)]] staff = [{'staff_id':'patient','health':0},{'staff_id':'carrier'},{'staff_id':'staff_gina','carried_inventory_id':'cargo:staff_gina'}] r = {'operations':{'enabled':operations},'guests':{'guests':guests},'staff':{'staff':staff,'tasks':[ {'task_id':'rescue','target_entity_id':'patient','rescue_facility_id':'clinic','state':'MOVING','assigned_staff_id':'carrier'}]}, 'construction':{'jobs':jobs,'facilities':facilities},'facilities':{'instances':copy.deepcopy(facilities)}, 'world':{'revisions':copy.deepcopy(REVISIONS['original_world_revisions']),'occupancies':[{'entity_id':j['job_id'],'cells':copy.deepcopy(j['reserved_cells'])} for j in jobs]+ [{'entity_id':f['instance_id'],'cells':copy.deepcopy(f['occupied_cells'])} for f in facilities]}, 'inventory':{'inventories':[{'owner_inventory_id':owner,'stacks':[{'stack_id':stack,'item_id':'ingredient/raw_food','quantity':4,'reserved_quantity':reserved}]} for owner,stack,reserved in [('facility.inventory:facility_00000005','stack_00000001',4),('cargo:staff_gina','stack_00000002',0)]], 'reservations':[{'reservation_id':req['reservation_ids'][0],'requester_id':req['request_id'], 'owner_inventory_id':req['source_inventory_id'],'stack_id':'stack_00000001','item_id':'ingredient/raw_food', 'quantity':4,'state':'CONSUMED' if req['state']=='DELIVERING' else 'ACTIVE'} for req in REQUESTS]}, 'logistics_requests':copy.deepcopy(REQUESTS), 'movement_credits':[{'credit':1000}], 'elevators':{'trips':[{'request_id':'ev'+str(n),'actor_id':'g'+str(n)} for n in range(5)],'shafts':[{'boarded':['g0','g1','g2','g3']}]}} return {'systems':{'runtime':r}} def phase(self,name,role,pid): folder = 'runs/'+name+'/'; test = 'tests/fixtures/p08-02/'+role+'.gd' command = ['synthetic-native','--script',test] start = {'actual_pid':pid,'command':command,'live_identity_exact':True,'event':'ENGINE_STARTED','owner_pid':1,'monotonic_ns':1} end = dict(start,event='ENGINE_ENDED',monotonic_ns=2,exit_code=0,engine_ended=True) self.put(folder+name+'/started.json',start); self.put(folder+name+'/ended.json',end) self.put(folder+name+'/outer-owned-finally.json',{'actual_captured_pids':[pid],'all_captured_ended':True,'outer_cleanup_errors':[],'foreign_process_or_group_signals':0}) self.put(folder+'raw.log',(BANNER+'P08 mixed '+role+': 1 checks, 0 failures\n').encode()) receipt = {'phase':name,'actual_pid':pid,'command':command,'exit_code':0,'engine_ended':True,'test':test, 'source_sha256':self.index['source_map'][test],'log':'raw.log', 'native_process':{'exit_code':0,'helper_exit_code':0,'timed_out':False,'lifecycle_error':None}, 'actual_private_profile':{'report':{'fresh_before_parent_initialize':True,'profile_writes_before_guard':0,'actual_user_data_dir':'diagnostic-private-'+name,'expected_user_data_dir':'diagnostic-private-'+name}}} self.put(folder+name+'-receipt.json',receipt) return folder+name+'-receipt.json' def case(self,name,number): operations = name.startswith('on-'); refusal = name.endswith('refusal'); prefix = 'pairs/'+name+'/' phases = [self.phase(name+'-'+role,role,100+2*number+i) for i,role in enumerate(('writer','reader'))] self.index['cases'].append({'id':name,'operations':'on' if operations else 'off','kind':'refusal' if refusal else 'positive','directory':prefix[:-1],'phases':phases}) self.index['phases'] += phases identity = {'starting_job':'j2','diagnostic_budget':{'seed':42588,'food_quantity':8,'food_cost':24,'signing':1,'costs':[{'cost':{'gold':2,'materials':1}}],'initial':{'gold':0,'materials':0},'needed':{'gold':27,'materials':1},'deltas':{'gold':27,'materials':1}},'jobs':['j1','j2'],'dorm':'dorm','rescue_task':'rescue','logistics_requests':['food'],'ev_requests':['ev'+str(n) for n in range(5)]} identity['logistics_requests'] = [req['request_id'] for req in REQUESTS] original = self.state(operations) self.variant(prefix,'identity',identity); self.variant(prefix,'original',original); self.variant(prefix,'after-live',original) before = {}; after = {} for member in c.MEMBERS: raw = ('synthetic '+member).encode() for package in ('original','result','backup'): self.put(prefix+package+'/formal/'+member,raw) before['/'+member] = after['/'+member] = hashlib.sha256(raw).hexdigest() before['.pre_migration.bak/'+member] = after['.pre_migration.bak/'+member] = hashlib.sha256(raw).hexdigest() self.variant(prefix,'before-members',before); self.variant(prefix,'after-members',after) checkpoint = copy.deepcopy(original) if not refusal: r = checkpoint['systems']['runtime'] r['world']['revisions'] = copy.deepcopy(REVISIONS['checkpoint_world_revisions']) for job in r['construction']['jobs']: job['reserved_cells'] = job['reserved_cells'][:4] for row in r['world']['occupancies'][:2]: row['cells'] = row['cells'][:4] events = [{'event_type':'ELEVATOR_ARRIVED','payload':{'request_id':rid}} for rid in identity['ev_requests']] for i,role in enumerate(('writer','reader')): self.variant(prefix,role+'-process',{'pid':100+2*number+i,'failures':0,'checks':1,'operations':operations,'refusal':refusal}) for label in c.LABELS: state = copy.deepcopy(checkpoint) if label == 'minute-160': state['systems']['runtime']['staff']['staff'][0]['health'] = 100 for req in state['systems']['runtime']['logistics_requests']: req['state'] = 'COMPLETED' self.variant(prefix,role+'-'+label,events if label=='events' else state) def os(self): proof = {'commit':self.index['commit'],'source_map':self.index['source_map'],'input_method':'native_os','setup':'controlled_old_save','actions':[]} for name in ('preview','cancel','confirm','changed_source_rejection'): row = {'action':name,'input_receipt':'os/'+name+'/input.json'} self.put(row['input_receipt'],{'method':'native_os','event_type':'mouse','target_pid':500,'timestamp_ns':1}) for field in ('screen_before','screen_after','model_before','model_after','formal_before','formal_after'): row[field] = 'os/'+name+'/'+field+'.json'; self.put(row[field],{'synthetic':'diagnostic'}) proof['actions'].append(row) self.put('os.json',proof) def regression(self): contracts = c.module('p08_synthetic_markers',ROOT/'tools/strict_test_markers.py').load_contracts(ROOT/c.CONTRACT) roster = self.runner.roster(ROOT); rows = [] for i,test in enumerate(roster): contract = contracts[test] def marker(template): for key in ('checks','passed'): template = template.replace('<'+key+'>','1') for key in ('failures','failed','skipped'): template = template.replace('<'+key+'>','0') return template+'\n' raw = BANNER+marker(contract['template']) for auxiliary in contract.get('auxiliary',[]): raw += marker(auxiliary['template']) if contract.get('json_report'): raw += contract['json_report']['prefix']+json.dumps({'check_summary':{'total':1,'passed':1,'failed':0,'skipped':0},'status':'PASS','runtime_errors':[],'unmet_fixture_fields':[]})+'\n' log = str(i)+'.log'; self.put('regression/'+log,raw.encode()) rows.append({'test':test,'log':log,'exit_code':0,'engine_ended':True}) self.put('regression/input.json',{'selected_source_map':self.index['source_map'],'roster':roster}) self.put('regression/ended.json',{'error':None,'cleanup_errors':[],'guard_errors':[],'phases':rows}) self.put('regression.json',{'commit':self.index['commit'],'directory':'regression','input':'regression/input.json','ended':'regression/ended.json'}) class EvidenceTests(unittest.TestCase): @classmethod def setUpClass(cls): cls.temp = tempfile.TemporaryDirectory(prefix='p08-checker-') cls.proof = Proof(Path(cls.temp.name)) cls.baseline = copy.deepcopy(cls.proof.index) @classmethod def tearDownClass(cls): cls.temp.cleanup() def test_positive_reaches_whole_check(self): self.assertEqual(c.check(ROOT,self.proof.path),[]) def test_strict_native_vectors(self): for kind,good,expected in ( ('Vector3i','Vector3i(0, -12, 6)',[0,-12,6]), ('Vector2i','Vector2i(-2, 3)',[-2,3]), ('Vector3','Vector3(1.5, -2e1, .25)',[1.5,-20,.25]), ('Vector2','Vector2(0, 1e-3)',[0,.001])): with self.subTest(kind=kind): self.assertEqual(c.decode({'type':kind,'text':good}),expected) invalid = [good.replace(kind,'WrongVector'),good+'tail',good.replace(',',',,',1), good[:good.rfind(',')]+')',good.replace(')',', 4)')] if kind.endswith('i'): invalid += [good.replace('(', '(1.0,',1), kind+'(1.0, '+', '.join(['2']*(len(expected)-1))+')'] for text in invalid: with self.subTest(text=text), self.assertRaisesRegex(ValueError,'invalid typed vector'): c.decode({'type':kind,'text':text}) def test_r04_local_deficits_through_whole_check(self): p = self.proof def mutate_state(node, change): state = c.decode(node); change(state['systems']['runtime']); return typed(state) def bad_vector(node): if node['type'] == 'Vector3i': node['text'] = 'Vector3i(0, , 1)'; return True children = ([n for pair in node['entries'] for n in pair] if node['type']=='Dictionary' else node.get('items',[])) return any(bad_vector(child) for child in children) corruptions = [ ('original','invalid typed vector',None), ('original','orphan cargo endpoint',lambda r:r['logistics_requests'][0].update(destination_inventory_id='facility.inventory:absent')), ('original','RESERVED missing/duplicate reservation',lambda r:r['inventory']['reservations'].pop()), ('original','DELIVERING carrier/cargo mismatch',lambda r:r['logistics_requests'][0].update(assigned_carrier_id='patient')), ('original','RESERVED carrier contradiction',lambda r:r['logistics_requests'][1].update(assigned_carrier_id='staff_gina')), ('original','DELIVERING reservation quantity mismatch',lambda r:r['inventory']['reservations'][0].update(quantity=3)), ('writer-checkpoint','support epoch/occupancy mismatch',lambda r:r['world']['revisions'].update(support=143)), ] for label,message,change in corruptions: path = 'pairs/off-positive/'+label+'.json' original = (p.path/path).read_bytes() try: node = json.loads(original) if change: node = mutate_state(node,change) else: self.assertTrue(bad_vector(node)) p.put(path,node); p.save() with self.subTest(message=message): self.assertTrue(any(message in deficit for deficit in c.check(ROOT,p.path)),message) finally: p.put(path,original); p.index = copy.deepcopy(self.baseline); p.save() for key in ('world','occupancy','support'): path = 'pairs/off-refusal/writer-checkpoint.json' original = (p.path/path).read_bytes() try: node = c.decode(json.loads(original)); revisions = node['systems']['runtime']['world']['revisions'] revisions[key] += 1 p.put(path,typed(node)); p.save() with self.subTest(refusal_revision=key): self.assertTrue(any('refusal revision changed' in deficit for deficit in c.check(ROOT,p.path))) finally: p.put(path,original); p.index = copy.deepcopy(self.baseline); p.save() def test_corruptions_cannot_be_accepted(self): p = self.proof mutations = { 'duplicate_actor':lambda r:r['guests']['guests'].append(copy.deepcopy(r['guests']['guests'][0])), 'orphan_occupancy':lambda r:r['world']['occupancies'][0].update(entity_id='orphan'), 'lost_patient':lambda r:r['staff']['staff'].pop(0), 'lost_cargo':lambda r:r['inventory']['inventories'].pop(), 'lost_ev':lambda r:r['elevators']['trips'].pop(), 'missing_start_stage':lambda r:r['construction']['jobs'][1].update(state='BUILDING',progress=1), } path = 'pairs/off-positive/original.json' original = (p.path/path).read_bytes() for label,mutate in mutations.items(): with self.subTest(label=label): node = c.decode(json.loads(original)); mutate(node['systems']['runtime']) p.put(path,typed(node)); p.save() self.assertTrue(c.check(ROOT,p.path),label) p.put(path,original); p.index = copy.deepcopy(self.baseline); p.save() path = 'pairs/off-positive/identity.json' original = (p.path/path).read_bytes() node = c.decode(json.loads(original)); node['diagnostic_budget']['deltas']['gold'] += 1 p.put(path,typed(node)); p.save() self.assertTrue(c.check(ROOT,p.path),'resource_delta_mismatch') p.put(path,original); p.index = copy.deepcopy(self.baseline); p.save() paths = [('partial_batch','pairs/off-refusal/after-live.json'),('unequal_cross_pid','pairs/on-positive/reader-checkpoint.json'), ('overwritten_original','pairs/off-refusal/original/formal/world.bin'),('overwritten_backup','pairs/off-refusal/backup/formal/world.bin')] for label,path in paths: with self.subTest(label=label): original = (p.path/path).read_bytes() p.put(path,b'{}'); p.save(); self.assertTrue(c.check(ROOT,p.path),label) p.put(path,original); p.index = copy.deepcopy(self.baseline); p.save() for label in ('missing_on','wrong_source','stale_source','fake_os_signal','missing_phase','duplicate_pid'): with self.subTest(label=label): changed = None if label=='missing_on': p.index['cases'] = p.index['cases'][:2] elif label=='wrong_source': p.index['source_map']['autoload/simulation_runtime.gd'] = '0'*64 elif label=='stale_source': del p.index['source_map']['tools/check_p08_migration_evidence.py'] elif label=='missing_phase': p.index['cases'][0]['phases'].pop() elif label=='fake_os_signal': changed = 'os/preview/input.json'; original = (p.path/changed).read_bytes() row = json.loads(original); row['method'] = 'signal_emitted'; p.put(changed,row) else: changed = p.index['cases'][0]['phases'][1]; original = (p.path/changed).read_bytes() row = json.loads(original); row['actual_pid'] = 100; p.put(changed,row) p.save(); self.assertTrue(c.check(ROOT,p.path),label) if changed: p.put(changed,original) p.index = copy.deepcopy(self.baseline); p.save() if __name__ == '__main__': unittest.main()