61 lines
4.5 KiB
Python
61 lines
4.5 KiB
Python
"""Read-only bundle lifecycle bridge; prepare before use.
|
|
|
|
Parent invokes after a native normal close, immediately before a fresh launch,
|
|
and after the final reopen/save close. This recorder compares every current saved
|
|
.arcadia file to original_after from the explicitly selected accepted v2 bundle
|
|
checkpoint, and binds supplied close/launch JSON records by exact bytes/hash.
|
|
It does not launch, close, compile, evaluate or save the model. A prelaunch record
|
|
attests to the observed instant, not future process behavior; bind the actual
|
|
launch separately at close. Existing launch/capture entry hashes stay unchanged.
|
|
Use a fresh --name and --event-record for each observation; failures are retained.
|
|
Example: python supplement-bundle-lifecycle.py --name run01-close --phase close
|
|
--checkpoint supplement-source-bundle-v2/checkpoints/03-redo/bundle-receipt.json
|
|
--event-record runs/run-01/normal-close.json
|
|
Relative paths resolve against this native root, independent of shell cwd.
|
|
"""
|
|
from pathlib import Path
|
|
import argparse,datetime,hashlib,importlib.util,json,re,sys
|
|
sys.dont_write_bytecode=True
|
|
ROOT=Path(__file__).resolve().parent
|
|
WRAPPER=ROOT/'supplement-source-bundle-v2/capture_bundle.py'
|
|
def digest(p):
|
|
b=p.read_bytes();return {'path':str(p),'bytes':len(b),'sha256':hashlib.sha256(b).hexdigest()}
|
|
def local(p):
|
|
p=p if p.is_absolute() else ROOT/p
|
|
p=p.absolute()
|
|
if not p.is_relative_to(ROOT):raise RuntimeError('Path outside native root')
|
|
q=p
|
|
while True:
|
|
if q.exists() and (q.is_symlink() or q.is_junction() or getattr(q.lstat(),'st_file_attributes',0)&1024):raise RuntimeError('Linked path component: '+str(q))
|
|
if q==ROOT:break
|
|
q=q.parent
|
|
if not p.resolve().is_relative_to(ROOT.resolve()):raise RuntimeError('Resolved path outside native root')
|
|
return p
|
|
def main():
|
|
ap=argparse.ArgumentParser();ap.add_argument('--name',required=True);ap.add_argument('--phase',choices=['close','prelaunch'],required=True);ap.add_argument('--checkpoint',required=True,type=Path);ap.add_argument('--event-record',required=True,action='append',type=Path);a=ap.parse_args()
|
|
if not re.fullmatch(r'[a-z0-9-]+',a.name):raise RuntimeError('Simple fresh observation name required')
|
|
checkpoint=local(a.checkpoint);events=[local(p) for p in a.event_record];out=local(Path('bundle-lifecycle')/(a.name+'.json'))
|
|
if out.exists():raise RuntimeError('Observation already exists')
|
|
wrapper=local(WRAPPER);work=local(Path('working'))
|
|
refs=[digest(checkpoint),*[digest(p) for p in events],digest(wrapper)]
|
|
expected=json.loads(checkpoint.read_text(encoding='utf-8-sig'))
|
|
if expected.get('accepted') is not True or 'original_after' not in expected:raise RuntimeError('Accepted complete-bundle checkpoint required')
|
|
if any(refs[-1][k]!=expected['wrapper'][k] for k in ('bytes','sha256')):raise RuntimeError('Inventory wrapper differs from accepted checkpoint')
|
|
if (ROOT/'supplement-source-bundle-v2/UNACCEPTED.json').exists():raise RuntimeError('Bundle trial quarantined')
|
|
for p in events:json.loads(p.read_text(encoding='utf-8-sig'))
|
|
spec=importlib.util.spec_from_file_location('bound_bundle_inventory',wrapper);module=importlib.util.module_from_spec(spec);spec.loader.exec_module(module)
|
|
observation={'started_utc':datetime.datetime.now(datetime.timezone.utc).isoformat(),'phase':a.phase,'name':a.name,'recorder':digest(Path(__file__).resolve()),'checkpoint':refs[0],'event_records':refs[1:-1],'inventory_implementation':refs[-1],'expected_inventory':expected['original_after'],'passed':False,'failure':None,'scope':'Saved bundle exact at observation time; process behavior only as supplied separately recorded event evidence'}
|
|
try:
|
|
actual=module.inventory(work);observation['actual_inventory']=actual
|
|
if actual!=expected['original_after']:raise RuntimeError('Complete saved bundle differs from accepted checkpoint')
|
|
if any(digest(Path(r['path']))!=r for r in refs):raise RuntimeError('Bound evidence changed during observation')
|
|
if module.inventory(work)!=actual:raise RuntimeError('Bundle changed during observation')
|
|
observation['passed']=True
|
|
except Exception as e:observation['failure']=str(e)
|
|
observation['finished_utc']=datetime.datetime.now(datetime.timezone.utc).isoformat()
|
|
out.parent.mkdir(exist_ok=True);local(out)
|
|
with out.open('x',encoding='utf-8') as f:json.dump(observation,f,indent=2);f.write('\n')
|
|
if not observation['passed']:raise SystemExit('Bundle lifecycle observation failed; retained '+str(out))
|
|
print('Complete saved bundle exact: '+str(out))
|
|
if __name__=='__main__':main()
|