Files
EscapistsReintroduced/tools/prison_held_action_evidence.py
T

163 lines
15 KiB
Python

"""Dormant independent raw-console held-F verifier. Never sends input.
Primary installed contracts: ScriptedUserAction.GetActionProgressScript adds
OnPostFrame timeSlice and returns seconds; SCR_InteractionHandlerComponent resets
progress in the same frame as DoPerformObjectAction; BaseWorld.GetWorldTime is
milliseconds. INPUT already records that world clock. Exact sampler/post-frame
ordering is not documented: timing checks conservatively allow ONE observed
adjacent world-sample interval plus2ms print rounding at each boundary, and
record that inference. They never replace duration with a guessed near4 cutoff
or wall-clock sleep. The source diagnostic skip-duration path would fail the
normal-duration timing checks. Full native object pins come from the unchanged
nonce-bound observer/source provenance; geometry alone cannot prove identity.
Earlier look/approach/facing/selection spans are revalidated without relaxing any
old validator. Only this held span permits phase2 and moving original-door
geometry. The runner still must verify exact process/display/source ownership,
actual F-only adapter requests, hard supervision and native keymap release.
"""
import math
import re
from ordinary_gate_choices import join_choice_snapshots, exact_prison_choices
from prison_action_selection_evidence import validate_selection_evidence
from prison_approach_evidence import number, need, yaw, angle, norm
SCOPE='one original local prison lock hold, authoritative phase transition and physical door response; no passage, full escape, multiplayer or FPS qualification'
def validate_held_action_evidence(content, nonce, look_proof, approach_proof, facing_proof, selection_proof, proof, *, facing_profile='standard'):
result={'verified':False,'held_action_verified':False,'escape_verified':False,'errors':[],
'scope':SCOPE,'unlock_observed':False,'gate_motion_observed':False,
'timing_basis':'native world milliseconds; one adjacent sampled world interval plus2ms allowance for undocumented post-frame/sampler ordering',
'adapter_release_confirmation_verified':False}
try:
selected=validate_selection_evidence(content,nonce,look_proof,approach_proof,facing_proof,selection_proof,facing_profile=facing_profile)
need(selected['verified'],'prior independent selection failed: '+str(selected['errors']))
required={'requested','completed','status','input_attempted','unlock_observed','gate_motion_observed','release_confirmed',
'unconfirmed_held_key','paired_release_errors','notification_errors','events','samples','progress_samples',
'transition_sample','control_sample','motion_sample','released_sample','initial_sample','final_sample',
'started_monotonic','finished_monotonic','press_attempted_monotonic','down_monotonic','up_monotonic',
'press_sample','original_deadline_monotonic','held_deadline_monotonic','nonce','geometry','actions',
'selection_final_sample','release_barrier_sample'}
need(type(proof) is dict and set(proof)==required,'unknown or missing held proof fields')
need(proof['requested'] is True and proof['completed'] is True and proof['status']=='verified' and proof['nonce']==nonce and
proof['input_attempted'] is True and proof['release_confirmed'] is True and proof['unlock_observed'] is True and
proof['gate_motion_observed'] is True and proof['unconfirmed_held_key'] is None and
proof['paired_release_errors']==[] and proof['notification_errors']==[], 'held proof did not complete cleanly')
data=join_choice_snapshots(content,nonce);frames=data['snapshots']
first=proof['initial_sample'];last=proof['final_sample'];anchor=selected['last_sample'];before=proof['press_sample']
need(all(type(i) is int for i in (first,last,anchor,before)) and first==anchor==proof['selection_final_sample'] and first<before<last<=450,
'held span or original selection handoff invalid')
need(proof['geometry']==selection_proof['geometry'] and proof['actions']==selection_proof['actions'],'held geometry/actions replaced selection binding')
refs=proof['samples'];need(type(refs) is list and 2<=len(refs)<=450,'held native references missing')
need([x['sample'] for x in refs]==list(range(first,last+1)),'held native references omitted, duplicated or reordered')
start=number(proof['started_monotonic']);finish=number(proof['finished_monotonic']);deadline=number(proof['original_deadline_monotonic'])
attempt=number(proof['press_attempted_monotonic']);down=number(proof['down_monotonic']);up=number(proof['up_monotonic']);lease=number(proof['held_deadline_monotonic'])
need(number(selection_proof['finished_monotonic'])<=start<=attempt<=down<up<finish<deadline and up<lease<=attempt+5 and lease<=deadline,
'held request, release or completion exceeded immutable deadline')
need(down+6<=deadline,'actual F request lacked six seconds of original observation budget')
queued=re.findall(r'ERR INPUT OBSERVE: run='+re.escape(nonce)+r' sample=0 real=(\d+) world=[^ ]+ QUEUED(?:\r?$)',content,re.MULTILINE)
need(len(queued)==1,'original QUEUED clock missing or ambiguous');queued=int(queued[0]);ready=data['ready']['real_ms']
original_host_end=number(approach_proof['deadline_monotonic'])-35+(45000-(ready-queued))/1000-.5
anchor_ref=selection_proof['samples'][-1]
anchor_end=number(anchor_ref['received_monotonic'])+(queued+45000-number(anchor_ref['real_ms']))/1000-.5
need(deadline<=min(original_host_end,anchor_end),'held stage renewed original observation clock')
events=proof['events'];need(type(events) is list and len(events)==2,'expected one F down and one F up only')
for e,is_down,at in zip(events,(True,False),(down,up)):
keys={'key','down','attempted_monotonic','confirmed','monotonic'} | (set() if is_down else {'cleanup'})
need(type(e) is dict and set(e)==keys and e['key']=='F' and e['down'] is is_down and e['confirmed'] is True and
number(e['monotonic'])==at and number(e['attempted_monotonic'])<=at and
(is_down or e['cleanup'] is False),'unexpected key, retry, cleanup or unconfirmed F event')
need(events[0]['attempted_monotonic']==attempt and down<=number(events[1]['attempted_monotonic'])<lease,'F attempted-event clocks disagree')
receipts={};prior_receipt=-1
for r in refs:
i=r['sample'];need(i in frames and type(r['real_ms']) is int and r['real_ms']==frames[i]['real_ms'],'held reference differs from native clock')
received=number(r['received_monotonic']);need(0<=received<=finish and received>=prior_receipt,'held receipts reversed or unbounded')
if 'world_ms' in r:need(number(r['world_ms'])==frames[i]['input']['world_ms'],'held world reference differs from native clock')
receipts[i]=received;prior_receipt=received
need(refs[0]['real_ms']==anchor_ref['real_ms'] and refs[0]['received_monotonic']==anchor_ref['received_monotonic'],'original selection receipt changed at held handoff')
need(0<=finish-receipts[last]<=.5 and receipts[before]<=attempt and down-receipts[before]<=.5,'held input/final native evidence is stale')
for cancellation in (data['cancelled'],data['look_cancelled']):
need(cancellation is None or cancellation['sample']>last,'original native observer cancelled within held span')
body=frames[anchor]['input']['origin'];direction=yaw(frames[anchor]['look']['direction']);geometry=proof['geometry'];actions=proof['actions']
need(actions[1]['id']==selected['selected_action_id'],'original selected lock ID differs')
progress=[];prior_progress=0.;pressed_seen=False;transition=None;control=None;motion=None;previous=None;last_positive=None
phase=1;allowance_max=0.;world_before=frames[before]['input']['world_ms']
# Check the two fresh, selected pre-input native samples themselves.
need(before-1>=first and 100<=frames[before]['real_ms']-frames[before-1]['real_ms']<=500,'F lacks a consecutive admission pair')
for i in range(first,last+1):
need(i in frames and 'choices' in frames[i],'held raw join/choice missing')
s=frames[i];a=s['action'];g=s['geometry'];native=s['input'];view=s['look']
need(queued<=s['real_ms']<queued+45000 and native['inventory'] is False,'held native lifetime/menu changed')
need(a['duration']==4 and 0<=a['progress']<=4,'original native four-second action duration/progress changed')
need(math.dist(body,native['origin'])<=.05 and abs(native['origin'][0]-geometry['prison'][0])<=5.25 and abs(native['origin'][2]-geometry['prison'][2])<=5.25 and
abs(native['forward'])<=.01 and norm(native['input'])<=.01 and norm(native['velocity'])<=.05 and norm(view['turn'])<=.01 and
abs(angle(yaw(view['direction']),direction))<=.2,'held body/view/movement changed')
need(a['phase'] in (1,2) and a['phase']>=phase,'held native phase reversed or advanced beyond RADIO_SEARCH')
need(math.dist(g['prison'],geometry['prison'])<=.01,'original prison geometry changed')
if previous:
need(0<s['real_ms']-previous['real_ms']<=500,'held native observation gap')
world_delta=(native['world_ms']-previous['input']['world_ms'])/1000
need(world_delta>=0,'native world clock reversed')
adjacent=world_delta
if i-2>=first:adjacent=max(adjacent,(previous['input']['world_ms']-frames[i-2]['input']['world_ms'])/1000)
else:world_delta=0.;adjacent=0.
tolerance=adjacent+.002;allowance_max=max(allowance_max,tolerance)
if a['phase']==1:
for name in ('gate','context'):need(math.dist(g[name],geometry[name])<=.01,'closed original door/context geometry changed')
need(math.dist(g['gate_forward'],geometry['gate_forward'])<=.001 and abs(g['radius']-geometry['radius'])<=.001 and g['door_state']==0 and g['door_control']==0,
'closed gate changed before accepted action transition')
c=exact_prison_choices(s)
need(c['actions']==actions and c['selected_id']==actions[1]['id'] and c['selected_class']==actions[1]['class'] and a['selected'] and a['context_selected'] and a['cached_performable'] and a['progress_bound'],
'held original lock selection/permission/binding lost')
if i<=before:
need(a['pressed']==0 and a['progress']==0 and not a['performing'],'held action appeared before own F request')
else:
need(receipts[i]>=down,'native action observation predates actual F request')
if a['pressed']>0:pressed_seen=True
if a['progress']>0:
need(a['performing'] and a['pressed']>0 and pressed_seen and a['progress']>=prior_progress,'native held action interrupted or progress decreased')
need(a['progress']<=4 and a['progress']<=(native['world_ms']-world_before)/1000+tolerance,
'native action progress exceeds available world time')
need(a['progress']-prior_progress<=world_delta+tolerance,'native action progress jumped beyond adjacent world interval')
if a['progress']>prior_progress:progress.append(i)
prior_progress=a['progress'];last_positive=i
elif prior_progress>0:raise ValueError('native action progress reset before phase transition')
if prior_progress>0:need(a['performing'] and a['pressed']>0,'native action interrupted before phase transition')
else:
need(i>before and receipts[i]>=down and pressed_seen and len(progress)>=2 and last_positive is not None,'phase2 lacks advancing bound progress from this F attempt')
if transition is None:
remaining=(native['world_ms']-frames[last_positive]['input']['world_ms'])/1000
need(prior_progress+remaining+tolerance>=4 and (native['world_ms']-world_before)/1000+tolerance>=4,
'phase2 occurs before four seconds of reachable native action progress')
transition=i;result['unlock_observed']=True
# Raw parser and native original-object pins still apply. Only
# closed geometry and selected/progress cache may now change.
if abs(g['door_control'])>.001 and control is None:control=i
if abs(g['door_state'])>.001 and motion is None:motion=i
phase=a['phase'];previous=s
need(transition is not None and control is not None and motion is not None,'original door transition/control/physical motion missing')
result['gate_motion_observed']=True
need(receipts[transition]<=number(events[1]['attempted_monotonic'])<=up,'normal F release preceded observed authoritative transition')
need(proof['progress_samples']==progress and proof['transition_sample']==transition and proof['control_sample']==control and proof['motion_sample']==motion,
'helper progress/transition/door references differ from raw observations')
barrier=proof['release_barrier_sample'];released=proof['released_sample']
need(type(barrier) is int and type(released) is int and transition<=barrier<released==last and barrier in receipts and released in receipts and
receipts[barrier]>=up and receipts[released]>=up,'release lacks post-keyup drain and newer native acknowledgement')
a=frames[released]['action'];need(a['phase']==2 and a['pressed']==0 and a['progress']==0 and not a['performing'],'native released action input not observed at final sample')
acknowledged=False
for i in range(barrier+1,last+1):
action=frames[i]['action'];neutral=action['pressed']==0 and action['progress']==0 and not action['performing']
if acknowledged:need(neutral,'native action input renewed after release acknowledgement')
if receipts[i]>=up and neutral:acknowledged=True
need(acknowledged,'newer native release acknowledgement missing')
need(any(receipts[i]>=down and frames[i]['action']['pressed']>0 for i in range(before+1,transition)), 'native PerformAction acknowledgement missing')
result.update(verified=True,held_action_verified=True,first_sample=first,last_sample=last,transition_sample=transition,
control_sample=control,motion_sample=motion,released_sample=released,selected_action_id=actions[1]['id'],
native_world_duration_ms=frames[transition]['input']['world_ms']-world_before,
maximum_timing_allowance_seconds=allowance_max,
release_confirmation_scope='actual native released input verified; native keymap/FFI release attestation remains separate runner responsibility')
except (ValueError,KeyError,TypeError,OverflowError,IndexError) as error:
result['errors'].append(str(error))
return result