Something went wrong. Try again.
Native control, inspection, UI automation and evidence capture for a running Witcher 3 process.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409"""Optional bounded host presentation-call timing. No per-frame bridge traffic."""import contextlibimport csvimport fcntlimport hashlibimport jsonimport mathimport mmapimport osfrom pathlib import Pathimport secretsimport structimport subprocessimport sysimport time
from . import supportfrom . import native
ROOT = Path(__file__).resolve().parent.parentACTIVE = support.STATE_ROOT / 'performance-active.json'LAUNCH = support.STATE_ROOT / 'performance-launch.json'LAYER = ROOT / 'presentation/build'HEADER = struct.Struct('<16Q')RECORD = struct.Struct('<6QiIQ')MAGIC = 0x31544e4553455250CAPACITY = 65536MARKER_LIMIT = 2*1024*1024SEMANTICS = 'Host CLOCK_MONOTONIC intervals between vkQueuePresentKHR entries, at this layer in the Proton Vulkan chain; not GPU completion, compositor/display timing or input latency.'
def digest(p): return hashlib.sha256(Path(p).read_bytes()).hexdigest()def write(p,v): support.atomic_json(Path(p),v)def read(p): return json.loads(Path(p).read_text())
@contextlib.contextmanagerdef lock(): with (support.STATE_ROOT/'.performance.lock').open('a') as f: fcntl.flock(f,fcntl.LOCK_EX) yield
def validate_number(value,low,high,name): if not math.isfinite(value) or not low <= value <= high:raise ValueError(f'{name} must be finite, {low}..{high}')
def launch(proton): """Temporary app-scoped Proton environment, with exact-byte rollback journal.""" from . import lifecycle lifecycle.verify();support.manage.ensure_game_closed() if LAUNCH.exists() and not read(LAUNCH).get('restored'):raise RuntimeError('Capture launch configuration exists; run perf-restore first') receipt=read(LAYER/'receipt.json') if digest(LAYER/'libPathPresent.so')!=receipt['library_sha256'] or digest(ROOT/'presentation/layer.cpp')!=receipt['source_sha256']: raise RuntimeError('Presentation library/source differs from its build receipt; rebuild with presentation/build.py') proton=Path(proton).resolve() if not (proton/'proton').is_file():raise ValueError('Expected the active Steam Proton directory containing proton') target=proton/'user_settings.py' if target.is_symlink():raise RuntimeError('Refusing symlinked Proton user_settings.py') before=target.read_bytes() if target.exists() else None folder=support.ARTIFACTS/('performance-launch-'+str(time.time_ns()));folder.mkdir() buffers=folder/'buffers';buffers.mkdir() env={'VK_LAYER_PATH':str(LAYER),'VK_INSTANCE_LAYERS':'VK_LAYER_PATHDEVKIT_present','PATHDEVKIT_PRESENT_DIR':str(buffers)} # Proton does not override an existing environment variable. Runtime discovery # verifies that this layer actually loaded, rather than treating launch as capture. addition='\n# PathDevkit temporary presentation capture\nimport os as _pd_os\nif _pd_os.environ.get("SteamAppId") == "292030":\n user_settings.update('+repr(env)+')\n' installed=(before or b'user_settings = {}\n')+addition.encode() if before is not None:(folder/'user_settings.before').write_bytes(before) (folder/'user_settings.installed').write_bytes(installed) journal=dict(path=str(target),folder=str(folder),before_exists=before is not None,installed_sha256=hashlib.sha256(installed).hexdigest(),env=env,restored=False) write(LAUNCH,journal);target.write_bytes(installed) try: from . import game_settings launched=game_settings.launch() except Exception: restore();raise return dict(ok=True,state='launch_requested',lab_mode=launched['lab_mode'],journal=str(LAUNCH),next='Wait for the real window; attach; perf-status must confirm a buffer owned by the game process. stop restores personal settings and capture configuration after exit.')
def restore(): with lock(): if not LAUNCH.exists():return dict(ok=True,restored=True,reason='no_owned_configuration') j=read(LAUNCH) if j.get('restored'):return dict(ok=True,restored=True,journal=str(LAUNCH)) p=Path(j['path']) if not p.exists() or digest(p)!=j['installed_sha256']:raise RuntimeError('Capture Proton configuration changed externally; refusing overwrite. Inspect performance-launch.json') if j['before_exists']:p.write_bytes((Path(j['folder'])/'user_settings.before').read_bytes()) else:p.unlink() j['restored']=True;j['restored_at']=support.now();write(LAUNCH,j);write(Path(j['folder'])/'cleanup.json',j) return dict(ok=True,restored=True,journal=str(LAUNCH),settings_bytes_restored=True)
def capability(): result=dict(ok=False,available=False,provider='PathPresent',version=1,semantics=SEMANTICS,clock='CLOCK_MONOTONIC',units='nanoseconds',capacity=CAPACITY,attribution={'gpu_time':False,'script_cpu_time':False,'displayed_frames':False}) try: identity=support.process();pid=identity['pid'];proc=Path('/proc')/str(pid) env=dict(x.split(b'=',1) for x in (proc/'environ').read_bytes().split(b'\0') if b'=' in x) directory=env.get(b'PATHDEVKIT_PRESENT_DIR') if not directory:raise RuntimeError('Game launched without capture environment; close it and use perf-launch') source=Path(os.fsdecode(directory))/f'{pid}.bin' maps=(proc/'maps').read_text() if str(LAYER/'libPathPresent.so') not in maps:raise RuntimeError('Presentation layer not mapped in game process') if not source.exists():raise RuntimeError('Layer mapped but no presentation buffer yet; wait for the real game window') header=HEADER.unpack(source.open('rb').read(HEADER.size)) if header[:5]!=(MAGIC,1,pid,CAPACITY,RECORD.size):raise RuntimeError('Invalid buffer identity/layout') same_clock=os.readlink(proc/'ns/time')==os.readlink('/proc/self/ns/time') if not same_clock:raise RuntimeError('Game and collector use different time namespaces; clock alignment unqualified') receipt=read(LAYER/'receipt.json') if digest(LAYER/'libPathPresent.so')!=receipt['library_sha256'] or digest(ROOT/'presentation/layer.cpp')!=receipt['source_sha256']:raise RuntimeError('Provider build receipt does not match source/library') if not any(int(line.split()[4])==(LAYER/'libPathPresent.so').stat().st_ino for line in maps.splitlines() if str(LAYER/'libPathPresent.so') in line):raise RuntimeError('Mapped provider inode differs from current library') result.update(ok=True,available=True,process=identity,buffer=str(source),presents_seen=header[11],active=bool(header[5]),same_time_namespace=True,clock_resolution_ns=int(time.get_clock_info("monotonic").resolution*1e9),alignment_offset_ns=0,measurement_uncertainty_ns=None,clock_alignment_note='Same kernel CLOCK_MONOTONIC domain; timestamps stored as integer ns. 1 ns is representation resolution, not clock accuracy or observer-effect bound.',build=receipt,runtime_libraries=sorted(set(line.split()[-1] for line in maps.splitlines() if any(n in line.lower() for n in ('vkd3d','winevulkan','libvulkan','libpathpresent'))))) except (OSError,RuntimeError,ValueError) as e:result['reason']=str(e) return result
def marker(kind,details=None): """Called by host control path only. Never queries or mutates the game.""" if not ACTIVE.exists():return state=read(ACTIVE) if state.get('finished'):return item=dict(monotonic_ns=time.monotonic_ns(),kind=kind,details=details or {}) path=Path(state['folder'])/'markers.jsonl' encoded=(json.dumps(item,separators=(',',':'))+'\n').encode() with path.open('ab') as f: fcntl.flock(f,fcntl.LOCK_EX) size=path.stat().st_size if size+len(encoded)>MARKER_LIMIT: write(path.parent/'marker-overflow.json',dict(limit_bytes=MARKER_LIMIT,retained_bytes=size,rejected_bytes=len(encoded),kind=str(kind)[:128])) raise RuntimeError('Performance marker buffer limit reached; perf-stop will retain a failed partial report') f.write(encoded)
def retained_markers(folder): """Read a bounded complete-line prefix, even for an externally overfilled log.""" path=folder/'markers.jsonl' with path.open('rb') as f: fcntl.flock(f,fcntl.LOCK_SH) size=os.fstat(f.fileno()).st_size;prefix=f.read(MARKER_LIMIT) complete=prefix[:prefix.rfind(b'\n')+1] markers=[];invalid=0 for line in complete.splitlines(): try: item=json.loads(line) if not isinstance(item,dict) or not isinstance(item.get('monotonic_ns'),int) or not isinstance(item.get('kind'),str) or not isinstance(item.get('details'),dict):raise ValueError('Invalid marker') markers.append(item) except (ValueError,UnicodeError):invalid+=1 return markers,dict(limit_bytes=MARKER_LIMIT,file_bytes=size,parsed_prefix_bytes=len(complete),omitted_bytes=size-len(complete),invalid_lines=invalid,overflow_recorded=(folder/'marker-overflow.json').exists())
def start(seconds,label,thresholds,budget_ms=None): validate_number(seconds,.1,120,'seconds') if not label or len(label)>128:raise ValueError('arm label required, <=128 characters') if not thresholds:raise ValueError('At least one explicit --threshold-ms is required') for t in thresholds:validate_number(t,.01,10000,'threshold-ms') if budget_ms is not None:validate_number(budget_ms,.01,10000,'budget-ms') with lock(): if ACTIVE.exists() and not read(ACTIVE).get('finished'):raise RuntimeError('Capture already exists; perf-stop preserves it before another start') cap=capability() if not cap['available']:raise RuntimeError(cap['reason']) session=native.state();snapshot=native.call('snapshot') if cap['process']!=session['process']:raise RuntimeError('Capture process differs from attached game') folder=support.ARTIFACTS/('performance-'+str(time.time_ns()));folder.mkdir() (folder/'markers.jsonl').touch() graphics={n:dict(sha256=digest(support.manage.DOCS/n),file=n) for n in ('user.settings','dx12user.settings')} for n in graphics:(folder/n).write_bytes((support.manage.DOCS/n).read_bytes()) state=dict(version=1,folder=str(folder),buffer=cap['buffer'],capability=cap,session=session['session'],generation=snapshot['generation'],bridge_build=session['build'],loaded_fixture=session.get('loaded_fixture'),process=cap['process'],label=label,thresholds_ms=sorted(set(thresholds)),budget_ms=budget_ms,seconds=seconds,snapshot_before=snapshot,graphics=graphics,finished=False,epoch=secrets.randbits(63),created=support.now(),collector_sources={n:digest(ROOT/'witcher_devkit'/n) for n in ('performance.py','native.py','cli_app.py','catalog.py')}) state['clock_identity']=dict(boot_id=Path('/proc/sys/kernel/random/boot_id').read_text().strip(),time_namespace=os.readlink('/proc/self/ns/time')) with open(cap['buffer'],'r+b') as f, mmap.mmap(f.fileno(),0) as mem: struct.pack_into('<Q',mem,40,0) if struct.unpack_from('<Q',mem,80)[0]:raise RuntimeError('Presentation still in flight; stop capture and retry after it returns') begin=time.monotonic_ns();state['begin_ns']=begin;state['deadline_ns']=begin+int(seconds*1e9) for offset,val in ((48,0),(56,0),(64,state['deadline_ns']),(72,state['epoch'])):struct.pack_into('<Q',mem,offset,val) write(folder/'capture.json',state);write(ACTIVE,state) struct.pack_into('<Q',mem,40,1) marker('capture_start',dict(label=label,snapshot=snapshot)) with (folder/'watch.log').open('w') as output: child=subprocess.Popen([sys.executable,'-m','witcher_devkit.performance','watch',str(folder)],cwd=ROOT,stdin=subprocess.DEVNULL,stdout=output,stderr=output,start_new_session=True) state['watch_pid']=child.pid;write(ACTIVE,state);write(folder/'capture.json',state) return dict(ok=True,folder=str(folder),deadline_ns=state['deadline_ns'],session=state['session'],generation=state['generation'],semantics=SEMANTICS)
def percentile(values,p): ordered=sorted(values);return ordered[max(0,math.ceil(p*len(ordered))-1)] if ordered else None
def summarize(rows,thresholds,budget_ms=None): values=[r['interval_ns']/1e6 for r in rows if r['interval_ns'] is not None] n=len(values) spikes=[] for threshold in thresholds: runs=[];run=[] for v in values+[0]: if v>threshold:run.append(v) elif run:runs.append(run);run=[] spikes.append(dict(threshold_ms=threshold,comparison='strictly_greater',frame_count=sum(len(r) for r in runs),episode_count=len(runs),total_frame_time_ms=sum(sum(r) for r in runs),total_excess_ms=sum(sum(v-threshold for v in r) for r in runs),longest_episode_ms=max([sum(r) for r in runs],default=0))) return dict(interval_count=n,duration_ms=sum(values),median_ms=percentile(values,.5),p95_ms=percentile(values,.95),p99_ms=percentile(values,.99),maximum_ms=max(values,default=None),percentile_method='nearest_rank: sorted[ceil(p*n)-1]',tail_estimates_qualified=n>=1000,tail_note='Require >=1000 intervals for this workflow (at least 10 observations in the top 1%); not a statistical confidence guarantee.',spikes=spikes,budget_ms=budget_ms,budget_exceeded_frames=sum(v>budget_ms for v in values) if budget_ms is not None else None)
def decode(raw,epoch): if len(raw)<HEADER.size:raise ValueError('Truncated header') h=HEADER.unpack_from(raw) if h[0]!=MAGIC or h[1]!=1 or h[3]!=CAPACITY or h[4]!=RECORD.size or h[6]>CAPACITY:raise ValueError('Invalid trace layout') if len(raw)!=HEADER.size+h[6]*RECORD.size:raise ValueError('Trace length mismatch') rows=[];previous=None for i in range(h[6]): seq,begin,end,swap,tid,ep,result,count,overhead=RECORD.unpack_from(raw,HEADER.size+i*RECORD.size) if seq!=i or ep!=epoch or begin>end or (previous and begin<previous['begin_ns']):raise ValueError('Invalid sequence, epoch or timestamps') interval=begin-previous['begin_ns'] if previous and previous['swapchain']==swap else None rows.append(dict(sequence=seq,begin_ns=begin,end_ns=end,swapchain=swap,thread_id=tid,epoch=ep,present_result=result,swapchain_count=count,observer_ns=overhead,interval_ns=interval));previous=rows[-1] return h,rows
def window_analysis(rows,begin,end,thresholds,budget_ms=None): """Completed intervals and coverage are separate; never fabricate a frame.""" if end<=begin:raise ValueError('Window end must follow begin') selected=[];boundary=[];gaps=[];cursor=begin;known=0 for row in rows: interval=row['interval_ns'] if interval is None:continue left=row['begin_ns']-interval;right=row['begin_ns'] lo=max(begin,left);hi=min(end,right) if hi<=lo:continue if lo>cursor:gaps.append((cursor,lo)) known+=hi-lo;cursor=hi if left>=begin and right<=end:selected.append(row) else:boundary.append(dict(begin_ns=left,end_ns=right,interval_ns=interval,overlap_ns=hi-lo,crosses_begin=left<begin,crosses_end=right>end,exceeds_thresholds_ms=[t for t in thresholds if interval>t*1e6])) if cursor<end:gaps.append((cursor,end)) censored=[] for lo,hi in gaps: censored.append(dict(begin_ns=lo,end_ns=hi,uncovered_ns=hi-lo,missing_previous_present=not rows or lo<rows[0]['begin_ns'],missing_next_present=not rows or hi>rows[-1]['begin_ns'],note='Uncovered time inside the window, not a completed frame duration.')) duration=end-begin;completed=sum(r['interval_ns'] for r in selected) limit=min(thresholds) reasons=[] if known!=duration:reasons.append('uncovered_or_censored_time') if any(b['interval_ns']>limit*1e6 for b in boundary):reasons.append('boundary_interval_exceeds_threshold') if not selected:reasons.append('no_completed_intervals') coverage=dict(version=1,window_ns=duration,completed_interval_ns=completed,known_interval_overlap_ns=known,boundary_overlap_ns=sum(b['overlap_ns'] for b in boundary),uncovered_ns=duration-known,uncovered_leading_ns=next((hi-lo for lo,hi in gaps if lo==begin),0),uncovered_trailing_ns=next((hi-lo for lo,hi in gaps if hi==end),0),completed_fraction=completed/duration,known_fraction=known/duration,boundary_intervals=boundary,censored_gaps=censored,acceptance_rule=dict(name='full_known_coverage_and_bounded_boundary_intervals_v1',required_known_fraction=1.0,max_boundary_interval_ms=limit,threshold_source='smallest declared spike threshold; not a universal performance budget',require_completed_intervals=True),acceptance_eligible=not reasons,rejection_reasons=reasons) return dict(summary=summarize(selected,thresholds,budget_ms),coverage=coverage)
def stop(reason='manual',expected_folder=None): with lock(): if not ACTIVE.exists():raise RuntimeError('No capture; use perf-start') state=read(ACTIVE);folder=Path(state['folder']) if expected_folder and str(folder)!=expected_folder:return dict(ok=False,reason='capture_replaced') if state.get('finished'):return read(folder/'report.json') with open(state['buffer'],'r+b') as f,mmap.mmap(f.fileno(),0) as mem: struct.pack_into('<Q',mem,40,0) wait_end=time.monotonic()+2 while struct.unpack_from('<Q',mem,80)[0] and time.monotonic()<wait_end:time.sleep(.005) in_flight=bool(struct.unpack_from('<Q',mem,80)[0]) # Only acquire committed prefix; an unfinished present slot is excluded. count=struct.unpack_from('<Q',mem,48)[0] raw=bytes(mem[:HEADER.size+count*RECORD.size]) (folder/'trace.bin').write_bytes(raw) h,rows=decode(raw,state['epoch']) final_marker_error=None try:marker('capture_stop',dict(reason=reason,in_flight=in_flight)) except (OSError,ValueError,RuntimeError) as error:final_marker_error=str(error)[:512] markers,marker_log=retained_markers(folder) marker_log['final_marker_error']=final_marker_error errors=[] if not rows:errors.append('empty_capture') if (folder/'marker-errors.jsonl').exists():errors.append('marker_recording_failed') if final_marker_error or marker_log['omitted_bytes'] or marker_log['invalid_lines'] or marker_log['overflow_recorded']:errors.append('marker_log_incomplete') if h[7]:errors.append('dropped_samples') if reason in ('window_failed','experiment_failed'):errors.append('experiment_failed_partial_capture') if in_flight:errors.append('present_still_in_flight_partial_prefix') if any(r['swapchain_count']!=1 for r in rows):errors.append('multiple_swapchains_in_present') if len({r['swapchain'] for r in rows})>1:errors.append('swapchain_changed_intervals_segmented') if any(r['present_result'] not in (0,1000001003) for r in rows):errors.append('present_returned_error') live=support.process_identity(state['process']['pid'])==state['process'] if not live:errors.append('game_exited_partial_capture') with (folder/'frames.csv').open('w') as f: cols=list(rows[0]) if rows else ['sequence','begin_ns','end_ns','interval_ns'] writer=csv.DictWriter(f,fieldnames=cols);writer.writeheader();writer.writerows(rows) windows=[] for m in markers: if m['kind']=='window_begin': ends=[e for e in markers if e['kind']=='window_end' and e['details'].get('name')==m['details'].get('name') and e['monotonic_ns']>=m['monotonic_ns']] if not ends:continue e=ends[0] if e['monotonic_ns']<=m['monotonic_ns']:continue windows.append(dict(name=m['details']['name'],kind=m['details'].get('kind'),begin_ns=m['monotonic_ns'],end_ns=e['monotonic_ns'],**window_analysis(rows,m['monotonic_ns'],e['monotonic_ns'],state['thresholds_ms'],state['budget_ms']))) report=dict(loaded_fixture=state.get('loaded_fixture'),ok=not errors,provider='PathPresent',semantics=SEMANTICS,folder=str(folder),label=state['label'],session=state['session'],generation=state['generation'],process=state['process'],stop_reason=reason,complete=not errors,errors=errors,dropped_samples=h[7],uncommitted_present=in_flight,capacity=CAPACITY,retained_records=len(rows),raw_sha256=digest(folder/'trace.bin'),units='ns in raw/CSV; ms in summaries',clock='CLOCK_MONOTONIC',clock_alignment=state['capability']['clock_alignment_note'],requested_seconds=state['seconds'],observed_span_ns=rows[-1]['begin_ns']-rows[0]['begin_ns'] if len(rows)>1 else 0,summary=summarize(rows,state['thresholds_ms'],state['budget_ms']),windows=windows,markers=markers,graphics=state['graphics'],observer_bookkeeping_ns=summarize([dict(interval_ns=r['observer_ns']) for r in rows],[],None),observer_note='Bookkeeping brackets exclude downstream Present; they also exclude dispatch lookup, initial enabled check, two clock reads, atomic commit and disabled-layer cost. Not total observer overhead.',exclusion_rule='Raw trace keeps all samples. Window summaries retain only intervals wholly inside explicit begin/end markers. Coverage separately exposes boundary-crossing intervals and censored gaps. complete is capture integrity, not coverage acceptance. Whole-trace summary includes pauses/menus/loading unless separately marked.',attribution={'gpu_time':None,'script_cpu_time':None,'displayed_frames':None,'input_latency':None}) report.update(coverage_version=1,marker_log=marker_log) write(folder/'report.json',report);state.update(finished=True,report=str(folder/'report.json'));write(folder/'capture.json',state);write(ACTIVE,state) return report
def watch(folder): state=read(Path(folder)/'capture.json') while time.monotonic_ns()<state['deadline_ns']: if read(ACTIVE).get('finished') or read(ACTIVE).get('folder')!=folder:return if support.process_identity(state['process']['pid'])!=state['process']: stop('game_exit',folder) if LAUNCH.exists() and read(LAUNCH)['env']['PATHDEVKIT_PRESENT_DIR']==str(Path(state['buffer']).parent):restore() return time.sleep(.05) stop('deadline',folder)
def window(seconds,warmup,label,thresholds,budget_ms=None,pause_seconds=0): """Sustained free-running gameplay, not successive tick-step actions.""" from . import experiments validate_number(seconds,1,60,'seconds');validate_number(warmup,0,20,'warmup');validate_number(pause_seconds,0,10,'pause-seconds') before=experiments.require_paused();lab=experiments.data('lab_state') identity=native.state()['process'];wid=support.window(identity) support.run('xdotool','windowactivate','--sync',wid) if support.run('xdotool','getactivewindow')!=wid:raise RuntimeError('Game foreground window could not be verified') total=seconds+warmup+pause_seconds+12 capture=start(total,label,thresholds,budget_ms);folder=Path(capture['folder']) write(folder/'fixture.json',dict(before=before,lab=lab,seconds=seconds,warmup=warmup,pause_seconds=pause_seconds)) failure=None try: marker('window_begin',dict(name='warmup',kind='warmup'));native.call('resume');time.sleep(warmup) marker('window_end',dict(name='warmup')) warm_snapshot=native.call('snapshot');write(folder/'snapshot-warm.json',warm_snapshot) if not warm_snapshot['data']['ready'] or warm_snapshot['data']['world']['paused'] or not warm_snapshot['data']['world']['player']['alive'] or warm_snapshot['data']['ui']['menus']: raise RuntimeError('Warm-up did not end in live unpaused gameplay') marker('window_begin',dict(name='gameplay',kind='active_gameplay'));time.sleep(seconds) marker('window_end',dict(name='gameplay'));native.call('pause') if pause_seconds: marker('window_begin',dict(name='pause',kind='paused'));time.sleep(pause_seconds);marker('window_end',dict(name='pause')) native.call('resume');marker('window_begin',dict(name='resumed',kind='active_gameplay'));time.sleep(1);marker('window_end',dict(name='resumed'));native.call('pause') after=native.call('snapshot');write(folder/'snapshot-after.json',after) if support.run('xdotool','getactivewindow')!=wid:failure='game_foreground_changed' if after['generation']!=before['generation']:failure='generation_changed' elif not after['data']['world']['player']['alive']:failure='player_died_terminal_time_not_observed' elif after['data']['ui']['menus']:failure='menu_present_at_end' except BaseException as error: failure=repr(error) try:native.call('release');native.call('pause') except Exception as cleanup:write(folder/'cleanup-error.json',dict(error=str(cleanup),uncertainty_preserved=True)) raise finally: try:marker('window_validity',dict(error=failure,continuous_state_coverage=False,note='Only boundary snapshots; unmarked transitions inside the window cannot be excluded automatically.')) except (OSError,ValueError,RuntimeError) as error:failure=failure or str(error) result=stop('window_finished' if failure is None else 'window_failed') result['ok'] = result['ok'] and failure is None result['window_validity']=dict(ok=failure is None,error=failure,boundary_checks_only=True) write(folder/'report.json',result) return result
if __name__=='__main__': if sys.argv[1]=='watch':watch(sys.argv[2])
def compare(plan_path): plan=read(plan_path) for field in ('changed_input','controls','pairs'): if not plan.get(field):raise ValueError('Comparison plan requires '+field) for key in ('fixture','actor_count','camera_input_protocol','graphics_frame_cap','capture_settings','warmup_exclusions','observer_cost_assessment'): if key not in plan['controls']:raise ValueError('Comparison controls require '+key) if len(plan['pairs'])<2:raise ValueError('At least two repeated baseline/treatment pairs required') pairs=[];reference=None;seen=[] for pair in plan['pairs']: audits={arm:verify_report(pair[arm]) for arm in ('baseline','treatment')} reports={arm:read(pair[arm]) for arm in ('baseline','treatment')} windows={} for arm,r in reports.items(): if not r['ok'] or not r['complete'] or not r.get('window_validity',{}).get('ok'):raise ValueError('Incomplete/invalid report: '+pair[arm]) window=next((w for w in r['windows'] if w['name']==pair.get('window','gameplay')),None) if not window or not window['summary']['interval_count']:raise ValueError('Missing populated gameplay window') metadata=read(Path(pair[arm]).parent/'capture.json');fixture=read(Path(pair[arm]).parent/'fixture.json') controls=dict(fixture=metadata.get('loaded_fixture'),graphics=r['graphics'],seconds=fixture['seconds'],warmup=fixture['warmup'],pause_seconds=fixture['pause_seconds'],thresholds=metadata['thresholds_ms'],budget=metadata['budget_ms'],provider_sha=metadata['capability']['build']['library_sha256'],actor_count=len(fixture['lab']['data']['owned'])) if reference is None:reference=controls if controls!=reference:raise ValueError('Measured controls differ between compared reports') if controls['fixture'] is not None and controls['fixture']!=plan['controls']['fixture']:raise ValueError('Declared fixture differs from recorded successful load') if controls['actor_count']!=plan['controls']['actor_count']:raise ValueError('Declared actor count differs from retained fixture') coverage=next(w['coverage'] for w in audits[arm]['window_coverage'] if w['name']==window['name'] and w['begin_ns']==window['begin_ns'] and w['end_ns']==window['end_ns']) if not coverage['acceptance_eligible']:raise ValueError('Window coverage rule failed: '+', '.join(coverage['rejection_reasons'])+'; '+pair[arm]) identity=dict(raw_sha256=r['raw_sha256'],capture=(metadata['session'],metadata['process'],metadata['epoch']),clock=metadata.get('clock_identity'),session=metadata['session'],process=metadata['process'],begin=window['begin_ns'],end=window['end_ns']) for previous in seen: if identity['raw_sha256']==previous['raw_sha256'] or identity['capture']==previous['capture']:raise ValueError('Reused capture/window identity: each arm and repetition requires a distinct capture, including copied files') same_clock=(identity['clock'] is not None and identity['clock']==previous['clock']) or identity['session']==previous['session'] or identity['process']==previous['process'] if same_clock and max(identity['begin'],previous['begin'])<min(identity['end'],previous['end']):raise ValueError('Overlapping measurement windows cannot qualify independent repetitions') seen.append(identity) windows[arm]=window['summary'] pairs.append(dict(reports=pair,summary=windows,coverage={arm:next(w['coverage'] for w in audits[arm]['window_coverage'] if w['name']==pair.get('window','gameplay')) for arm in reports},delta_treatment_minus_baseline_ms={metric:windows['treatment'][metric]-windows['baseline'][metric] for metric in ('median_ms','p95_ms','p99_ms','maximum_ms')})) return dict(fixture_identity_verified=reference['fixture'] is not None,ok=True,changed_input=plan['changed_input'],controls=plan['controls'],verified_controls=reference,pairs=pairs,interpretation='Descriptive paired window differences, not significance or causal attribution. Frame caps can mask headroom loss. A/A smoke controls do not measure an unimplemented policy.',source_reports=[dict(path=p[arm],sha256=digest(p[arm])) for p in plan['pairs'] for arm in ('baseline','treatment')])
def verify_report(path): """Recompute all timing summaries from the retained binary and window bounds.""" path=Path(path);folder=path.parent;r=read(path);capture=read(folder/'capture.json') if digest(folder/'trace.bin')!=r['raw_sha256']:raise ValueError('Raw trace hash mismatch') _,rows=decode((folder/'trace.bin').read_bytes(),capture['epoch']) for field in ('session','process','generation'): if r[field]!=capture[field]:raise ValueError('Report/capture identity differs: '+field) if r['summary']!=summarize(rows,capture['thresholds_ms'],capture['budget_ms']):raise ValueError('Whole-trace summary differs from raw intervals') coverage=[];names=set() for w in r['windows']: if w['name'] in names:raise ValueError('Ambiguous duplicate window name: '+w['name']) names.add(w['name']) analysis=window_analysis(rows,w['begin_ns'],w['end_ns'],capture['thresholds_ms'],capture['budget_ms']) if w['summary']!=analysis['summary']:raise ValueError('Window summary differs from raw intervals: '+w['name']) if ('coverage' in w or r.get('coverage_version')==1) and w.get('coverage')!=analysis['coverage']:raise ValueError('Window coverage differs from raw intervals: '+w['name']) coverage.append(dict(name=w['name'],begin_ns=w['begin_ns'],end_ns=w['end_ns'],coverage=analysis['coverage'])) with (folder/'frames.csv').open() as f: decoded=list(csv.DictReader(f)) if len(decoded)!=len(rows):raise ValueError('CSV/raw sample count differs') for a,b in zip(decoded,rows): if {k:None if v=='' else int(v) for k,v in a.items()}!=b:raise ValueError('CSV/raw record differs') if len(rows)>1 and len({v['swapchain'] for v in rows})==1: if sum(v['interval_ns'] or 0 for v in rows)!=rows[-1]['begin_ns']-rows[0]['begin_ns']:raise ValueError('Trace duration mismatch') return dict(ok=True,report=str(path),raw_hash=r['raw_sha256'],records=len(rows),windows_checked=len(r['windows']),window_coverage=coverage,csv_matches_raw=True,summary_matches_raw=True,scope='Content and arithmetic integrity, not proof of gameplay validity or causal attribution. Legacy report coverage is recomputed without rewriting evidence; comparison enforces the current coverage rule.')