jakub
cab998fb64
Merge owned systems into one addon while preserving resource identities. Gate player admission on the replicated prison scene, restore the original framed door, and hold enemy AI until escape. Correct scenario branding and author metadata, include publishing assets, and retain regression and native verification evidence.
1512 lines
96 KiB
Python
1512 lines
96 KiB
Python
#!/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+'-<name> 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') == 4:
|
|
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') == 4:
|
|
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') == 4:
|
|
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') == 4:
|
|
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 <CLI profile>/profile, so this exact observed
|
|
# engine alias denotes <CLI profile>/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', ('2528466DE220D91C' if packed.get('schema_version') == 4 else PACKED_PROJECTS[project][1])]
|
|
elif project != CONTROL:
|
|
project_text = (root/'addons/core/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 ')) or bool(re.search(r'Re(?:re)?introduced \[mission\]', content))
|
|
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)
|
|
# Accept the historical spelling when reviewing immutable older evidence.
|
|
ready = header_verified and participant_verified and bool(re.search(r'Re(?:re)?introduced \[mission\]: phase 0 -> 1', content))
|
|
unexpected_fixture = 'ERR SMOKE ' in content
|
|
failed = (bool(report['engine_errors'] or report['script_or_runtime_failures'])
|
|
or bool(re.search(r'Re(?:re)?introduced: scenario preparation failed', 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') == 4:
|
|
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='Packed addon root; with --external-addons expects the single combined core folder (v4), otherwise historical five-addon v2; requires --project chat')
|
|
parser.add_argument('--external-addons', type=Path, help='Explicit DEVELOPMENT combined packed mod v4 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-<name> or erri-packed-chat-<name> 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())
|