268 lines
18 KiB
Python
268 lines
18 KiB
Python
"""Dormant one-hold prison policy; no OS, input backend or native action calls.
|
|
|
|
perform_held_action accepts only explicit adapters:
|
|
* guard.check(): original private window/focus/process/display ownership check.
|
|
* require_idle(): native all-key/button idle check, used before pressing only.
|
|
* key.down_f(guard, *, deadline, not_after) -> exactly
|
|
{key:'F', down_monotonic:actual_request_time}. The adapter must freshly prove
|
|
the SAME owner/connection and idle state immediately before the request, and
|
|
refuse after not_after (which includes six-second admission and pose freshness).
|
|
* key.up_f(guard, *, deadline, cleanup=False) -> exactly
|
|
{key:'F', up_monotonic:actual_request_time, released:True}. The adapter freshly
|
|
proves the same owner before key-up and confirms native keymap idle afterward.
|
|
cleanup=True may release an uncertain F after expiry ONLY through the same
|
|
freshly proven connection; it cannot renew the lease or qualify a late result.
|
|
No adapter exposes another key, wheel, mouse, action invocation or pose setter.
|
|
* notify('held-key-pending', key='F', deadline_monotonic=...) MUST establish an
|
|
independent hard supervisor before any uncertain down. Its maximum lease is
|
|
five seconds from that attempt, further shortened by the frozen observation
|
|
deadline. 'held-key-release-confirmed' clears only that lease. The wrapper owns
|
|
failure containment; this pure module cannot guarantee release after lost
|
|
ownership and retains uncertainty rather than redirecting input.
|
|
* observe(): all new consecutive native INPUT/LOOK/GATE/CHOICE joined snapshots,
|
|
or an unchanged latest sample. Begin with the committed selection final sample.
|
|
Keep original receipts; reject missing BASIS, nonce/world/body/life/account/
|
|
camera/gate/action/context identity changes, menus or observer cancellation.
|
|
Before phase2 snapshot fields match the selection adapter, plus turn=[x,y,z].
|
|
After phase2 original observer object pins remain mandatory, but live door,
|
|
context, gate and forward geometry may move; selected/permission/progress may
|
|
disappear. Never substitute another native identity based on similar geometry.
|
|
* calibration: nonce, observer_start_real_ms (QUEUED), ready_real_ms,
|
|
ready_monotonic, optional observer_deadline_monotonic <= original bound.
|
|
* selection_evidence is independently revalidated raw one-wheel evidence:
|
|
verified, last_sample, selected_action_id, wheel_button=5. report must retain
|
|
action_selection_proof and its full in-memory native samples/immutable refs.
|
|
|
|
Only six seconds of genuinely remaining ORIGINAL45s time admits a press. The
|
|
previous facing/selection8s lease is closed, never renewed or borrowed. One F
|
|
attempt only, <=5s hard lease, .75s reserved for guarded release, two advancing
|
|
bound native progress samples, monotonic phase1->2, separate original-door
|
|
control/physical motion, and native released-input observation are required.
|
|
No exact4.0 progress sample is required: native completion can reset in one frame.
|
|
Results distinguish refused-no-input, failed-after-input, unlock-observed-motion-
|
|
unverified, and verified. Even verified means this local action/door observation
|
|
only, not passage through the gate, a full escape, multiplayer or FPS evidence.
|
|
Importing this file does nothing; no production/helper/runner activates it.
|
|
"""
|
|
import copy
|
|
import math
|
|
from prison_approach_proof import number, vector, length, yaw, delta, inside, target_for
|
|
from prison_action_selection_proof import choices_for
|
|
|
|
|
|
class Refused(ValueError):
|
|
pass
|
|
|
|
|
|
def need(value, message):
|
|
if not value:
|
|
raise Refused(message)
|
|
|
|
|
|
def perform_held_action(guard, require_idle, key, observe, geometry, calibration, selection_evidence, report, *, clock, sleep, notify):
|
|
need('held_action_proof' not in report, 'Held action attempt already exists')
|
|
state = {'requested':True,'completed':False,'status':'refused-no-input',
|
|
'input_attempted':False,'unlock_observed':False,'gate_motion_observed':False,
|
|
'release_confirmed':False,'unconfirmed_held_key':None,'paired_release_errors':[],
|
|
'notification_errors':[],'events':[],'samples':[],'progress_samples':[],'transition_sample':None,
|
|
'control_sample':None,'motion_sample':None,'released_sample':None,
|
|
'initial_sample':None,'final_sample':None,'started_monotonic':clock(),
|
|
'finished_monotonic':None,'press_attempted_monotonic':None,
|
|
'down_monotonic':None,'up_monotonic':None,'press_sample':None,
|
|
'original_deadline_monotonic':None,'held_deadline_monotonic':None}
|
|
report['held_action_proof'] = state
|
|
held = False; down = None; previous = None; phase = 1; progress = 0.; pressed_ack = None
|
|
deadline = None; lease = None; release_due = None; release_attempts = 0; release_barrier = None
|
|
|
|
def release(cleanup=False):
|
|
nonlocal held, release_attempts
|
|
if not held: return
|
|
need(release_attempts < 2, 'Bounded F release attempts exhausted')
|
|
release_attempts += 1
|
|
event = {'key':'F','down':False,'attempted_monotonic':clock(),'cleanup':cleanup,'confirmed':False}
|
|
state['events'].append(event)
|
|
guard.check() # Expiry never excuses original ownership, even in cleanup.
|
|
result = key.up_f(guard, deadline=lease, cleanup=cleanup)
|
|
need(type(result) is dict and set(result)=={'key','up_monotonic','released'} and
|
|
result['key']=='F' and result['released'] is True, 'Native F release is unconfirmed')
|
|
up = number(result['up_monotonic']); returned = clock()
|
|
need(event['attempted_monotonic'] <= up <= returned, 'Native F release timestamps invalid')
|
|
event.update(monotonic=up, confirmed=True)
|
|
state['up_monotonic']=up; state['release_confirmed']=True
|
|
state['unconfirmed_held_key']=None; held=False
|
|
notify({'kind':'held-key-release-confirmed','key':'F'})
|
|
if not cleanup:
|
|
need(returned < lease and returned < deadline, 'F release exceeded immutable lease')
|
|
|
|
try:
|
|
pinned = copy.deepcopy(geometry); _, prison = target_for(pinned)
|
|
state['nonce']=pinned['nonce']; state['geometry']=pinned
|
|
selected = report.get('action_selection_proof', {})
|
|
need(type(selection_evidence) is dict and selection_evidence.get('verified') is True and
|
|
selection_evidence.get('wheel_button')==5 and selected.get('completed') is True and
|
|
selected.get('nonce')==pinned['nonce'] and selected.get('geometry')==pinned and
|
|
selection_evidence.get('last_sample')==selected.get('final_sample'),
|
|
'Independently accepted same-run one-wheel selection required')
|
|
prior = selected.get('samples', [])
|
|
need(type(prior) is list and len(prior)>=2 and prior[-1]['sample']==selected['final_sample'],
|
|
'Original full selection samples unavailable')
|
|
anchor = copy.deepcopy(prior[-1]); body=vector(anchor['origin']); view=yaw(anchor['direction'])
|
|
actions = choices_for(anchor,pinned,selected_lock=True)
|
|
need(selected.get('actions')==actions and type(selection_evidence.get('selected_action_id')) is int and
|
|
selection_evidence['selected_action_id']==actions[1]['id'], 'Original selected lock identity mismatch')
|
|
need(selected.get('selection_samples')==[prior[-2]['sample'],anchor['sample']] and
|
|
anchor['sample']==prior[-2]['sample']+1 and 100<=anchor['real_ms']-prior[-2]['real_ms']<=500,
|
|
'Committed native selection pair missing')
|
|
choices_for(prior[-2],pinned,selected_lock=True)
|
|
queued=number(calibration['observer_start_real_ms']); ready=number(calibration['ready_real_ms'])
|
|
ready_at=number(calibration['ready_monotonic']); started=state['started_monotonic']
|
|
need(calibration.get('nonce')==pinned['nonce'] and 0<=queued and 2000<=ready-queued<45000 and
|
|
0<=ready_at<=started, 'Original QUEUED/READY binding invalid')
|
|
original=ready_at+(45000-(ready-queued))/1000-.5
|
|
supplied=number(calibration.get('observer_deadline_monotonic',original))
|
|
need(supplied<=original, 'Supplied observation deadline renews original lifetime')
|
|
deadline=min(original,supplied,number(anchor['received_monotonic'])+(queued+45000-number(anchor['real_ms']))/1000-.5)
|
|
state['original_deadline_monotonic']=deadline;state['actions']=actions
|
|
state['selection_final_sample']=anchor['sample']
|
|
need(started<deadline, 'Original observation lifetime expired')
|
|
|
|
def consume(raw):
|
|
nonlocal previous, phase, progress, pressed_ack
|
|
s=copy.deepcopy(raw); index=s['sample'];real=number(s['real_ms']);receipt=number(s['received_monotonic'])
|
|
need(type(index) is int and anchor['sample']<=index<=450 and queued<=real<queued+45000 and
|
|
0<=receipt<=clock(), 'Held native index/clock invalid')
|
|
if previous:
|
|
need(index>=previous['sample'] and real>=previous['real_ms'] and receipt>=previous['received_monotonic'],
|
|
'Held native samples or receipts reversed')
|
|
if index==previous['sample']:
|
|
need(s==previous, 'Repeated held sample changed')
|
|
return s
|
|
need(index==previous['sample']+1 and 0<real-previous['real_ms']<=500, 'Held native history has a gap')
|
|
else: need(index==anchor['sample'], 'Held handoff omitted selection final sample')
|
|
for name in ('origin','camera','direction','body_forward','input','velocity','turn'):vector(s[name])
|
|
need(inside(s['origin'],prison) and math.dist(body,s['origin'])<=.05 and
|
|
abs(number(s['forward']))<=.01 and length(s['input'])<=.01 and length(s['velocity'])<=.05 and
|
|
length(s['turn'])<=.01 and abs(delta(yaw(s['direction']),view))<=.2 and s['inventory'] is False,
|
|
'Original stationary inside-prison body/view/input changed')
|
|
need(type(s['phase']) is int and s['phase'] in (1,2) and s['phase']>=phase,
|
|
'Held phase reversed or left PRISON/RADIO_SEARCH')
|
|
for flag in ('selected','context_selected','cached_performable','progress_bound','performing'):
|
|
need(type(s[flag]) is bool, 'Malformed held action flag')
|
|
value=number(s['progress']);pressed=number(s['pressed']);door=number(s['door_state']);control=number(s['door_control'])
|
|
need(0<=value<=4 and 0<=pressed<=1 and number(s['duration'])==4 and -1<=door<=1 and -1<=control<=1,
|
|
'Invalid native action/door observation')
|
|
geo=s['geometry'];need(type(geo) is dict and set(geo)==set(pinned) and geo['nonce']==pinned['nonce'] and
|
|
geo['basis']==pinned['basis'] and math.dist(vector(geo['prison']),prison)<=.01,
|
|
'Original native gate geometry binding changed')
|
|
for name in ('gate','gate_forward','context'):vector(geo[name])
|
|
need(.5<=length(geo['gate_forward'])<=2 and 0<number(geo['radius'])<=10,'Invalid live original door geometry')
|
|
if s['phase']==1:
|
|
need(geo==pinned and door==0 and control==0, 'Closed gate changed before accepted transition')
|
|
observed=choices_for(s,pinned,selected_lock=True)
|
|
need(observed==actions, 'Original selected action identities changed')
|
|
if down is None or index<=state['press_sample']:
|
|
need(pressed==0 and value==0 and not s['performing'], 'Action/input appeared before this F attempt')
|
|
else:
|
|
need(receipt>=down, 'Action sample predates actual F request')
|
|
if pressed>0: pressed_ack=index
|
|
if value>0:
|
|
need(s['performing'] and pressed>0 and pressed_ack is not None, 'Progress lacks bound native input/performing proof')
|
|
need(value>=progress, 'Native action progress decreased before transition')
|
|
if value>progress: state['progress_samples'].append(index)
|
|
progress=value
|
|
elif progress>0:
|
|
raise Refused('Native action progress reset before transition')
|
|
if progress>0:need(s['performing'] and pressed>0,'Native action interrupted before transition')
|
|
else:
|
|
need(down is not None and index>state['press_sample'] and receipt>=down and
|
|
pressed_ack is not None and len(state['progress_samples'])>=2,
|
|
'Phase transition lacks advancing bound progress from this F hold')
|
|
if phase==1:
|
|
state['transition_sample']=index;state['unlock_observed']=True
|
|
state['status']='unlock-observed-motion-unverified'
|
|
if abs(control)>.001 and state['control_sample'] is None:state['control_sample']=index
|
|
if abs(door)>.001 and state['motion_sample'] is None:state['motion_sample']=index
|
|
state['gate_motion_observed']=state['control_sample'] is not None and state['motion_sample'] is not None
|
|
# After the full post-keyup drain, every subsequent native sample
|
|
# is evidence. A neutral observation cannot be reused after renewed
|
|
# action input, even if the reversal shares one buffered batch.
|
|
if state['released_sample'] is not None:
|
|
need(pressed==0 and value==0 and not s['performing'], 'Native action input renewed after release acknowledgement')
|
|
if (release_barrier is not None and index>release_barrier and receipt>=state['up_monotonic'] and
|
|
pressed==0 and value==0 and not s['performing']):
|
|
state['released_sample']=index
|
|
phase=s['phase'];previous=s
|
|
state['samples'].append(copy.deepcopy(s));state['final_sample']=index
|
|
if state['initial_sample'] is None:state['initial_sample']=index
|
|
return s
|
|
|
|
def read(idle=False):
|
|
need(clock()<deadline, 'Original held observation deadline expired')
|
|
guard.check()
|
|
if idle:require_idle()
|
|
batch=observe();need(type(batch) is list and 1<=len(batch)<=450,'Missing ordered held snapshots')
|
|
for raw in batch:current=consume(raw)
|
|
need(0<=clock()-current['received_monotonic']<=.5 and clock()<deadline,'Held native evidence stale or expired')
|
|
return current
|
|
|
|
notify({'kind':'held-action-pending','deadline_monotonic':deadline})
|
|
current=read(idle=True)
|
|
while True:
|
|
samples=state['samples']
|
|
if len(samples)>=2 and samples[-1]['sample']==samples[-2]['sample']+1 and 100<=samples[-1]['real_ms']-samples[-2]['real_ms']<=500:
|
|
break
|
|
need(clock()+6<=deadline,'held-action-not-started: insufficient-original-budget')
|
|
sleep(.025);current=read(idle=True)
|
|
guard.check();require_idle()
|
|
attempted=clock()
|
|
not_after=min(deadline-6,current['received_monotonic']+.5)
|
|
need(attempted<=not_after,'held-action-not-started: insufficient-original-budget-or-stale-pose')
|
|
lease=min(deadline,attempted+5);release_due=lease-.75
|
|
state.update(press_attempted_monotonic=attempted,held_deadline_monotonic=lease,press_sample=current['sample'],
|
|
input_attempted=True,status='failed-after-input',unconfirmed_held_key='F')
|
|
held=True
|
|
event={'key':'F','down':True,'attempted_monotonic':attempted,'confirmed':False}
|
|
state['events'].append(event)
|
|
notify({'kind':'held-key-pending','key':'F','deadline_monotonic':lease})
|
|
result=key.down_f(guard,deadline=lease,not_after=not_after)
|
|
need(type(result) is dict and set(result)=={'key','down_monotonic'} and result['key']=='F','Unexpected held key adapter result')
|
|
down=number(result['down_monotonic'])
|
|
need(attempted<=down<=clock() and down<=not_after and clock()<release_due,'Actual F request exceeded admission/lease')
|
|
event.update(monotonic=down,confirmed=True);state['down_monotonic']=down
|
|
while True:
|
|
if held and clock()>=release_due:
|
|
raise Refused('Completion not observed before guarded F release reserve')
|
|
current=read()
|
|
if state['release_confirmed']:
|
|
# Drain the entire first post-keyup batch before accepting a
|
|
# newer native input acknowledgement. Never credit buffered
|
|
# completion/reset samples merely received after our release.
|
|
if release_barrier is None:
|
|
release_barrier=current['sample'];state['release_barrier_sample']=release_barrier
|
|
elif current['sample']>release_barrier and current['received_monotonic']>=state['up_monotonic'] and current['pressed']==0 and not current['performing'] and current['progress']==0:
|
|
state['released_sample']=current['sample']
|
|
if state['unlock_observed'] and held:release()
|
|
if (state['unlock_observed'] and state['gate_motion_observed'] and state['release_confirmed'] and
|
|
state['released_sample']==current['sample'] and current['pressed']==0 and current['progress']==0 and not current['performing']):
|
|
state['completed']=True;state['status']='verified'
|
|
return state
|
|
sleep(min(.025,max(0,deadline-clock())))
|
|
except Exception as error:
|
|
state['error']=str(error)
|
|
raise
|
|
finally:
|
|
if held:
|
|
try:release(cleanup=True)
|
|
except Exception as error:state['paired_release_errors'].append(str(error))
|
|
if state['unlock_observed'] and state['gate_motion_observed'] and not state['completed']:
|
|
state['status']='failed-after-input'
|
|
state['finished_monotonic']=clock()
|
|
try:
|
|
notify({'kind':'held-action-finished','successful':state['completed']})
|
|
except Exception as error:
|
|
state['notification_errors'].append(str(error))
|
|
if 'error' not in state:
|
|
state['completed']=False;state['status']='failed-after-input' if state['input_attempted'] else 'refused-no-input'
|
|
state['error']=str(error)
|
|
raise
|