#!/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 time import uuid from native_provenance import snapshot, persist, bindings, seal_run from native_report import analyze 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 = re.fullmatch(r'\s*GameProject\s*\{\s*ID\s+"'+name+r'"\s+GUID\s+"'+guid+r'"\s+TITLE\s+"[^"\r\n]+"\s+Dependencies\s*\{((?:\s*"[0-9A-F]{16}"\s*)+)\}\s*\}\s*', text) actual = re.findall(r'"([0-9A-F]{16})"', match[1]) if match else [] expected = {PACKED_BASE, *(PACKED_PROJECTS[key][1] for key in dependencies)} if not match 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): 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.""" 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): 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.""" 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): 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 = ('WINEPREFIX='+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 if target 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): if project not in {'core', 'chat', CONTROL}: refuse('Unsupported ordinary game project/control') command = ['gamescope', '--backend', 'headless', '-W', '1280', '-H', '720', '-w', '1280', '-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'] 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) 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 <= 30: refuse('Use failure-grace 0..30 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 run(a): packed = packed_request(a) 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 <= 30: refuse('Use failure-grace 0..30 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) 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 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 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 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) 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: try: cleanup(child, proton, env) assert_prefix_idle(compat) except (OSError, ValueError, subprocess.SubprocessError) as error: cleanup_ok = False result = 'cleanup-failed' runner_error = str(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' 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, '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', '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 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('--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('--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..30 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())