# coding: utf-8
"""One owned Guest batch ABI/local/affected-S3 execution; no full regression or release acceptance."""
from pathlib import Path
import hashlib,importlib.util,json,os,shutil,sys,uuid
sys.dont_write_bytecode=True
R=Path(__file__).resolve().parents[2];A=R/'artifacts/original-completion-2026-10-02';O=A/'p23-02-native-hot-loop-local-r329';D=R/'docs/plans/original-completion-2026-10-02'
NAMES=['IMPLEMENTATION_PLAN.md','PROGRESS_INTEGRATION.md','work_packages.json','progress_snapshot.json']
def mod(name,p):
 s=importlib.util.spec_from_file_location(name,p);m=importlib.util.module_from_spec(s);s.loader.exec_module(m);return m
def main():
 r=mod('guest_local_runner',R/'tools/run_all_tests.py');outer=mod('guest_local_owned',r.SUPPORT/'owned_outer.py');plan=r.read(O/'input.json');fixed=plan['pins']
 assert not (O/'execution.json').exists()
 stage=Path(plan['stage']);parent=Path(plan['parent']);public=(stage/'project.godot').read_bytes();private=Path(fixed['private_library']['path']).parent
 source=r.source_refs(R);player=r.default_player(R);docs={n:r.sha(D/n) for n in NAMES}
 assert source==plan['source_before'] and player==plan['default_before'] and docs==plan['doc4_before']
 def protect():
  assert r.source_refs(R)==source and r.default_player(R)==player and {n:r.sha(D/n) for n in NAMES}==docs
  assert r.inventory(parent,stage=True)==plan['parent_stage_inventory'] and r.inventory(stage,stage=True)==plan['expected_stage']
  assert r.inventory(r.SUPPORT)==plan['support_inventory'] and r.inventory(private)==plan['private_library_inventory']
  for k,q in fixed.items():
   p=Path(q['path']);assert p.stat().st_size==q['bytes'] and r.sha(p)==q['sha256'],k
 def growth():return sum(p.stat().st_size for p in O.rglob('*') if p.is_file() and p.relative_to(O).parts[0]!='stage')
 protect();host=r.read(fixed['host_review']['path']);assert host['host_build_accepted'] and not host['blockers']
 prepath=A/'p23-02-native-guest-bulk-local-preexecution-independent-review-r315.json';pre=r.read(prepath);assert pre['execution_allowed'] and not pre['blockers']
 for group in ['evidence_sha256','canonical_doc4_context_pins']:
  for q in pre[group].values():
   p=Path(q['path']);assert p.stat().st_size==q['bytes'] and r.sha(p)==q['sha256']
 root=r.read(O/'preexecution-root-verification.json');assert root['passed'] and root['review_sha256']==r.sha(prepath)
 phases=[];samples=[];error=None;cleanup=[];cleanup_errors=[];report=None
 try:
  protect();assert not outer.native.helper['live_engines']() and shutil.disk_usage(A).free>=plan['capacity_floor_bytes'] and growth()<=plan['maximum_measurement_growth_bytes']
  output=O/'local-measurement';output.mkdir();ident=uuid.uuid4().hex;startup='TheTree-PCK-tests-'+ident+'-guest-startup';profile='TheTree-PCK-tests-'+ident+'-guest-runtime'
  assert not r.profile_path(startup).exists() and not r.profile_path(profile).exists()
  env=r.fresh_environment(os.environ);env.update(THE_TREE_PACK_PROFILE_NAME=profile,THE_TREE_PACK_PROFILE_EXPECTED=str(r.profile_path(profile)),THE_TREE_PACK_PROFILE_REPORT=str(O/'local-profile.json'),THE_TREE_RUNTIME_MONITOR_OUTPUT=str(output))
  (stage/'project.godot').write_bytes(r.private_project(public,startup));phase=O/'local'
  argv=[fixed['engine']['path'],'--headless','--path',str(stage),'--script',fixed['starter']['path'],'--verbose','--log-file',str(phase/'engine-native.log')]
  result,capture=outer.run_phase(argv,phase,90,env,now_fn=r.now,dump_fn=r.write,sha_fn=r.sha,live_engines_fn=outer.native.helper['live_engines'])
  phases.append(dict(name='local',result=result,capture=capture));r.write(O/'local-phase.json',phases[0])
  report=r.read(output/'report.json') if (output/'report.json').exists() else None
  if report:samples.append(dict(name='local',report=report))
  assert result is not None and result['exit_code']==0 and not result['error_lines'] and not result['timed_out'] and capture['passed']
  logs=''.join((phase/n).read_text() for n in ['stdout.log','stderr.log','engine-native.log'])
  assert 'Orphan StringName:' not in logs and 'unclaimed string names' not in logs
  assert report and report['passed'] and report['failures']==0 and report['pid']==result['pid']
  count=plan['expected_case_counts'];assert report['pure_case_count']==count['pure'] and report['decision_case_count']==count['decision'] and report['custom_fallback_case_count']==count['custom_fallback'] and report['original_s3_invocations']==count['original_s3_invocations']
  stdout=(phase/'stdout.log').read_text();assert stdout.count('Guest bulk local: %d checks, 0 failures'%report['checks'])==1 and stdout.count('NATIVE_ADAPTER_LIBRARY_REGISTERED')==1
  fixture=output/'fixtures.varbytes';assert fixture.stat().st_size==report['fixture_bytes'] and r.sha(fixture)==report['fixture_sha256']
  protocol=r.read(output/'protocol-report.json');assert protocol['passed'] and not protocol['failures'] and len(protocol['cases'])==50 and len(protocol['cached_cases'])==50 and protocol['checks']==report['protocol_checks']
  assert fixture.read_bytes()==Path(plan['original_Guest_fixture']['path']).read_bytes(), 'Unchanged full Guest53 original fixture bytes'
  r.profile_gate(r.profile_path(profile),O/'local-profile.json',profile)
  assert growth()<=plan['maximum_measurement_growth_bytes']
  (stage/'project.godot').write_bytes(public);protect()
 except BaseException as caught:error=type(caught).__name__+': '+str(caught)
 finally:
  cleanup,cleanup_errors=outer.finish_captured(outer.OWNED)
  if all(q['engine_ended'] is True for q in cleanup):(stage/'project.godot').write_bytes(public)
  else:cleanup_errors.append('Restoration refused while actual owned child remains live')
 guard_errors=[]
 try:protect()
 except BaseException as caught:guard_errors.append(type(caught).__name__+': '+str(caught))
 passed=error is None and not guard_errors and not cleanup_errors and len(phases)==1 and len(samples)==1
 r.write(O/'execution.json',dict(passed=passed,error=error,guard_errors=guard_errors,phases=phases,samples=samples,cleanup=cleanup,cleanup_errors=cleanup_errors,input_source_default_doc4_parent_support_private_preserved=not guard_errors,compiler_processes=0,full_regression_runs=0,candidate_selected=False,candidate_adopted=False,unit_accepted=False,performance_accepted=False,normal_OS_accepted=False,Windows_accepted=False,limitations=['Finite Guest53 and18 affected original S3 methods plus frozen50 protocols/shrink/ownership/reentrant collision on same new bridge. Literal original Guest source oracle removes class_name line only. No Runtime cold646→647 whole state, ordinary OS/reentry domain, arbitrary conversion or memory failure, final full regression, performance budget/P95/D10, Windows or adoption/unit acceptance.']))
 print(json.dumps(dict(passed=passed,error=error,guard_errors=guard_errors)),flush=True);return 0 if passed else 1
if __name__=='__main__':raise SystemExit(main())