272 lines
17 KiB
Python
272 lines
17 KiB
Python
"""Actual dormant held policy on simulated native snapshots; no input/runtime."""
|
|
import copy
|
|
import math
|
|
from pathlib import Path
|
|
import sys
|
|
import unittest
|
|
sys.path.insert(0,str(Path(__file__).resolve().parents[1]/'tools'))
|
|
import prison_held_action_proof as policy
|
|
|
|
|
|
class Scene:
|
|
def __init__(self):
|
|
self.now=30.;self.next_sample=30.1;self.guard_cost=.02
|
|
self.geometry={'nonce':'N','basis':'registered-prison-lock-native-interaction','prison':[100.,0.,100.],
|
|
'gate':[100.,0.,94.],'gate_forward':[0.,0.,1.],'context':[102.7,1.2,94.],'radius':2.}
|
|
self.origin=[102.7,0.,95.5];self.camera=[102.7,1.7,95.4];self.direction=[0.,0.,-1.]
|
|
self.actions=[{'class':'ERR_PrisonDoorAction','id':8,'permitted':False},{'class':'ERR_UnlockPrisonAction','id':3,'permitted':True}]
|
|
self.focus=True;self.native_held=False;self.down=None;self.up=None;self.events=[];self.notifications=[]
|
|
self.fail_down=False;self.fail_up=False;self.unknown_down=False;self.release_late=False;self.native_release_missing=False
|
|
self.phase_delay=4.1;self.motion_delay=.15;self.never_complete=False;self.no_motion=False;self.no_control=False
|
|
self.mutate=None;self.stale=False;self.history=[];self.last_read=299
|
|
self.snapshot=self.make();self.history=[copy.deepcopy(self.snapshot)]
|
|
prior=copy.deepcopy(self.snapshot);prior.update(sample=299,real_ms=59900,received_monotonic=29.9);prior['choices'].update(sample=299,real_ms=59900)
|
|
self.report={'action_selection_proof':{'completed':True,'nonce':'N','geometry':copy.deepcopy(self.geometry),
|
|
'final_sample':300,'selection_samples':[299,300],'samples':[prior,copy.deepcopy(self.snapshot)],'actions':copy.deepcopy(self.actions),
|
|
'deadline_monotonic':30.,'finished_monotonic':30.}}
|
|
self.calibration={'nonce':'N','observer_start_real_ms':30000,'ready_real_ms':32000,'ready_monotonic':2.,'observer_deadline_monotonic':44.5}
|
|
self.evidence={'verified':True,'last_sample':300,'selected_action_id':3,'wheel_button':5}
|
|
def clock(self):return self.now
|
|
def make(self):
|
|
elapsed=None if self.down is None else self.now-self.down
|
|
phase=2 if elapsed is not None and elapsed>=self.phase_delay and not self.never_complete else 1
|
|
performing=phase==1 and elapsed is not None and elapsed>=.1 and self.native_held
|
|
progress=min(3.98,max(0,elapsed-.1)) if performing else 0.
|
|
pressed=1. if self.native_held or (self.native_release_missing and self.down is not None) else 0.
|
|
geo=copy.deepcopy(self.geometry)
|
|
if phase==2:geo['context']=[102.,1.2,94.7];geo['gate_forward']=[.3,0.,.953939]
|
|
selected=phase==1
|
|
s={'sample':int(round(self.now*10)),'real_ms':30000+int(round(self.now*1000)),
|
|
'received_monotonic':self.now-(1 if self.stale else 0),'geometry':geo,'phase':phase,
|
|
'origin':list(self.origin),'camera':list(self.camera),'direction':list(self.direction),
|
|
'body_forward':[0.,0.,-1.],'input':[0.,0.,0.],'forward':0.,'velocity':[0.,0.,0.],'turn':[0.,0.,0.],
|
|
'inventory':False,'door_state':.25 if phase==2 and elapsed>=self.phase_delay+self.motion_delay and not self.no_motion else 0.,
|
|
'door_control':1. if phase==2 and not self.no_control else 0.,'selected':selected,'context_selected':selected,
|
|
'cached_performable':selected,'progress_bound':selected,'performing':performing,'progress':progress,'pressed':pressed,'duration':4.}
|
|
s['choices']={'nonce':'N','sample':s['sample'],'real_ms':s['real_ms'],'selected_id':3 if selected else -1,
|
|
'selected_class':'ERR_UnlockPrisonAction' if selected else 'none','actions':copy.deepcopy(self.actions) if selected else []}
|
|
if self.mutate:self.mutate(s,elapsed)
|
|
return s
|
|
def sleep(self,seconds):
|
|
end=self.now+seconds
|
|
while self.now<end-1e-8:
|
|
self.now+=min(.005,end-self.now)
|
|
if self.now>=self.next_sample-1e-8:
|
|
self.snapshot=self.make();self.history.append(copy.deepcopy(self.snapshot));self.next_sample+=.1
|
|
def check(self):
|
|
if not self.focus:raise policy.Refused('original private owner lost')
|
|
self.sleep(self.guard_cost)
|
|
def idle(self):
|
|
if self.native_held:raise policy.Refused('native keys not idle')
|
|
def observe(self):
|
|
batch=[x for x in self.history if x['sample']>self.last_read] or [self.snapshot]
|
|
self.last_read=batch[-1]['sample'];return copy.deepcopy(batch)
|
|
def down_f(self,guard,*,deadline,not_after):
|
|
guard.check();self.idle()
|
|
if self.now>not_after or self.now>=deadline:raise policy.Refused('adapter admission expired')
|
|
self.native_held=True;self.down=self.now;self.events.append(('F',True,self.now,deadline))
|
|
if self.unknown_down:raise policy.Refused('down request outcome uncertain')
|
|
if self.fail_down:raise policy.Refused('down refused')
|
|
return {'key':'F','down_monotonic':self.now}
|
|
def up_f(self,guard,*,deadline,cleanup=False):
|
|
guard.check()
|
|
if self.fail_up:raise policy.Refused('up refused')
|
|
if self.release_late and not cleanup:self.sleep(2.)
|
|
self.native_held=False;self.up=self.now;self.events.append(('F',False,self.now,deadline))
|
|
return {'key':'F','up_monotonic':self.now,'released':True}
|
|
def run(self):
|
|
return policy.perform_held_action(self,self.idle,self,self.observe,self.geometry,self.calibration,self.evidence,self.report,
|
|
clock=self.clock,sleep=self.sleep,notify=self.notifications.append)
|
|
|
|
|
|
class HeldTests(unittest.TestCase):
|
|
def test_one_real_policy_hold_progress_reset_transition_motion_and_release(self):
|
|
s=Scene();r=s.run()
|
|
self.assertTrue(r['completed']);self.assertEqual(r['status'],'verified')
|
|
self.assertEqual([(e[0],e[1]) for e in s.events],[('F',True),('F',False)])
|
|
self.assertGreaterEqual(len(r['progress_samples']),2);self.assertTrue(r['unlock_observed'] and r['gate_motion_observed'])
|
|
self.assertTrue(r['release_confirmed']);self.assertIsNone(r['unconfirmed_held_key'])
|
|
self.assertEqual(r['paired_release_errors'],[]);self.assertLess(r['released_sample'],451)
|
|
self.assertGreater(r['released_sample'],r['release_barrier_sample']);self.assertLess(s.now,44.5)
|
|
self.assertEqual(r['original_deadline_monotonic'],44.5)
|
|
self.assertLessEqual(r['held_deadline_monotonic']-r['press_attempted_monotonic'],5)
|
|
self.assertLess(r['up_monotonic'],r['held_deadline_monotonic'])
|
|
frames={x['sample']:x for x in r['samples']}
|
|
self.assertTrue(frames[r['transition_sample']]['progress']==0 and not frames[r['transition_sample']]['selected'])
|
|
self.assertNotEqual(frames[r['motion_sample']]['geometry']['context'],s.geometry['context'])
|
|
def test_does_not_renew_previous_selection_eight_second_lease(self):
|
|
s=Scene();self.assertEqual(s.report['action_selection_proof']['deadline_monotonic'],30.)
|
|
r=s.run();self.assertTrue(r['completed']);self.assertEqual(r['original_deadline_monotonic'],44.5)
|
|
self.assertEqual(s.report['action_selection_proof']['deadline_monotonic'],30.)
|
|
def test_insufficient_original_budget_sends_nothing(self):
|
|
s=Scene();s.calibration['observer_deadline_monotonic']=35.9
|
|
with self.assertRaisesRegex(ValueError,'insufficient-original-budget'):s.run()
|
|
self.assertEqual(s.events,[]);self.assertEqual(s.report['held_action_proof']['status'],'refused-no-input')
|
|
def test_no_verified_selection_no_press(self):
|
|
for fault in ('verdict','nonce','action','pair','deadline'):
|
|
s=Scene()
|
|
if fault=='verdict':s.evidence['verified']=False
|
|
if fault=='nonce':s.calibration['nonce']='foreign'
|
|
if fault=='action':s.evidence['selected_action_id']=8
|
|
if fault=='pair':s.report['action_selection_proof']['selection_samples']=[298,300]
|
|
if fault=='deadline':s.calibration['observer_deadline_monotonic']=60
|
|
with self.subTest(fault=fault),self.assertRaises(ValueError):s.run()
|
|
self.assertEqual(s.events,[])
|
|
def test_early_reset_or_missing_native_input_refuses_and_releases(self):
|
|
for fault in ('reset','pressed','performing','decrease','unbound','selection','permission'):
|
|
s=Scene()
|
|
def mutate(x,e):
|
|
if e is None or e<1 or x['phase']!=1:return
|
|
if fault=='reset':x['progress']=0
|
|
if fault=='pressed':x['pressed']=0
|
|
if fault=='performing':x['performing']=False
|
|
if fault=='decrease':x['progress']=.01
|
|
if fault=='unbound':x['progress_bound']=False
|
|
if fault=='selection':x['selected']=False
|
|
if fault=='permission':x['cached_performable']=False
|
|
s.mutate=mutate
|
|
with self.subTest(fault=fault),self.assertRaises(ValueError):s.run()
|
|
self.assertEqual([e[1] for e in s.events],[True,False]);self.assertFalse(s.native_held)
|
|
self.assertFalse(s.report['held_action_proof']['completed'])
|
|
def test_phase_without_advancing_progress_is_not_completion(self):
|
|
s=Scene();s.phase_delay=.1
|
|
with self.assertRaisesRegex(ValueError,'advancing bound progress'):s.run()
|
|
self.assertEqual([e[1] for e in s.events],[True,False])
|
|
def test_intermediate_identity_geometry_body_input_phase_refusals(self):
|
|
for fault in ('phase0','phase3','body','turn','input','menu','geometry','gap','nonce'):
|
|
s=Scene()
|
|
def mutate(x,e):
|
|
if e is None or not .8<e<1.:return
|
|
if fault=='phase0':x['phase']=0
|
|
if fault=='phase3':x['phase']=3
|
|
if fault=='body':x['origin'][0]+=.06
|
|
if fault=='turn':x['turn'][0]=.02
|
|
if fault=='input':x['input'][2]=1
|
|
if fault=='menu':x['inventory']=True
|
|
if fault=='geometry':x['geometry']['context'][0]+=.1
|
|
if fault=='gap':x['sample']+=1;x['choices']['sample']=x['sample']
|
|
if fault=='nonce':x['geometry']['nonce']='foreign'
|
|
s.mutate=mutate
|
|
with self.subTest(fault=fault),self.assertRaises(ValueError):s.run()
|
|
self.assertEqual([e[1] for e in s.events],[True,False])
|
|
def test_phase_two_reversal_refuses_even_after_release(self):
|
|
s=Scene();s.motion_delay=1.
|
|
def mutate(x,e):
|
|
if e is not None and e>4.3:x['phase']=1
|
|
s.mutate=mutate
|
|
with self.assertRaisesRegex(ValueError,'phase reversed'):s.run()
|
|
self.assertTrue(s.report['held_action_proof']['unlock_observed']);self.assertFalse(s.report['held_action_proof']['completed'])
|
|
def test_unlock_without_motion_or_control_is_not_full_proof(self):
|
|
for fault in ('no_motion','no_control'):
|
|
s=Scene();setattr(s,fault,True)
|
|
with self.subTest(fault=fault),self.assertRaises(ValueError):s.run()
|
|
r=s.report['held_action_proof'];self.assertTrue(r['unlock_observed']);self.assertFalse(r['completed'])
|
|
self.assertEqual(r['status'],'unlock-observed-motion-unverified');self.assertFalse(s.native_held)
|
|
def test_renewed_native_input_after_acknowledgement_refuses_before_delayed_motion(self):
|
|
for buffered in (False,True):
|
|
s=Scene();s.motion_delay=.8
|
|
def mutate(x,e):
|
|
if e is not None and 4.35<e<4.45:x['pressed']=1.
|
|
s.mutate=mutate
|
|
if buffered:
|
|
observe=s.observe
|
|
def batched():
|
|
if s.report.get('held_action_proof',{}).get('release_barrier_sample') is not None and not hasattr(s,'extra_batch_delay'):
|
|
s.extra_batch_delay=True;s.sleep(.35)
|
|
return observe()
|
|
s.observe=batched
|
|
with self.subTest(buffered=buffered),self.assertRaisesRegex(ValueError,'renewed after release'):
|
|
s.run()
|
|
self.assertFalse(s.report['held_action_proof']['completed']);self.assertFalse(s.native_held)
|
|
self.assertEqual([e[1] for e in s.events],[True,False])
|
|
def test_delayed_motion_success_has_latest_neutral_release_acknowledgement(self):
|
|
s=Scene();s.motion_delay=.8;r=s.run()
|
|
self.assertTrue(r['completed']);self.assertEqual(r['released_sample'],r['final_sample'])
|
|
self.assertEqual(r['samples'][-1]['pressed'],0)
|
|
|
|
def test_missing_native_released_input_refuses(self):
|
|
s=Scene();s.native_release_missing=True
|
|
with self.assertRaises(ValueError):s.run()
|
|
r=s.report['held_action_proof'];self.assertTrue(r['release_confirmed']);self.assertIsNone(r['released_sample']);self.assertFalse(r['completed'])
|
|
def test_incomplete_action_releases_before_original_five_second_lease(self):
|
|
s=Scene();s.never_complete=True
|
|
with self.assertRaisesRegex(ValueError,'release reserve'):s.run()
|
|
r=s.report['held_action_proof'];self.assertEqual([e[1] for e in s.events],[True,False]);self.assertLess(r['up_monotonic'],r['held_deadline_monotonic']);self.assertFalse(r['completed'])
|
|
def test_uncertain_down_always_attempts_up_and_preserves_primary_error(self):
|
|
s=Scene();s.unknown_down=True
|
|
with self.assertRaisesRegex(ValueError,'down request outcome uncertain'):s.run()
|
|
r=s.report['held_action_proof'];self.assertEqual([e[1] for e in s.events],[True,False]);self.assertTrue(r['release_confirmed']);self.assertFalse(r['completed'])
|
|
s=Scene();s.unknown_down=True;s.fail_up=True
|
|
with self.assertRaisesRegex(ValueError,'down request outcome uncertain'):s.run()
|
|
r=s.report['held_action_proof'];self.assertEqual(r['unconfirmed_held_key'],'F');self.assertIn('up refused',r['paired_release_errors'])
|
|
def test_lost_owner_never_redirects_uncertain_release(self):
|
|
s=Scene();original=s.down_f
|
|
def down(*a,**kw):result=original(*a,**kw);s.focus=False;return result
|
|
s.down_f=down
|
|
with self.assertRaisesRegex(ValueError,'original private owner lost'):s.run()
|
|
self.assertEqual([e[1] for e in s.events],[True]);self.assertEqual(s.report['held_action_proof']['unconfirmed_held_key'],'F')
|
|
def test_late_release_cannot_qualify(self):
|
|
s=Scene();s.release_late=True
|
|
with self.assertRaisesRegex(ValueError,'release exceeded'):s.run()
|
|
self.assertFalse(s.report['held_action_proof']['completed']);self.assertFalse(s.native_held)
|
|
def test_final_adapter_guard_rechecks_six_seconds_before_any_down(self):
|
|
s=Scene();s.calibration['observer_deadline_monotonic']=36.3
|
|
original=s.down_f
|
|
def slow(*a,**kw):s.sleep(.4);return original(*a,**kw)
|
|
s.down_f=slow
|
|
with self.assertRaisesRegex(ValueError,'adapter admission expired'):s.run()
|
|
self.assertFalse(any(event[1] for event in s.events))
|
|
# A down adapter refusal remains uncertain until guarded cleanup proves up.
|
|
self.assertEqual([event[1] for event in s.events],[False])
|
|
self.assertTrue(s.report['held_action_proof']['release_confirmed'])
|
|
|
|
def test_constant_progress_never_substitutes_for_two_advances(self):
|
|
s=Scene()
|
|
def mutate(x,e):
|
|
if e is not None and x['phase']==1 and x['progress']>0:x['progress']=.5
|
|
s.mutate=mutate
|
|
with self.assertRaisesRegex(ValueError,'advancing bound progress'):s.run()
|
|
self.assertFalse(s.report['held_action_proof']['unlock_observed'])
|
|
self.assertEqual([x[1] for x in s.events],[True,False])
|
|
|
|
def test_first_post_release_buffered_reset_is_not_release_acknowledgement(self):
|
|
s=Scene();s.no_motion=True
|
|
original=s.observe
|
|
def observe():
|
|
batch=original()
|
|
if s.up is not None:
|
|
# Freeze the first post-release batch forever: all are reset,
|
|
# but none is newer than the full drained acknowledgement barrier.
|
|
if not hasattr(s,'frozen_after_up'):s.frozen_after_up=copy.deepcopy(batch)
|
|
return copy.deepcopy(s.frozen_after_up[-1:]) if s.last_read>s.frozen_after_up[-1]['sample'] else batch
|
|
return batch
|
|
s.observe=observe
|
|
with self.assertRaises(ValueError):s.run()
|
|
r=s.report['held_action_proof'];self.assertTrue(r['release_confirmed']);self.assertIsNone(r['released_sample'])
|
|
|
|
def test_primary_failure_survives_notification_cleanup_failure(self):
|
|
s=Scene();s.unknown_down=True
|
|
def notify(event):
|
|
s.notifications.append(event)
|
|
if event['kind']=='held-action-finished':raise policy.Refused('supervisor pipe closed')
|
|
with self.assertRaisesRegex(ValueError,'down request outcome uncertain'):
|
|
policy.perform_held_action(s,s.idle,s,s.observe,s.geometry,s.calibration,s.evidence,s.report,
|
|
clock=s.clock,sleep=s.sleep,notify=notify)
|
|
r=s.report['held_action_proof'];self.assertEqual(r['error'],'down request outcome uncertain')
|
|
self.assertEqual(r['notification_errors'],['supervisor pipe closed'])
|
|
self.assertEqual([x[1] for x in s.events],[True,False])
|
|
|
|
def test_retry_cannot_press_again(self):
|
|
s=Scene();s.run();count=len(s.events)
|
|
with self.assertRaisesRegex(ValueError,'already exists'):s.run()
|
|
self.assertEqual(len(s.events),count)
|
|
def test_stale_snapshot_and_old_initial_phase_refuse_without_input(self):
|
|
for fault in ('stale','phase'):
|
|
s=Scene()
|
|
if fault=='stale':s.stale=True;s.sleep(.1)
|
|
else:s.history[0]['phase']=2
|
|
with self.subTest(fault=fault),self.assertRaises(ValueError):s.run()
|
|
self.assertEqual(s.events,[])
|
|
|
|
if __name__=='__main__':unittest.main()
|