Prepare independent successor playtest with spawn recovery and casualty feedback

tools/headless_input_proof.py

@@ -1,6 +1,7 @@
 #!/usr/bin/env python3
 """Opt-in bounded private XTest keyboard and optional relative-look experiment. Not a launcher or gameplay proof."""
 import argparse
+import copy
 import ctypes as C
 import hashlib
 import json
@@ -46,9 +47,19 @@ def option(args, name):
 def gate_options(native, args):
     # Two exact read-only options; a prefix match must never admit an alias,
     # duplicate, foreign nonce or extra diagnostic in keyboard-only mode.
+    held = getattr(args, 'enable_private_held_action_proof', False)
+    held_tokens = [value for value in native if value.startswith('-errOrdinaryHeld')]
+    if type(held) is not bool or (held and not (getattr(args, 'enable_private_coarse_facing_proof', False) and getattr(args, 'enable_private_selection_proof', False))):
+        raise Refused('Held action requires explicit coarse facing and selection')
+    if held_tokens != (['-errOrdinaryHeldAction'] if held else []):
+        raise Refused('Unexpected, missing or repeated native held observer option')
+    if held and option(native, '-errOrdinaryHeldAction') != args.nonce:
+        raise Refused('Native held observer nonce mismatch')
     tokens = [value for value in native if value.startswith('-errOrdinaryGate')]
     if any(value not in {'-errOrdinaryGateObserve', '-errOrdinaryGateChoices'} for value in tokens):
         raise Refused('Unknown native gate observer option')
+    if getattr(args, 'enable_private_coarse_facing_proof', False) and not (getattr(args, 'enable_private_coarse_approach_proof', False) and getattr(args, 'enable_private_selection_proof', False)):
+        raise Refused('Coarse facing requires coarse approach and selection')
     approach = getattr(args, 'enable_private_approach_proof', False)
     if getattr(args, 'enable_private_coarse_approach_proof', False) and not approach:
         raise Refused('Coarse approach requires explicit approach proof')
@@ -136,9 +147,14 @@ def yaw_delta(first, second):
 
 
 class RelativeMouse:
-    def __init__(self, query, keyboard, owner, enable_approach=False, enable_alignment=False, enable_coarse_approach=False):
+    def __init__(self, query, keyboard, owner, enable_approach=False, enable_alignment=False, enable_coarse_approach=False, enable_coarse_facing=False, enable_selection=False):
         if type(enable_coarse_approach) is not bool or (enable_coarse_approach and not enable_approach):
             raise Refused("Coarse approach requires explicit approach capability")
+        if type(enable_coarse_facing) is not bool or (enable_coarse_facing and not (enable_coarse_approach is True and enable_alignment is True and enable_selection is True)):
+            raise Refused('Coarse facing requires explicit selection and coarse approach capability')
+        self.enable_coarse_facing = enable_coarse_facing
+        self.alignment_pixels = 0
+        self.alignment_requests = 0
         self.enable_coarse_approach = enable_coarse_approach
         self.query, self.keyboard, self.owner = query, keyboard, owner
         self.enable_approach = enable_approach
@@ -189,13 +205,19 @@ class RelativeMouse:
 
     def alignment_move(self, dx, dy, guard, *, deadline, clock=time.monotonic):
         if (self.enable_alignment is not True or self.enable_approach is not True or type(dx) is not int
-                or not 1 <= abs(dx) <= 128 or type(dy) is not int or dy != 0):
+                or not 1 <= abs(dx) <= (384 if self.enable_coarse_facing else 128) or type(dy) is not int or dy != 0):
             raise Refused('Unsupported or unenabled alignment relative step')
         def within_lease():
-            if clock() >= deadline:
+            if type(deadline) not in (int, float) or not math.isfinite(deadline) or not 0 <= clock() < deadline:
                 raise Refused('Original alignment deadline exceeded before relative input')
         within_lease()
-        self._move(dx, dy, guard, before_send=within_lease)
+        if self.alignment_requests >= 6 or self.alignment_pixels+abs(dx) > 768:
+            raise Refused('Alignment request or cumulative pixel budget exhausted')
+        def commit():
+            within_lease()
+            self.alignment_requests += 1
+            self.alignment_pixels += abs(dx)
+        self._move(dx, dy, guard, before_send=commit)
 
     def _move(self, dx, dy, guard, before_send=None):
         target = guard.check()
