Something went wrong. Try again.
Native control, inspection, UI automation and evidence capture for a running Witcher 3 process.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237"""Bounded startup evidence, without executing scripts before a real game window."""import jsonfrom pathlib import Pathimport queueimport reimport socketimport threadingimport time
from . import supportfrom . import wirefrom .debug_journal import durable_jsonfrom .diagnostics import DiagnosticError
def read(folder): return json.loads((Path(folder)/'startup.json').read_text())
def update(folder, **fields): value = read(folder) value.update(fields) durable_json(Path(folder)/'startup.json', value) return value
def manifests(): return dict(saves=support.manage.files(support.manage.DOCS/'gamesaves'), mods=support.manage.files(support.manage.GAME/'mods'), settings={name: support.manage.digest(support.manage.DOCS/name) for name in ('user.settings', 'dx12user.settings')})
def processes(): return [identity for entry in Path('/proc').iterdir() if entry.name.isdigit() if (identity := support.process_identity(entry.name))]
def windows(process): """Inventory exact-PID windows; never adopt a similarly named foreign window.""" result = support.run('xdotool', 'search', '--pid', str(process['pid']), timeout=1) rows = [] for window in result.split()[:32]: try: title = support.run('xdotool', 'getwindowname', window, timeout=.5) rows.append(dict(id=window, title=title)) except RuntimeError: rows.append(dict(id=window, title=None)) return rows
class CompilerListener: """Subscribe only. Never reload, exec, acquire a native session, or retry work.""" def __init__(self, deadline): self.deadline = deadline self.stopped = threading.Event() self.connection = None self.queue = queue.Queue(maxsize=2048) self.bytes = 0 self.connected = False self.dropped = False self.error = None self.thread = threading.Thread(target=self.run, daemon=True)
def run(self): while not self.stopped.is_set() and time.monotonic() < self.deadline: connection = None try: connection = wire.Connection(min(.2, max(.01, self.deadline-time.monotonic()))) self.connection = connection self.connected = True connection.deadline = self.deadline while not self.stopped.is_set(): packet = connection.receive() if packet[:1] != ['ScriptCompiler']: continue self.bytes += len(json.dumps(packet).encode()) if self.bytes > 256*1024: self.dropped = True return try: self.queue.put_nowait(packet) except queue.Full: self.dropped = True return except (OSError, RuntimeError, ValueError) as error: if not self.stopped.is_set(): self.error = str(error) finally: if connection: connection.close() self.stopped.wait(.2)
def close(self): self.stopped.set() connection = self.connection if connection: try: connection.socket.shutdown(socket.SHUT_RDWR) except OSError: pass self.thread.join(timeout=1)
def log_state(): path = support.manage.DOCS/'scriptslog.txt' try: with path.open('rb') as stream: data = stream.read(128*1024) # Read a bounded prefix and detect append versus replacement independently. return dict(path=str(path), bytes=path.stat().st_size, prefix=data) except OSError: return dict(path=str(path), bytes=0, prefix=b'')
def fresh_log(before): current = log_state() prefix = current['prefix'] if before['bytes'] <= len(prefix) and prefix.startswith(before['prefix']): prefix = prefix[before['bytes']:] elif prefix == before['prefix']: prefix = b'' text = prefix.decode('utf-8', errors='replace') errors = [] # Only an explicit fresh script-file error is compiler-error evidence. for line in text.splitlines(): match = re.match(r'^\s*Error\s+(.+?\.ws)\((\d+)\)\s*:\s*(.*)$', line, re.I) if match: errors.append(['ScriptCompiler', 'error', int(match[2]), match[1], match[3]]) return dict(path=current['path'], fresh_text=text, errors=errors, bounded_prefix_bytes=131072, truncated=current['bytes'] > 131072)
def fail(folder, code, category, detail): value = read(folder) if value.get('failure'): return value failure = dict(code=code, category=category, detail=detail) return update(folder, phase='failed', failure=failure, cleanup_required=True, audits_final=False, recovery='Do not launch/attach/reload or remove locks. If the recorded lab process is live, ' 'confirm its PID/start/executable before external closure. After confirmed exit, ' 'run debug recover for this startup baseline and settings/save/mod audits.')
def wait_for_window(folder, seconds, launch): before_log = log_state() update(folder, phase='launching', before=manifests(), process=None, windows=[], compiler=dict(errors=[], packets=[], connected=False, coverage='not_observed')) update(folder, launch=launch()) deadline = time.monotonic()+seconds listener = CompilerListener(deadline) listener.thread.start() packets = [] observed = None try: while True: try: while True: packets.append(listener.queue.get_nowait()) except queue.Empty: pass log = fresh_log(before_log) errors = [packet for packet in packets if packet[:2] == ['ScriptCompiler', 'error']] errors += log['errors'] compiler = dict(errors=errors, packets=packets, connected=listener.connected, dropped=listener.dropped, listener_error=listener.error, coverage='available_packets_only_not_replayed', log=log) found = processes() inventory, window_error = [], None if len(found) == 1: if observed and found[0] != observed: update(folder, processes=found) fail(folder, 'E2003', 'process_identity_changed', 'Game process identity changed during startup.') raise DiagnosticError('E2003', 'Game process identity changed during startup.') observed = found[0] try: inventory = windows(observed) except (OSError, RuntimeError) as error: window_error = str(error) update(folder, phase='waiting_for_window', process=observed, processes=found, windows=inventory, window_error=window_error, compiler=compiler, seconds_remaining=max(0, deadline-time.monotonic())) compiler_windows = [row for row in inventory if row['title'] and 'Script Compilation Errors' in row['title']] if errors or compiler_windows: fail(folder, 'E5001', 'script_compiler_error', 'Script compiler errors observed.' if errors else 'Script Compilation Errors window observed; file/line diagnostics unavailable.') raise DiagnosticError('E5001', 'Startup script compilation failed.', read(folder)) if len(found) > 1: fail(folder, 'E2003', 'ambiguous_game_process', 'Multiple game processes; no process adopted.') raise DiagnosticError('E2003', 'Multiple game processes during startup.', read(folder)) if observed and not found: fail(folder, 'E2003', 'game_exited', 'The exact observed game process exited before readiness.') raise DiagnosticError('E2003', 'Game exited during startup.', read(folder)) main = [row for row in inventory if row['title'] == 'The Witcher 3'] if len(main) == 1: update(folder, phase='attaching', window=main[0], compiler=compiler) return if time.monotonic() >= deadline: category = 'game_absent' if not observed else 'window_mismatch' if inventory else 'startup_timeout' code = 'E2003' if category != 'startup_timeout' else 'E3003' fail(folder, code, category, 'Startup budget expired. Process/window evidence is retained; absent compiler evidence ' 'does not establish successful compilation or a mod-loading cause.') raise DiagnosticError(code, 'Startup budget expired: '+category, read(folder)) time.sleep(.2) finally: listener.close()
def recover_closed(folder): """Caller holds installation exclusion. Never replay or change native receipts.""" support.manage.ensure_game_closed() value = read(folder) if value.get('phase') == 'recovered': return dict(value['recovery_result'], historical=True, already_recovered=True) if not value.get('before'): raise DiagnosticError('E9001', 'Startup baseline unavailable; no audit inferred.', value) from . import game_settings from . import lifecycle result = dict(process_confirmed_closed=True, failures={}) for name, action in (('game_settings', game_settings.reset), ('installed_files', lifecycle.verify)): try: result[name] = action() except Exception as error: result['failures'][name] = str(error) after = manifests() result.update(saves_unchanged=after['saves'] == value['before']['saves'], mods_unchanged=after['mods'] == value['before']['mods'], settings_restored=after['settings'] == value['before']['settings']) result['ok'] = not result['failures'] and all(result[key] for key in ('saves_unchanged','mods_unchanged','settings_restored')) update(folder, phase='recovered', after=after, recovery_result=result, cleanup_required=False, audits_final=True) return result