"""Synthetic checker specimens only. No native/OS execution or acceptance claim. The JSON, Variant stand-ins, executable and lifecycle records here are deliberately fabricated unit-test data, never collector output. Semantic corruption tests repin their bytes, so they cannot pass merely by detecting a stale artifact checksum. """ import copy import json import hashlib import json from pathlib import Path import shutil import struct import zlib import subprocess def put(path, data): path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(data, ensure_ascii=False, indent=2)+'\n') def typed(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(v) for v in value]} name = 'bool' if isinstance(value, bool) else 'int' if isinstance(value, int) else 'float' if isinstance(value, float) else 'Nil' if value is None else 'String' return {'type': name, 'text': json.dumps(value, ensure_ascii=False)} def owner_proof(folder, pid, checker): """Fabricated weak-reference records, only for the whole-parser control.""" directory = folder/'teardown'; audio = directory/'owned-audio' audio.mkdir(parents=True) before = {'owner_ids':[1], 'owner_meta_present':True, 'closing':False, 'clock_process':True,'clock_physics_process':False,'clock_process_mode':0, 'root_gui_disable_input':False,'auto_accept_quit':False,'original_auto_accept_quit':True} after = dict(before, owner_ids=[], owner_meta_present=False, auto_accept_quit=True) control = {'owned_engine_pid':pid,'owner_instance_id':1,'passed':True, 'maximum_wait_us':500000,'whole_Runtime_field_deletions':0, 'before':before,'after':after,'expected_remaining_owner_ids':[]} for key in ('ordinary_owner_release_verified','clock_process_flags_unchanged', 'root_gui_input_flag_unchanged','audio_teardown_passed'): control[key] = True for key in ('gameplay_expectations_changed','clock_flags_intentionally_modified','application_close_called', 'SceneTree_quit_overridden','WindowQuit_verified','formal_acceptance'): control[key] = False put(directory/'control.json', control) targets = [{'kind':kind,'role':role,'instance_id':i*4+j+1,'alive':True} for i,kind in enumerate(('stream','packet_sequence')) for j,role in enumerate(('click','success','refusal','notification'))] proof = {'owned_engine_pid':pid,'owner_instance_id':1,'passed':True,'maximum_wait_us':500000, 'waited_actual_us':100,'deadline_reached':False,'initial_targets':targets, 'final_targets':[dict(t,alive=False) for t in targets], 'full_typed_snapshots':{}} for key in ('owner_freed','all_owned_targets_released','whole_runtime_unchanged'): proof[key] = True for key in ('bus_deleted','cues_muted_for_test','actual_human_hearing','WindowQuit_verified','formal_acceptance'): proof[key] = False for part in ('before','after'): path = audio/('owner_1-'+part+'.varbytes') path.write_bytes(b'SYNTHETIC typed bytes; not engine proof') proof['full_typed_snapshots'][part] = {'path':str(path),'field_deletions':0,'bytes':path.stat().st_size,'sha256':checker.sha(path)} put(audio/'owner_1.json', proof) owners = checker.module('synthetic_owner_parser', checker.HERE/'regression/ordinary_proof_gate.py') return owners.ordinary_gate(directory,pid,folder,['single']) def make(base, source, checker): root, out = base/'source', base/'p08-normal-placement-evidence' for folder in ('autoload', 'scripts', 'scenes', 'data', 'tests', 'assets', 'tools'): (root/folder).mkdir(parents=True, exist_ok=True) for name in ('project.godot', 'export_presets.cfg', 'Makefile', 'AGENTS.md', checker.SPEC, checker.CONTRACT, checker.TEST, checker.RECORDER, checker.RECORDER+".uid", "tools/run_p08_normal_placement.py", 'docs/plans/original-completion-2026-10-02/IMPLEMENTATION_PLAN.md', 'docs/traceability/original/BASELINE.txt', 'data/facilities/reception.tres', 'data/facilities/standard_room.tres', 'data/facilities/infrastructure/guest_stairs.tres'): target = root/name target.parent.mkdir(parents=True, exist_ok=True) shutil.copyfile(source/name, target) (root/'tests/fixtures/p08-03').mkdir(parents=True) out.mkdir() engine = base/'synthetic-engine-not-executable' engine.write_text('SYNTHETIC TEST DATA; NEVER EXECUTE\n') sha = checker.sha source_map = {n: v['sha256'] for n, v in checker.runner.source_refs(root).items()} index = {'schema': 'p08-normal-placement-v1', 'commit': subprocess.check_output(['git','-C',str(source),'rev-parse','HEAD'],text=True).strip(), 'source_map': source_map, 'inputs': checker.input_map(root), 'spec_sha256': checker.SPEC_SHA, 'marker_contract_sha256': sha(root/checker.CONTRACT), 'conditions': checker.CONDITIONS, 'collection_error': None, 'acceptance': 'pending', 'regression': None} put(out/'runner-input.json', {'selected_source_map': source_map, 'commit': index['commit'], 'engine': str(engine), 'engine_sha256': sha(engine)}) put(out/'runner-ended.json', {'error': None, 'cleanup_errors': [], 'synthetic': True}) def phase(label, pid): prefix = 'runs/'+label+'/' command = [str(engine), '--headless', '--path', str(root), '--script', label] raw = {'actual_pid': pid, 'command': command, 'live_identity_exact': True, 'engine_ended': True, 'exit_code': 0, 'synthetic': True} for name in ('started.json', 'ended.json'): put(out/(prefix+label+'/'+name), raw) put(out/(prefix+label+'/outer-owned-finally.json'), {'actual_captured_pids': [pid], 'all_captured_ended': True, 'outer_cleanup_errors': [], 'foreign_process_or_group_signals': 0}) (out/(prefix+'stdout.log')).write_text('Godot Engine v4.7.2.stable.official.ed1daf0bf - https://godotengine.org\nP08 normal placement: 100 checks, 0 failures\n') receipt = {'actual_pid': pid, 'command': command, 'phase': label, 'test': checker.TEST, 'source_sha256': source_map[checker.TEST], 'log': 'stdout.log', 'exit_code': 0, 'engine_ended': True, 'failed': False, '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': label, 'expected_user_data_dir': label}}} name = prefix+label+'-receipt.json' put(out/name, receipt) return name pair_phases = [phase('writer', 101), phase('reader', 102)] os_phases = [] index['cases'] = [{'directory': 'pair', 'phases': pair_phases}] index['phases'] = pair_phases index['os_proof'] = 'os.json' put(out/'os.json', {'source_map': source_map, 'inputs': index['inputs'], 'input_method': 'native_os', 'observer_mutations': [], 'phases': os_phases, 'directory': 'os-pair', 'synthetic': True}) def pair(directory, pids, method): folder = out/directory folder.mkdir() runtime = {k: {} for k in sorted(checker.RUNTIME_KEYS)} runtime.update(world={'occupancies': [], 'bounds':{'min_x':-10,'max_y':10}}, construction={'facilities': [], 'jobs': []}, id_generator={'version':1,'counters':{'domain_event':7}}, domain_events={'version':1,'published_ids':[],'recent_events':[]}, facilities={'instances': []}, next_command_id=1, command_queue={'processed': [], 'pending': [], 'in_flight': [], 'accepted': [], 'idempotency': {}}) model = {'schema_version': 1, 'clock': {'absolute_minute': 360, 'paused': True}, 'game_state': {'resources': {'gold': 5000, 'materials': 50}}, 'systems': {'runtime': runtime}} runtime['domain_events']['published_ids'] = ['domain_event.%012d' % n for n in range(1,8)] runtime['domain_events']['recent_events'] = [ {'event_id':'domain_event.%012d' % n, 'event_type':'SYNTHETIC_HISTORY', 'absolute_minute':360, 'source_service':'synthetic', 'entity_ids':[], 'payload':{'ordinal':n}, 'causation_command_id':0} for n in range(1,8)] target = [-8, 0, 0] cost = {'reception': {'gold': 500, 'materials': 8}, 'guest_stairs': {'gold': 40, 'materials': 0}, 'standard_room':{'gold':350,'materials':5}} rooms = 0 def snap(name): # Synthetic controls follow the same split as Runtime: stairs are # construction instances with infrastructure, never operational rooms. projections = [] entities = []; connectors = [] for f in runtime['construction']['facilities']: if f['definition_id'] != 'guest_stairs': continue f.update(rotation=0, layer=0, completed_at=360) fid = f['instance_id']; origin = f['origin'] p = {k:copy.deepcopy(f[k]) for k in ('instance_id','definition_id','origin','source_job_id','rotation','layer','completed_at')} p.update(kind='guest_stairs', route_entity_id='infra.route:'+fid, connector_id='infra.connector:'+fid, cells=f['occupied_cells'], staff_only=False, pass_through=True, footprint=[1,1], definition_tags=['guest_access','movement','route_subslot','schema_facility_proxy','stairs','vertical_connector']) canonical = 'projection|'+''.join('%d:%s|' % (len(p[k]),p[k]) for k in ('instance_id','definition_id','kind','route_entity_id','connector_id','source_job_id')) canonical += 'layer=0|staff=0|pass=1|footprint=1,1|rotation=0|completed=360|' canonical += 'origin=%d,%d,%d|tags=' % tuple(origin) canonical += ''.join('%d:%s;' % (len(t),t) for t in p['definition_tags'])+'|cells=' canonical += ''.join('%d,%d,%d;' % tuple(c) for c in p['cells']) p['record_fingerprint'] = hashlib.sha256(canonical.encode()).hexdigest() projections.append(p) entities.append({'entity_id':p['route_entity_id'],'category':'ROUTE','cells':p['cells']}) connectors.append({'id':p['connector_id'],'from':origin,'to':[origin[0],origin[1]-1,origin[2]], 'cost':150,'bidirectional':True,'access_id':'','enabled':True,'movement_type':'CONNECTOR', 'dynamic_cost':0,'operational_state':'OPERATIONAL','cabin_capacity':8,'queue_capacity':256, 'cabin_state':'IDLE','current_floor':origin[1],'travel_minutes_per_floor':1,'request_queue':[]}) runtime['facility_infrastructure'] = {'projections':projections} runtime['spatial_subslots'] = {'entities':entities} route_cells = [c for p in projections for c in p['cells']] walkable = [[4,0,0]]+route_cells runtime['route_projection'] = {'base_walkable_cells':[[4,0,0]],'projected_route_cells':route_cells,'projected_walkable_cells':walkable} runtime['path'] = {'connectors':connectors,'vertical_walk_blockers':[], 'cell_traffic':[{'cell':c} for c in walkable]} put(folder/(name+'.json'), typed(model)) (folder/(name+'.var')).write_bytes(json.dumps(model, sort_keys=True).encode()) return name def position(cell): return {'world_cell': cell[:], 'selected_cell': cell[:], 'bounds': {'min_x': -10, 'max_y': 10}, 'map_position': [0, 0], 'cell_dimensions': [10, 10], 'local': [(cell[0]+10.5)*10, (10.5-cell[1])*10], 'viewport_transform':[1,0,0,1,0,0], 'screen_transform':[1,0,0,1,0,0], 'screen':[(cell[0]+10.5)*10, (10.5-cell[1])*10], 'viewport': [(cell[0]+10.5)*10, (10.5-cell[1])*10], 'zoom': 1.0, 'pan': [0, 0]} def result(code, details=None, message=''): row = {'code': code, 'details': details or {}, 'message': message, 'command_id': runtime['next_command_id'], 'target_id': '', 'is_replay': False, 'resolution_hints': []} runtime['next_command_id'] += 1 runtime['command_queue']['processed'].append(copy.deepcopy(row)) return row os_sequences = {} def os_meta(row, pid): row['input_method'] = method if method != 'native_os': return p = row['position'] p['dispatch'] = {k:copy.deepcopy(v) for k,v in p.items() if k != 'selected_cell'} occupant = next((o['entity_id'] for o in runtime['world']['occupancies'] if p['world_cell'] in o['cells']), '') p['selection'] = {'required':row['id'] == 'pan_zoom', 'entity_id':occupant, 'occupant_id':occupant} p['diagnostics'] = {'getter_cell':[]} requests = [{'action':'cell', 'world_cell':row['position']['world_cell'], 'viewport':row['position']['viewport'], 'screen':row['position']['screen'], 'viewport_to_screen':[1,0,0,1,0,0], 'guide_visible':False, 'guide_rect':[0,0,0,0]}] if row['id'] in ('unsupported','resume_overlap'): requests.append({'action':'load_menu'}) if row['id'] == 'pan_zoom': requests.append({'action':'button','node':'/root/Main/ManualSaveButton'}) requests += [{'action':'wheel','ctrl':False,'move_mode':True}] requests += [{'action':'wheel','ctrl':False,'move_mode':False} for _ in range(18)] if row['id'] == 'supported_connected': geometry = {'target_rect':[120,0,20,20], 'target_screen_rect':[120,0,20,20], 'palette_rect':[0,0,100,30], 'thumb_rect':[0,20,50,10], 'viewport_to_screen':[1,0,0,1,0,0], 'bar':{'rect':[0,20,100,10], 'visible':True, 'min_value':0, 'max_value':200, 'page':100, 'value':0}} requests = [dict(geometry, action='reveal_palette', node='/root/Main/Palette/Target', direction='right', scroll_before=0), {'action':'button','node':'/root/Main/Palette/Target'}] + requests row['requests'] = [] for local_sequence, request in enumerate(requests, 1): sequence = os_sequences.get(pid, 0)+1 os_sequences[pid] = sequence prefix = directory+'/'+row['id']+'-'+str(local_sequence)+'/' request.update(target_pid=pid, sequence=sequence) put(out/(prefix+'request.json'), request) if request['action'] == 'reveal_palette': after = copy.deepcopy(geometry) after.update(target_rect=[60,0,20,20], target_screen_rect=[60,0,20,20], thumb_rect=[30,20,50,10], scroll_after=60, visible_after=True) after['bar']['value'] = 60 put(out/(prefix+'reveal.json'), after) events = [{'synthetic':True}] if request['action'] == 'cell': events = [{'synthetic':True, 'button':1, 'pressed':True, 'position':p['dispatch']['viewport'], 'dispatch':copy.deepcopy(p['dispatch'])}] put(out/(prefix+'observed.json'), {'pid':pid,'sequence':sequence,'events':events}) receipt = directory+'/'+row['id']+'-os.json' put(out/receipt, {'method':'native_os','target_pid':pid,'timestamp_ns':10000, 'collector_pid':999,'tool_name':'mock CUA','tool_call_id':'synthetic-only', 'events':[{'synthetic':True}]}) put(out/(prefix+'ack.json'), {'target_pid':pid,'sequence':sequence, 'request_sha256':sha(out/(prefix+'request.json')), 'input_receipt':receipt, 'screen_before':prefix+'screen-before.png','screen_after':prefix+'screen-after.png'}) row['requests'].append(prefix+'ack.json') def chunk(kind, body): return struct.pack('>I',len(body))+kind+body+struct.pack('>I',zlib.crc32(kind+body)) screenshot = (b'\x89PNG\r\n\x1a\n'+ chunk(b'IHDR',struct.pack('>IIBBBBB',1,1,8,2,0,0,0))+ chunk(b'IDAT',zlib.compress(b'\0\0\0\0'))+chunk(b'IEND',b'')) for ack_name in row['requests']: for side in ('before', 'after'): (out/Path(ack_name).parent/('screen-'+side+'.png')).write_bytes(screenshot) actions = [] for label in checker.WRITER: cell = [4, 1, 0] if label in ('lower_stair', 'refused_demolition') else [4, 2, 0] if label == 'upper_stair' else target[:] definition = 'guest_stairs' if label in ('lower_stair', 'upper_stair', 'missing_connection') else 'reception' if label == 'unsupported': definition = 'standard_room' row = {'id': label, 'position': position(cell), 'expected': '', 'results': [], 'status': '', 'kind': 'build', 'definition': definition, 'before': snap('writer-'+label+'-before')} report = {'can_build': True, 'error_codes': [], 'total_cost': cost[definition], 'connection': {'status': 'CONNECTED'}} row['report'] = report if label in checker.REASONS: code, message = checker.REASONS[label] report.update(can_build=False, error_codes=[code]) row.update(expected=code, status=message) elif label == 'missing_connection': row['expected'] = 'STAIR_LANDING_MISSING' report.update(can_build=False, error_codes=[row['expected']], reasons=[{'code':row['expected'], 'message':'先に階段の一つ下の階に通路を作ってください。', 'cells':[], 'details':{'reason':row['expected']}}]) row['status'] = '配置不可: '+report['reasons'][0]['message'] elif label == 'cancel_demolition': row.update(kind='cancel', status='撤去を取り消しました。') elif label in ('demolition', 'refused_demolition'): fid = 'f1' if label == 'demolition' else 'f2' row.update(kind='demolish', instance_id=fid) if label == 'refused_demolition': row['expected'] = 'STAIR_LANDING_IN_USE' row['results'] = [result('VALIDATION_FAILED', {'reason': row['expected']}, 'dependent stair')] else: row['results'] = [result('OK', {'refund': {'materials': 2}, 'instance_id': fid})] model['game_state']['resources']['materials'] += 2 runtime['construction']['facilities'] = [v for v in runtime['construction']['facilities'] if v['instance_id'] != fid] runtime['construction']['jobs'] = [v for v in runtime['construction']['jobs'] if v.get('facility_instance_id') != fid] runtime['facilities']['instances'] = [v for v in runtime['facilities']['instances'] if v['instance_id'] != fid] runtime['world']['occupancies'] = [v for v in runtime['world']['occupancies'] if v['entity_id'] != fid] r = row['results'][0] row['status'] = '撤去できません: '+r['message'] if row['expected'] else '撤去しました（返金: 資材 2）' elif label == 'pan_zoom': row.update(kind='camera', original_local=[1, 2], ctrl_pressed=False, pan_before=[0,0], pan_after=[0,10], steps=[0.5]*9+[1, 1.5, 2, 2.5, 3, 3.5, 4, 4, 4]) else: rooms += 1 fid, jid = 'f'+str(rooms), 'j'+str(rooms) row['results'] = [result('OK', {'job_id': jid})] for k, v in cost[definition].items(): model['game_state']['resources'][k] -= v cells = [[cell[0]+dx,cell[1]+dy,cell[2]] for dx in range(4 if definition == 'reception' else 1) for dy in range(2 if definition == 'reception' else 1)] job = {'job_id': jid, 'state': 'PLANNED', 'reserved_cells': cells} runtime['construction']['jobs'].append(job) runtime['world']['occupancies'].append({'entity_id': jid, 'cells': cells}) row['after'] = snap('writer-'+label+'-after') job.update(state='COMPLETED', facility_instance_id=fid) facility = {'instance_id': fid, 'definition_id': definition, 'occupied_cells': cells, 'origin': cell[:], 'source_job_id': jid} runtime['construction']['facilities'].append(facility) if definition != 'guest_stairs': runtime['facilities']['instances'].append(copy.deepcopy(facility)) runtime['world']['occupancies'][-1]['entity_id'] = fid row['completed'] = snap('writer-'+label+'-completed') if 'after' not in row: row['after'] = snap('writer-'+label+'-after') os_meta(row, pids[0]) actions.append(row) budget = {'setup': 'diagnostic', 'costs': [{'definition': d, 'cost': cost[d]} for d in ('reception', 'reception', 'guest_stairs', 'guest_stairs')], 'needed': {'gold': 1080, 'materials': 16}, 'initial': {'gold': 5000, 'materials': 50}, 'deltas': {}} put(folder/'writer-actions.json', {'setup': 'diagnostic', 'actions': actions, 'budget': budget}) snap('writer-presave') snap('writer-formal-model') snap('reader-formal-model') formal_model = typed(model) formal = folder/'formal' formal.mkdir() for member in checker.MEMBERS: (formal/member).write_text('synthetic '+member+'\n') shutil.copytree(formal, folder/'formal.bak') shutil.copyfile(formal/'manifest.json', folder/'formal-slot-metadata.json') profile = '/synthetic-private-profile' absolute = profile+'/saves/manual_ui' put(folder/'formal-export.json', {'pid':pids[0], 'requested_path':'user://saves/manual_ui', 'absolute_path':absolute, 'user_data_dir':profile, 'packages':{name:{'present':True, 'source_path':absolute+suffix, 'members':{n:{'sha256':sha(folder/name/n),'bytes':(folder/name/n).stat().st_size} for n in checker.MEMBERS}} for name,suffix in [('formal',''),('formal.bak','.bak')]}, 'slot_metadata':{'file':'formal-slot-metadata.json','sha256':sha(formal/'manifest.json')}}) for role,pid,op in [('writer',pids[0],'save'),('reader',pids[1],'recovery')]: returned = {'ok':True,'operation':op,'save_path':'user://saves/manual_ui', 'code':'OK','message':'synthetic result','duration_us':13,'runtime_duration_us':17, 'manifest':{'save_id':'synthetic-save','absolute_minute':360}} if role == 'reader': returned['selected'] = {'id':'formal','kind':'formal','generation':0} expected = checker.expected_operation(formal_model,returned, 'SAVE_CHECKPOINT_SUCCEEDED' if role == 'writer' else 'SAVE_RECOVERY_LOADED') for label in ('checkpoint','expected-operation','operation-actual'): put(folder/(role+'-'+label+'.json'),expected) (folder/(role+'-'+label+'.var')).write_bytes(json.dumps(expected,sort_keys=True).encode()) count = len(model['systems']['runtime']['domain_events']['recent_events']) put(folder/(role+'-operation-comparison.json'), {'pid':pid,'delta_kind':'source_defined_notification', 'formal_model':role+'-formal-model','actual_model':role+'-checkpoint', 'expected_model':role+'-expected-operation','returned_result':typed(returned), 'saved_recent_count':count,'expected_recent_count':count+1,'actual_recent_count':count+1, 'whole_raw_equal':True}) stem = role+'-save-result-0001' (folder/(stem+'.var')).write_bytes(b'synthetic raw Dictionary stand-in') put(folder/(stem+'.json'), {'pid':pid,'operation':'runtime_operation', 'requested_path':'user://saves/manual_ui','absolute_path':absolute,'user_data_dir':profile, 'result':typed(returned), 'raw_sha256':sha(folder/(stem+'.var'))}) inspect_stem = role+'-save-result-0002' (folder/(inspect_stem+'.var')).write_bytes(b'synthetic inspection Dictionary stand-in') put(folder/(inspect_stem+'.json'), {'pid':pid,'operation':'formal_inspect', 'requested_path':'user://saves/manual_ui','absolute_path':absolute,'user_data_dir':profile, 'result':typed({'ok':True,'read_only':True,'snapshot':copy.deepcopy(model)}), 'raw_sha256':sha(folder/(inspect_stem+'.var'))}) put(folder/(role+'-save-results.json'), {'pid':pid,'records':[stem,inspect_stem]}) if method == "native_os": prefix = "writer-bootstrap-" returned = {"ok":True,"operation":"recovery","code":"OK","save_path":"user://saves/manual_ui"} expected = checker.expected_operation(formal_model,returned,"SAVE_RECOVERY_LOADED") for label,value in [("formal-model",formal_model),("returned-result",typed(returned)), ("checkpoint",expected),("expected-operation",expected),("operation-actual",expected)]: put(folder/(prefix+label+".json"),value) (folder/(prefix+label+".var")).write_bytes(json.dumps(value,sort_keys=True).encode()) records = [] for suffix,operation,value in [ ("0000-bootstrap","runtime_operation",returned), ("0000-inspect","initial_inspect",{"ok":True,"read_only":True,"snapshot":copy.deepcopy(model)})]: stem = "writer-save-result-"+suffix (folder/(stem+".var")).write_bytes(b"synthetic bootstrap Dictionary") put(folder/(stem+".json"), {"pid":pids[0],"operation":operation, "requested_path":"user://saves/manual_ui","absolute_path":absolute,"user_data_dir":profile, "result":typed(value),"raw_sha256":sha(folder/(stem+".var"))}) records.append(stem) log = json.loads((folder/"writer-save-results.json").read_text()) log["records"] = records+log["records"] put(folder/"writer-save-results.json",log) count = len(model["systems"]["runtime"]["domain_events"]["recent_events"]) put(folder/(prefix+"operation-comparison.json"), { "pid":pids[0],"delta_kind":"source_defined_notification", "formal_model":prefix+"formal-model","actual_model":prefix+"checkpoint", "expected_model":prefix+"expected-operation","returned_result":typed(returned), "saved_recent_count":count,"expected_recent_count":count+1,"actual_recent_count":count+1, "whole_raw_equal":True}) put(folder/'save.json', {'members': {n: sha(formal/n) for n in checker.MEMBERS}, 'pid': pids[0], 'cell': target, 'instance_id': 'f4'}) row = {'id': 'resume_overlap', 'kind': 'build', 'position': position(target), 'definition':'reception', 'before': snap('reader-resume-before'), 'after': snap('reader-resume-after'), 'results': [], 'report': {'can_build': False, 'error_codes': ['OVERLAP']}, 'status': checker.REASONS['overlap'][1]} os_meta(row, pids[1]) put(folder/'reader-actions.json', {'setup': 'diagnostic', 'actions': [row]}) if method == 'native_os': put(folder/'writer-requests.json', {'pid':pids[0], 'requests':[name for action in actions for name in action['requests']]}) put(folder/'reader-requests.json', {'pid':pids[1],'requests':row['requests']}) for role, pid in zip(('writer', 'reader'), pids): put(folder/(role+'-process.json'), {'pid': pid, 'engine': {'hash': 'synthetic', 'string': 'synthetic'}, 'checks': 100, 'failures': 0}) pair('pair', [101, 102], 'handler') pair('os-pair', [201, 202], 'native_os') shutil.copytree(out/'pair/formal', out/'pair/initial-formal') initial = copy.deepcopy(json.loads((out/'pair/formal-export.json').read_text())) initial.update(requested_path='user://saves/p08_initial',absolute_path='/synthetic-private-profile/saves/p08_initial') initial['packages'] = {'initial-formal':initial['packages']['formal'], 'initial-formal.bak':{'present':False,'source_path':initial['absolute_path']+'.bak','members':{}}} initial['packages']['initial-formal']['source_path'] = initial['absolute_path'] shutil.copyfile(out/'pair/initial-formal/manifest.json',out/'pair/initial-formal-slot-metadata.json') initial['slot_metadata']['file'] = 'initial-formal-slot-metadata.json' put(out/'pair/initial-formal-export.json',initial) for role, pid in (('writer',201),('reader',202)): prefix = 'os-runs/'+role+'/' folder = out/prefix folder.mkdir(parents=True) shutil.copyfile(source/checker.RECORDER, folder/'passive_os.gd') profile = base/('synthetic-profile-'+role) command = [str(engine),'--path',str(root),'--script',str(folder/'passive_os.gd')] formal = out/('pair/initial-formal' if role == 'writer' else 'os-pair/formal') members = {n:{'bytes':(formal/n).stat().st_size,'sha256':sha(formal/n)} for n in checker.MEMBERS} record = {'protocol':'p08-placement-native-os-v1','role':role,'commit':index['commit'], 'source_map':source_map,'stage':str(root),'profile':str(profile),'command':command, 'engine_sha256':sha(engine),'controller_sha256':source_map['tools/run_p08_normal_placement.py'], 'recorder_sha256':sha(source/checker.RECORDER),'formal_input':members, 'actual_pid':pid,'monotonic_ns':1,'exit_code':0,'engine_ended':True,'passed':True, 'error':None,'guard_errors':[],'cleanup_errors':[], 'owned_cleanup':[{'pid':pid,'engine_ended':True,'outer_cleanup_errors':[],'foreign_process_or_group_signals':0}], 'stage_before':{},'stage_after':{},'default_before':{},'default_after':{}, 'source_after':{n:{'sha256':v} for n,v in source_map.items()}, 'profile_before':{'existed':False,'files':{}},'ordinary_owner_proof':owner_proof(folder,pid,checker)} for name in ('started.json','process-started.json','ended.json'): put(folder/name, record) (folder/'engine-native.log').write_text('P08 passive OS placement: 100 checks, 0 failures\n') put(out/('os-pair/'+role+'-profile.json'), {'pid':pid,'user_data_dir':str(profile),'display':'synthetic'}) os_phases.append(prefix+'ended.json') put(out/'os.json', {'source_map':source_map,'inputs':index['inputs'],'input_method':'native_os', 'observer_mutations':[],'phases':os_phases,'directory':'os-pair','synthetic':True}) put(out/'index.json', index) repin(out) return root, out def repin(out): index = json.loads((out/'index.json').read_text()) index['artifacts'] = {p.relative_to(out).as_posix(): {'bytes': p.stat().st_size, 'sha256': hashlib.sha256(p.read_bytes()).hexdigest()} for p in out.rglob('*') if p.is_file() and p != out/'index.json'} put(out/'index.json', index)