#!/usr/bin/env python3
"""Offline source/wire verification. Private natural values never printed."""
import hashlib,json,pathlib,sys
HERE=pathlib.Path(__file__).resolve().parent
def sha_file(name):return {x.split(None,1)[1]:x.split(None,1)[0] for x in (HERE/name).read_text().splitlines()}
ORIGINAL=sha_file('SOURCE_SHA256.txt');CARD=sha_file('CARD_GREEN_SOURCE_SHA256.txt');FINAL=sha_file('RESTAURANT_GREEN_SOURCE_SHA256.txt')
FIRST=['SEARCH_SOURCE','BILL_SOURCE','SEARCH_MISSING','BILL_MISSING','CARDS_1','CARDS_2','LOGS_1','LOGS_2'];REST=['RESTAURANT_LIST','RESTAURANT_LIST_2','RESTAURANT_READ','RESTAURANT_DEFAULT']
def read(root,name):return json.loads((pathlib.Path(root)/name).read_bytes())
def strict(a,b,path='$'):
 assert type(a) is type(b),path+':type'
 if isinstance(a,dict):
  assert set(a)==set(b),path+':keys'
  for k in a:strict(a[k],b[k],path+'.'+str(k))
 elif isinstance(a,list):
  assert len(a)==len(b),path+':length'
  for i,(x,y) in enumerate(zip(a,b)):strict(x,y,path+f'[{i}]')
 else:assert a==b,path+':value'
def manifest(root):
 root=pathlib.Path(root);m=read(root,'EVIDENCE_MANIFEST.json');assert {p.name for p in root.iterdir()}==set(m)|{'EVIDENCE_MANIFEST.json'}
 for name,proof in m.items():
  assert pathlib.Path(name).name==name;raw=(root/name).read_bytes();assert len(raw)==proof['bytes'] and hashlib.sha256(raw).hexdigest()==proof['sha256']
def credential_free(value):
 if isinstance(value,dict):
  for key,item in value.items():
   if key.lower() in ['pwd','password','salt','zpkey','zpchargekey']:assert item is None or item=='','credential-value'
   credential_free(item)
 elif isinstance(value,list):
  for item in value:credential_free(item)
def stage(root,prefix,names,sha,red=(),single=False,raw_prefix=None):
 root=pathlib.Path(root);spec=read(root,prefix+'spec-private.json');spec=[spec] if single else spec;after=read(root,prefix+'after-private.json');after={spec[0]['name']:after} if single else after
 report=read(root,prefix+'report-private.json');strict(report['source_sha256'],sha);assert report['actor_unchanged'] is True
 assert report['http_dml']==report['new_fixture_dml']==report['ddl']==0
 cases={x['case']:x for x in report['cases']};assert set(cases)==set(names) and {x['name'] for x in spec}==set(names)
 for entry in spec:
  name=entry['name'];case=cases[name];assert case['method']==entry['method']=='POST' and case['path']==entry['path']
  assert case['http_status']==200 and case['fatal'] is False and case['code']==0
  before=entry['before'];post=after[name];strict(before['source_sets'],post['source_sets']);assert before['source_digest']==post['source_digest'] and case['source_window_unchanged'] is True
  credential_free(before);credential_free(post)
  actor=before['source_sets']['actor'][0];assert set(actor)=={'id','uuid','account','type','status','full_row_digest'} and actor['type']==actor['status']==0
  assert actor['full_row_digest']['rows']==1 and len(actor['full_row_digest']['sha256'])==64
  raw=(root/('raw-'+(prefix if raw_prefix is None else raw_prefix)+name+'-private.json')).read_bytes();assert len(raw)==case['wire_bytes'] and hashlib.sha256(raw).hexdigest()==case['wire_sha256'];body=json.loads(raw);credential_free(body)
  if name in red:
   assert case['source_pass'] is False and case['differences']
   try:strict(before['body'],body)
   except AssertionError:pass
   else:raise AssertionError('red erased')
  else:assert case['source_pass'] is True and case['differences']==[];strict(before['body'],body)
  if name in ['SEARCH_SOURCE','BILL_SOURCE','CARDS_1','LOGS_1','RESTAURANT_LIST','RESTAURANT_READ']:assert case['positive'] is True
 return cases