@@ -304,7 +326,11 @@ class Observer:
     """Only records appended after this exact owned console was opened count."""
     pattern = re.compile(r'ERR INPUT OBSERVE: run=([A-Za-z0-9]{1,32}) sample=(\d+) real=(\d+) world=(\d+(?:\.\d+)?) (.+)$')
 
-    def __init__(self, path, nonce, clock=time.monotonic, enable_look=False, enable_approach=False, enable_selection=False):
+    def __init__(self, path, nonce, clock=time.monotonic, enable_look=False, enable_approach=False, enable_selection=False, enable_held=False):
+        if type(enable_held) is not bool or (enable_held and not (enable_selection and enable_approach and enable_look)):
+            raise Refused('Held observation requires all original observation capabilities')
+        self.enable_held = enable_held
+        self.held_previous = None; self.held_anchor = None
         self.path, self.nonce, self.clock = path, nonce, clock
         self.fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW)
         info = os.fstat(self.fd)
@@ -476,8 +502,42 @@ class Observer:
     def selection_snapshots(self):
         return self._snapshot_batch('selection_previous')
 
-    def _snapshot_batch(self, cursor):
+    def begin_held(self, final_sample):
+        if self.enable_held is not True or self.held_anchor is not None or type(final_sample) is not int:
+            raise Refused('Held observation cannot be started or re-adopted')
         _, data = self.approach_data()
+        if final_sample not in data['snapshots'] or final_sample < data['ready']['sample']:
+            raise Refused('Held observation lacks original selection anchor')
+        self.held_anchor = final_sample; self.held_previous = final_sample-1
+
+    def held_data(self):
+        # Native observer retains original world/player/action/door objects; the
+        # same acknowledged stream may show their moving geometry after phase2.
+        # Do not run the closed-yard geometry validator on that moving door.
+        from ordinary_gate_choices import join_choice_snapshots
+        self.fresh()
+        if (self.enable_held is not True or self.held_anchor is None or self.events != ['READY'] or
+                not self.gate_complete_end or any(r['kind'] == 'cancelled' for r in self.gate_records+self.look_records)):
+            raise Refused('Original held observation context ended or was not adopted')
+        content = '\n'.join(self.approach_lines[:self.gate_complete_end])
+        data = join_choice_snapshots(content, self.nonce)
+        latest = max(data['snapshots'])
+        if (latest < self.held_anchor or self.clock()-self.gate_received[latest] > .5 or
+                self.input_samples[-1]['real_ms']-data['snapshots'][latest]['real_ms'] > 500):
+            raise Refused('Original held native snapshot is stale or reversed')
+        phase = 1
+        for index in range(self.held_anchor, latest+1):
+            item = data['snapshots'][index]
+            if item['action']['phase'] < phase or item['input']['inventory']:
+                raise Refused('Original held phase or menu context changed')
+            phase = item['action']['phase']
+        return content, data
+
+    def held_snapshots(self):
+        return self._snapshot_batch('held_previous')
+
+    def _snapshot_batch(self, cursor):
+        _, data = self.held_data() if cursor == 'held_previous' else self.approach_data()
         snapshots = data['snapshots']; latest = max(snapshots)
         previous = getattr(self, cursor)
         first = latest if previous is None else min(previous + 1, latest)
@@ -493,9 +553,11 @@ class Observer:
                 'inventory': native['inventory'], 'phase': action['phase'], 'door_state': geo['door_state'], 'door_control': geo['door_control'],
                 'geometry': {'nonce': self.nonce, 'basis': 'registered-prison-lock-native-interaction',
                     **{key: geo[key] for key in ('prison', 'gate', 'gate_forward', 'context', 'radius')}}})
+            if self.enable_held:
+                result[-1].update(world_ms=native['world_ms'], turn=pose['turn'])
             if self.enable_selection:
                 choice = item['choices']
-                if cursor != 'approach_previous' and action['context_selected']:
+                if cursor != 'approach_previous' and action['phase'] == 1 and action['context_selected']:
                     from ordinary_gate_choices import exact_prison_choices
                     choice = exact_prison_choices(item)
                 result[-1]['choices'] = {key: choice[key] for key in ('nonce','sample','real_ms','selected_id','selected_class','actions')}
@@ -545,6 +607,89 @@ class Keyboard:
         self.query.x.XFlush(self.query.d)
 
 
