Something went wrong. Try again.
Native control, inspection, UI automation and evidence capture for a running Witcher 3 process.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124"""Temporary evaluation with durable, hash-guarded source restoration.
Restoring source never rolls back arbitrary effects of the evaluated body.Receipts survive interruption; recovery refuses unrelated edits."""import collectionsimport hashlibimport jsonfrom pathlib import Pathimport time
from . import native
def digest(data): return hashlib.sha256(data).hexdigest()
def diagnostic_delta(baseline, current): def packets(result): path = result.get('artifacts') if not path or not Path(path).is_file(): return None return [p for p in json.loads(Path(path).read_text()).get('packets', []) if p[:2] in (['ScriptCompiler', 'warn'], ['ScriptCompiler', 'error'])] before, after = packets(baseline), packets(current) if before is None or after is None: return {'available': False, 'reason': 'compiler packets unavailable'} a = collections.Counter(json.dumps(p) for p in before) b = collections.Counter(json.dumps(p) for p in after) return {'available': True, 'added': [json.loads(p) for p in (b-a).elements()], 'removed': [json.loads(p) for p in (a-b).elements()], 'comparison': 'exact compiler packet multiset, including file/line'}
def recover(path): path = Path(path).resolve() journal = json.loads(path.read_text()) target = Path(journal['target']) receipt_path = native.STATE_ROOT/'deployment.json' receipt = json.loads(receipt_path.read_text()) meta = receipt['evaluation'] backup = path.parent/'previous.ws' previous = backup.read_bytes() if digest(previous) != journal['previous_sha256']: raise RuntimeError('Scoped evaluation backup changed; refusing restoration') if Path(meta['path']) != target or meta['sha256'] not in (journal['previous_sha256'], journal['temporary_sha256']): raise RuntimeError('Evaluation receipt changed outside scope; refusing restoration') if digest(target.read_bytes()) not in (journal['previous_sha256'], journal['temporary_sha256']): raise RuntimeError('Evaluation source changed outside scope; refusing restoration') if journal.get('status') == 'restored': if digest(target.read_bytes()) != journal['previous_sha256']: raise RuntimeError('Previously restored evaluation changed; refusing restoration') return {'ok': True, 'already_restored': True, 'receipt': str(path)} target.write_bytes(previous) meta['sha256'] = journal['previous_sha256'] native.legacy.atomic_json(receipt_path, receipt) journal.update(status='source_restored_compile_pending', source_restored=True) native.legacy.atomic_json(path, journal) try: compilation = native.recompile() except (OSError, RuntimeError, ValueError) as error: compilation = {'ok': False, 'error': str(error)} result = {'ok': compilation['ok'], 'source_restored': True, 'compile': compilation, 'receipt': str(path), 'runtime_effects_rolled_back': False} journal.update(status='restored' if result['ok'] else 'source_restored_compile_unverified', restoration=result) native.legacy.atomic_json(path, journal) if not result['ok']: try: s = native.state() s['uncertain'] = {'command': 'evaluate-scoped', 'reason': 'restoration compile unverified', 'receipt': str(path)} native.legacy.atomic_json(native.STATE, s) except (OSError, RuntimeError): pass return result
def scoped(path): body = Path(path).read_text(encoding='utf-8') if len(body.encode()) > 16384: raise ValueError('Evaluation source exceeds 16KiB') native.state() receipt = json.loads((native.STATE_ROOT/'deployment.json').read_text()) meta = receipt['evaluation'] target = Path(meta['path']) previous = target.read_bytes() if digest(previous) != meta['sha256']: raise RuntimeError('Evaluation source changed outside devkit') folder = native.legacy.ARTIFACTS/('scoped-evaluation-'+str(time.time_ns())) folder.mkdir(parents=True) (folder/'previous.ws').write_bytes(previous) temporary = ('function PDEvaluateBody() : string\n{\n'+body+'\n}\n').encode() journal = {'version': 1, 'status': 'prepared', 'target': str(target), 'previous_sha256': digest(previous), 'temporary_sha256': digest(temporary), 'probe_sha256': digest(body.encode()), 'session': native.state()['session']} recovery_path = folder/'recovery.json' native.legacy.atomic_json(recovery_path, journal) baseline = native.recompile() if not baseline['ok']: return {'ok': False, 'phase': 'baseline_compile', 'compile': baseline, 'executed': False, 'recovery_receipt': str(recovery_path)} result = None try: result = native.evaluate(path) result['diagnostic_delta'] = diagnostic_delta(baseline, result.get('compile', {})) return result except (OSError, RuntimeError, ValueError) as error: error.evidence = dict(error=str(error), recovery_receipt=str(recovery_path), runtime_effects_rolled_back=False) native.legacy.atomic_json(folder/'result.json',dict(ok=False,**error.evidence)) raise finally: try: restoration = recover(recovery_path) except (OSError, RuntimeError, ValueError) as error: error.evidence = dict(error=str(error), recovery_receipt=str(recovery_path), source_restored=False) native.legacy.atomic_json(folder/'restoration-failed.json',error.evidence) raise if result is not None: result['evaluation_ok'] = result['ok'] result['ok'] = result['ok'] and restoration['ok'] result['restoration'] = restoration result['recovery_receipt'] = str(recovery_path) native.legacy.atomic_json(folder/'result.json', result)