#!/usr/bin/env python3 """Bounded ordinary-game startup check on a private hidden and muted display. No Workbench plugin, scripted smoke fixture, Steam profile, or existing process is used. """ import argparse import fcntl import hashlib import json import os from pathlib import Path import re import shutil import signal import subprocess import sys import time import uuid from native_provenance import snapshot, persist, bindings, seal_run from native_report import analyze from project_schema import parse_project ROOT = Path(__file__).resolve().parents[1] PROFILE_MARKER = '.erri-game-profile.json' PROFILE_ID = {'kind': 'erri-ordinary-game-disposable-profile', 'version': 1} CONTROL = 'vanilla-control' CONTROL_MARKER = '.erri-vanilla-control.json' CONTROL_ID = {'kind': 'erri-vanilla-only-startup-control', 'version': 1} GAME = re.compile(r'^armareforgersteam(?:diag)?(?:\.exe)?$', re.I) def refuse(message): raise ValueError(message) PACKED_FILES = {'addon.gproj', 'data.pak', 'resourceDatabase.rdb'} PACKED_GUID = '2528466DE220D91C' PACKED_MANIFEST = 'packed-input-provenance.json' PACKED_BASE = '58D0FB3206B6F859' PACKED_PROJECTS = { 'core': ('ERR_Core', PACKED_GUID, ()), 'ai': ('ERR_Ai', '5192138CB9D5B911', ('core',)), 'intel': ('ERR_Intel', 'D31039C7C53E348A', ('core',)), 'map': ('ERR_Map', '9AAEE9041D79809E', ('core',)), 'chat': ('ERR_Chat', 'FA804F411BDDF2C6', ('core', 'ai', 'intel', 'map')), } def packed_inventory(folder, project='core'): """Exact runtime payload, including files deliberately excluded by source manifests.""" if project not in PACKED_PROJECTS: refuse('Unknown packed addon project') folder = Path(folder) if folder.is_symlink() or not folder.is_dir() or folder != folder.resolve(): refuse('Packed candidate must be a canonical real directory') if {p.name for p in folder.iterdir()} != PACKED_FILES: refuse('Packed addon requires exactly addon.gproj, data.pak and resourceDatabase.rdb') result = {} for name in sorted(PACKED_FILES): path = folder/name if path.is_symlink() or not path.is_file() or path.stat().st_size <= 0: refuse('Packed input must be a nonempty regular file: '+name) digest = hashlib.sha256() with path.open('rb') as stream: for chunk in iter(lambda: stream.read(1024*1024), b''): digest.update(chunk) result[name] = {'bytes': path.stat().st_size, 'sha256': digest.hexdigest()} name, guid, dependencies = PACKED_PROJECTS[project] text = (folder/'addon.gproj').read_text() match = parse_project(text, allow_core_menu=project == 'core') actual = re.findall(r'"([0-9A-F]{16})"', match[4]) if match else [] expected = {PACKED_BASE, *(PACKED_PROJECTS[key][1] for key in dependencies)} if not match or match[1] != name or match[2] != guid or set(actual) != expected or len(actual) != len(expected): refuse('Packed project identity or exact declared dependencies do not match '+project) return result def packed_payload_inventory(folder, project='core'): """Core keeps its v1 three-file map; Chat binds all fifteen closure artifacts.""" if project == 'core': return packed_inventory(folder) if project != 'chat': refuse('Packed runtime supports only Core or the complete Chat closure') folder = Path(folder) if folder.is_symlink() or not folder.is_dir() or folder != folder.resolve(): refuse('Packed closure must be a canonical real directory') if {p.name for p in folder.iterdir()} != set(PACKED_PROJECTS): refuse('Packed Chat requires exactly core, ai, intel, map and chat directories') return {project+'/'+name: record for project in PACKED_PROJECTS for name, record in packed_inventory(folder/project, project).items()} def packed_directories(packed): if packed.get('project', 'core') == 'core': return {'core': packed['input']} if packed.get('project') != 'chat': refuse('Unknown packed runtime project') return {project: packed['input']/project for project in PACKED_PROJECTS} def packed_request(a): if getattr(a, 'external_addons', None) is not None: from packed_external import request return request(a, ROOT) candidate, session = getattr(a, 'packed_core', None), getattr(a, 'packed_session', None) addons = getattr(a, 'packed_addons', None) if candidate is None and addons is None and session is None: return None if candidate is not None and addons is not None: refuse('--packed-core and --packed-addons are mutually exclusive') project = 'chat' if addons is not None else 'core' candidate = addons if addons is not None else candidate if candidate is None or session is None or a.project != project: refuse('Packed input and --packed-session require their matching --project '+project) if a.profile != Path('/tmp/erri-game-profile'): refuse('Packed mode owns --packed-session/profile; do not supply --profile') if a.settle != 15: refuse('Packed mode requires the canonical 15-second readiness settling window') session = Path(session) if (not session.is_absolute() or session.parent != Path('/tmp') or session != session.resolve() or not re.fullmatch('erri-packed-'+project+r'-[A-Za-z0-9_-]+', session.name) or session.exists() or session.is_symlink()): refuse('Packed session must be a new canonical /tmp/erri-packed-'+project+'- directory') candidate = Path(candidate) manifest = packed_payload_inventory(candidate, project) target = session/'input/addons' if project == 'chat' else session/'input/addons/core' return {'candidate': candidate, 'session': session, 'input': target, 'manifest': manifest, 'project': project} def prepare_packed(packed): """Called only while holding the ordinary runner's existing prefix lock.""" if packed.get('schema_version') == 3: from packed_external import prepare return prepare(packed) packed['session'].mkdir(exist_ok=False) target = packed['input'] target.mkdir(parents=True, exist_ok=False) project = packed.get('project', 'core') if packed_payload_inventory(packed['candidate'], project) != packed['manifest']: refuse('Packed candidate changed before private copy') for name in sorted(packed['manifest']): # Exclusive creation; never hardlink the candidate or redirect the copy. (target/name).parent.mkdir(parents=True, exist_ok=True) with (packed['candidate']/name).open('rb') as source, (target/name).open('xb') as output: shutil.copyfileobj(source, output) if packed_payload_inventory(target, project) != packed['manifest'] or packed_payload_inventory(packed['candidate'], project) != packed['manifest']: refuse('Private packed copy or candidate changed during copy') def packed_snapshot(packed, command, profile): if packed.get('schema_version') == 3: from packed_external import snapshot return snapshot(packed, command, profile) project = packed.get('project', 'core') manifest = packed_payload_inventory(packed['input'], project) if manifest != packed['manifest'] or packed_payload_inventory(packed['candidate'], project) != manifest: refuse('Packed input changed before launch') value = {'schema_version': 1, 'kind': 'packed-core-runtime-input', 'launch_id': str(uuid.uuid4()), 'captured_unix': time.time(), 'candidate_directory': str(packed['candidate']), 'runtime_directory': str(packed['input']), 'profile': str(profile), 'command': list(command), 'artifacts': manifest, 'build_qualification': 'unqualified candidate; packaging verdict is unchanged'} if project == 'chat': value.update(schema_version=2, kind='packed-chat-runtime-input', project='chat', addon_order=list(PACKED_PROJECTS)) raw = (json.dumps(value, indent=2)+'\n').encode() (packed['session']/PACKED_MANIFEST).write_bytes(raw) return raw def packed_seal(packed, raw, session): """Fail closed on changed/deleted inputs, retaining a useful failure record.""" if packed.get('schema_version') == 3: from packed_external import seal return seal(packed, raw, session) metadata = json.loads(raw) unchanged, error = False, None try: project = packed.get('project', 'core') unchanged = (packed_payload_inventory(packed['input'], project) == metadata['artifacts'] and packed_payload_inventory(packed['candidate'], project) == metadata['artifacts']) except (OSError, ValueError) as caught: error = str(caught) console = session/'console.log' if session else None return {'packed_only': True, 'packed_input_provenance': PACKED_MANIFEST, 'packed_input_provenance_sha256': hashlib.sha256(raw).hexdigest(), 'packed_launch_id': metadata['launch_id'], 'stopped_unix': time.time(), 'packed_inputs_unchanged_after_run': unchanged, 'packed_input_error': error, 'console_sha256': hashlib.sha256(console.read_bytes()).hexdigest() if console and console.is_file() else None, 'build_qualified': False, 'release_qualified': False} def native_path(value, game_root): if value.startswith('$'): return value # Unknown engine aliases must not normalize into allowed paths. value = value.replace('\\', '/') if value[:3].lower() == 'z:/': value = value[2:] if not value.startswith('/'): value = str(game_root/value) return os.path.normpath(value) def packed_evidence(content, packed, steam, profile): if packed.get('schema_version') == 3: from packed_external import evidence return evidence(content, packed, steam, profile, sys.modules[__name__]) project = packed.get('project', 'core') report = game_evidence(content, project) directories = packed_directories(packed) game_root = steam/'steamapps/common/Arma Reforger' expected = {'5614BBCCBB55ED1C': str(game_root/'addons/core/core.gproj'), '58D0FB3206B6F859': str(game_root/'addons/data/ArmaReforger.gproj')} expected.update({PACKED_PROJECTS[key][1]: str(folder/'addon.gproj') for key, folder in directories.items()}) blocks = re.findall(r"Loaded addons:\r?\n((?:[^\n]*ENGINE[^\n]*gproj:[^\n]*\n)+)", content) complete = False violations = [] for block in blocks: entries = re.findall(r"gproj: '([^']+)' guid: '([0-9A-Fa-f]{16})'", block) if len(entries) != len(block.splitlines()): violations.append('Malformed loaded addon entry') seen = set() for path, guid in entries: guid = guid.upper() if guid in seen or guid not in expected or native_path(path, game_root) != expected.get(guid): violations.append('Unexpected, duplicate or redirected loaded addon: '+path+' '+guid) seen.add(guid) complete = seen == set(expected) mounts = re.findall(r"FileSystem: Adding package '([^']+)' \(pak count: (\d+)\) to filesystem under name ([^\s]+)", content) expected_mounts = {'core': str(game_root/'addons/core'), 'ArmaReforger': str(game_root/'addons/data')} expected_databases = {PACKED_PROJECTS[key][0]: str(folder/'resourceDatabase.rdb') for key, folder in directories.items()} expected_mounts.update({PACKED_PROJECTS[key][0]: str(folder) for key, folder in directories.items()}) mounted = set() for path, count, name in mounts: if name not in expected_mounts or native_path(path, game_root) != expected_mounts.get(name) or int(count) < 1 or (name in expected_databases and int(count) != 1): violations.append('Unexpected packed mount: '+path+' '+name) else: mounted.add(name) for path, name in re.findall(r"FileSystem: Adding relative directory '([^']+)' to filesystem under name ([^\s]+)", content): if name.startswith('ERR_'): violations.append('Loose successor directory mount: '+path) addon_root = packed['input'] if project == 'chat' else packed['input'].parent allowed_dirs = {str(addon_root), str(game_root/'addons'), str(profile/'addons')} dirs = re.findall(r"ENGINE\s*:\s*dir: '([^']+)'", content) for path in dirs: # Native $profile: maps to /profile, so this exact observed # engine alias denotes /addons. No general alias/traversal # expansion is accepted here or in loaded project/package paths. resolved = str(profile/'addons') if path == '$profile:../addons/' else native_path(path, game_root) if resolved not in allowed_dirs: violations.append('Unexpected addon search directory: '+path) database_paths = re.findall(r'ResourceDB: loading cache \(id=\d+ name=(ERR_[^\s]+) path=([^\r\n]+)\)', content) databases = set() for name, path in database_paths: if name not in expected_databases or native_path(path, game_root) != expected_databases.get(name): violations.append('Unexpected or redirected successor resource database: '+name+' '+path) else: databases.add(name) database_ok = databases == set(expected_databases) modules = re.findall(r'Module: (GameLib|Game);[^\n]*defines:\s*"([^"]*)"', content) normal_modules = {name for name, flags in modules} == {'GameLib', 'Game'} and all(not {'WORKBENCH', 'ENF_WB'}.intersection(flags.split(',')) for _, flags in modules) if any({'WORKBENCH', 'ENF_WB'}.intersection(flags.split(',')) for _, flags in modules): violations.append('Workbench module defines in ordinary packed run') path_proof = complete and mounted == set(expected_mounts) and database_ok and bool(dirs) failed = report['failed'] or bool(violations) return {**report, 'packed_only': True, 'packed_loaded_paths_verified': path_proof, 'packed_input_violations': violations, 'packed_normal_modules': normal_modules, 'failed': failed, 'ready': report['ready'] and path_proof and normal_modules and not failed, 'build_qualified': False, 'release_qualified': False, 'scope': 'packed-only '+project+' ordinary local startup; explicit noSplash; no packaging, visible UI, escape, multiplayer or release qualification'} def assert_no_game(proc_root=Path('/proc')): """Read-only and fail closed; also recognizes Wine's truncated comm name.""" for proc in proc_root.iterdir(): if not proc.name.isdigit(): continue try: args = (proc/'cmdline').read_bytes().split(b'\0') comm = (proc/'comm').read_text().strip() except (FileNotFoundError, ProcessLookupError): continue # process exited during enumeration except OSError as error: refuse(f'Cannot exclude an existing game process {proc.name}: {error}') names = [re.split(r'[\\/]', arg.decode(errors='replace'))[-1] for arg in args] if any(GAME.fullmatch(name) for name in names) or comm.lower().startswith('armareforgerste'): refuse(f'An Arma Reforger game process already exists (PID {proc.name}); leave it untouched') def prefix_process_candidate(proc): """Classify readable process metadata before accessing protected environ. Linux services such as systemd can be non-dumpable even to their UID. Their environment is irrelevant. Windows/Wine/runtime candidates still fail closed. """ args = [value.decode(errors='replace') for value in (proc/'cmdline').read_bytes().split(b'\0') if value] comm = (proc/'comm').read_text().strip() try: executable = os.readlink(proc/'exe') except (FileNotFoundError, PermissionError): executable = '' values = args + [comm, executable] for value in values: name = re.split(r'[\\/]', value)[-1].lower() if (name.endswith('.exe') or name.startswith(('wine', 'proton', 'gamescope')) or name in {'reaper', 'pressure-vessel-wrap', 'pv-bwrap'} or name.startswith('armareforger')): return True # systemd's PAM child overwrites both names, and its non-dumpable exe is # inaccessible. This exact native helper is positively identified by both. if comm == '(sd-pam)' and args == ['(sd-pam)']: return False # Observed host Apache workers are also non-dumpable; accept only their # exact native foreground-service signature, after all runtime checks. if comm == 'apache2' and args == ['apache2', '-DFOREGROUND']: return False # A readable native executable is conclusive; otherwise require an absolute # Linux argv[0] corroborated by the kernel's (possibly truncated) comm name. if executable.startswith('/'): return False if args and args[0].startswith('/'): basename = Path(args[0]).name if comm and (comm == basename or comm == basename[:15]): return False argv0 = args[0] if args else '' refuse(f'Cannot classify possible Wine/runtime process {proc.name} (comm={comm!r}, argv0={argv0!r})') def assert_prefix_idle(compat, proc_root=Path('/proc')): target = str(compat/'pfx').encode() failures = [] for proc in proc_root.iterdir(): if not proc.name.isdigit() or int(proc.name) == os.getpid(): continue try: if proc.stat().st_uid != os.getuid(): continue if not prefix_process_candidate(proc): continue environment = (proc/'environ').read_bytes().split(b'\0') except (FileNotFoundError, ProcessLookupError): continue except (OSError, ValueError) as error: failures.append(f'Cannot verify disposable prefix is idle (PID {proc.name}): {error}') continue # Proton assigns pfx/; only trailing separators are equivalent. Do not # resolve untrusted process paths or accept sibling/symlink spellings. if any(value.startswith(b'WINEPREFIX=') and value[len(b'WINEPREFIX='):].rstrip(b'/') == target for value in environment): failures.append(f'Disposable prefix already has a process (PID {proc.name}); leave it untouched') if failures: refuse('\n'.join(failures)) def validate_paths(root, steam, compat, profile): if compat.is_relative_to(steam) or not (compat/'pfx/system.reg').is_file(): refuse('--compat must be an existing disposable cloned prefix outside Steam') if any(profile.is_relative_to(path) or path.is_relative_to(profile) for path in (root, steam, compat)): refuse('--profile must be separate from source, Steam and disposable prefix trees') if profile.exists(): marker = profile/PROFILE_MARKER if marker.is_symlink() or not marker.is_file() or json.loads(marker.read_text()) != PROFILE_ID: refuse('Existing profile is not marked as this runner\'s disposable game profile') for path in profile.rglob('*'): if path.is_symlink() and path != profile/'addons/ArmaReforger': refuse(f'Unexpected profile symlink: {path}') def command_for(root, steam, compat, profile, project, packed=None, capture_prison_view=False, private_width=1280): if project not in {'core', 'chat', CONTROL}: refuse('Unsupported ordinary game project/control') if capture_prison_view and (project == CONTROL or packed is not None): refuse('Prison view capture requires loose successor source with its opt-in observer') if type(private_width) is not int or private_width not in (1280, 2560) or (private_width != 1280 and (project == CONTROL or packed is not None)): refuse('Unsupported private headless viewport') command = ['gamescope', '--backend', 'headless', '-W', str(private_width), '-H', '720', '-w', str(private_width), '-h', '720', '-r', '30', '--', str(steam/'steamapps/common/Proton - Experimental/proton'), 'run', str(steam/'steamapps/common/Arma Reforger/ArmaReforgerSteam.exe')] if packed is not None: if project != packed.get('project', 'core') or project not in {'core', 'chat'}: refuse('Packed input project does not match requested ordinary scenario') addon_root = packed['input'] if project == 'chat' else packed['input'].parent command += ['-addonsDir', 'Z:'+str(addon_root), '-addons', PACKED_PROJECTS[project][1]] elif project != CONTROL: project_text = (root/'addons'/project/'addon.gproj').read_text() guid = re.search(r'\bGUID "([A-F0-9]{16})"', project_text)[1] command += ['-addonsDir', 'Z:'+str(root/'addons'), '-addons', guid] command += ['-profile', 'Z:'+str(profile)] if project != CONTROL: command += ['-errOrdinaryScenario', project, '-noSplash'] if capture_prison_view: command += ['-errOrdinaryScreenshot', '1'] return command + ['-noBackend', '-noSound', '-noThrow', '-disableCrashReporter', '-logconsole'] def validate_control_profile(profile): """Never reuse the successor or user's profile as a vanilla control.""" if profile.exists(): marker = profile/CONTROL_MARKER if marker.is_symlink() or not marker.is_file() or json.loads(marker.read_text()) != CONTROL_ID: refuse('Vanilla control requires a new or explicitly control-owned disposable profile') def vanilla_evidence(content): report = analyze(content, 'console.log') modules = re.findall(r'Module: (GameLib|Game);[^\n]*defines:\s*"([^"]*)"', content) ordinary = {name for name, flags in modules} == {'GameLib', 'Game'} bad_defines = any({'WORKBENCH', 'ENF_WB'}.intersection(flags.split(',')) for _, flags in modules) # Only the engine's Loaded-addons blocks count, not its available inventory. blocks = re.findall(r"Loaded addons:\r?\n((?:[^\n]*ENGINE[^\n]*gproj:[^\n]*\n)+)", content) guids = [guid.upper() for block in blocks for guid in re.findall(r"guid: '([0-9A-Fa-f]{16})'", block)] allowed = {'5614BBCCBB55ED1C', '58D0FB3206B6F859'} wrong_addons = bool(set(guids) - allowed) base_only = bool(blocks) and set(guids) == allowed and not wrong_addons successor = any(token in content for token in ('ERR ORDINARY:', 'ERR SMOKE ', 'Rereintroduced [mission]')) menu = "Entities load 'worlds/MainMenuWorld/MainMenuWorld.ent'" in content and 'Entered main menu.' in content failed = bool(report['engine_errors'] or report['script_or_runtime_failures']) or bad_defines or wrong_addons or successor return {**report, 'control_only': True, 'loaded_addon_guids': guids, 'vanilla_addons_only': base_only, 'game_module_defines': modules, 'workbench_undefined': ordinary and not bad_defines, 'main_menu_initialized': menu, 'unexpected_successor_activity': successor, 'normal_mission_ready': False, 'local_participant_verified': False, 'scenario_header_verified': False, 'failed': failed, 'ready': ordinary and base_only and menu and not failed, 'scope': 'vanilla-only ordinary main-menu startup control; no successor mission, input, multiplayer or release qualification'} def game_evidence(content, project=None): if project == CONTROL: return vanilla_evidence(content) report = analyze(content, 'console.log') defines = re.findall(r'Module: Game;[^\n]*defines:\s*"([^"]*)"', content) non_workbench = bool(defines) and all(not {'WORKBENCH', 'ENF_WB'}.intersection(value.split(',')) for value in defines) headers = re.findall(r'ERR ORDINARY: header verified project=(core|chat) systems=MissionSystems', content) header_verified = bool(headers) and (project is None or project in headers) participants = re.findall(r'ERR ORDINARY: local participant verified project=(core|chat) phase=prison', content) participant_verified = bool(participants) and (project is None or project in participants) ready = header_verified and participant_verified and 'Rereintroduced [mission]: phase 0 -> 1' in content unexpected_fixture = 'ERR SMOKE ' in content failed = (bool(report['engine_errors'] or report['script_or_runtime_failures']) or 'Rereintroduced: scenario preparation failed' in content or 'Can\'t compile "Game"' in content or unexpected_fixture or any({'WORKBENCH', 'ENF_WB'}.intersection(value.split(',')) for value in defines)) return {**report, 'local_participant_verified': participant_verified, 'scenario_header_verified': header_verified, 'verified_header_projects': headers, 'game_module_defines': defines, 'workbench_undefined': non_workbench, 'normal_mission_ready': ready, 'unexpected_smoke_fixture': unexpected_fixture, 'failed': failed, 'ready': non_workbench and ready and not failed, 'scope': 'ordinary local game startup only; no player input, full escape, multiplayer or FPS proof'} def cleanup(child, proton, env): # Caller owns the prefix lock and verified no pre-existing prefix processes. try: subprocess.run([str(proton.parent/'files/bin/wineserver'), '-k'], env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=10) except (OSError, subprocess.TimeoutExpired): pass try: os.killpg(child.pid, signal.SIGTERM) except ProcessLookupError: pass try: child.wait(timeout=10) except subprocess.TimeoutExpired: pass finally: try: os.killpg(child.pid, signal.SIGKILL) except ProcessLookupError: pass child.wait(timeout=10) def cleanup_owned_run(child, proton, env, compat): """After a launched owned session, while the caller still holds the prefix lock. Noncanonical private prefixes retain normal shutdown plus strict idle proof; adopted-helper signalling is restricted to its one explicitly owned prefix. """ report = {'cleanup_confirmed':False, 'idle_verified':False, 'errors':[], 'orphan_cleanup':{'attempted':False,'reason':'noncanonical-private-prefix'}} try: cleanup(child,proton,env) except (OSError,ValueError,subprocess.SubprocessError) as error: report['errors'].append('normal shutdown: '+str(error)) canonical = Path('/tmp/erri-headless-compat') if Path(compat) == canonical and Path(compat).resolve() == canonical: report['orphan_cleanup'] = {'attempted':True} try: from private_prefix_cleanup import cleanup_private_prefix details = cleanup_private_prefix(compat,proton) report['orphan_cleanup']['report'] = details if not isinstance(details,dict) or details.get('cleared') is not True: raise ValueError('Private helper cleanup did not prove completion') except (OSError,ValueError,ImportError,subprocess.SubprocessError) as error: report['orphan_cleanup']['error'] = str(error) details = getattr(error,'report',None) if details is not None: report['orphan_cleanup']['report'] = details report['errors'].append('private helper cleanup: '+str(error)) try: assert_prefix_idle(compat) report['idle_verified'] = True except (OSError,ValueError,subprocess.SubprocessError) as error: report['errors'].append('idle verification: '+str(error)) report['cleanup_confirmed'] = report['idle_verified'] and not report['errors'] return report class StartupMonitor: """Latch failures while allowing a bounded diagnostic drain; no native I/O.""" def __init__(self, started, timeout, settle, failure_grace=0): if not 0 <= failure_grace <= 120: refuse('Use failure-grace 0..120 seconds') self.started = started self.deadline = started + timeout self.settle = settle self.failure_grace = failure_grace self.first_failure_at = None self.ready_at = None def observe(self, evidence, now, process_exited=False): if evidence.get('failed') and self.first_failure_at is None: self.first_failure_at = now if self.first_failure_at is not None: # Repeated errors and subsequent readiness cannot renew this lease. cutoff = min(self.deadline, self.first_failure_at + self.failure_grace) if process_exited or now >= cutoff: return 'game-failed' return None if now >= self.deadline: return 'timeout' if process_exited: return 'process-exited' if evidence.get('ready'): if self.ready_at is None: self.ready_at = now if now - self.ready_at >= self.settle: return 'game-started' else: self.ready_at = None return None def timeout_result(self): return 'game-failed' if self.first_failure_at is not None else 'timeout' def annotate(self, final): # Keep raw final error counts intact, but never lose an earlier failure # if the process truncated/replaced the log during the grace interval. final['failure_observed_during_polling'] = self.first_failure_at is not None if self.first_failure_at is not None: final['failed'] = True final['ready'] = False return final def display_probe_command(command, root, compat, profile, executable, run_id, host_display): """Wrap only the existing child of headless Gamescope; preserve Proton argv.""" separator = command.index('--') if command[:separator].count('--backend') != 1 or command[command.index('--backend')+1] != 'headless': refuse('Display inspection requires the explicit headless Gamescope boundary') helper = root/'tools/headless_display_probe.py' report = profile/('display-ownership-'+run_id+'.json') if helper.is_symlink() or not helper.is_file() or report.exists() or report.is_symlink(): refuse('Display inspection requires a regular helper and a new report') child = command[separator+1:] context = {'report': str(report), 'helper_sha256': hashlib.sha256(helper.read_bytes()).hexdigest(), 'prefix': str(compat/'pfx'), 'profile': str(profile), 'game_executable': str(executable), 'command': child, 'host_display': host_display} wrapper = [sys.executable, str(helper), '--host-display', host_display, '--prefix', context['prefix'], '--profile', str(profile), '--game-executable', str(executable), '--report', str(report), '--observe-seconds', '30', '--'] return command[:separator+1] + wrapper + child, context def display_probe_result(context, session): """Retain exact observation bytes after cleanup; an absent/unverified report fails closed.""" result = {'requested': True, 'verified': False, 'read_only': True, 'input_sent': False} try: source = Path(context['report']) if source.is_symlink() or not source.is_file() or not 0 < source.stat().st_size <= 1024*1024: refuse('Missing or invalid bounded display ownership report') raw = source.read_bytes() report = json.loads(raw) result['sha256'] = hashlib.sha256(raw).hexdigest() if session: with (session/'display-ownership.json').open('xb') as output: output.write(raw) result['archive_file'] = 'display-ownership.json' if not isinstance(report, dict): refuse('Display ownership report is not an object') for key in ('helper_sha256', 'prefix', 'profile', 'game_executable', 'report', 'command'): if report.get(key) != context[key]: refuse('Display report binding mismatch: '+key) if (report.get('schema') != 1 or report.get('kind') != 'headless-display-read-only' or report.get('read_only') is not True or report.get('input_sent') is not False or report.get('observation_finished') is not True or report.get('owned_game_window_verified') is not True): refuse('Private display game-window ownership was not verified') from headless_display_probe import validate_display, shared_relationship_valid bound = report.get('binding', {}) number = validate_display(bound.get('display'), context['host_display']) sockets = bound.get('sockets') allowed = {f'/tmp/.X11-unix/X{number}', f'@/tmp/.X11-unix/X{number}'} if not isinstance(sockets, list) or not sockets or len(sockets) > 2: refuse('Missing private X11 socket ownership') paths = set() for sock in sockets: if not isinstance(sock, dict) or sock.get('path') not in allowed or sock['path'] in paths or type(sock.get('inode')) is not int or sock['inode'] <= 0: refuse('Invalid private X11 socket ownership') paths.add(sock['path']) for identity in [report.get('wrapper'), bound.get('gamescope'), bound.get('xwayland')]: if not isinstance(identity, dict) or identity.get('uid') != os.getuid() or any(type(identity.get(k)) is not int or identity[k] <= 0 for k in ('pid', 'start')): refuse('Missing owned process binding') relationship = bound.get('relationship') if not isinstance(relationship, dict): refuse('Missing private display process relationship') if relationship.get('mode') == 'descendant': if relationship.get('descendant') is not True: refuse('Native descendant relationship was not verified') elif relationship.get('mode') == 'shared-listeners': checks = relationship.get('checks') if (not shared_relationship_valid(checks) or checks['listener_inodes'] != [sock['inode'] for sock in sockets] or bound['xwayland']['start'] < bound['gamescope']['start']): refuse('Shared-listener process relationship was not verified') else: refuse('Unknown private display process relationship') windows = report.get('owned_windows') if not isinstance(windows, list) or not windows: refuse('No verified game window records') for window in windows: identity = window.get('process', {}) if (type(window.get('window')) is not int or window['window'] <= 0 or type(window.get('server_pid')) is not int or type(identity.get('pid')) is not int or identity.get('uid') != os.getuid() or type(identity.get('start')) is not int or identity['start'] <= 0 or type(window.get('pid')) is not int or window['pid'] <= 0 or window['pid'] != window.get('server_pid') or window['pid'] != identity.get('pid')): refuse('Window lacks consistent native and server process ownership') result['verified'] = True except (OSError, ValueError, TypeError, AttributeError) as error: result['error'] = str(error) return result def validate_input_observer(a): if getattr(a, 'observe_prison_choices', False) and not getattr(a, 'observe_prison_gate', False): refuse('Prison action choices require the existing gate observer') if getattr(a, 'observe_prison_gate', False) and not getattr(a, 'observe_player_input', False): refuse('Prison gate observation requires the existing input observer') if not getattr(a, 'observe_player_input', False): return if a.project == CONTROL or getattr(a, 'packed_core', None) or getattr(a, 'packed_addons', None): refuse('Input observation requires loose Core or Chat, not packed input or vanilla control') if a.settle < 50: refuse('Input observation requires --settle >=50 for the bounded 45-second sampler') def validate_input_exercise(a): if type(getattr(a, 'exercise_prison_held_action', False)) is not bool or (getattr(a, 'exercise_prison_held_action', False) and not (getattr(a, 'exercise_prison_selection', False) and getattr(a, 'exercise_prison_coarse_facing', False))): refuse('Prison held action requires explicit selection and coarse facing') if type(getattr(a, 'exercise_prison_coarse_facing', False)) is not bool or (getattr(a, 'exercise_prison_coarse_facing', False) and not (getattr(a, 'exercise_prison_coarse_approach', False) and getattr(a, 'exercise_prison_selection', False))): refuse('Coarse facing requires explicit coarse approach and selection') if getattr(a, 'exercise_prison_selection', False) and not (getattr(a, 'exercise_prison_alignment', False) and getattr(a, 'observe_prison_choices', False)): refuse('Prison action selection requires explicit alignment and native choices') if getattr(a, 'observe_prison_choices', False) and getattr(a, 'exercise_player_input', False) and not getattr(a, 'exercise_prison_approach', False): refuse('Input exercise with gate choices requires explicit approach') if type(getattr(a, 'exercise_prison_coarse_approach', False)) is not bool or (getattr(a, 'exercise_prison_coarse_approach', False) and not getattr(a, 'exercise_prison_approach', False)): refuse('Coarse approach requires explicit prison approach') if getattr(a, 'exercise_prison_alignment', False) and not getattr(a, 'exercise_prison_approach', False): refuse('Prison lock alignment requires the explicit prison approach experiment') if getattr(a, 'approach_wide_display', False) and not getattr(a, 'exercise_prison_approach', False): refuse('Wide private viewport requires the explicit prison approach experiment') if getattr(a, 'exercise_prison_approach', False) and not (getattr(a, 'exercise_relative_look', False) and getattr(a, 'observe_prison_gate', False)): refuse('Prison approach requires explicit relative look and gate observation') if getattr(a, 'exercise_relative_look', False) and not getattr(a, 'exercise_player_input', False): refuse('Relative look requires the existing explicit keyboard exercise') if not getattr(a, 'exercise_player_input', False): return if (a.project not in {'core', 'chat'} or getattr(a, 'packed_core', None) or getattr(a, 'packed_addons', None) or getattr(a, 'packed_session', None)): refuse('Input exercise requires loose Core or Chat') if not getattr(a, 'observe_player_input', False) or not getattr(a, 'inspect_headless_display', False): refuse('Input exercise requires --observe-player-input and --inspect-headless-display') if a.settle < 50: refuse('Input exercise requires --settle >=50') def input_observer_command(command, project, packed, nonce, look=False, gate=False, choices=False, held_action=False): if project not in {'core', 'chat'} or packed is not None: refuse('Input observation requires loose Core or Chat') if not re.fullmatch(r'[A-Za-z0-9]{1,32}', nonce): refuse('Input observer nonce must contain 1..32 ASCII letters/digits') if type(look) is not bool: refuse('Relative look selection must be boolean') if type(choices) is not bool or (choices and not gate): refuse('Prison choices require explicit boolean gate observation') if type(gate) is not bool: refuse('Prison gate observation selection must be boolean') if type(held_action) is not bool or (held_action and not (look and gate and choices)): refuse('Held action observation requires explicit look, gate and choices') if any(value.startswith(('-errOrdinaryInput', '-errOrdinaryLook', '-errOrdinaryGate', '-errOrdinaryHeld')) for value in command): refuse('Input observer command already has observer options') options = ['-errOrdinaryInputObserve', nonce, '-errOrdinaryInputScreenshots', '1'] if look: options += ['-errOrdinaryLookObserve', nonce] if gate: options += ['-errOrdinaryGateObserve', nonce] if choices: options += ['-errOrdinaryGateChoices', nonce] if held_action: options += ['-errOrdinaryHeldAction', nonce] return command + options def input_observer_evidence(content, nonce): """Strictly summarize native sampler records; never infer input delivery.""" import math errors, records, samples = [], [], [] prefix = 'ERR INPUT OBSERVE: ' number = r'-?(?:\d+(?:\.\d*)?|\.\d+)(?:[eE][+-]?\d+)?' header = re.compile(r'run=([A-Za-z0-9]{1,32}) sample=(\d+) real=(\d+) world=('+number+r') (.+)') sample_pattern = re.compile(r'SAMPLE origin=(.+?) forward=('+number+r') input=(.+?) velocity=(.+?) inventory=(0|1|true|false)') def vector_value(value): if value.startswith('<') and value.endswith('>'): value = value[1:-1] if not re.fullmatch(number+r'(?:\s*,\s*|\s+)'+number+r'(?:\s*,\s*|\s+)'+number, value.strip()): raise ValueError('invalid vector') result = [float(x) for x in re.findall(number, value)] if not all(math.isfinite(x) for x in result): raise ValueError('nonfinite vector') return result lines = [line.split(prefix, 1)[1].strip() for line in content.splitlines() if prefix in line] if not re.fullmatch(r'[A-Za-z0-9]{1,32}', nonce): errors.append('invalid expected nonce') if len(lines) > 460: errors.append('observer record bound exceeded') for line in lines[:460]: match = header.fullmatch(line) if not match: errors.append('malformed observer record') continue actual, index, real, world, event = match.groups() index, real, world = int(index), int(real), float(world) if actual != nonce: errors.append('observer nonce mismatch') continue if not math.isfinite(world) or world < 0 or index > 450 or real > 2147483647: errors.append('observer sample/clock bound invalid') continue record = {'sample': index, 'real_ms': real, 'world_ms': world, 'event': event} if records and (real < records[-1]['real_ms'] or world < records[-1]['world_ms'] or index < records[-1]['sample']): errors.append('observer records reversed') if event.startswith('SAMPLE '): sample = sample_pattern.fullmatch(event) if not sample: errors.append('malformed observer sample') continue origin, forward, movement, velocity, opened = sample.groups() try: value = {**record, 'origin': vector_value(origin), 'forward': float(forward), 'input': vector_value(movement), 'velocity': vector_value(velocity), 'inventory': opened in {'1', 'true'}} if not math.isfinite(value['forward']): raise ValueError('nonfinite forward') except ValueError: errors.append('invalid observer sample values') continue if index != len(samples)+1: errors.append('observer sample sequence is not contiguous') samples.append(value) elif event not in {'QUEUED', 'READY', 'INVENTORY_OPEN', 'INVENTORY_CLOSED', 'CAPTURE_REQUEST baseline', 'CAPTURE_REQUEST open', 'CAPTURE_REQUEST closed', 'FINISHED observation-bound', 'FINISHED sample-bound'} and not re.fullmatch(r'CANCELLED [a-z-]+', event): errors.append('unknown observer event') elif index != len(samples): errors.append('observer transition sample does not match last sample') records.append(record) def matching(name): return [r for r in records if r['event'] == name or r['event'].startswith(name+' ')] queued, ready, finished, cancelled = (matching(name) for name in ('QUEUED', 'READY', 'FINISHED', 'CANCELLED')) if len(queued) != 1 or not records or records[0] not in queued or records[0]['sample'] != 0: errors.append('missing/duplicate/misordered QUEUED') if len(ready) != 1: errors.append('missing/duplicate READY') if len(finished) != 1 or not records or records[-1] not in finished: errors.append('missing/duplicate/misordered FINISHED') if cancelled: errors.append('observer cancelled') if not samples: errors.append('no native samples') if queued: start = queued[0]['real_ms'] if any(s['real_ms'] < start or s['real_ms']-start >= 45000 for s in samples): errors.append('native sample outside real-time observation window') if ready and (not samples or ready[0]['real_ms'] < samples[0]['real_ms']+2000): errors.append('READY before stationary baseline interval') if finished: end = finished[0] if end['event'] == 'FINISHED observation-bound' and end['real_ms']-start < 45000: errors.append('FINISHED before real-time bound') if end['event'] == 'FINISHED sample-bound' and len(samples) != 450: errors.append('FINISHED before sample bound') opens, closes = matching('INVENTORY_OPEN'), matching('INVENTORY_CLOSED') if len(opens) > 1 or len(closes) > 1: errors.append('duplicate inventory transition') by_index = {s['sample']: s for s in samples} for transition, expected in [(r, True) for r in opens] + [(r, False) for r in closes]: sample = by_index.get(transition['sample']) if not ready or records.index(transition) < records.index(ready[0]) or not sample or sample['inventory'] is not expected: errors.append('inventory transition lacks matching post-READY sample') if closes and (not opens or records.index(closes[0]) <= records.index(opens[0])): errors.append('inventory close lacks prior observed open') captures = matching('CAPTURE_REQUEST') for suffix, transition, following in [('baseline', ready, opens), ('open', opens, closes), ('closed', closes, finished)]: requested = [r for r in captures if r['event'] == 'CAPTURE_REQUEST '+suffix] if len(requested) > 1 or (requested and len(transition) != 1): errors.append('capture request lacks its unique transition') continue if not requested: continue # superseded/invalid pending captures are legitimately skipped capture, begin = requested[0], transition[0] capture_index, begin_index = records.index(capture), records.index(begin) current = by_index.get(capture['sample']) if (capture_index <= begin_index or capture['real_ms']-begin['real_ms'] < 300 or not current or current['inventory'] is not (suffix == 'open')): errors.append('delayed capture lacks matching native state and minimum delay') if following and capture_index >= records.index(following[0]): errors.append('delayed capture crossed its transition interval') if queued and not 0 <= capture['real_ms']-queued[0]['real_ms'] < 45000: errors.append('delayed capture outside observation window') interval = [sample for sample in samples if begin['sample'] <= sample['sample'] <= capture['sample']] if any(sample['inventory'] is not (suffix == 'open') for sample in interval): errors.append('delayed capture native inventory state changed') if suffix == 'baseline': baseline = by_index.get(begin['sample']) if not baseline or any(abs(sample['forward']) > .01 or math.sqrt(sum(value*value for value in sample['input'])) > .01 or math.sqrt(sum(value*value for value in sample['velocity'])) > .05 or math.dist(sample['origin'], baseline['origin']) > .05 for sample in interval): errors.append('delayed baseline capture lost stationary state') # Native movement proof belongs to the observer, independently of whether # an external helper claims to have sent keys. Restrict it to READY→open. movement = {'action_observed': False, 'physical_motion_observed': False, 'released_input_observed': False, 'max_horizontal_displacement': 0.0} if len(ready) == 1: baseline = by_index.get(ready[0]['sample']) end_index = opens[0]['sample'] if len(opens) == 1 else 0 if baseline and end_index: for sample in samples: if not ready[0]['sample'] < sample['sample'] <= end_index: continue distance = math.hypot(sample['origin'][0]-baseline['origin'][0], sample['origin'][2]-baseline['origin'][2]) movement['max_horizontal_displacement'] = max(movement['max_horizontal_displacement'], distance) if sample['forward'] > 0.1 and math.hypot(sample['input'][0], sample['input'][2]) > 0.1: movement['action_observed'] = True if distance > 0.05 and math.hypot(sample['velocity'][0], sample['velocity'][2]) > 0.05: movement['physical_motion_observed'] = True if len(closes) == 1: movement['released_input_observed'] = any( sample['sample'] > closes[0]['sample'] and abs(sample['forward']) <= 0.01 and math.sqrt(sum(value*value for value in sample['input'])) <= 0.01 for sample in samples) return {'nonce': nonce, 'read_only': True, 'input_sent': False, 'sampler_complete': not errors, 'queued': len(queued) == 1, 'ready': len(ready) == 1, 'finished': len(finished) == 1, 'cancelled': bool(cancelled), 'sample_count': len(samples), 'record_count': len(lines), 'inventory_open_observed': len(opens) == 1, 'inventory_close_observed': len(closes) == 1, 'capture_requests': [r['event'].split()[-1] for r in captures], 'movement': movement, 'transitions': [r for r in records if not r['event'].startswith('SAMPLE ')], 'errors': list(dict.fromkeys(errors)), 'scope': 'native read-only sampler lifecycle only; no generated input, movement success, inventory usability, screenshot file verification, manual playtest or multiplayer claim'} def input_observer_result(result, content, nonce): evidence = input_observer_evidence(content, nonce) if result == 'game-started' and not evidence['sampler_complete']: result = 'observation-incomplete' return result, evidence def approach_module_hashes(root): hashes = {} for name in ('prison_approach_proof.py', 'ordinary_gate_evidence.py', 'ordinary_look_evidence.py', 'prison_approach_evidence.py'): path = root/'tools'/name if path.is_symlink() or not path.is_file(): refuse('Approach requires regular local proof modules') hashes[name] = hashlib.sha256(path.read_bytes()).hexdigest() return hashes def alignment_module_hashes(root): hashes = {} for name in ('prison_alignment_proof.py', 'prison_alignment_evidence.py'): path = root/'tools'/name if path.is_symlink() or not path.is_file(): refuse('Alignment requires regular local proof modules') hashes[name] = hashlib.sha256(path.read_bytes()).hexdigest() return hashes def selection_module_hashes(root): hashes = {} for name in ('ordinary_gate_choices.py', 'prison_action_selection_proof.py', 'prison_action_selection_evidence.py'): path = root/'tools'/name if path.is_symlink() or not path.is_file(): refuse('Action selection requires regular local proof modules') hashes[name] = hashlib.sha256(path.read_bytes()).hexdigest() return hashes def held_module_hashes(root): hashes = {} for name in ('prison_held_action_proof.py', 'prison_held_action_evidence.py'): path = root/'tools'/name if path.is_symlink() or not path.is_file(): refuse('Held action requires regular local proof modules') hashes[name] = hashlib.sha256(path.read_bytes()).hexdigest() return hashes def input_exercise_command(command, root, compat, profile, executable, run_id, host_display, display_context, look=False, approach=False, alignment=False, selection=False, coarse_approach=False, coarse_facing=False, held_action=False): """Put the input supervisor around the display inspector, inside Gamescope.""" separator = command.index('--') if command[:separator].count('--backend') != 1 or command[command.index('--backend')+1] != 'headless': refuse('Input exercise requires the explicit headless Gamescope boundary') helper = root/'tools/headless_input_proof.py' ownership = root/'tools/headless_display_probe.py' child = command[separator+1:] if (not display_context or len(child) < 3 or child[:2] != [sys.executable, str(ownership)] or not re.fullmatch(r'[A-Za-z0-9]{1,32}', run_id)): refuse('Input exercise requires the existing scoped display wrapper and observer nonce') if type(held_action) is not bool or (held_action and not (selection and coarse_facing)): refuse('Held action requires explicit selection and coarse facing') if type(coarse_facing) is not bool or (coarse_facing and not (coarse_approach and selection)): refuse('Coarse facing requires explicit coarse approach and selection') if type(coarse_approach) is not bool or (coarse_approach and not approach): refuse('Coarse approach requires explicit approach') if type(selection) is not bool or (selection and not alignment): refuse('Action selection requires explicit alignment') if type(alignment) is not bool or (alignment and not approach): refuse('Prison alignment requires explicit prior approach') if type(approach) is not bool or (approach and not look): refuse('Prison approach requires explicit relative look') held_tokens = [i for i,value in enumerate(child) if value.startswith('-errOrdinaryHeld')] if held_action: if len(held_tokens) != 1 or child[held_tokens[0]] != '-errOrdinaryHeldAction' or child[held_tokens[0]+1:held_tokens[0]+2] != [run_id]: refuse('Held action requires its exact native observer nonce') elif held_tokens: refuse('Unexpected native held-action option') gate_tokens = [value for value in child if value.startswith('-errOrdinaryGate')] if selection and '-errOrdinaryGateChoices' not in gate_tokens: refuse('Action selection requires exact native choices') if any(value not in {'-errOrdinaryGateObserve', '-errOrdinaryGateChoices'} for value in gate_tokens): refuse('Unknown native gate observer option') if approach and '-errOrdinaryGateObserve' not in gate_tokens: refuse('Prison approach requires its exact native gate observer nonce') if '-errOrdinaryGateChoices' in gate_tokens and not approach: refuse('Input exercise with gate choices requires explicit approach') for name in set(gate_tokens): index = child.index(name) if child.count(name) != 1 or child[index+1:index+2] != [run_id]: refuse('Native gate observer option is repeated or has a mismatched nonce') if type(look) is not bool: refuse('Relative look selection must be boolean') look_options = [i for i,value in enumerate(child) if value.startswith('-errOrdinaryLook')] if look: if len(look_options) != 1 or child[look_options[0]] != '-errOrdinaryLookObserve' or child[look_options[0]+1:look_options[0]+2] != [run_id]: refuse('Relative look requires its exact native observer nonce') elif look_options: refuse('Unexpected native relative-look observer') report = profile/('input-proof-'+run_id+'.json') if helper.is_symlink() or not helper.is_file() or report.exists() or report.is_symlink(): refuse('Input exercise requires a regular helper and a new report') context = {'report': str(report), 'helper_sha256': hashlib.sha256(helper.read_bytes()).hexdigest(), 'ownership_helper_sha256': hashlib.sha256(ownership.read_bytes()).hexdigest(), 'prefix': str(compat/'pfx'), 'profile': str(profile), 'game_executable': str(executable), 'command': child, 'nonce': run_id, 'display_report': display_context['report']} wrapper = [sys.executable, str(helper), '--enable-private-keyboard-proof', '--host-display', host_display, '--prefix', context['prefix'], '--profile', str(profile), '--game-executable', str(executable), '--report', str(report), '--nonce', run_id, '--'] if look: context['look_requested'] = True wrapper.insert(-1, '--enable-private-look-proof') if approach: context['approach_requested'] = True context['approach_module_sha256'] = approach_module_hashes(root) wrapper.insert(-1, '--enable-private-approach-proof') if coarse_facing: context['coarse_facing_requested'] = True wrapper.insert(-1, '--enable-private-coarse-facing-proof') if coarse_approach: context['coarse_approach_requested'] = True wrapper.insert(-1, '--enable-private-coarse-approach-proof') if alignment: context['alignment_requested'] = True context['alignment_module_sha256'] = alignment_module_hashes(root) wrapper.insert(-1, '--enable-private-alignment-proof') if selection: context['selection_requested'] = True context['selection_module_sha256'] = selection_module_hashes(root) wrapper.insert(-1, '--enable-private-selection-proof') if held_action: context['held_action_requested'] = True context['held_module_sha256'] = held_module_hashes(root) wrapper.insert(-1, '--enable-private-held-action-proof') return command[:separator+1] + wrapper + child, context def input_exercise_poll_failure(context): """Read only a complete, context-bound failure; an in-progress write is pending.""" if not context: return False try: source = Path(context['report']) if source.is_symlink() or not source.is_file() or not 0 < source.stat().st_size <= 1024*1024: return False with source.open('rb') as stream: raw = stream.read(1024*1024+1) if len(raw) > 1024*1024: return False report = json.loads(raw) if not isinstance(report, dict) or report.get('schema') != 1 or report.get('kind') != 'headless-private-keyboard-experiment': return False for key in ('helper_sha256', 'ownership_helper_sha256', 'prefix', 'profile', 'game_executable', 'report', 'command', 'nonce'): if report.get(key) != context[key]: return False if type(report.get('held_action_requested', False)) is not bool or report.get('held_action_requested', False) != context.get('held_action_requested', False): return False if context.get('held_action_requested') and report.get('held_module_sha256') != context.get('held_module_sha256'): return False if context.get('selection_requested') and report.get('selection_module_sha256') != context.get('selection_module_sha256'): return False if report.get('selection_requested', False) != context.get('selection_requested', False): return False if context.get('alignment_requested') and report.get('alignment_module_sha256') != context.get('alignment_module_sha256'): return False if report.get('alignment_requested', False) != context.get('alignment_requested', False): return False if context.get('approach_requested') and report.get('approach_module_sha256') != context.get('approach_module_sha256'): return None if report.get('approach_requested', False) != context.get('approach_requested', False): return None if type(report.get('coarse_approach_requested', False)) is not bool or report.get('coarse_approach_requested', False) != context.get('coarse_approach_requested', False): return False if type(report.get('coarse_facing_requested', False)) is not bool or report.get('coarse_facing_requested', False) != context.get('coarse_facing_requested', False): return False if report.get('look_requested', False) != context.get('look_requested', False): return False return bool(report.get('error') or report.get('supervision_failed') is True or ( report.get('sequence_completed') is False and report.get('whole_process_group_cleanup_verified') is False and type(report.get('input_sent')) is bool)) except (OSError, ValueError, TypeError, KeyError): return False def input_exercise_poll_result(context, outcome, native_failure): if not input_exercise_poll_failure(context): return outcome # Native failures already latched by StartupMonitor remain authoritative, # even during its diagnostic grace period. Never wait for settle on failure. if native_failure: return 'game-failed' if outcome is not None and outcome != 'game-started': return outcome return 'input-exercise-failed' def input_exercise_result(result, context, session, observation, display_result): """Bind immutable helper bytes and require independent native player evidence.""" proof = {'requested': True, 'verified': False, 'input_sent': None, 'gameplay_qualified': False, 'scope': 'bounded private keyboard delivery plus native movement/inventory observations; no full escape or multiplayer qualification'} try: source = Path(context['report']) if source.is_symlink() or not source.is_file() or not 0 < source.stat().st_size <= 1024*1024: refuse('Missing or invalid bounded input report') raw = source.read_bytes() proof['sha256'] = hashlib.sha256(raw).hexdigest() if not session: refuse('Input experiment lacks an archived native session') with (session/'input-proof.json').open('xb') as output: output.write(raw) proof['archive_file'] = 'input-proof.json' report = json.loads(raw) if not isinstance(report, dict): refuse('Input report is not an object') for key in ('helper_sha256', 'ownership_helper_sha256', 'prefix', 'profile', 'game_executable', 'report', 'command', 'nonce'): if report.get(key) != context[key]: refuse('Input report binding mismatch: '+key) held_action_requested = context.get('held_action_requested', False) if type(held_action_requested) is not bool or type(report.get('held_action_requested', False)) is not bool or report.get('held_action_requested', False) != held_action_requested: refuse('Held action request does not match launch context') if held_action_requested and (context.get('selection_requested') is not True or context.get('coarse_facing_requested') is not True or report.get('held_module_sha256') != context.get('held_module_sha256') or context.get('held_module_sha256') != held_module_hashes(ROOT)): refuse('Held action requires explicit coarse selection and unchanged module sources') if not held_action_requested and (report.get('held_action_proof') is not None or report.get('held_events')): refuse('Unexpected held action evidence or F input') if report.get('held_release_errors', []) != []: refuse('Held F release is unconfirmed') selection_requested = context.get('selection_requested', False) if type(report.get('selection_requested', False)) is not bool or report.get('selection_requested', False) != selection_requested: refuse('Action selection request does not match launch context') if selection_requested and (not context.get('alignment_requested') or report.get('selection_module_sha256') != context.get('selection_module_sha256') or context.get('selection_module_sha256') != selection_module_hashes(ROOT)): refuse('Selection requires explicit alignment and unchanged module sources') if not selection_requested and any(report.get(k) is not None for k in ('facing_proof', 'action_selection_proof')): refuse('Unexpected action-selection proof') if not selection_requested and report.get('wheel_events'): refuse('Unexpected wheel input') if report.get('unconfirmed_held_button') is not None or report.get('wheel_release_errors', []) != []: refuse('Wheel release is unconfirmed') alignment_requested = context.get('alignment_requested', False) if type(report.get('alignment_requested', False)) is not bool or report.get('alignment_requested', False) != alignment_requested: refuse('Prison alignment request does not match launch context') if alignment_requested and (not context.get('approach_requested') or report.get('alignment_module_sha256') != context.get('alignment_module_sha256') or context.get('alignment_module_sha256') != alignment_module_hashes(ROOT)): refuse('Alignment requires prior approach and unchanged module sources') if not alignment_requested and report.get('alignment_proof') is not None: refuse('Unexpected alignment evidence') approach_requested = context.get('approach_requested', False) if type(report.get('approach_requested', False)) is not bool or report.get('approach_requested', False) != approach_requested: refuse('Prison approach request does not match launch context') if approach_requested and (report.get('approach_module_sha256') != context.get('approach_module_sha256') or context.get('approach_module_sha256') != approach_module_hashes(ROOT)): refuse('Approach module sources changed or differ from launch hashes') if not approach_requested and report.get('approach_proof') is not None: refuse('Unexpected approach evidence in fixed input run') coarse_approach_requested = context.get('coarse_approach_requested', False) if type(coarse_approach_requested) is not bool or type(report.get('coarse_approach_requested', False)) is not bool or report.get('coarse_approach_requested', False) != coarse_approach_requested or (coarse_approach_requested and not approach_requested): refuse('Coarse approach request does not match explicit launch context') if approach_requested and report.get('approach_proof', {}).get('profile', 'standard') != ('coarse' if coarse_approach_requested else 'standard'): refuse('Approach profile does not match explicit launch context') coarse_facing_requested = context.get('coarse_facing_requested', False) if type(coarse_facing_requested) is not bool or type(report.get('coarse_facing_requested', False)) is not bool or report.get('coarse_facing_requested', False) != coarse_facing_requested or (coarse_facing_requested and not (coarse_approach_requested and selection_requested)): refuse('Coarse facing request does not match explicit launch context') facing_profile = 'coarse' if coarse_facing_requested else 'standard' if report.get('facing_proof', {}).get('profile', 'standard') != facing_profile: refuse('Facing profile does not match explicit launch context') look_requested = context.get('look_requested', False) if type(report.get('look_requested', False)) is not bool or report.get('look_requested', False) != look_requested: refuse('Relative look request does not match launch context') if not look_requested and report.get('look_proof') is not None: refuse('Unexpected relative mouse evidence in keyboard-only run') if type(report.get('input_sent')) is bool: proof['input_sent'] = report['input_sent'] if (report.get('schema') != 1 or report.get('kind') != 'headless-private-keyboard-experiment' or report.get('sequence_completed') is not True or report.get('input_sent') is not True or report.get('hard_supervision') is not True or report.get('supervision_failed') or report.get('all_attempted_presses_released') is not True or report.get('error') or report.get('paired_release_errors') != [] or report.get('unconfirmed_held_key') is not None): refuse('Private input sequence did not complete with confirmed paired release') import math events = report.get('events', []) expected = [('W', True), ('W', False), ('Tab', True), ('Tab', False), ('Tab', True), ('Tab', False)] if approach_requested: if not isinstance(events, list) or not 2 <= len(events) <= (32 if coarse_approach_requested else 24) or len(events)%2: refuse('Prison approach requires bounded W pairs') expected = [('W', i%2 == 0) for i in range(len(events))] if (not isinstance(events, list) or len(events) != len(expected) or any(not isinstance(e, dict) or type(e.get('down')) is not bool or (e.get('key'), e['down']) != pair or e.get('finally') for e, pair in zip(events, expected))): refuse('Input experiment does not contain the exact fixed key pairs') times = [e.get('monotonic') for e in events] if any(type(t) not in {int, float} or not math.isfinite(t) or t < 0 for t in times) or times != sorted(times) or times[-1]-times[0] > (35 if approach_requested else 25): refuse('Input event clock is invalid or outside its bound') if not display_result or display_result.get('verified') is not True: refuse('Input experiment lacks verified private display ownership') display = json.loads((session/'display-ownership.json').read_bytes()) if report.get('binding') != display.get('binding') or report.get('owner') not in [ {key: window[key] for key in ('window', 'pid', 'server_pid', 'process')} for window in display.get('owned_windows', [])]: refuse('Input owner does not match the independently observed private game window') native = report.get('observer', {}) if (report.get('console') != str(session/'console.log') or native.get('events') != (['READY'] if approach_requested else ['READY', 'INVENTORY_OPEN', 'INVENTORY_CLOSED'])): refuse('Input experiment lacks the exact native session/transition handshake') if (not observation or observation.get('nonce') != context['nonce'] or not observation.get('sampler_complete') or (not approach_requested and (not observation.get('inventory_open_observed') or not observation.get('inventory_close_observed'))) or (not approach_requested and (not observation.get('movement', {}).get('action_observed') or not observation.get('movement', {}).get('physical_motion_observed') or not observation.get('movement', {}).get('released_input_observed')))): refuse('Native movement and inventory open/close proof is incomplete') if approach_requested: from prison_approach_evidence import validate_approach_evidence route = report.get('approach_proof', {}) route_events = route.get('events', []) if len(route_events) != len(events) or any(not isinstance(a, dict) or a.get('key') != b.get('key') or a.get('down') is not b.get('down') or type(a.get('monotonic')) not in (int, float) or not math.isfinite(a['monotonic']) or abs(a['monotonic']-b['monotonic']) > .01 for a,b in zip(route_events,events)): refuse('Approach adapter and route key evidence differ') proof['approach_proof'] = validate_approach_evidence((session/'console.log').read_text(errors='replace'), context['nonce'], report.get('look_proof'), route) proof['scope'] = 'private relative mouse and W input with independently observed interior prison approach; no held action, gate exit, full escape or multiplayer qualification' if proof['approach_proof'].get('verified') is not True: refuse('Independent native prison approach evidence is incomplete') if selection_requested: from prison_action_selection_evidence import validate_selection_evidence selected = report.get('action_selection_proof', {}) wheel = selected.get('wheel', {}) buttons = report.get('wheel_events', []) if not isinstance(buttons,list) or len(buttons) != 2: refuse('Selection requires exactly one wheel pair') for event, down, key in zip(buttons,(True,False),('down_monotonic','up_monotonic')): if (not isinstance(event,dict) or type(event.get('button')) is not int or event['button'] != 5 or event.get('down') is not down or event.get('sent') is not True or event.get('finally') or type(event.get('monotonic')) not in (int,float) or not math.isfinite(event['monotonic']) or type(wheel.get(key)) not in (int,float) or not math.isfinite(wheel[key]) or abs(event['monotonic']-wheel[key]) > .01): refuse('Selection wheel adapter and policy pair disagree') proof['action_selection_proof'] = validate_selection_evidence((session/'console.log').read_text(errors='replace'),context['nonce'],report.get('look_proof'),route,report.get('facing_proof'),selected,facing_profile=facing_profile) proof['scope'] = 'ordinary private approach, stationary facing and one wheel pair with independent native original-lock selection; no F hold, unlock, exit or multiplayer qualification' if proof['action_selection_proof'].get('verified') is not True: refuse('Independent native action-selection evidence is incomplete') if held_action_requested: from prison_held_action_evidence import validate_held_action_evidence held = report.get('held_action_proof', {}) held_events = report.get('held_events', []) policy_events = held.get('events', []) if (held.get('completed') is not True or held.get('status') != 'verified' or held.get('input_attempted') is not True or held.get('release_confirmed') is not True or held.get('unconfirmed_held_key') is not None or held.get('paired_release_errors') != [] or held.get('notification_errors') != [] or held.get('error') or not isinstance(held_events, list) or len(held_events) != 2 or not isinstance(policy_events, list) or len(policy_events) != 2): refuse('Held action lacks exactly one confirmed F pair') for event, policy_event, down, field in zip(held_events,policy_events,(True,False),('down_monotonic','up_monotonic')): stamp = event.get('monotonic') if isinstance(event,dict) else None if (not isinstance(event,dict) or set(event) != {'key','down','monotonic','sent'} or event.get('key') != 'F' or event.get('down') is not down or event.get('sent') is not True or type(stamp) not in (int,float) or not math.isfinite(stamp) or stamp < 0 or not isinstance(policy_event,dict) or policy_event.get('key') != 'F' or policy_event.get('down') is not down or policy_event.get('confirmed') is not True or policy_event.get('cleanup') or type(policy_event.get('monotonic')) not in (int,float) or policy_event['monotonic'] != stamp or type(held.get(field)) not in (int,float) or held[field] != stamp): refuse('Held F adapter and policy pair disagree') if not held_events[0]['monotonic'] < held_events[1]['monotonic'] or held_events[1]['monotonic']-held_events[0]['monotonic'] > 5 or held_events[0]['monotonic'] <= buttons[-1]['monotonic']: refuse('Held F pair violates order or five-second bound') proof['held_action_proof'] = validate_held_action_evidence((session/'console.log').read_text(errors='replace'),context['nonce'],report.get('look_proof'),route,report.get('facing_proof'),selected,held,facing_profile=facing_profile) proof['scope'] = 'ordinary private approach, selection and one held F action with independent native unlock, original door motion and released-input evidence; no gate passage, full escape or multiplayer qualification' if any(proof['held_action_proof'].get(key) is not True for key in ('verified','held_action_verified','unlock_observed','gate_motion_observed')): refuse('Independent native held-action evidence is incomplete') elif alignment_requested: from prison_alignment_evidence import validate_alignment_evidence proof['alignment_proof'] = validate_alignment_evidence((session/'console.log').read_text(errors='replace'), context['nonce'], report.get('look_proof'), route, report.get('alignment_proof')) proof['scope'] = 'private approach and stationary yaw alignment with independently observed native prison lock selection; no F hold, unlock, exit or multiplayer qualification' if proof['alignment_proof'].get('verified') is not True: refuse('Independent native prison lock alignment evidence is incomplete') elif look_requested: from ordinary_look_evidence import validate_look_evidence proof['look_proof'] = validate_look_evidence((session/'console.log').read_text(errors='replace'), context['nonce'], report.get('look_proof'), events) proof['scope'] = 'bounded private relative mouse and keyboard delivery with native view/movement/inventory evidence; no gate, full escape or multiplayer qualification' if proof['look_proof'].get('verified') is not True: refuse('Independent native relative-look evidence is incomplete') proof['verified'] = True except (OSError, ValueError, TypeError, AttributeError, KeyError) as error: proof['error'] = str(error) if result == 'game-started' and not proof['verified']: result = 'input-exercise-incomplete' return result, proof def run(a): validate_input_exercise(a) validate_input_observer(a) packed = packed_request(a) if getattr(a, 'capture_prison_view', False) and (a.project == CONTROL or packed is not None): refuse('Prison view capture requires loose successor source with its opt-in observer') if getattr(a, 'inspect_headless_display', False) and (a.project == CONTROL or packed is not None): refuse('Display ownership inspection currently requires loose Core or Chat') steam, compat = a.steam.resolve(), a.compat.resolve() profile = packed['session']/'profile' if packed else a.profile.resolve() if not 60 <= a.timeout <= 1800 or not 5 <= a.settle <= 120: refuse('Use timeout 60..1800 seconds and settle 5..120 seconds') if not 0 <= a.failure_grace <= 120: refuse('Use failure-grace 0..120 seconds') assert_no_game() if a.project == CONTROL: if a.timeout > 120 or a.settle > 30: refuse('Vanilla control is bounded to timeout 60..120 and settle 5..30 seconds') validate_control_profile(profile) elif (profile/CONTROL_MARKER).exists(): refuse('Successor runs must not reuse the vanilla-control profile') validate_paths(ROOT, steam, compat, profile) # Use the SAME prefix lock as headless_workbench.py: never run them together. with (compat/'.erri-workbench.lock').open('a') as prefix_lock: try: fcntl.flock(prefix_lock, fcntl.LOCK_EX | fcntl.LOCK_NB) except BlockingIOError: refuse('Another test runner owns this disposable prefix') assert_prefix_idle(compat) return run_locked(a, steam, compat, profile, packed) def run_locked(a, steam, compat, profile, packed=None): command = command_for(ROOT, steam, compat, profile, a.project, packed, getattr(a, 'capture_prison_view', False), 2560 if getattr(a, 'approach_wide_display', False) else 1280) separator = command.index("--") proton, executable = Path(command[separator+1]), Path(command[separator+3]) if not proton.is_file() or not executable.is_file(): refuse('Installed ordinary ArmaReforgerSteam.exe and Proton Experimental are required') if not shutil.which('gamescope') or not shutil.which('pactl'): refuse('gamescope and pactl are required') # Refuse launcher/profile reuse; repeat immediately before any launch setup. assert_no_game() if packed: prepare_packed(packed) profile.mkdir(parents=True, exist_ok=True) marker = profile/PROFILE_MARKER if not marker.exists(): marker.write_text(json.dumps(PROFILE_ID)+'\n') if a.project == CONTROL: (profile/CONTROL_MARKER).write_text(json.dumps(CONTROL_ID)+'\n') addons = profile/'addons' addons.mkdir(exist_ok=True) vanilla = steam/'steamapps/common/Arma Reforger/addons/data' link = addons/'ArmaReforger' if not link.exists() and not link.is_symlink(): link.symlink_to(vanilla, target_is_directory=True) if packed and packed.get('schema_version') == 3: from packed_external import prepare_profile prepare_profile(packed, steam, profile) elif link.resolve() != vanilla.resolve() or any(x.name not in {'ArmaReforger', 'saves'} for x in addons.iterdir()) or any((addons/'saves').rglob('*.gproj')): refuse('Private profile addons must contain only the installed vanilla link') run_id = uuid.uuid4().hex observation_nonce = run_id if getattr(a, 'observe_player_input', False) else None if observation_nonce: command = input_observer_command(command, a.project, packed, observation_nonce, getattr(a, 'exercise_relative_look', False), getattr(a, 'observe_prison_gate', False), getattr(a, 'observe_prison_choices', False), getattr(a, 'exercise_prison_held_action', False)) display_context = None if getattr(a, 'inspect_headless_display', False): command, display_context = display_probe_command(command, ROOT, compat, profile, executable, run_id, os.environ.get('DISPLAY', '')) exercise_context = None if getattr(a, 'exercise_player_input', False): command, exercise_context = input_exercise_command(command, ROOT, compat, profile, executable, run_id, os.environ.get('DISPLAY', ''), display_context, getattr(a, 'exercise_relative_look', False), getattr(a, 'exercise_prison_approach', False), getattr(a, 'exercise_prison_alignment', False), getattr(a, 'exercise_prison_selection', False), getattr(a, 'exercise_prison_coarse_approach', False), getattr(a, 'exercise_prison_coarse_facing', False), getattr(a, 'exercise_prison_held_action', False)) silent_sink = 'erri_game_'+run_id application_id = 'org.escapistsrereintroduced.game.'+run_id env = os.environ.copy() env.update(STEAM_COMPAT_DATA_PATH=str(compat), STEAM_COMPAT_CLIENT_INSTALL_PATH=str(steam), WINEPREFIX=str(compat/'pfx'), PULSE_SINK=silent_sink, PULSE_PROP='application.id='+application_id) sink_module = None child = None session = None captured = None if packed else snapshot(ROOT, command) packed_raw = packed_snapshot(packed, command, profile) if packed else None started = time.time() monitor = StartupMonitor(time.monotonic(), a.timeout, a.settle, a.failure_grace) result = 'timeout' runner_error = None cleanup_ok = True cleanup_report = None log_root = profile/'logs' previous = set(log_root.glob('logs_*')) launch_log = profile/('game-launch-'+run_id+'.log') try: sink_module = subprocess.check_output(['pactl', 'load-module', 'module-null-sink', 'sink_name='+silent_sink], text=True).strip() subprocess.run(['pactl', 'set-sink-mute', silent_sink, '1'], check=True) subprocess.run(['pactl', 'set-sink-volume', silent_sink, '0%'], check=True) assert_no_game() with launch_log.open('x') as output: child = subprocess.Popen(command, cwd=executable.parent, env=env, stdout=output, stderr=subprocess.STDOUT, start_new_session=True) while time.monotonic() < monitor.deadline: # -noSound prevents game audio; this catches a Proton startup stream # or a saved explicit Wine device. Only our fresh application ID. streams = json.loads(subprocess.check_output(['pactl', '-f', 'json', 'list', 'sink-inputs'], text=True)) for stream in streams: if stream.get('properties', {}).get('application.id') != application_id: continue index = str(stream['index']) subprocess.run(['pactl', 'move-sink-input', index, silent_sink], check=True) subprocess.run(['pactl', 'set-sink-input-mute', index, '1'], check=True) candidates = [x for x in log_root.glob('logs_*') if x not in previous] if len(candidates) > 1: result = 'unexpected-multiple-sessions'; break evidence = {} if candidates: if session is None: session = candidates[0] if packed: (session/PACKED_MANIFEST).write_bytes(packed_raw) else: persist(session, captured) console = session/'console.log' content = console.read_text(errors='replace') if console.is_file() else '' evidence = packed_evidence(content, packed, steam, profile) if packed else game_evidence(content, a.project) outcome = monitor.observe(evidence, time.monotonic(), child.poll() is not None) outcome = input_exercise_poll_result(exercise_context, outcome, monitor.first_failure_at is not None) if outcome is not None: result = outcome; break # Grace never extends the original absolute run deadline. cutoff = monitor.deadline if monitor.first_failure_at is not None: cutoff = min(cutoff, monitor.first_failure_at + a.failure_grace) time.sleep(max(0, min(1, cutoff - time.monotonic()))) else: result = monitor.timeout_result() except KeyboardInterrupt: result = 'interrupted' except (OSError, ValueError, subprocess.SubprocessError) as error: result = 'runner-failed' runner_error = str(error) finally: if child is not None: cleanup_report = cleanup_owned_run(child,proton,env,compat) cleanup_ok = cleanup_report['cleanup_confirmed'] if not cleanup_ok: result = 'cleanup-failed' cleanup_error = '; '.join(cleanup_report['errors']) runner_error = (runner_error + '; ' if runner_error else '') + cleanup_error if sink_module is not None: try: subprocess.run(['pactl', 'unload-module', sink_module], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=5) except (OSError, subprocess.SubprocessError): pass # Never replace the user's audio defaults to compensate. if packed: sealed = packed_seal(packed, packed_raw, session) else: sealed = seal_run(ROOT, session, captured) if cleanup_ok and session and (session/'console.log').is_file() else {} content = (session/'console.log').read_text(errors='replace') if session and (session/'console.log').is_file() else '' if packed: final = packed_evidence(content, packed, steam, profile) else: final = game_evidence(content, a.project) if session and (session/'console.log').is_file() else {} final = monitor.annotate(final) if final.get('failed'): result = 'game-failed' if packed and not sealed['packed_inputs_unchanged_after_run']: result = 'packed-input-changed' if not packed and sealed and not sealed['source_unchanged_after_run']: result = 'source-changed' if a.project == CONTROL and result == 'game-started': result = 'vanilla-control-started' if packed and result == 'game-started': result = 'packed-'+packed.get('project', 'core')+'-started' if final.get('ready') is True else 'packed-evidence-incomplete' display_result = None startup_result = result if display_context: display_result = display_probe_result(display_context, session) if not display_result['verified'] and result == 'game-started': result = 'display-probe-incomplete' observation_result = None if observation_nonce: result, observation_result = input_observer_result(result, content, observation_nonce) final['input_observation'] = observation_result exercise_result = None if exercise_context: result, exercise_result = input_exercise_result(result, exercise_context, session, observation_result, display_result) final['input_exercise'] = exercise_result final['scope'] = 'ordinary game startup with an explicitly requested bounded private input experiment; see input_exercise for sent-input and native-proof results; no full escape, multiplayer or FPS qualification' source_bindings = {} if packed else bindings(captured) summary = {**sealed, **source_bindings, 'started_unix': started, 'result': result, 'session': str(session) if session else None, 'project': a.project, 'cleanup_confirmed': cleanup_ok, 'cleanup_report':cleanup_report, 'runner_error': runner_error, 'failure_grace_seconds': a.failure_grace, 'failure_observed_during_polling': monitor.first_failure_at is not None, 'first_failure_elapsed_seconds': (monitor.first_failure_at - monitor.started) if monitor.first_failure_at is not None else None, 'executable': str(executable), 'dedicated_server': False, 'display': 'headless Gamescope', 'private_viewport': {'width':2560 if getattr(a, 'approach_wide_display', False) else 1280,'height':720}, 'audio': '-noSound plus own silent sink', 'control_only': a.project == CONTROL, 'source_manifest_scope': 'none: packed input provenance is separate; loose source manifests cannot qualify this run' if packed else 'successor addon snapshot retained for provenance bookkeeping only; none loaded by vanilla control' if a.project == CONTROL else 'successor addon sources', 'scope': final.get('scope', 'ordinary game compile/startup; no scripted fixture, full escape or multiplayer acceptance')} if observation_result is not None: summary['input_observation'] = observation_result if exercise_result is not None: summary['input_exercise'] = exercise_result if display_result is not None: summary['display_ownership'] = display_result summary['startup_result_before_display_check'] = startup_result if packed: summary['packed_consumption_verified'] = result == 'packed-'+packed.get('project', 'core')+'-started' and cleanup_ok and final.get('ready') is True and sealed['packed_inputs_unchanged_after_run'] final['packed_consumption_verified'] = summary['packed_consumption_verified'] (packed['session']/'packed-runner.json').write_text(json.dumps(summary, indent=2)+'\n') (packed['session']/'packed-startup-report.json').write_text(json.dumps(final, indent=2)+'\n') if session: (session/'runner.json').write_text(json.dumps(summary, indent=2)+'\n') (session/'game-startup-report.json').write_text(json.dumps(final, indent=2)+'\n') shutil.copyfile(launch_log, session/'headless-launch.log') print(json.dumps(summary, indent=2), flush=True) return 0 if result in {'game-started', 'vanilla-control-started', 'packed-core-started', 'packed-chat-started'} else 1 def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--steam', type=Path, default=Path.home()/'.local/share/Steam') parser.add_argument('--compat', type=Path, required=True) parser.add_argument('--profile', type=Path, default=Path('/tmp/erri-game-profile')) parser.add_argument('--project', choices=['core', 'chat', CONTROL], default='core') packed_options = parser.add_mutually_exclusive_group() packed_options.add_argument('--packed-core', type=Path, help='Unqualified Core candidate directory containing exactly addon.gproj, data.pak and resourceDatabase.rdb') packed_options.add_argument('--packed-addons', type=Path, help='Unqualified complete core/ai/intel/map/chat packed directories; requires --project chat') parser.add_argument('--external-addons', type=Path, help='Explicit DEVELOPMENT packed Chat v3 only: audited installed RHS/WCS/BMP root; six private links, no broad workshop mount') parser.add_argument('--packed-session', type=Path, help='New /tmp/erri-packed-core- or erri-packed-chat- session; owns a fresh input copy and profile') parser.add_argument('--inspect-processes', action='store_true', help='Read-only process preflight report; never launch, lock, write a profile or configure audio') parser.add_argument('--inspect-headless-display', action='store_true', help='Read-only ownership inspection within the hidden display; no input sent, loose Core/Chat only') parser.add_argument('--capture-prison-view', action='store_true', help='Opt-in single genuine local prison screenshot; loose successor source only, no camera or input changes') parser.add_argument('--observe-player-input', action='store_true', help='Read-only native local-input sampler; loose Core/Chat only, requires --settle >=50; sends no input') parser.add_argument('--approach-wide-display', action='store_true', help='Use a disposable 2560x720 headless viewport for the explicit approach experiment; no physical display changes') parser.add_argument('--exercise-prison-held-action', action='store_true', help='Explicit single held-F unlock experiment after coarse facing and selection; requires independent native progress, original door motion and release evidence') parser.add_argument('--exercise-prison-selection', action='store_true', help='Explicit facing and single wheel selection trial; requires alignment and native choice diagnostics; no F or gate operation') parser.add_argument('--exercise-prison-alignment', action='store_true', help='Explicit stationary yaw-only alignment to native prison lock after verified approach; no F or gate operation') parser.add_argument('--exercise-prison-coarse-facing', action='store_true', help='Explicit coarse yaw facing; requires coarse approach and selection, unchanged total pixels and clocks') parser.add_argument('--exercise-prison-coarse-approach', action='store_true', help='Explicit coarse stationary-heading approach profile; requires approach and retains original clocks; no F') parser.add_argument('--exercise-prison-approach', action='store_true', help='Explicit bounded relative-mouse/W approach inside prison; requires relative look and gate observer; never operates gate') parser.add_argument('--observe-prison-choices', action='store_true', help='Read-only native selected/ordered gate action diagnostics; requires --observe-prison-gate') parser.add_argument('--observe-prison-gate', action='store_true', help='Read-only native registered prison lock/context/progress telemetry; requires --observe-player-input') parser.add_argument('--exercise-relative-look', action='store_true', help='Explicit tiny relative right/return mouse experiment before W/Tab; requires --exercise-player-input and native view proof') parser.add_argument('--exercise-player-input', action='store_true', help='Explicit bounded private W/Tab keyboard experiment; requires observer, display inspection, loose Core/Chat and settle >=50') parser.add_argument('--timeout', type=int, default=180) parser.add_argument('--settle', type=int, default=15) parser.add_argument('--failure-grace', type=int, default=0, help='Keep a failed run alive for diagnostic logs only, at most 0..120 seconds within the original timeout') args = parser.parse_args() try: if args.inspect_processes: report = {'read_only': True, 'game_process_check': 'clear', 'prefix_process_check': 'clear'} for key, check in [('game_process_check', assert_no_game), ('prefix_process_check', lambda: assert_prefix_idle(args.compat.resolve()))]: try: check() except (OSError, ValueError) as error: report[key] = str(error) print(json.dumps(report, indent=2)) return 0 if report['game_process_check'] == report['prefix_process_check'] == 'clear' else 1 return run(args) except (OSError, ValueError, subprocess.SubprocessError) as error: parser.error(str(error)) if __name__ == '__main__': raise SystemExit(main())