"""Opt-in held runner wiring and independent proof boundary; no runtime or input.""" import copy import hashlib import json from pathlib import Path import sys import tempfile from types import SimpleNamespace, ModuleType import unittest from unittest.mock import patch sys.path[:0]=[str(Path(__file__).resolve().parents[1]/'tools'),str(Path(__file__).resolve().parent)] import headless_game as runner import test_relative_look_runner as selection_fixtures import test_headless_input_exercise_runner as exercise_fixtures from test_ordinary_input_runner import NONCE MODULES=('headless_input_proof.py','headless_display_probe.py','prison_approach_proof.py','ordinary_gate_evidence.py','ordinary_look_evidence.py','prison_approach_evidence.py','prison_alignment_proof.py','prison_alignment_evidence.py','ordinary_gate_choices.py','prison_action_selection_proof.py','prison_action_selection_evidence.py','prison_held_action_proof.py','prison_held_action_evidence.py') class HeldRunnerTests(unittest.TestCase): def options(self): return dict(exercise_prison_held_action=True,exercise_prison_coarse_facing=True,exercise_prison_coarse_approach=True,exercise_prison_selection=True,exercise_prison_alignment=True,observe_prison_choices=True,exercise_prison_approach=True,exercise_relative_look=True,observe_prison_gate=True,exercise_player_input=True,observe_player_input=True,inspect_headless_display=True,project='chat',settle=50) def test_all_prior_capabilities_required_before_filesystem_or_launch(self): good=self.options();runner.validate_input_exercise(SimpleNamespace(**good));runner.validate_input_exercise(SimpleNamespace()) for change in [{key:False} for key in good if key not in ('exercise_prison_held_action','project','settle')]+[{'exercise_prison_held_action':1},{'exercise_prison_held_action':'yes'},{'settle':49},{'project':runner.CONTROL},{'packed_core':Path('/packed')},{'packed_addons':Path('/packed')},{'packed_session':Path('/packed')}]: with self.subTest(change=change),patch.object(runner,'packed_request',side_effect=AssertionError('preflight reached filesystem')): with self.assertRaises(ValueError):runner.run(SimpleNamespace(**(good|change))) def test_actual_cli_parser_forwards_optin_and_preserves_default(self): for enabled in (False,True): argv=['headless_game.py','--compat','/private/unlaunched'] if enabled: argv+=['--'+key.replace('_','-') for key,value in self.options().items() if value is True] argv+=['--project','chat','--settle','50'] captured=[] def fake_run(args):runner.validate_input_exercise(args);captured.append(args);return 17 with patch.object(sys,'argv',argv),patch.object(runner,'run',side_effect=fake_run): self.assertEqual(runner.main(),17) self.assertEqual(captured[0].exercise_prison_held_action,enabled) def setup_command(self,root,held=True): (root/'tools').mkdir();profile=root/'profile';profile.mkdir() for name in MODULES:(root/'tools'/name).write_text('# private stub '+name) native=['gamescope','--backend','headless','--','/proton','run','/game'] ordinary=runner.input_observer_command(native,'chat',None,NONCE,True,True,True,held_action=held) wrapped,display=runner.display_probe_command(ordinary,root,Path('/compat'),profile,Path('/game'),NONCE,':0') return native,ordinary,wrapped,(wrapped,root,Path('/compat'),profile,Path('/game'),NONCE,':0',display) def test_native_and_helper_flags_exact_nonce_inside_existing_boundary(self): with tempfile.TemporaryDirectory() as folder: root=Path(folder);native,ordinary,wrapped,args=self.setup_command(root) cmd,context=runner.input_exercise_command(*args,look=True,approach=True,alignment=True,selection=True,coarse_approach=True,coarse_facing=True,held_action=True) self.assertEqual(ordinary[-2:],['-errOrdinaryHeldAction',NONCE]);self.assertEqual(cmd[-len(wrapped[4:]):],wrapped[4:]);self.assertEqual(context['command'],wrapped[4:]) self.assertEqual(cmd.count('--enable-private-held-action-proof'),1);self.assertTrue(context['held_action_requested']) self.assertEqual(context['held_module_sha256'],{name:hashlib.sha256((root/'tools'/name).read_bytes()).hexdigest() for name in MODULES[-2:]}) normal=runner.input_observer_command(native,'chat',None,NONCE,True,True,True) self.assertEqual(ordinary,normal+['-errOrdinaryHeldAction',NONCE]) for tail in ([],['-errOrdinaryHeldAction'],['-errOrdinaryHeldAction','foreign'],['-errOrdinaryHeldAction',NONCE]*2,['-errOrdinaryHeldAlias',NONCE]): altered=wrapped[:-2]+tail;changed=(altered,*args[1:]) with self.subTest(tail=tail),self.assertRaises(ValueError):runner.input_exercise_command(*changed,look=True,approach=True,alignment=True,selection=True,coarse_approach=True,coarse_facing=True,held_action=True) with self.assertRaises(ValueError):runner.input_exercise_command(*args,look=True,approach=True,alignment=True,selection=True,coarse_approach=True,coarse_facing=True) with self.assertRaises(ValueError):runner.input_exercise_command(*args,look=True,approach=True,alignment=True,selection=True,coarse_approach=True,held_action=True) def test_no_implicit_native_or_helper_activation_and_module_refusal(self): with tempfile.TemporaryDirectory() as folder: root=Path(folder);native,ordinary,wrapped,args=self.setup_command(root,False) cmd,context=runner.input_exercise_command(*args,look=True,approach=True,alignment=True,selection=True,coarse_approach=True,coarse_facing=True) self.assertNotIn('--enable-private-held-action-proof',cmd);self.assertNotIn('held_module_sha256',context);self.assertFalse(context.get('held_action_requested',False)) for flags in ((False,True,True),(True,False,True),(True,True,False)): with self.assertRaises(ValueError):runner.input_observer_command(native,'chat',None,NONCE,*flags,held_action=True) for duplicate in ('-errOrdinaryHeldAction','-errOrdinaryHeldAlias'): with self.assertRaises(ValueError):runner.input_observer_command(native+[duplicate,NONCE],'chat',None,NONCE,True,True,True,held_action=True) for name in MODULES[-2:]: path=root/'tools'/name;old=path.read_bytes();path.unlink() with self.assertRaises(ValueError):runner.held_module_hashes(root) path.symlink_to(root/'tools'/MODULES[0]) with self.assertRaises(ValueError):runner.held_module_hashes(root) path.unlink();path.write_bytes(old) def fixture(self,root): session,context,report,observation=selection_fixtures.SelectionRunner.make_selection(self,root) context.update(held_action_requested=True,coarse_facing_requested=True,coarse_approach_requested=True,held_module_sha256=runner.held_module_hashes(runner.ROOT)) report.update({k:context[k] for k in ('held_action_requested','coarse_facing_requested','coarse_approach_requested','held_module_sha256')}) report['approach_proof']['profile']='coarse';report['facing_proof']['profile']='coarse' held={'requested':True,'completed':True,'input_attempted':True,'status':'verified','down_monotonic':32.,'up_monotonic':36.2,'release_confirmed':True,'unconfirmed_held_key':None,'paired_release_errors':[],'notification_errors':[], 'events':[{'key':'F','down':down,'monotonic':at,'attempted_monotonic':at-.01,'confirmed':True} for down,at in ((True,32.),(False,36.2))]} report.update(held_action_proof=held,held_release_errors=[],held_events=[{'key':'F','down':down,'monotonic':at,'sent':True} for down,at in ((True,32.),(False,36.2))]) return session,context,report,observation def result(self,root,change=None,accepted=True,prior='game-started'): session,context,report,observation=self.fixture(root) if change:change(context,report,observation) raw=(json.dumps(report,indent=2)+'\n').encode();Path(context['report']).write_bytes(raw) approach=ModuleType('prison_approach_evidence');approach.validate_approach_evidence=lambda *a:{'verified':True} selection=ModuleType('prison_action_selection_evidence');selection.validate_selection_evidence=lambda *a,**kw:{'verified':True} held=ModuleType('prison_held_action_evidence');calls=[] def validate(*args,**kwargs):calls.append((args,kwargs));return {'verified':accepted,'held_action_verified':accepted,'unlock_observed':accepted,'gate_motion_observed':accepted} held.validate_held_action_evidence=validate with patch.dict(sys.modules,{'prison_approach_evidence':approach,'prison_action_selection_evidence':selection,'prison_held_action_evidence':held}): status,proof=runner.input_exercise_result(prior,context,session,observation,{'verified':True}) self.assertEqual((session/'input-proof.json').read_bytes(),raw) return status,proof,calls,context,report def test_raw_validator_receives_exact_archived_content_and_coarse_authority(self): for accepted,prior in ((True,'game-started'),(False,'game-started'),(True,'game-failed'),(True,'cleanup-failed')): with tempfile.TemporaryDirectory() as folder: status,proof,calls,context,report=self.result(Path(folder),accepted=accepted,prior=prior) self.assertEqual(calls,[(('native selection',NONCE,report['look_proof'],report['approach_proof'],report['facing_proof'],report['action_selection_proof'],report['held_action_proof']),{'facing_profile':'coarse'})]) self.assertEqual(proof['verified'],accepted);self.assertFalse(proof['gameplay_qualified']) self.assertEqual(status,'input-exercise-incomplete' if not accepted and prior=='game-started' else prior) self.assertIn('no gate passage',proof['scope']) def test_wrong_hash_scope_or_release_and_extra_input_never_reach_raw_credit(self): changes=[lambda c,p,o:p.update(held_action_requested=False),lambda c,p,o:p.update(held_action_requested=1),lambda c,p,o:c.update(held_action_requested=1),lambda c,p,o:p.update(held_module_sha256={}),lambda c,p,o:c.update(held_module_sha256={}),lambda c,p,o:c.update(coarse_facing_requested=False),lambda c,p,o:p.update(held_release_errors=['failed']),lambda c,p,o:p.update(unconfirmed_held_key='F'),lambda c,p,o:p.update(all_attempted_presses_released=False),lambda c,p,o:p['held_events'].append(copy.deepcopy(p['held_events'][1])),lambda c,p,o:p['held_events'][0].update(key='W'),lambda c,p,o:p['held_events'][0].update(down=1),lambda c,p,o:p['held_events'][0].update(monotonic=True),lambda c,p,o:p['held_events'][0].update(monotonic=float('nan')),lambda c,p,o:p['held_events'][0].update(sent=False),lambda c,p,o:p['held_events'][1].update(finally_=True),lambda c,p,o:p['held_action_proof'].update(release_confirmed=False),lambda c,p,o:p['held_action_proof'].update(notification_errors=['failed']),lambda c,p,o:p['held_action_proof']['events'][1].update(cleanup=True),lambda c,p,o:p['held_action_proof']['events'][1].update(monotonic=36.199),lambda c,p,o:p['held_action_proof'].update(up_monotonic=36.199),lambda c,p,o:p['held_action_proof'].update(status='failed-after-input'),lambda c,p,o:o.update(sampler_complete=False)] for change in changes: with self.subTest(change=change),tempfile.TemporaryDirectory() as folder: status,proof,calls,*_=self.result(Path(folder),change) self.assertEqual(status,'input-exercise-incomplete');self.assertFalse(proof['verified']);self.assertEqual(calls,[]) def test_late_or_reversed_pair_refuses_even_with_internally_equal_timestamps(self): for down,up in ((32.,37.001),(32.,31.),(31.,35.)): def change(c,p,o): for i,(name,value) in enumerate((('down_monotonic',down),('up_monotonic',up))): p['held_events'][i]['monotonic']=value;p['held_action_proof'][name]=value;p['held_action_proof']['events'][i]['monotonic']=value with tempfile.TemporaryDirectory() as folder: status,proof,calls,*_=self.result(Path(folder),change);self.assertFalse(proof['verified']);self.assertEqual(calls,[]) def test_unrequested_F_or_held_result_refuses_and_exact_failure_poll_is_bound(self): for payload in ({'held_events':[{'key':'F'}]},{'held_action_proof':{}},{'held_action_requested':True},{'held_release_errors':['uncertain']}): with tempfile.TemporaryDirectory() as folder: session,context,report,observation=exercise_fixtures.ExerciseTests.make_proof(self,Path(folder));report.update(payload);Path(context['report']).write_text(json.dumps(report)) status,proof=runner.input_exercise_result('game-started',context,session,observation,{'verified':True});self.assertFalse(proof['verified']) with tempfile.TemporaryDirectory() as folder: session,context,report,observation=self.fixture(Path(folder));report.update(error='held release failed',sequence_completed=False) Path(context['report']).write_text(json.dumps(report));self.assertTrue(runner.input_exercise_poll_failure(context));self.assertEqual(runner.input_exercise_poll_result(context,None,True),'game-failed') for update in ({'held_action_requested':False},{'held_action_requested':1},{'held_module_sha256':{}}): Path(context['report']).write_text(json.dumps(report|update));self.assertFalse(runner.input_exercise_poll_failure(context)) class FullRawHeldRunnerTests(unittest.TestCase): @classmethod def setUpClass(cls): from test_prison_held_action_evidence import build cls.native_fixture=build('coarse') def test_actual_policy_chain_through_all_raw_validators_and_runner(self): from test_prison_held_action_evidence import raw for fault in (None,'native_progress','native_door_motion','adapter_timestamp','expected_profile'): with self.subTest(fault=fault),tempfile.TemporaryDirectory() as folder: frames,look,route,facing,selected,held=copy.deepcopy(self.native_fixture) session,context,report,observation=HeldRunnerTests.fixture(self,Path(folder)) context['nonce']='test';report['nonce']='test';observation['nonce']='test' if fault=='native_progress':frames[held['progress_samples'][0]]['progress']=3.9 if fault=='native_door_motion': for index in range(held['transition_sample'],held['final_sample']+1):frames[index]['door_state']=0 report.update(look_proof=look,approach_proof=route,facing_proof=facing,action_selection_proof=selected,held_action_proof=held) report['events']=copy.deepcopy(route['events']) report['wheel_events']=[{'button':5,'down':down,'sent':True,'monotonic':selected['wheel'][field]} for down,field in ((True,'down_monotonic'),(False,'up_monotonic'))] report['held_events']=[{'key':'F','down':down,'sent':True,'monotonic':held[field]} for down,field in ((True,'down_monotonic'),(False,'up_monotonic'))] if fault=='adapter_timestamp':report['held_events'][1]['monotonic']+=.001 if fault=='expected_profile':context['coarse_facing_requested']=False;report['coarse_facing_requested']=False # Match the real wrapper's compact serialization, leaving raw native logs # as the independent source for every state/position/progress measurement. for record in (route,facing,selected,held): record['samples']=[{key:x[key] for key in ('sample','real_ms','received_monotonic')} for x in record['samples']] content=raw(frames);(session/'console.log').write_text(content) original=(json.dumps(report,indent=1)+'\n').encode();Path(context['report']).write_bytes(original) status,result=runner.input_exercise_result('game-started',context,session,observation,{'verified':True}) self.assertEqual((session/'input-proof.json').read_bytes(),original) if fault is None: self.assertEqual(status,'game-started',result);self.assertTrue(result['verified'],result) self.assertTrue(result['held_action_proof']['held_action_verified']);self.assertFalse(result['held_action_proof']['escape_verified']) self.assertEqual(result['held_action_proof']['transition_sample'],held['transition_sample']) else:self.assertEqual(status,'input-exercise-incomplete',result);self.assertFalse(result['verified'],result) if __name__=='__main__':unittest.main()