#!/usr/bin/env python3 """Deterministic SHA-256-addressed patch object store, retention graph, and materializer. Local byte-integrity evidence only. This tool does not establish signing, authorship, timestamp authority, semantic truth, public witness consensus, production disaster recovery, live deployment, domain control, or real-world authorization. """ from __future__ import annotations from pathlib import Path, PurePosixPath from typing import Any import argparse, hashlib, json, shutil, tempfile ROOT = Path(__file__).resolve().parents[1] STORE_SCHEMA = 'memory-patch-object-catalog/1' REF_SCHEMA = 'memory-patch-object-references/1' RETENTION_SCHEMA = 'memory-patch-object-retention/1' MATERIAL_SCHEMA = 'memory-patch-object-materialization-results/1' GC_SCHEMA = 'memory-patch-object-gc-plan/1' CHECKPOINT_SCHEMA = 'memory-patch-object-checkpoint/1' PROTECTED = {'.uai/totem.uai', '.uai/taboo.uai', '.uai/talisman.uai'} BUNDLES = [ ('assets/data/patch-bundles/memory-patch-bundle-v16.4-to-v16.5.json', 'assets/patches/memory-v16.4-to-v16.5/payload'), ('assets/data/patch-bundles/memory-patch-bundle-v16.5-to-v16.6.json', 'assets/patches/memory-v16.5-to-v16.6/payload'), ('assets/data/patch-bundles/memory-patch-bundle-v16.4-to-v16.6-composed.json', 'assets/patches/memory-v16.4-to-v16.6-composed/payload'), ] SNAPSHOTS = [ 'assets/data/snapshots/memory-state-snapshot-v2026.08.16.4.json', 'assets/data/snapshots/memory-state-snapshot-v2026.08.16.5.json', 'assets/data/snapshots/memory-state-snapshot-v2026.08.16.6.json', ] LINEAGE = 'assets/data/memory-patch-lineage.json' STORE_DIR = 'assets/patch-object-store/sha256' class ObjectStoreError(RuntimeError): pass def canonical_bytes(obj: Any) -> bytes: return json.dumps(obj, ensure_ascii=False, sort_keys=True, separators=(',', ':')).encode('utf-8') def sha_bytes(data: bytes) -> str: return hashlib.sha256(data).hexdigest() def sha_file(path: Path) -> str: h=hashlib.sha256() with path.open('rb') as f: for chunk in iter(lambda:f.read(1024*1024), b''): h.update(chunk) return h.hexdigest() def load(path: Path) -> dict[str, Any]: return json.loads(path.read_text(encoding='utf-8')) def safe_rel(value: str) -> str: if not isinstance(value,str) or not value or '\\' in value or '\x00' in value: raise ObjectStoreError('unsafe_path') p=PurePosixPath(value) if p.is_absolute() or any(x in ('','.','..') for x in p.parts) or p.as_posix()!=value: raise ObjectStoreError('unsafe_path') return value def object_rel(digest: str) -> str: if len(digest)!=64 or any(c not in '0123456789abcdef' for c in digest): raise ObjectStoreError('bad_digest') return f'{STORE_DIR}/{digest[:2]}/{digest}' def write_object(root: Path, digest: str, data: bytes) -> Path: if sha_bytes(data)!=digest: raise ObjectStoreError('digest_mismatch') rel=object_rel(digest); p=root/rel; p.parent.mkdir(parents=True,exist_ok=True) if p.exists(): existing=p.read_bytes() if sha_bytes(existing)!=digest or existing!=data: raise ObjectStoreError('digest_collision_or_corrupt_existing_object') else: p.write_bytes(data) return p def verify_object(root: Path, meta: dict[str,Any]) -> None: digest=meta['sha256']; expected=object_rel(digest) if meta.get('storage_path')!=expected: raise ObjectStoreError('wrong_object_path') p=root/expected if not p.is_file(): raise ObjectStoreError('missing_object:'+digest) if p.stat().st_size!=meta.get('size_bytes'): raise ObjectStoreError('truncated_object:'+digest) if sha_file(p)!=digest: raise ObjectStoreError('object_digest_mismatch:'+digest) def bundle_payload_source(root: Path, payload_root: str, action: dict[str,Any]) -> Path: pp=action.get('payload_path') if not isinstance(pp,str) or not pp.startswith('payload/'): raise ObjectStoreError('bundle_payload_path') rel=safe_rel(pp[len('payload/'):]) if rel!=safe_rel(action['path']): raise ObjectStoreError('bundle_path_substitution') return root/payload_root/rel def validate_bundle_payloads(root: Path, bundle_path: str, payload_root: str) -> dict[str,Any]: bundle=load(root/bundle_path); actions=bundle.get('actions') or [] if actions!=sorted(actions,key=lambda a:a['path']): raise ObjectStoreError('bundle_action_order') seen=set(); refs=[] for a in actions: rel=safe_rel(a['path']) if rel in seen: raise ObjectStoreError('duplicate_bundle_path') seen.add(rel) if rel in PROTECTED: raise ObjectStoreError('protected_anchor_payload_conflict:'+rel) if a.get('classification')=='removed': if a.get('payload_path') is not None: raise ObjectStoreError('removed_payload_forbidden') continue src=bundle_payload_source(root,payload_root,a) if not src.is_file(): raise ObjectStoreError('missing_preserved_payload:'+rel) data=src.read_bytes(); dg=sha_bytes(data) if dg!=a.get('payload_sha256') or dg!=a.get('after_sha256'): raise ObjectStoreError('preserved_payload_digest:'+rel) if len(data)!=a.get('after_size_bytes'): raise ObjectStoreError('preserved_payload_size:'+rel) refs.append((a,src,data,dg)) # Reuse the predecessor transaction verifier so manifest mutation/digest changes fail closed. import sys sys.path.insert(0,str(root/'tools')) import memory_repair_transaction as tx tx.verify_bundle(bundle,root/payload_root) return {'bundle':bundle,'refs':refs} def build_catalog(root: Path, write_objects: bool=True) -> tuple[dict[str,Any],dict[str,Any]]: objects: dict[str,dict[str,Any]]={}; bundles=[]; total_reference_bytes=0 for bundle_path,payload_root in BUNDLES: checked=validate_bundle_payloads(root,bundle_path,payload_root); b=checked['bundle']; bref=[] for a,src,data,dg in checked['refs']: if write_objects: write_object(root,dg,data) ref={'bundle_id':b['bundle_id'],'bundle_manifest':bundle_path,'path':a['path'],'payload_path':a['payload_path'],'target_sha256':a['after_sha256'],'object_sha256':dg,'protected_anchor':False} bref.append(ref); total_reference_bytes+=len(data) obj=objects.setdefault(dg,{'sha256':dg,'size_bytes':len(data),'storage_path':object_rel(dg),'references':[]}) if obj['size_bytes']!=len(data): raise ObjectStoreError('digest_collision_size') obj['references'].append(ref) bundles.append({'bundle_id':b['bundle_id'],'bundle_digest_sha256':b['bundle_digest_sha256'],'bundle_manifest':bundle_path,'base_snapshot_id':b['base_snapshot_id'],'target_snapshot_id':b['target_snapshot_id'],'reference_count':len(bref),'references':bref}) obj_list=[] for dg,obj in sorted(objects.items()): obj['references']=sorted(obj['references'],key=lambda r:(r['bundle_id'],r['path'])) obj['reference_count']=len(obj['references']); obj_list.append(obj) bundle_refs=sorted(bundles,key=lambda b:b['bundle_id']) cat_core={'schema':STORE_SCHEMA,'digest_algorithm':'sha256','object_addressing_rule':'assets/patch-object-store/sha256//; object identity is payload bytes only, never source path','objects':obj_list} cat_digest=sha_bytes(canonical_bytes(cat_core)) unique_bytes=sum(o['size_bytes'] for o in obj_list) catalog={**cat_core,'catalog_digest_sha256':cat_digest,'catalog_id':'object-catalog-'+cat_digest[:20],'summary':{'bundle_count':len(bundle_refs),'payload_reference_count':sum(b['reference_count'] for b in bundle_refs),'unique_object_count':len(obj_list),'deduplicated_reference_count':sum(b['reference_count'] for b in bundle_refs)-len(obj_list),'reference_bytes':total_reference_bytes,'unique_object_bytes':unique_bytes,'deduplicated_bytes_saved':total_reference_bytes-unique_bytes},'truth_boundary':'Local SHA-256 content-addressed deduplication evidence only; not signing, authorship, timestamp authority, semantic truth, public witness consensus, deployment, domain control, or real-world authorization.'} refs_core={'schema':REF_SCHEMA,'catalog_id':catalog['catalog_id'],'catalog_digest_sha256':cat_digest,'bundles':bundle_refs} refs_digest=sha_bytes(canonical_bytes(refs_core)); references={**refs_core,'reference_digest_sha256':refs_digest,'reference_id':'object-refs-'+refs_digest[:20]} return catalog,references def validate_catalog(root: Path,catalog: dict[str,Any],references: dict[str,Any]) -> dict[str,Any]: if catalog.get('schema')!=STORE_SCHEMA or references.get('schema')!=REF_SCHEMA: raise ObjectStoreError('schema') objects=catalog.get('objects') or [] if objects!=sorted(objects,key=lambda o:o['sha256']): raise ObjectStoreError('catalog_order') by={}; for o in objects: if o['sha256'] in by: raise ObjectStoreError('duplicate_digest') verify_object(root,o);by[o['sha256']]=o core={k:catalog[k] for k in ('schema','digest_algorithm','object_addressing_rule','objects')} dg=sha_bytes(canonical_bytes(core)) if dg!=catalog.get('catalog_digest_sha256') or catalog.get('catalog_id')!='object-catalog-'+dg[:20]: raise ObjectStoreError('catalog_digest') if references.get('catalog_id')!=catalog['catalog_id'] or references.get('catalog_digest_sha256')!=dg: raise ObjectStoreError('stale_catalog_reference') refs_core={k:references[k] for k in ('schema','catalog_id','catalog_digest_sha256','bundles')} rd=sha_bytes(canonical_bytes(refs_core)) if rd!=references.get('reference_digest_sha256') or references.get('reference_id')!='object-refs-'+rd[:20]: raise ObjectStoreError('reference_digest') expected=[] for bundle_path,payload_root in BUNDLES: checked=validate_bundle_payloads(root,bundle_path,payload_root);b=checked['bundle'] expected.append((b['bundle_id'],b['bundle_digest_sha256'],bundle_path,[(a['path'],dg) for a,_,_,dg in checked['refs']])) actual=[] for b in references.get('bundles') or []: actual.append((b['bundle_id'],b['bundle_digest_sha256'],b['bundle_manifest'],[(r['path'],r['object_sha256']) for r in b['references']])) for r in b['references']: if r['object_sha256'] not in by: raise ObjectStoreError('orphan_reference') if r['target_sha256']!=r['object_sha256']: raise ObjectStoreError('target_object_mismatch') if r['path'] in PROTECTED: raise ObjectStoreError('protected_anchor_payload_conflict') if sorted(actual)!=sorted(expected): raise ObjectStoreError('stale_or_mutated_bundle_reference') return {'pass':True,'catalog_digest_sha256':dg,'reference_digest_sha256':rd,'object_count':len(objects),'bundle_count':len(expected)} def build_retention(root: Path,catalog: dict[str,Any],references: dict[str,Any]) -> dict[str,Any]: lineage=load(root/LINEAGE); snapshots=[load(root/p) for p in SNAPSHOTS] roots=[] all_objects=sorted(o['sha256'] for o in catalog['objects']) by_bundle={b['bundle_id']:b for b in references['bundles']} for b in references['bundles']: roots.append({'root_id':'bundle:'+b['bundle_id'],'kind':'bundle','identifier':b['bundle_id'],'retains_objects':sorted(r['object_sha256'] for r in b['references'])}) for s in snapshots: touching=sorted(e['bundle_id'] for e in lineage['edges'] if s['snapshot_id'] in (e['base_snapshot_id'],e['target_snapshot_id'])) retains=sorted({r['object_sha256'] for bid in touching for r in by_bundle.get(bid,{}).get('references',[])}) roots.append({'root_id':'snapshot:'+s['snapshot_id'],'kind':'snapshot','identifier':s['snapshot_id'],'retains_bundle_ids':touching,'retains_objects':retains}) roots.append({'root_id':'lineage:'+lineage['lineage_id'],'kind':'lineage','identifier':lineage['lineage_id'],'retains_bundle_ids':sorted(by_bundle),'retains_objects':all_objects}) roots=sorted(roots,key=lambda r:r['root_id']) live={o:sorted(r['root_id'] for r in roots if o in r['retains_objects']) for o in all_objects} core={'schema':RETENTION_SCHEMA,'catalog_id':catalog['catalog_id'],'reference_id':references['reference_id'],'roots':roots,'object_liveness':[{'object_sha256':o,'retained_by_roots':live[o]} for o in all_objects]} dg=sha_bytes(canonical_bytes(core)) return {**core,'retention_digest_sha256':dg,'retention_id':'retention-'+dg[:20],'summary':{'root_count':len(roots),'retained_object_count':len(all_objects),'unretained_object_count':sum(1 for o in live if not live[o])},'rule':'Historical bundles, snapshots, and lineage metadata are retention roots. Protected anchors and durable historical evidence are never automatic garbage-collection targets.'} def validate_retention(catalog:dict[str,Any],ret:dict[str,Any])->dict[str,Any]: if ret.get('schema')!=RETENTION_SCHEMA or ret.get('catalog_id')!=catalog.get('catalog_id'): raise ObjectStoreError('retention_schema_or_catalog') core={k:ret[k] for k in ('schema','catalog_id','reference_id','roots','object_liveness')};dg=sha_bytes(canonical_bytes(core)) if dg!=ret.get('retention_digest_sha256') or ret.get('retention_id')!='retention-'+dg[:20]: raise ObjectStoreError('retention_digest') cat=set(o['sha256'] for o in catalog['objects']); live={x['object_sha256'] for x in ret['object_liveness']} if cat!=live: raise ObjectStoreError('retention_object_set') for x in ret['object_liveness']: if not x['retained_by_roots']: raise ObjectStoreError('unretained_live_catalog_object') return {'pass':True,'retention_digest_sha256':dg,'root_count':len(ret['roots']),'object_count':len(cat)} def build_gc_plan(catalog:dict[str,Any],retention:dict[str,Any])->dict[str,Any]: live={x['object_sha256'] for x in retention['object_liveness'] if x['retained_by_roots']}; all_obj={o['sha256'] for o in catalog['objects']};unreachable=sorted(all_obj-live) core={'schema':GC_SCHEMA,'catalog_id':catalog['catalog_id'],'retention_id':retention['retention_id'],'unreachable_objects':unreachable,'automatic_delete_objects':[],'manual_review_objects':unreachable,'policy':{'automatic_historical_evidence_deletion':False,'automatic_protected_anchor_deletion':False,'automatic_snapshot_or_bundle_deletion':False,'rule':'Report unreachable objects only. Automatic deletion is always empty for checked-in historical evidence.'}} dg=sha_bytes(canonical_bytes(core));return {**core,'gc_plan_digest_sha256':dg,'gc_plan_id':'gc-plan-'+dg[:20],'summary':{'unreachable_count':len(unreachable),'automatic_delete_count':0,'manual_review_count':len(unreachable)}} def validate_gc(plan:dict[str,Any])->dict[str,Any]: if plan.get('schema')!=GC_SCHEMA or plan.get('automatic_delete_objects')!=[]: raise ObjectStoreError('gc_automatic_delete_forbidden') core={k:plan[k] for k in ('schema','catalog_id','retention_id','unreachable_objects','automatic_delete_objects','manual_review_objects','policy')};dg=sha_bytes(canonical_bytes(core)) if dg!=plan.get('gc_plan_digest_sha256') or plan.get('gc_plan_id')!='gc-plan-'+dg[:20]: raise ObjectStoreError('gc_digest') return {'pass':True,'gc_plan_digest_sha256':dg,'unreachable_count':len(plan['unreachable_objects']),'automatic_delete_count':0} def materialize_bundle(root:Path,bundle_manifest:str,out_dir:Path,catalog:dict[str,Any],references:dict[str,Any]) -> dict[str,Any]: bundle=load(root/bundle_manifest); match=[b for b in references['bundles'] if b['bundle_id']==bundle['bundle_id']] if len(match)!=1: raise ObjectStoreError('bundle_reference_missing_or_duplicate') refmap={r['path']:r for r in match[0]['references']};out_dir.mkdir(parents=True,exist_ok=True) for a in bundle['actions']: if a.get('classification')=='removed': continue rel=safe_rel(a['path']) if rel in PROTECTED: raise ObjectStoreError('protected_anchor_payload_conflict:'+rel) r=refmap.get(rel) if not r or r['object_sha256']!=a['payload_sha256']: raise ObjectStoreError('stale_bundle_reference:'+rel) obj=root/object_rel(r['object_sha256']); data=obj.read_bytes() if sha_bytes(data)!=r['object_sha256']: raise ObjectStoreError('materialize_object_hash:'+rel) dst=out_dir/rel;dst.parent.mkdir(parents=True,exist_ok=True);dst.write_bytes(data) return {'bundle_id':bundle['bundle_id'],'materialized_file_count':sum(1 for a in bundle['actions'] if a.get('payload_path'))} def compare_tree_bytes(a:Path,b:Path)->tuple[bool,int,str]: pa=sorted(p.relative_to(a).as_posix() for p in a.rglob('*') if p.is_file());pb=sorted(p.relative_to(b).as_posix() for p in b.rglob('*') if p.is_file()) if pa!=pb:return False,0,'path_set_mismatch' h=hashlib.sha256(); for rel in pa: ba=(a/rel).read_bytes();bb=(b/rel).read_bytes() if ba!=bb:return False,len(pa),'byte_mismatch:'+rel h.update(rel.encode());h.update(b'\0');h.update(hashlib.sha256(ba).digest()) return True,len(pa),h.hexdigest() def build_materialization_results(root:Path,catalog:dict[str,Any],references:dict[str,Any])->dict[str,Any]: results=[] with tempfile.TemporaryDirectory(prefix='patch-objects-materialize-') as td: base=Path(td) for i,(bundle_path,payload_root) in enumerate(BUNDLES): out=base/f'bundle-{i}' m=materialize_bundle(root,bundle_path,out,catalog,references) ok,count,dg=compare_tree_bytes(out,root/payload_root) if not ok: raise ObjectStoreError('materialization_parity:'+dg) results.append({**m,'bundle_manifest':bundle_path,'preserved_payload_root':payload_root,'byte_for_byte_parity':True,'tree_digest_sha256':dg,'file_count':count}) core={'schema':MATERIAL_SCHEMA,'catalog_id':catalog['catalog_id'],'reference_id':references['reference_id'],'bundles':results} dg=sha_bytes(canonical_bytes(core));return {**core,'materialization_digest_sha256':dg,'materialization_id':'materialize-'+dg[:20],'pass':all(x['byte_for_byte_parity'] for x in results)} def merkle_root(leaves:list[dict[str,Any]])->str: if not leaves:return sha_bytes(b'') level=[bytes.fromhex(x['leaf_digest_sha256']) for x in leaves] while len(level)>1: if len(level)%2: level.append(level[-1]) level=[hashlib.sha256(level[i]+level[i+1]).digest() for i in range(0,len(level),2)] return level[0].hex() def build_checkpoint(root:Path,catalog:dict[str,Any],references:dict[str,Any],retention:dict[str,Any])->dict[str,Any]: lineage=load(root/LINEAGE);leaves=[] for o in catalog['objects']: value={'sha256':o['sha256'],'size_bytes':o['size_bytes'],'storage_path':o['storage_path'],'reference_count':o['reference_count']} vd=sha_bytes(canonical_bytes(value));key='object:'+o['sha256'];leaves.append({'key':key,'value_digest_sha256':vd,'leaf_digest_sha256':sha_bytes((key+'\0'+vd).encode())}) for b in references['bundles']: value={k:b[k] for k in ('bundle_id','bundle_digest_sha256','base_snapshot_id','target_snapshot_id','reference_count')} vd=sha_bytes(canonical_bytes(value));key='bundle:'+b['bundle_id'];leaves.append({'key':key,'value_digest_sha256':vd,'leaf_digest_sha256':sha_bytes((key+'\0'+vd).encode())}) value={'lineage_id':lineage['lineage_id'],'lineage_digest_sha256':lineage['lineage_digest_sha256'],'retention_id':retention['retention_id'],'catalog_id':catalog['catalog_id']} vd=sha_bytes(canonical_bytes(value));key='lineage:'+lineage['lineage_id'];leaves.append({'key':key,'value_digest_sha256':vd,'leaf_digest_sha256':sha_bytes((key+'\0'+vd).encode())}) leaves=sorted(leaves,key=lambda x:x['key']);rootdg=merkle_root(leaves) core={'schema':CHECKPOINT_SCHEMA,'hash_algorithm':'sha256','leaf_rule':'leaf = SHA256(UTF8(key + NUL + SHA256(canonical JSON value))); parent = SHA256(left_digest_bytes || right_digest_bytes); duplicate final node on odd levels','catalog_id':catalog['catalog_id'],'reference_id':references['reference_id'],'retention_id':retention['retention_id'],'lineage_id':lineage['lineage_id'],'leaves':leaves,'merkle_root_sha256':rootdg} dg=sha_bytes(canonical_bytes(core));return {**core,'checkpoint_digest_sha256':dg,'checkpoint_id':'object-checkpoint-'+dg[:20],'leaf_count':len(leaves)} def validate_checkpoint(cp:dict[str,Any])->dict[str,Any]: if cp.get('schema')!=CHECKPOINT_SCHEMA: raise ObjectStoreError('checkpoint_schema') leaves=cp.get('leaves') or [] if leaves!=sorted(leaves,key=lambda x:x['key']): raise ObjectStoreError('checkpoint_order') for x in leaves: calc=sha_bytes((x['key']+'\0'+x['value_digest_sha256']).encode()) if calc!=x['leaf_digest_sha256']: raise ObjectStoreError('checkpoint_leaf') root=merkle_root(leaves) if root!=cp.get('merkle_root_sha256'): raise ObjectStoreError('checkpoint_root') core={k:cp[k] for k in ('schema','hash_algorithm','leaf_rule','catalog_id','reference_id','retention_id','lineage_id','leaves','merkle_root_sha256')};dg=sha_bytes(canonical_bytes(core)) if dg!=cp.get('checkpoint_digest_sha256') or cp.get('checkpoint_id')!='object-checkpoint-'+dg[:20]: raise ObjectStoreError('checkpoint_digest') return {'pass':True,'merkle_root_sha256':root,'checkpoint_digest_sha256':dg,'leaf_count':len(leaves)} def write_json(root:Path,rel:str,obj:dict[str,Any])->None: p=root/rel;p.parent.mkdir(parents=True,exist_ok=True);p.write_text(json.dumps(obj,ensure_ascii=False,indent=2)+'\n',encoding='utf-8') def build_all(root:Path)->dict[str,Any]: catalog,refs=build_catalog(root,True);validate_catalog(root,catalog,refs) retention=build_retention(root,catalog,refs);validate_retention(catalog,retention) gc=build_gc_plan(catalog,retention);validate_gc(gc) mat=build_materialization_results(root,catalog,refs) cp=build_checkpoint(root,catalog,refs,retention);validate_checkpoint(cp) write_json(root,'assets/data/memory-patch-object-catalog.json',catalog) write_json(root,'assets/data/memory-patch-object-references.json',refs) write_json(root,'assets/data/memory-patch-object-retention.json',retention) write_json(root,'assets/data/memory-patch-object-gc-plan.json',gc) write_json(root,'assets/data/memory-patch-object-materialization-results.json',mat) write_json(root,'assets/data/memory-patch-object-checkpoint.json',cp) return {'pass':True,'catalog_id':catalog['catalog_id'],'catalog_digest_sha256':catalog['catalog_digest_sha256'],'unique_object_count':catalog['summary']['unique_object_count'],'payload_reference_count':catalog['summary']['payload_reference_count'],'deduplicated_reference_count':catalog['summary']['deduplicated_reference_count'],'deduplicated_bytes_saved':catalog['summary']['deduplicated_bytes_saved'],'retention_id':retention['retention_id'],'unreachable_count':gc['summary']['unreachable_count'],'automatic_delete_count':0,'materialization_pass':mat['pass'],'checkpoint_id':cp['checkpoint_id'],'merkle_root_sha256':cp['merkle_root_sha256']} def main()->None: ap=argparse.ArgumentParser();ap.add_argument('--root',default=str(ROOT));ap.add_argument('--verify-only',action='store_true');args=ap.parse_args();root=Path(args.root).resolve() if args.verify_only: catalog=load(root/'assets/data/memory-patch-object-catalog.json');refs=load(root/'assets/data/memory-patch-object-references.json');ret=load(root/'assets/data/memory-patch-object-retention.json');gc=load(root/'assets/data/memory-patch-object-gc-plan.json');cp=load(root/'assets/data/memory-patch-object-checkpoint.json') result={'catalog':validate_catalog(root,catalog,refs),'retention':validate_retention(catalog,ret),'gc':validate_gc(gc),'checkpoint':validate_checkpoint(cp),'materialization':build_materialization_results(root,catalog,refs)} else: result=build_all(root) print(json.dumps(result,ensure_ascii=False)) if __name__=='__main__': main()