74 lines
6.4 KiB
Python
74 lines
6.4 KiB
Python
# -*- coding: utf-8 -*-
|
|
from pathlib import Path
|
|
import hashlib, importlib.util, json, os, shutil, subprocess, sys
|
|
sys.dont_write_bytecode = True
|
|
R = Path(__file__).resolve().parents[2]
|
|
A = R / 'artifacts/original-completion-2026-10-02'
|
|
OUT = A / 'p04-idempotency-paid-midnight-native-r01'
|
|
QA = A / 'p04-idempotency-paid-midnight-prep-r01/idempotency-paid-midnight.gd'
|
|
TEST = 'tests/s3/test_runtime_staff_assignment.gd'
|
|
def module(name, path):
|
|
spec = importlib.util.spec_from_file_location(name, path)
|
|
m = importlib.util.module_from_spec(spec); spec.loader.exec_module(m); return m
|
|
def main():
|
|
r = module('order_repro_runner', R / 'tools/run_all_tests.py')
|
|
root = r.source_refs(R); source = {n:v['sha256'] for n,v in root.items()}
|
|
source_path = A / 'p04-idempotency-paid-midnight-prep-r01/SOURCE_MAP.json'
|
|
assert source == r.read(source_path) and len(source) == 1338
|
|
protected = r.default_player(R)
|
|
docs = {n:r.sha(R / 'docs/plans/original-completion-2026-10-02' / n) for n in ['IMPLEMENTATION_PLAN.md','PROGRESS_INTEGRATION.md','work_packages.json','progress_snapshot.json']}
|
|
old = Path('/Users/korizatou114/Projects/TheTree-managed-worktrees/p18-01-model/.runtime/P18-01-CODE/build/p18-relationship-evidence')
|
|
assert r.read(old / 'runner-ended.json')['passed']
|
|
assert r.read(old / 'runner-input.json')['selected_source_map'] == source
|
|
assert shutil.disk_usage(A).free >= 1024**3 + Path('/Applications/Godot.app/Contents/MacOS/Godot').stat().st_size + 136*1024**2
|
|
outer = module('order_literal_owned', r.SUPPORT / 'owned_outer.py')
|
|
assert not outer.native.helper['live_engines']()
|
|
refs = dict(root); refs[TEST] = dict(path=str(QA), bytes=QA.stat().st_size, sha256=r.sha(QA))
|
|
expected = {n:dict(bytes=p['bytes'],sha256=p['sha256']) for n,p in refs.items()}
|
|
selected = dict(root_before=root,map={n:p['sha256'] for n,p in refs.items()},timeout=1200)
|
|
assert not OUT.exists(); OUT.mkdir(); stage=OUT/'stage'; stage.mkdir()
|
|
for leaf in ['starters','bootstrap']: (OUT/leaf).mkdir()
|
|
for n,p in r.read(r.SUPPORT/'bootstrap-scripts.json').items():
|
|
data=p['text'].encode(); assert hashlib.sha256(data).hexdigest()==p['sha256']; (OUT/'bootstrap'/n).write_bytes(data)
|
|
for n,p in refs.items():
|
|
dst=stage/n; dst.parent.mkdir(parents=True,exist_ok=True); r.copy_source(Path(p['path']),dst)
|
|
assert r.inventory(stage,stage=True)==expected
|
|
cache=old/'stage/.godot'; cache_before=r.inventory(cache)
|
|
for n in cache_before:
|
|
src=cache/n; dst=stage/'.godot'/n; dst.parent.mkdir(parents=True,exist_ok=True)
|
|
subprocess.run(['/bin/cp','-c',str(src),str(dst)],check=True); assert src.stat().st_ino != dst.stat().st_ino
|
|
assert r.inventory(stage/'.godot')==cache_before
|
|
r.write(OUT/'cache-reuse.json',dict(ended_run=str(old),ended_receipt_sha256=r.sha(old/'runner-ended.json'),whole_cache=cache_before,independent_inode=True,ordinary_import_required=True,ordinary_parse_required=True))
|
|
paths=['res://'+n for n in sorted(refs) if n.endswith('.gd')]; r.write(OUT/'parse-paths.json',paths)
|
|
markers=module('order_literal_markers',R/'tools/strict_test_markers.py')
|
|
contracts=markers.load_contracts(R/'docs/plans/original-completion-2026-10-02/execution/completion_marker_contracts.json')
|
|
contracts[TEST]=dict(test=TEST,template='Midnight boundary diagnostic: <checks> checks, <failures> failures',required=True)
|
|
policy=r.read(r.SUPPORT/'suite-policy.json'); assert TEST not in policy['suite_policies']
|
|
r.write(OUT/'contracts.json',contracts)
|
|
r.write(OUT/'runner-input.json',dict(classification='P04-01 same-minute public service request resends with full typed state equality and existing paid catalog midnight/dependency scenario; snapshots are exact typed evidence and JSON mirrors lose Variant key/value types; no formal save, Windows or whole-unit acceptance',owner_runner_pid=os.getpid(),wrapper_sha256=r.sha(__file__),QA=dict(path=str(QA),sha256=r.sha(QA),bytes=QA.stat().st_size),runner_sha256=r.sha(R/'tools/run_all_tests.py'),source_before=root,selected_source_map=selected['map'],source_map_authority=str(source_path),default_player_before=protected,doc4_before=docs,source_adoption=False,QA_overrides=[TEST],ordinary_disk_floor_bytes=r.MIN_SPACE,functional_watchdog_seconds=1200))
|
|
diagnostics=module('order_literal_diagnostics',r.SUPPORT/'diagnostic_gate.py'); owners=module('order_literal_owners',r.SUPPORT/'ordinary_proof_gate.py'); audio=module('order_literal_audio',r.SUPPORT/'audio_evidence_gate.py')
|
|
original=outer.run_phase
|
|
def run_phase(argv,folder,timeout,env,**kwargs):
|
|
env=dict(env)
|
|
if folder.name=='repro': env['THE_TREE_ORDER_EVIDENCE']=str(OUT/'order-evidence')
|
|
return original(argv,folder,timeout,env,**kwargs)
|
|
outer.run_phase=run_phase
|
|
phases=[]; previous={}; error=None; public=(stage/'project.godot').read_bytes()
|
|
try:
|
|
for phase,test in [('import',None),('parse',None),('repro',TEST)]:
|
|
row=r.execute_phase('/Applications/Godot.app/Contents/MacOS/Godot',stage,OUT,phase,test,selected,expected,previous,policy,contracts,paths,public,protected,R,outer,markers,diagnostics,owners,audio)
|
|
phases.append(row); r.write(OUT/(phase+'-actual-phase.json'),row)
|
|
print(json.dumps(dict(phase=phase,passed=row['passed'],checks=row.get('reported_checks'),error=row.get('validation_error'))),flush=True)
|
|
if not row['passed']:
|
|
error=phase+' strict phase failure'; break
|
|
except BaseException as caught: error=type(caught).__name__+': '+str(caught)
|
|
finally: cleanup,cleanup_errors=outer.finish_captured(outer.OWNED)
|
|
assert r.source_refs(R)==root and r.default_player(R)==protected
|
|
assert {n:r.sha(R/'docs/plans/original-completion-2026-10-02'/n) for n in docs}==docs
|
|
assert r.inventory(stage,stage=True)==dict(expected,**previous) and r.inventory(cache)==cache_before
|
|
passed=error is None and len(phases)==3 and all(p['passed'] for p in phases) and not cleanup_errors and all(p['engine_ended'] for p in cleanup)
|
|
r.write(OUT/'execution.json',dict(passed=passed,error=error,phases=phases,actual_captured_PIDs=[p.pid for p in outer.OWNED],cleanup=cleanup,cleanup_errors=cleanup_errors,all_source_unchanged=True,all_default_player_unchanged=True,doc4_unchanged=True,actual_native=len(outer.OWNED),whole_goal_accepted=False,unit_accepted=False))
|
|
print(json.dumps(dict(passed=passed,error=error,actual_native=len(outer.OWNED))),flush=True)
|
|
return 0 if passed else 1
|
|
if __name__=='__main__': raise SystemExit(main())
|