"""Optional bounded host presentation-call timing. No per-frame bridge traffic."""
import contextlib
import csv
import fcntl
import hashlib
import json
import math
import mmap
import os
from pathlib import Path
import secrets
import struct
import subprocess
import sys
import time

from . import support
from . import native

ROOT = Path(__file__).resolve().parent.parent
ACTIVE = 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 = 0x31544e4553455250
CAPACITY = 65536
MARKER_LIMIT = 2*1024*1024
SEMANTICS = '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.contextmanager
def 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.')