+class HeldFKey:
+    """One F-only attempt on the captured private connection; never creates one."""
+    def __init__(self, guard, keyboard, mouse, report, clock, sleep, original_deadline, *, enabled=False):
+        if enabled is not True or type(original_deadline) not in (int,float) or not math.isfinite(original_deadline):
+            raise Refused('Held F requires explicit capability and original deadline')
+        self.guard, self.keyboard, self.mouse = guard, keyboard, mouse
+        self.query, self.xt = keyboard.query, keyboard.xt
+        self.owner = copy.deepcopy(mouse.owner)
+        self.report, self.clock, self.sleep = report, clock, sleep
+        self.original_deadline = original_deadline
+        self.attempted = False; self.pending = False; self.lease = None; self.up_attempts = 0
+        self._context(guard)
+        self.code = int(self.query.x.XKeysymToKeycode(self.query.d, 0x66))
+        if (not 8 <= self.code <= 255 or self.query.x.XKeycodeToKeysym(self.query.d, self.code, 0) != 0x66 or
+                self.code in keyboard.codes.values()):
+            raise Refused('Default distinct unmodified F mapping unavailable')
+        report.setdefault('held_events', []); report.setdefault('held_release_errors', [])
+
+    def _context(self, guard):
+        if (guard is not self.guard or self.guard.query is not self.query or self.keyboard.query is not self.query or
+                self.mouse.query is not self.query or self.mouse.keyboard is not self.keyboard or
+                self.keyboard.xt is not self.xt or self.mouse.owner != self.owner or self.guard.owner != self.owner):
+            raise Refused('Captured F owner or connection changed')
+
+    def _owned(self, guard, *, idle):
+        self._context(guard)
+        target = guard.check()
+        if idle: self.keyboard.require_idle()
+        self.mouse.validate_pointer(target,0,0)
+        target = guard.check()
+        if idle: self.keyboard.require_idle()
+        self.mouse.validate_pointer(target,0,0)
+        self._context(guard)
+
+    def down_f(self, guard, *, deadline, not_after):
+        if (self.attempted or type(deadline) not in (int,float) or not math.isfinite(deadline) or
+                type(not_after) not in (int,float) or not math.isfinite(not_after) or
+                not 0 <= self.clock() <= not_after <= self.original_deadline-6 or
+                not self.clock() < deadline <= min(self.original_deadline,self.clock()+5)):
+            raise Refused('Repeated, expired or unqualified held F admission')
+        # The numeric original keycode is retained for cleanup even if its
+        # mapping later changes; never redirect key-up to a new F keycode.
+        self.attempted = True; self.pending = True; self.lease = deadline
+        self.report['unconfirmed_held_key'] = 'F'
+        self._owned(guard,idle=True)
+        if self.query.x.XKeycodeToKeysym(self.query.d,self.code,0) != 0x66:
+            raise Refused('Original F mapping changed before down')
+        now = self.clock()
+        if not 0 <= now <= not_after or now >= self.lease:
+            raise Refused('Six-second admission or native pose freshness expired before F')
+        event = {'key':'F','down':True,'monotonic':now,'sent':False}
+        self.report['held_events'].append(event); self.report['input_attempted'] = True
+        if not self.xt.XTestFakeKeyEvent(self.query.d,self.code,1,0):
+            raise Refused('Native held F down request uncertain')
+        self.query.x.XFlush(self.query.d)
+        event['sent'] = True; self.report['input_sent'] = True
+        return {'key':'F','down_monotonic':now}
+
+    def up_f(self, guard, *, deadline, cleanup=False):
+        try:
+            if (type(cleanup) is not bool or not self.attempted or not self.pending or self.up_attempts >= 2 or
+                    type(deadline) not in (int,float) or not math.isfinite(deadline) or deadline != self.lease):
+                raise Refused('Unowned, repeated or renewed held F release')
+            self.up_attempts += 1
+            self._owned(guard,idle=False)
+            now = self.clock()
+            if not math.isfinite(now) or now < 0 or (not cleanup and now >= min(self.lease,self.original_deadline)):
+                raise Refused('Original F release deadline expired')
+            event = {'key':'F','down':False,'monotonic':now,'sent':False}
+            if cleanup: event['finally'] = True
+            self.report['held_events'].append(event); self.report['input_attempted'] = True
+            if not self.xt.XTestFakeKeyEvent(self.query.d,self.code,0,0):
+                raise Refused('Native held F up request uncertain')
+            self.query.x.XFlush(self.query.d)
+            event['sent'] = True; self.report['input_sent'] = True
+            confirm_release(guard,self.keyboard,self.report,'F',self.clock,self.sleep)
+            self.pending = False; self.report['unconfirmed_held_key'] = None
+            return {'key':'F','up_monotonic':now,'released':True}
+        except Exception as error:
+            self.report['held_release_errors'].append(str(error))
+            raise
+
+
 class Guard:
     def __init__(self, args, bound, query, owner):
         self.args, self.bound, self.query, self.owner = args, bound, query, owner
