Something went wrong. Try again.
Native control, inspection, UI automation and evidence capture for a running Witcher 3 process.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343#!/usr/bin/env python3"""Direct REDengine network driver. No debug-console UI or settings-file transport."""import hashlibimport jsonimport secretsimport shutilimport sysimport timefrom pathlib import Path
from . import support as legacyfrom . import wirefrom .arguments import BUTTONS, validate
ROOT = Path(__file__).resolve().parent.parentSTATE_ROOT = legacy.STATE_ROOTSTATE = STATE_ROOT / 'native-session.json'READ_ONLY = {'profile','ability','lab_state','event_page','actor','snapshot', 'ui', 'ui_items', 'events', 'saves','ui_scan','ui_rows','ui_node'}READ_ONLY |= {'combat_state', 'encounter_scan', 'encounter_page', 'encounter', 'sample_page'}READ_ONLY |= {'water_depth', 'water_depth_actor', 'teleport_observer_status'}READ_ONLY |= {'request_algorithm_status', 'request_algorithm_page'}READ_ONLY |= {'crossbow_state'}READ_ONLY |= {'overlay_status'}COMMANDS = legacy.COMMANDS | {'lab_config','profile','ability','lab_state','event_page','actor','ui', 'ui_items', 'ui_click', 'menu', 'advance', 'input', 'skip_intro', 'pad', 'release', 'evaluate','ui_scan','ui_rows','ui_node','ui_dispatch'}COMMANDS |= {'combat_state', 'encounter_scan', 'encounter_page', 'encounter', 'sample_arm', 'sample_stop', 'sample_page'}COMMANDS |= {'water_depth', 'water_depth_actor', 'teleport_observer_arm', 'teleport_observer_stop', 'teleport_observer_status'}COMMANDS |= {'request_algorithm_arm', 'request_algorithm_stop', 'request_algorithm_status', 'request_algorithm_page'}COMMANDS |= {'crossbow_state', 'quickslot', 'control', 'heal'}COMMANDS |= {'overlay_status','overlay_config','overlay_set','overlay_remove','overlay_list','overlay_list_clear'}
def parse_reply(output, token, request): prefix = f'PDPART:{token}:{request}:' parts = {} count = None for line in output.splitlines(): if line.startswith(prefix): index, value = line[len(prefix):].split(':',1) if int(index) in parts:raise ValueError('Duplicate response part') parts[int(index)] = value elif line.startswith(f'PDEND:{token}:{request}:'): if count is not None:raise ValueError('Duplicate response commit') count=int(line.rsplit(':',1)[1]) if count is None or not 1 <= count <= 128 or set(parts) != set(range(count)): raise RuntimeError('Missing or incomplete bridge acknowledgement; no mutation replayed. '+output[:150]) r=json.loads(''.join(parts[i] for i in range(count))) if r.get('session') != token or r.get('request') != request: raise RuntimeError('Response identity mismatch') return r
def state(): if not STATE.exists(): from .diagnostics import DiagnosticError raise DiagnosticError('E2004', 'No recorded session. Launch the verified installation, wait for its real window, then attach.') s = json.loads(STATE.read_text()) if legacy.process_identity(s['process']['pid']) != s['process']: raise RuntimeError('Recorded game ended; attach explicitly to the new process') return s
def record(s, kind, data): legacy.record(s, kind, data) from . import performance if performance.ACTIVE.exists(): try: # Request/response brackets are host markers, never frame samples. if kind == 'request': detail = dict(session=s['session'], request=data.get('request'), command=data.get('command'), generation=s.get('generation')) else: reply = data.get('reply', {}) payload = reply.get('data') if isinstance(reply.get('data'), dict) else {} world = payload.get('world') or (payload if 'player' in payload else {}) detail = dict(session=s['session'], request=reply.get('request'), generation=reply.get('generation'), engine_time=reply.get('engine_time'), latency_ms=data.get('latency_ms'), ok=reply.get('ok'), ticks=payload.get('ticks'), ready=payload.get('ready'), loading_video=payload.get('loading_video'), ui=payload.get('ui'), paused=world.get('paused'), devkit_paused=world.get('devkit_paused'), player_alive=world.get('player',{}).get('alive')) performance.marker('native_'+kind, detail) except (OSError, ValueError, RuntimeError) as error: legacy.record(s, 'performance_marker_error', {'error': str(error)}) try: active = json.loads(performance.ACTIVE.read_text()) # One bounded receipt; repeated marker failures must not grow a log. performance.write(Path(active['folder'])/'marker-errors.jsonl', {'error': str(error)[:512], 'at': time.monotonic_ns()}) except (OSError, ValueError, KeyError): pass # The session trace above retains the failure even if capture storage failed.
def call(command, args='', timeout=5, connection=None): from .debugger import reject_legacy reject_legacy() validate(command, args) if command in ('teleport_observer_arm', 'teleport_observer_stop', 'request_algorithm_arm', 'request_algorithm_stop', 'quickslot', 'heal', 'overlay_config','overlay_set','overlay_remove','overlay_list','overlay_list_clear'): from .debugger import active_owner, identity owner = active_owner() if not owner or owner['broker'] != identity(): operation = ('overlay' if command.startswith('overlay_') else 'heal' if command == 'heal' else 'quickslot' if command == 'quickslot' else 'request-algorithm' if command.startswith('request_algorithm_') else 'observe-teleport') code = ('overlay_owner_required' if command.startswith('overlay_') else 'heal_owner_required' if command == 'heal' else 'observer_owner_required') raise RuntimeError(f'{code}: use debug {operation} through the session owner') s = state() if s.get('uncertain') and command not in READ_ONLY: raise RuntimeError('A mutation has an uncertain outcome; inspect snapshot/events and recover explicitly') binding = None if command == 'input' and args.split('|')[4] == 'Crossbow': from .crossbow_input import verify_bindings binding = verify_bindings(s.get('input_settings_sha256')) s['request'] += 1 if command not in READ_ONLY: s['uncertain'] = {'request': s['request'], 'command': command, 'args': args} legacy.atomic_json(STATE, s) expression = f"pathdevkit('{s['session']}',{s['request']},'{command}','{args}')" record(s, 'request', {'request': s['request'], 'command': command, 'args': args, 'transport': 'native_tcp', **({'input_binding': binding} if binding else {})}) started = time.monotonic() own = connection is None c = connection or wire.Connection(timeout) try: packet, packets = c.execute(expression, markers=[f"PDPART:{s['session']}:{s['request']}:0:"]) output = packet[2] commits = [line for line in output.splitlines() if line.startswith(f"PDEND:{s['session']}:{s['request']}:")] if len(commits) == 1: count = int(commits[0].rsplit(':', 1)[1]) if not 1 <= count <= 128: raise ValueError('Response part count exceeds bound') for index in range(1, count): c.timeout = timeout - (time.monotonic()-started) if c.timeout <= 0: raise TimeoutError('Complete response deadline elapsed; cached response not replayed') part, _ = c.execute(f"pathdevkitpart('{s['session']}',{s['request']},{index})", markers=[f"PDPART:{s['session']}:{s['request']}:{index}:", f"PDERR:{s['session']}:{s['request']}:{index}:"]) output += part[2] reply = parse_reply(output, s['session'], s['request']) if reply.get('build') != s['build']: raise RuntimeError('Running build differs from session build; no mutation replayed') if command not in READ_ONLY: s['uncertain'] = None s['generation'] = reply.get('generation') s['last_response'] = reply s['last_response_parts'] = count s['latency_ms'] = (time.monotonic()-started)*1000 legacy.atomic_json(STATE, s) record(s, 'response', {'reply': reply, 'latency_ms': s['latency_ms']}) return reply finally: if own: c.close()
def attach(token=None): from . import lifecycle lifecycle.verify() identity = legacy.process() # The network socket opens before script RTTI is ready. Executing while the # startup splash is present stalled an otherwise healthy cold launch. legacy.window(identity) if STATE.exists(): old = json.loads(STATE.read_text()) if legacy.process_identity(old['process']['pid']) == old['process']: raise RuntimeError('Already attached to this live process') receipt = json.loads((STATE_ROOT/'deployment.json').read_text()) if legacy.manage.digest(Path(receipt['mod_path'])) != receipt['mod_sha256']: raise RuntimeError('Installed source differs from deployment receipt') s = {'session': token or secrets.token_hex(12), 'request': 1 if token else 0, 'process': identity, 'build': receipt['build'], 'created': legacy.now(), 'uncertain': None} from .crossbow_input import capture_bindings_hash s['input_settings_sha256'] = capture_bindings_hash() folder = legacy.ARTIFACTS/s['session'];folder.mkdir(parents=True, exist_ok=True) legacy.atomic_json(folder/'saves-before.json', legacy.manage.files(legacy.manage.DOCS/'gamesaves')) for name in ('user.settings', 'dx12user.settings'): shutil.copy2(legacy.manage.DOCS/name, folder/('before-'+name)) legacy.atomic_json(STATE,s) return call('hello')
def wait_ready(timeout=180, generation=None): end = time.monotonic()+timeout last_error = None while time.monotonic() < end: try: r=call('snapshot', timeout=min(5,max(.1,end-time.monotonic()))) if r.get('ok') and r['data'].get('ready') and (generation is None or r['generation'] != generation): return r except (TimeoutError, ConnectionError, OSError) as e: last_error=str(e) time.sleep(.1) raise TimeoutError('Gameplay readiness not observed; no mutation retried. '+str(last_error))
def load(filename): before = call('snapshot')['generation'] reply = call('load',filename) if not reply.get('ok'): return reply result = wait_ready(generation=before) current = state(); current['loaded_fixture'] = filename legacy.atomic_json(STATE, current) return result
def bounded_action(command, args, timeout=15): validate(command, args) before = call('snapshot') if not before['data']['ready']: raise RuntimeError('Gameplay is not ready') if not before['data']['world']['devkit_paused']: raise RuntimeError('Pause the game before a bounded action') ticks = int(args.split('|')[-1]) target = before['data']['ticks'] + ticks reply = call(command,args) if not reply.get('ok'):return reply end = time.monotonic()+timeout while time.monotonic() < end: after = call('snapshot',timeout=min(5,end-time.monotonic())) if after['generation'] != before['generation']: raise RuntimeError('World generation changed during action') data = after['data'] if data['world'].get('player', {}).get('alive') is False: # A death screen can pause the engine before the requested budget. # Cancel the remaining lease; never restart or replay the input. release = call('release') pause = call('pause') if not release.get('ok') or not pause.get('ok'): raise RuntimeError('Player died but input release/pause was not acknowledged') after = call('snapshot') if after['generation'] != before['generation'] or not after['data']['world']['devkit_paused'] or after['data'].get('input_ticks',0): raise RuntimeError('Death termination cleanup could not be verified') observed = after['data']['ticks'] - before['data']['ticks'] return dict(ok=True,before=before,after=after,ticks_observed=observed,ticks_requested=ticks, completed_requested_ticks=observed==ticks,terminal='player_death', input_delivered=command=='input',effect_verified=False,render_frame_determinism=False) if data['remaining_ticks'] == 0 and data['world']['devkit_paused']: if data['ticks'] != target or data.get('input_ticks',0) != 0: raise RuntimeError('Action completed with unexpected tick count') return {'ok':True,'before':before,'after':after,'ticks_observed':ticks, 'input_delivered':command=='input','effect_verified':False, 'render_frame_determinism':False} time.sleep(.03) raise TimeoutError('Bounded action completion not observed; no mutation retried')
def stop(): s=state();r=call('exit') if not r.get('ok'):return r end=time.monotonic()+15 while legacy.process_identity(s['process']['pid']) == s['process']: if time.monotonic()>end:raise TimeoutError('Exit acknowledged but process still running') time.sleep(.1) folder=legacy.ARTIFACTS/s['session'] result={'ok':True,'state':'stopped'} from . import performance from . import game_settings from .diagnostics import DiagnosticError # One cleanup failure must not prevent the other independent restorations. errors={} def cleanup(name, action): try: result[name]=action() if isinstance(result[name],dict) and result[name].get('ok') is False: errors[name]='Cleanup returned ok=false; inspect the retained result.' except Exception as error:errors[name]=str(error) cleanup('save_files_unchanged',lambda: json.loads((folder/'saves-before.json').read_text())==legacy.manage.files(legacy.manage.DOCS/'gamesaves')) def finish_capture(): if performance.ACTIVE.exists() and not performance.read(performance.ACTIVE).get('finished'): return performance.stop('game_exit') return dict(ok=True,reason='no_active_capture') cleanup('performance_capture',finish_capture) cleanup('performance_configuration',performance.restore) cleanup('game_settings',game_settings.reset) if errors: result.update(ok=False,cleanup_errors=errors) legacy.atomic_json(folder/'native-stop.json',result) if errors: raise DiagnosticError('E9001','Game stopped, but cleanup is incomplete. Inspect cleanup_errors; ' 'use reset-game-settings or perf-restore after resolving the reported conflict.',result) return result
def capture(): s=state();r=call('snapshot');stamp=str(time.time_ns()) folder=legacy.ARTIFACTS/s['session'];path=folder/(stamp+'.png') legacy.run('spectacle','-b','-n','-f','-o',str(path)) legacy.atomic_json(folder/(stamp+'.json'),r) return {'ok':True,'path':str(path),'snapshot':r,'capture_scope':'desktop','same_render_frame':False}
def recompile(): from . import performance performance.marker('compile_begin') legacy.window(legacy.process()) with wire.Connection(60) as c: result=c.recompile() performance.marker('compile_end', {'ok': result['ok']}) path=legacy.ARTIFACTS/('native-compile-'+str(time.time_ns())+'.json') legacy.atomic_json(path,result) return {'ok':result['ok'],'artifacts':str(path),'errors':[p for p in result['packets'] if p[:2]==['ScriptCompiler','error']], 'warnings':sum(p[:2]==['ScriptCompiler','warn'] for p in result['packets'])}
def evaluate(path): """Compile a trusted WitcherScript function body, then invoke it once.
The body must return a JSON string. This is a powerful escape hatch, not a sandbox; keep save handling and irreversible operations out of experiments. """ body=Path(path).read_text() if len(body.encode())>16384:raise ValueError('Evaluation source exceeds 16KiB') state() # Must be attached before changing script source. receipt=json.loads((STATE_ROOT/'deployment.json').read_text()) meta=receipt['evaluation'];target=Path(meta['path']) if legacy.manage.digest(target)!=meta['sha256']:raise RuntimeError('Evaluation file changed outside devkit') previous=target.read_bytes() folder=legacy.ARTIFACTS/('evaluation-'+str(time.time_ns()));folder.mkdir() (folder/'previous.ws').write_bytes(previous) source='function PDEvaluateBody() : string\n{\n'+body+'\n}\n' (folder/'source.ws').write_text(source);target.write_text(source) try: compile_result=recompile() except (OSError,RuntimeError,ValueError) as error: compile_result={'ok':False,'phase':'compile_outcome_unknown','error':str(error)} if not compile_result['ok']: target.write_bytes(previous) try:rollback=recompile() except (OSError,RuntimeError,ValueError) as error:rollback={'ok':False,'error':str(error)} result={'ok':False,'phase':'compile','compile':compile_result,'rollback':rollback,'executed':False} if not rollback['ok']: s=state();s['uncertain']={'command':'evaluate','reason':'rollback compilation unverified','artifacts':str(folder)} legacy.atomic_json(STATE,s) else: meta['sha256']=legacy.manage.digest(target) legacy.atomic_json(STATE_ROOT/'deployment.json',receipt) result={'ok':True,'phase':'executed','compile':compile_result,'revision':meta['sha256'],'result':call('evaluate')} result['ok']=result['result']['ok'] result['artifacts']=str(folder);legacy.atomic_json(folder/'result.json',result) return result
def main(argv=None): from .cli_app import main as cli_main return cli_main(argv)
if __name__=='__main__':sys.exit(main())