def verify(b,c):
 b=pathlib.Path(b);c=pathlib.Path(c)
 stage(b,'',FIRST,ORIGINAL);stage(c,'',FIRST,ORIGINAL,['CARDS_1','LOGS_1'])
 stage(b,'restaurant-',REST,ORIGINAL,raw_prefix='');stage(c,'restaurant-',REST,ORIGINAL,['RESTAURANT_LIST','RESTAURANT_READ'],raw_prefix='')
 stage(c,'card-green-',['CARDS_1','LOGS_1'],CARD);stage(c,'restaurant-green-',['RESTAURANT_LIST','RESTAURANT_READ'],FINAL)
 for root,lane,requests,assertions,expected_failures in [(b,'B',13,26,3),(c,'C',17,34,4)]:
  dining=stage(root,'dining-',['DINING_TODAY'],ORIGINAL,single=True,raw_prefix='')['DINING_TODAY'];assert dining['actual_dining_nonempty'] is False
  source=read(root,'dining-after-private.json')['source_sets'];assert source['actual_today_menus']==[] and len(source['leaf_departments'])==(77 if lane=='B' else 23)
  boundary=read(root,'boundary-private.json')
  for key in ['private_source_reflection','effective_local_or_dm_config_verified','legacy_ai_fallback_disabled','file_cache','redis_disabled','external_stubs']:assert boundary[key] is True
  assert boundary['real_queue'] is False and boundary['json_encode_param']==256
  native=read(root,'native-stats-private.json');assert native['requests']==requests and native['assertions']==assertions and sum(x['failures'] for x in native['stages'].values())==expected_failures
  if lane=='C':assert native['stages']['card-green']['failures']==native['stages']['restaurant-green']['failures']==0
  cleanup=read(root,'cleanup-private.json');assert cleanup['port_listening'] is cleanup['worker_state_exists'] is False
  assert cleanup['private_cache_files_remaining']==cleanup['token_artifacts_remaining']==cleanup['credential_bearing_source_remaining']==0
  assert cleanup['new_pdo_owned_actor_verified'] is cleanup['owned_restaurant_verified'] is True
  assert all(x['empty'] is True for x in cleanup['integration_key_flags'].values())
  assert cleanup['child_business_dml']==cleanup['child_ddl']==cleanup['business_rows_deleted']==0 and cleanup['root_approved_ddl']==(3 if lane=='C' else 0)
 # Truly identical no-match/default contracts; never equate natural snapshots.
 for name in ['SEARCH_MISSING','BILL_MISSING','CARDS_2','LOGS_2','RESTAURANT_LIST_2','RESTAURANT_DEFAULT']:strict(read(b,'raw-'+name+'-private.json'),read(c,'raw-'+name+'-private.json'))
 final=read(c,'root-column-final.json');assert final['passed'] is True and final['old_35_column_rows_unchanged'] is True and final['business_DML']==0
 assert read(c,'metadata-after-root-column-private.json')['restaurant_star_column_names_equal'] is True
 return {'status':'pass','methods':7,'bounded_partial':6,'limited_dining_zero_menus':1,'business_http_requests':{'B':13,'C':17},'child_DML_DDL':0,'root_approved_DDL':3,'full_natural_cross_engine_equal_claimed':False,'wide_precision_P0_closed':False}
if __name__=='__main__':
 if len(sys.argv)!=3:raise SystemExit('usage: verify_read7.py B-private-evidence C-private-evidence')
 manifest(sys.argv[1]);manifest(sys.argv[2]);print(json.dumps(verify(sys.argv[1],sys.argv[2]),ensure_ascii=False,indent=2))