@@ -595,9 +740,13 @@ def confirm_release(guard, keyboard, report, key, clock=time.monotonic, sleep=ti
         observation['end'] = clock()
 
 
-def sequence(guard, keyboard, observer, report, clock=time.monotonic, sleep=time.sleep, mouse=None, approach=False, alignment=False, selection=False, coarse_approach=False):
+def sequence(guard, keyboard, observer, report, clock=time.monotonic, sleep=time.sleep, mouse=None, approach=False, alignment=False, selection=False, coarse_approach=False, coarse_facing=False, held_action=False):
     if type(coarse_approach) is not bool or (coarse_approach and not approach) or report.get("coarse_approach_requested", False) is not coarse_approach:
         raise Refused("Coarse approach requires matching explicit capability")
+    if type(coarse_facing) is not bool or (coarse_facing and not (selection and coarse_approach)) or report.get('coarse_facing_requested', False) is not coarse_facing:
+        raise Refused('Coarse facing requires matching explicit selection capability')
+    if type(held_action) is not bool or report.get('held_action_requested', False) is not held_action or (held_action and not (selection and coarse_facing and getattr(observer, 'enable_held', False))):
+        raise Refused('Held action requires matching explicit selection and observer capabilities')
     deadline = clock() + 70
     held = None
     approach_completed = False
@@ -665,6 +814,9 @@ def sequence(guard, keyboard, observer, report, clock=time.monotonic, sleep=time
             notify({'kind': 'approach-pending', 'deadline_monotonic': deadline})
         if observer.events != ['READY']:
             raise Refused('Inventory was already changed before input')
+        if held_action:
+            original_end = observer.ready_at + (45000-(observer.ready_real-observer.queued_real))/1000-.5
+            notify({'kind':'held-observer-deadline','deadline_monotonic':original_end})
         if mouse is not None:
             perform_look(guard, keyboard, observer, mouse, report, clock, sleep)
             if not approach:
@@ -679,6 +831,8 @@ def sequence(guard, keyboard, observer, report, clock=time.monotonic, sleep=time
                 notify({'kind': 'approach-finished', 'successful': True})
                 if selection:
                     run_selection(guard, keyboard, observer, mouse, report, clock, sleep)
+                    if held_action:
+                        run_held_action(guard, keyboard, observer, mouse, report, clock, sleep)
                 else:
                     run_alignment(guard, keyboard, observer, mouse, report, clock, sleep)
             report["sequence_completed"] = True
@@ -714,9 +868,9 @@ def sequence(guard, keyboard, observer, report, clock=time.monotonic, sleep=time
         if approach:
             route = report.get('approach_proof', {})
             report['paired_release_errors'].extend(route.get('paired_release_errors', []))
-            report['unconfirmed_held_key'] = route.get('unconfirmed_held_key')
+            report['unconfirmed_held_key'] = report.get('unconfirmed_held_key') or route.get('unconfirmed_held_key')
             notify({'kind': 'approach-finished', 'successful': approach_completed if alignment else report['sequence_completed']})
-        report['all_attempted_presses_released'] = not report.get('unconfirmed_held_key') and not report['paired_release_errors'] and not report.get('unconfirmed_held_button') and not report.get('wheel_release_errors')
+        report['all_attempted_presses_released'] = not report.get('unconfirmed_held_key') and not report['paired_release_errors'] and not report.get('unconfirmed_held_button') and not report.get('wheel_release_errors') and not report.get('held_release_errors')
 
 
 class ApproachKeyboard:
@@ -926,6 +1080,10 @@ def run_selection(guard, keyboard, observer, mouse, report, clock, sleep):
     from prison_alignment_proof import perform_facing
     from prison_alignment_evidence import validate_facing_evidence
     from prison_action_selection_proof import perform_selection
+    coarse_facing = report.get('coarse_facing_requested', False)
+    if type(coarse_facing) is not bool or coarse_facing is not getattr(mouse, 'enable_coarse_facing', False):
+        raise Refused('Facing profile differs from owned mouse capability')
+    facing_profile = 'coarse' if coarse_facing else 'standard'
     from prison_approach_evidence import validate_approach_evidence
     if observer.ready_at is None or observer.ready_real is None or observer.queued_real is None:
         raise Refused('Original native alignment clocks missing')
@@ -960,13 +1118,13 @@ def run_selection(guard, keyboard, observer, mouse, report, clock, sleep):
         # is handed to this stage. It may only assert that all keys are idle.
         perform_facing(guard, keyboard.require_idle, AlignmentMouse(mouse, report, clock, deadline),
             observer.alignment_snapshots, route['geometry'], calibration, evidence, report,
-            clock=clock, sleep=sleep, notify=alignment_notify)
+            clock=clock, sleep=sleep, notify=alignment_notify, profile=facing_profile)
         facing = report.get('facing_proof', {})
         if not facing.get('completed'):
             raise Refused('Facing module did not complete')
         deadline = min(deadline, facing['deadline_monotonic'])
         content, data = observer.approach_data()
-        facing_evidence = validate_facing_evidence(content, observer.nonce, look, route, facing)
+        facing_evidence = validate_facing_evidence(content, observer.nonce, look, route, facing, profile=facing_profile)
         if not facing_evidence.get('verified'):
             raise Refused('Independent native facing prerequisite refused')
         if clock() >= deadline:
@@ -987,12 +1145,54 @@ def run_selection(guard, keyboard, observer, mouse, report, clock, sleep):
         notify({'kind': 'alignment-finished', 'successful': successful})
 
 
+def run_held_action(guard, keyboard, observer, mouse, report, clock, sleep):
+    from prison_action_selection_evidence import validate_selection_evidence
+    from prison_held_action_proof import perform_held_action
+    from prison_held_action_evidence import validate_held_action_evidence
+    if (report.get('held_action_requested') is not True or report.get('coarse_facing_requested') is not True or
+            observer.enable_held is not True or mouse.enable_coarse_facing is not True):
+        raise Refused('Held action lacks explicit native/input capabilities')
+    content, _ = observer.approach_data()
+    selected = report.get('action_selection_proof')
+    accepted = validate_selection_evidence(content,observer.nonce,report.get('look_proof'),report.get('approach_proof'),
+        report.get('facing_proof'),selected,facing_profile='coarse')
+    if accepted.get('verified') is not True:
+        raise Refused('Independent native one-wheel selection prerequisite refused')
+    if observer.ready_at is None or observer.ready_real is None or observer.queued_real is None:
+        raise Refused('Original held observation clock missing')
+    original_end = observer.ready_at+(45000-(observer.ready_real-observer.queued_real))/1000-.5
+    calibration = {'nonce':observer.nonce,'ready_real_ms':observer.ready_real,'ready_monotonic':observer.ready_at,
+        'observer_start_real_ms':observer.queued_real,'observer_deadline_monotonic':original_end}
+    observer.begin_held(accepted['last_sample'])
+    key = HeldFKey(guard,keyboard,mouse,report,clock,sleep,original_end,enabled=True)
+    try:
+        perform_held_action(guard,keyboard.require_idle,key,observer.held_snapshots,selected['geometry'],calibration,accepted,report,
+            clock=clock,sleep=sleep,notify=notify)
+    finally:
+        held = report.get('held_action_proof',{})
+        report['paired_release_errors'].extend(held.get('paired_release_errors',[]))
+        if held.get('unconfirmed_held_key') is not None:
+            report['unconfirmed_held_key'] = held['unconfirmed_held_key']
+    # Live success still belongs to the independent raw parser. No action/input
+    # can follow this final read-only observation and full policy verification.
+    content, _ = observer.held_data()
+    verified = validate_held_action_evidence(content,observer.nonce,report.get('look_proof'),report.get('approach_proof'),
+        report.get('facing_proof'),selected,report.get('held_action_proof'),facing_profile='coarse')
+    if verified.get('verified') is not True:
+        raise Refused('Independent native held-action evidence refused')
+    guard.check(); keyboard.require_idle()
+    if clock() >= original_end:
+        raise Refused('Original held observation lifetime expired before completion')
+
+
 def arguments(argv=None):
     p = argparse.ArgumentParser(description=__doc__)
     p.add_argument('--enable-private-keyboard-proof', action='store_true', required=True)
+    p.add_argument('--enable-private-held-action-proof', action='store_true', help='One bounded F hold after independently accepted selection; requires explicit coarse facing')
     p.add_argument('--enable-private-selection-proof', action='store_true', help='After facing, try one button5 wheel selection; never perform the action')
     p.add_argument('--enable-private-look-proof', action='store_true')
     p.add_argument('--enable-private-alignment-proof', action='store_true', help='After successful approach, attempt stationary native lock selection using horizontal relative mouse only')
+    p.add_argument('--enable-private-coarse-facing-proof', action='store_true', help='Explicit coarse facing only; requires coarse approach and selection, no renewed clocks')
     p.add_argument('--enable-private-coarse-approach-proof', action='store_true', help='Explicit larger stationary-yaw profile; requires approach, original clocks unchanged')
     p.add_argument('--enable-private-approach-proof', action='store_true', help='After look calibration, replace W/Tab cycle with bounded interior approach only')
     for name in ('prefix', 'game-executable', 'profile', 'report'):
@@ -1002,6 +1202,10 @@ def arguments(argv=None):
     p.add_argument('--nonce', required=True)
     p.add_argument('command', nargs=argparse.REMAINDER)
     args = p.parse_args(argv)
+    if args.enable_private_held_action_proof and not (args.enable_private_selection_proof and args.enable_private_coarse_facing_proof):
+        p.error('Held action requires selection and coarse facing')
+    if args.enable_private_coarse_facing_proof and not (args.enable_private_coarse_approach_proof and args.enable_private_selection_proof):
+        p.error('Coarse facing requires coarse approach and selection')
     if args.enable_private_coarse_approach_proof and not args.enable_private_approach_proof:
         p.error('Coarse approach proof requires explicit approach proof')
     if args.enable_private_selection_proof and not (args.enable_private_alignment_proof and args.enable_private_approach_proof and args.enable_private_look_proof):
@@ -1067,6 +1271,7 @@ APPROACH_MODULES = ('prison_approach_proof.py', 'ordinary_gate_evidence.py',
 
 ALIGNMENT_MODULES = ('prison_alignment_proof.py', 'prison_alignment_evidence.py')
 SELECTION_MODULES = ('ordinary_gate_choices.py', 'prison_action_selection_proof.py', 'prison_action_selection_evidence.py')
+HELD_MODULES = ('prison_held_action_proof.py', 'prison_held_action_evidence.py')
 
 
 def approach_module_hashes():
@@ -1088,6 +1293,8 @@ def report_for_write(report):
         names.append('alignment_proof')
     if report.get('selection_requested') is True:
         names += ['facing_proof', 'action_selection_proof']
+    if report.get('held_action_requested') is True:
+        names.append('held_action_proof')
     for name in names:
         route = report.get(name)
         if isinstance(route, dict):
@@ -1095,7 +1302,7 @@ def report_for_write(report):
             samples = route.get('samples')
             if isinstance(samples, list):
                 result[name]['samples'] = [
-                    {key: sample.get(key) for key in ('sample', 'real_ms', 'received_monotonic')}
+                    {key: sample.get(key) for key in (('sample', 'real_ms', 'world_ms', 'received_monotonic') if name == 'held_action_proof' else ('sample', 'real_ms', 'received_monotonic'))}
                     if isinstance(sample, dict) else {'invalid_reference': True} for sample in samples]
     return result
 
@@ -1111,7 +1318,9 @@ def main(argv=None):
         console_path(args)
     bound = display.binding(args.host_display, args.prefix)
     report = {'schema': 1, 'kind': 'headless-private-keyboard-experiment', 'input_sent': False,
+              'held_action_requested': getattr(args, 'enable_private_held_action_proof', False),
               'look_requested': args.enable_private_look_proof,
+              'coarse_facing_requested': getattr(args, 'enable_private_coarse_facing_proof', False),
               'coarse_approach_requested': getattr(args, 'enable_private_coarse_approach_proof', False),
               'approach_requested': getattr(args, 'enable_private_approach_proof', False),
               'alignment_requested': getattr(args, 'enable_private_alignment_proof', False),
@@ -1128,6 +1337,9 @@ def main(argv=None):
     if report['selection_requested']:
         report['selection_module_sha256'] = module_hashes(SELECTION_MODULES)
         report.update(wheel_events=[], wheel_release_errors=[], unconfirmed_held_button=None)
+    if report['held_action_requested']:
+        report['held_module_sha256'] = module_hashes(HELD_MODULES)
+        report.update(held_events=[], held_release_errors=[])
     fd = os.open(args.report, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
     info = os.fstat(fd)
     notify({'kind': 'report', 'path': str(args.report), 'device': info.st_dev, 'inode': info.st_ino,
@@ -1165,7 +1377,7 @@ def main(argv=None):
                 raise Refused('New native console deadline exceeded')
         report['console'] = str(args.console)
         observer = Observer(args.console, args.nonce, enable_look=args.enable_private_look_proof,
-                            enable_approach=getattr(args, "enable_private_approach_proof", False), enable_selection=getattr(args, "enable_private_selection_proof", False))
+                            enable_approach=getattr(args, "enable_private_approach_proof", False), enable_selection=getattr(args, "enable_private_selection_proof", False), enable_held=getattr(args, "enable_private_held_action_proof", False))
         query = display.X11(bound['display'])
         while time.monotonic() < deadline:
             if child is not None and child.poll() is not None:
@@ -1198,8 +1410,8 @@ def main(argv=None):
         guard.check()
         keyboard = Keyboard(query)
         keyboard.require_idle()
-        mouse = RelativeMouse(query, keyboard, owner, enable_approach=getattr(args, "enable_private_approach_proof", False), enable_alignment=getattr(args, "enable_private_alignment_proof", False), enable_coarse_approach=getattr(args, "enable_private_coarse_approach_proof", False)) if args.enable_private_look_proof else None
-        sequence(guard, keyboard, observer, report, mouse=mouse, approach=getattr(args, "enable_private_approach_proof", False), alignment=getattr(args, "enable_private_alignment_proof", False), selection=getattr(args, "enable_private_selection_proof", False), coarse_approach=getattr(args, "enable_private_coarse_approach_proof", False))
+        mouse = RelativeMouse(query, keyboard, owner, enable_approach=getattr(args, "enable_private_approach_proof", False), enable_alignment=getattr(args, "enable_private_alignment_proof", False), enable_coarse_approach=getattr(args, "enable_private_coarse_approach_proof", False), enable_coarse_facing=getattr(args, 'enable_private_coarse_facing_proof', False), enable_selection=getattr(args, 'enable_private_selection_proof', False)) if args.enable_private_look_proof else None
+        sequence(guard, keyboard, observer, report, mouse=mouse, approach=getattr(args, "enable_private_approach_proof", False), alignment=getattr(args, "enable_private_alignment_proof", False), selection=getattr(args, "enable_private_selection_proof", False), coarse_approach=getattr(args, "enable_private_coarse_approach_proof", False), coarse_facing=getattr(args, 'enable_private_coarse_facing_proof', False), held_action=getattr(args, 'enable_private_held_action_proof', False))
     except (OSError, ValueError, UnicodeError, subprocess.SubprocessError) as error:
         report['error'] = str(error)
     finally:
@@ -1296,6 +1508,50 @@ def supervision_failure(record, key, reason, button=None):
         os.close(fd)
 
 
+class HeldSupervisor:
+    """Independent absolute observer/F deadlines; other key leases stay1.5s."""
+    def __init__(self, enabled):
+        self.enabled = enabled; self.observer_deadline = None
+        self.key_deadline = None; self.key = None; self.attempted = False; self.failed = False
+
+    def consume(self, event, now, *, other_pending=False):
+        kind = event.get('kind')
+        if not isinstance(kind,str) or not kind.startswith('held-'): return False
+        if self.enabled is not True or type(now) not in (int,float) or not math.isfinite(now):
+            self.failed = True; return True
+        value = event.get('deadline_monotonic')
+        numeric = type(value) in (int,float) and math.isfinite(value)
+        if kind == 'held-observer-deadline':
+            if self.observer_deadline is not None or not numeric or not now < value <= now+45:
+                self.failed = True
+            else:self.observer_deadline = value
+        elif kind == 'held-action-pending':
+            if self.observer_deadline is None or not numeric or not now < value <= self.observer_deadline:
+                self.failed = True
+            else:self.observer_deadline = value
+        elif kind == 'held-key-pending':
+            if (event.get('key') != 'F' or self.attempted or other_pending or self.observer_deadline is None or
+                    not numeric or not now < value <= min(now+5,self.observer_deadline)):
+                self.failed = True
+            else:
+                self.attempted = True; self.key = 'F'; self.key_deadline = value
+        elif kind == 'held-key-release-confirmed':
+            if event.get('key') != 'F' or self.key != 'F':self.failed = True
+            else:
+                if now >= self.key_deadline:self.failed = True
+                self.key = None; self.key_deadline = None
+        elif kind != 'held-action-finished':self.failed = True
+        return True
+
+    def expired(self, now, proof_finished=False):
+        return (self.failed or (self.key_deadline is not None and now >= self.key_deadline) or
+                (not proof_finished and self.observer_deadline is not None and now >= self.observer_deadline))
+
+    def cleanup_ack(self, event):
+        if event.get('kind') == 'held-key-release-confirmed' and event.get('key') == self.key == 'F':
+            self.key = None; self.key_deadline = None
+
+
 def supervise(argv=None):
     # Fork before loading Xlib or opening any display. The worker owns the one
     # connection. Parent never opens a second connection or retries an event.
@@ -1313,6 +1569,8 @@ def supervise(argv=None):
     approach_deadline = None; approach_finished = False
     button_at = None; held_button = None
     look_at = None; held_at = None; held_key = None; owned_child = None; report_record = None; proof_finished = False; expired = False
+    raw_args = sys.argv[1:] if argv is None else argv
+    held_scope = HeldSupervisor(raw_args.count('--enable-private-held-action-proof') == 1)
     deadline = time.monotonic() + 190
     try:
         child.start(); child_pipe.close()
@@ -1322,7 +1580,9 @@ def supervise(argv=None):
                     event = parent.recv()
                 except EOFError:
                     event = None
-                if event and event['kind'] == 'alignment-pending':
+                if event and str(event.get('kind','')).startswith('held-') and held_scope.consume(event,time.monotonic(),other_pending=held_key is not None or held_button is not None or look_at is not None):
+                    pass
+                elif event and event['kind'] == 'alignment-pending':
                     proposed = event.get('deadline_monotonic')
                     current = time.monotonic()
                     if type(proposed) not in (int, float) or not math.isfinite(proposed) or proposed > current + 8:
@@ -1359,7 +1619,7 @@ def supervise(argv=None):
                 elif event and event['kind'] == 'proof-finished':
                     proof_finished = event['successful']
             now = time.monotonic()
-            if received or now >= deadline or (alignment_deadline is not None and not alignment_finished and now >= alignment_deadline) or (approach_deadline is not None and not approach_finished and now >= approach_deadline) or (button_at is not None and now - button_at > 1.5) or (held_at is not None and now - held_at > 1.5) or (look_at is not None and now - look_at > 4):
+            if received or now >= deadline or held_scope.expired(now,proof_finished) or (alignment_deadline is not None and not alignment_finished and now >= alignment_deadline) or (approach_deadline is not None and not approach_finished and now >= approach_deadline) or (button_at is not None and now - button_at > 1.5) or (held_at is not None and now - held_at > 1.5) or (look_at is not None and now - look_at > 4):
                 expired = True
                 child.terminate()  # asks worker finally to release if still provable
                 child.join(1)
@@ -1372,6 +1632,7 @@ def supervise(argv=None):
                         last = parent.recv()
                     except EOFError:
                         break
+                    held_scope.cleanup_ack(last)
                     if last['kind'] == 'release-confirmed':
                         held_key = None
                     elif last['kind'] == 'button-release-confirmed' and last['button'] == held_button:
@@ -1380,14 +1641,14 @@ def supervise(argv=None):
                         report_record = last
                     elif last['kind'] == 'proof-finished':
                         proof_finished = last['successful']
-                if not (received and proof_finished and held_key is None and held_button is None):
-                    supervision_failure(report_record, held_key, 'Private input worker timed out or was interrupted', button=held_button)
+                if not (received and proof_finished and held_key is None and held_scope.key is None and held_button is None):
+                    supervision_failure(report_record, held_scope.key or held_key, 'Private input worker timed out or was interrupted', button=held_button)
                 break
         child.join(2)
         if child.exitcode != 0 and not expired:
             stop_exact_child(owned_child)
-            if not (received and proof_finished and held_key is None and held_button is None):
-                supervision_failure(report_record, held_key, 'Private input worker exited unsuccessfully', button=held_button)
+            if not (received and proof_finished and held_key is None and held_scope.key is None and held_button is None):
+                supervision_failure(report_record, held_scope.key or held_key, 'Private input worker exited unsuccessfully', button=held_button)
         return 1 if expired or child.exitcode != 0 else 0
     finally:
         if child.pid and child.is_alive():