"""Snapshot a saved source and compare its complete Project with a predeclared checkpoint.""" from pathlib import Path import argparse, datetime, json, subprocess, sys sys.dont_write_bytecode=True from stage import HERE, child_env, copy_new, read, record, verify, write def main(): parser=argparse.ArgumentParser(description=__doc__) parser.add_argument('--expected',required=True,choices=['B','N1','N2','N3']) parser.add_argument('--output-prefix',required=True,type=Path) parser.add_argument('--actual',type=Path,default=HERE/'working/attachments.arcadia') args=parser.parse_args(); prefix=args.output_prefix.resolve() outputs={suffix:Path(str(prefix)+suffix) for suffix in ('.arcadia','.project.json','.stdout.log','.stderr.log','.receipt.json')} if any(path.exists() for path in outputs.values()): raise RuntimeError('Fresh output prefix required') stage=read(HERE/'staging-record.json'); verify(stage['manifest']); manifest=read(HERE/'input-manifest.json') if stage['source_commit']!=manifest['source_commit']: raise RuntimeError('Stage pin mismatch') for item in manifest['immutable_files']: verify(item) expected=read(HERE/'expected/catalog.json')['checkpoints'][args.expected]; verify(expected) prefix.parent.mkdir(parents=True,exist_ok=True) actual=record(args.actual); copy_new(args.actual,outputs['.arcadia']) started=datetime.datetime.now(datetime.timezone.utc).isoformat(); failure=None; exit_code=None; exact=False; unchanged=False try: with outputs['.stdout.log'].open('xb') as stdout, outputs['.stderr.log'].open('xb') as stderr: proc=subprocess.run([manifest['evaluator']['path'],str(outputs['.arcadia']),str(outputs['.project.json'])],cwd=HERE/'unrelated-cwd',env=child_env(),stdout=stdout,stderr=stderr) exit_code=proc.returncode if exit_code: raise RuntimeError('Snapshot evaluator failed') if outputs['.stderr.log'].stat().st_size: raise RuntimeError('Snapshot stderr is not empty') # Compare complete JSON including exact fields/types; formatting/key order alone are ignored. canonical=lambda value:json.dumps(value,sort_keys=True,ensure_ascii=False,separators=(',',':'),allow_nan=False) exact=canonical(read(outputs['.project.json']))==canonical(read(expected['path'])) if not exact: failure='Complete Project differs from predeclared '+args.expected verify(actual) for item in manifest['immutable_files']: verify(item) verify(stage['manifest']); unchanged=True except Exception as error: failure=str(error) artifacts=[record(path) for suffix,path in outputs.items() if suffix!='.receipt.json' and path.exists()] write(outputs['.receipt.json'],{'source_commit':manifest['source_commit'],'evaluator_source_commit':manifest['evaluator_source_commit'],'started_utc':started,'finished_utc':datetime.datetime.now(datetime.timezone.utc).isoformat(),'expected':args.expected,'expected_file':expected,'input':actual,'evaluator':manifest['evaluator'],'manifest':stage['manifest'],'exit_code':exit_code,'failure':failure,'inputs_unchanged':unchanged,'full_project_exact':exact,'artifacts':artifacts,'scope':'Complete saved Project only; no native action/editor/history/camera claim'}) if failure or not exact or not unchanged: raise SystemExit('Comparison failed; preserve outputs: '+str(failure)) print(args.expected+': complete saved Project exact; inputs unchanged') if __name__=='__main__':main()