"""Pure offline full original DTO/native/source verification. Never SQL/HTTP or mutate raw."""
from pathlib import Path
import json,hashlib,re,copy
from oracle_lib import expected
HERE=Path(__file__).resolve().parent;RAW=Path('/tmp/cdx-ai-api-customer-read12-20261011')
checks=0
def need(ok,msg):
 global checks
 checks+=1
 if not ok:raise AssertionError(msg)
def load(p):return json.loads(p.read_text())
def lines(p):return [json.loads(v) for v in p.read_text().splitlines()]
def same(a,b,path=''):
 need(type(a) is type(b),'Exact JSON type '+path)
 if type(a) is dict:
  need(list(a)==list(b),'Complete field set/order '+path)
  for k in a:same(a[k],b[k],path+'.'+k)
 elif type(a) is list:
  need(len(a)==len(b),'Complete array length '+path)
  for i,x in enumerate(a):same(x,b[i],path+str(i))
 else:need(a==b,'Exact source/table value '+path)
def phptype(v):return 'NULL' if v is None else 'boolean' if type(v) is bool else 'integer' if type(v) is int else 'double' if type(v) is float else 'string' if type(v) is str else 'array'
pins=load(HERE/'source-frozen-sha.json')['files'];results={}
for lane in ['B','C']:
 p=RAW/f'private-{lane}-final';pre=load(p/'preflight-read.json');cases=load(p/'cases.json');exp=expected(pre,cases);frozen=load(p/'expected.json');same(exp,frozen,'pre-HTTP expected all DTOs')
 need(hashlib.sha256((p/'expected.json').read_bytes()).hexdigest()==load(p/'oracle-ready.json')['expected_SHA'],'Expected original pre-HTTP freeze')
 final=load(p/'persistence-final.json');need(pre['source_data']==final['source_data'] and pre['protection']==final['protection'] and final['unchanged'] is True,'Whole original source/user/financial rows conserved')
 budget=lines(p/'sql-budget.jsonl');need(len(budget)<=600 and [r['n'] for r in budget]==list(range(1,len(budget)+1)),'All readonly SQL including failed prep counted/cap600')
 events=lines(p/'business-native.jsonl');executed=[e for e in events if e['layer']=='actual PDOStatement.execute'];need(all(e['operation'] in ['SELECT','metadata_SHOW','metadata_CALL'] for e in executed),'0 native business/prepare DMLDDL')
 need(not any(e.get('layer') in ['actual PDO.exec transaction','actual PDO.beginTransaction','actual PDO.commit','actual PDO.rollBack'] for e in events),'0 business transactions')
 plans={};pairs=0
 for e in events:
  key=(e['scenario'],e['connection'],e.get('queryString'))
  if e['layer']=='original framework bind plan':plans[key]=e['bind_plan']
  if e['layer']!='actual PDOStatement.execute':continue
  need(key in plans,'Native original framework binding exists')
  for k,v in (plans[key].items() if type(plans[key]) is dict else enumerate(plans[key])):
   param=str(k+1) if type(k) is int else ':'+k
   if type(v) is list:
    val,typ=v
    if typ==1 and val=='':val=0
    elif typ==21:
     if type(val) is str:val=float(val)
     typ=2
   else:val,typ=v,2
   b=e['bindValue'][param];need(type(b['value']) is type(val) and b['value']==val and b['pdo_type']==typ and b['php_type']==phptype(val),'Every native value/PHPtype/PDOtype exact');pairs+=1
 outcomes={}
 for label,c in cases.items():
  r=load(p/f'response-{label}.json');client=load(p/f'client-{label}.json');body=json.loads(r['raw_response_body']);need(r['raw_response_body']==client['body_raw'] and r['http_status']==client['http_status'],'Original saved→sent/client raw bytes '+label)
  req=load(p/f'request-{label}.json');need(req['case']==c and load(p/f'client-attempt-{label}.json')['query_raw']==c['query_raw'],'Exact authorized GET/params once '+label)
  inc=load(p/f'included-{label}.json')
  for n,h in inc['files'].items():need(pins.get(n)==h,'Every actualincluded source pin '+label)
  route_events=[e for e in events if e['scenario']==label];errors=[e for e in route_events if 'error' in e]
  if label=='WEIGHT':
   need(errors and any('yoshop_staff' in e.get('queryString','') for e in errors),'Exact tool-only missing staff metadata scope')
   need(list(body)==['message','status','data','trace_id'] and body['status']==500 and body['data']==[] and body['message']==errors[-1]['error']['message'],'Full actual blocked response, not business failure')
   outcomes[label]={'result':'TOOL_BLOCKED_METADATA_SCOPE','business_DB_pass':False,'native_SQL':sum(e['layer']=='actual PDOStatement.execute' for e in route_events),'blocked_sql':errors[-1]['queryString'],'message':body['message']}
  else:
   need(errors==[],'No swallowed original/native error in successful route '+label)
   need(r['http_status']==exp[label]['http_status'] and list(body)==['status','message','data','trace_id'],'Complete original envelope/status '+label)
   same(body['status'],exp[label]['status'],label+'.status');same(body['message'],exp[label]['message'],label+'.message');same(body['data'],exp[label]['data'],label+'.data')
   outcomes[label]={'result':exp[label]['classification'],'full_DTO_pass':True,'business_DB_pass':label!='INDEX','native_SQL':sum(e['layer']=='actual PDOStatement.execute' for e in route_events)}
  need(type(body['trace_id']) is str and re.fullmatch('[0-9a-f]{13}',body['trace_id']) is not None,'Original traceid type/format')
  need(not (p/f'observer-error-{label}.json').exists() and not (p/f'harness-error-{label}.json').exists(),'Observer cannot alter return/no actual observer failure')
 need(len(cases)==load(p/'run-complete.json')['actual_HTTP']==10,'Exactly ten actual HTTP/no success replay')
 need(load(p/'cleanup.json')['tokens']=={'U1':'revoked'} and load(p/'worker-stopped.json')['port_closed'] is True,'Own token/worker/port cleaned')
 cc=load(p/'cache-cleanup.json');need(cc['shared_cache_touched'] is False and len(cc['own_cache_files_removed'])==2,'Only two exact own cache files removed')
 src=load(p/'source-actual-fileset.json');need(set(src['files'])==set(pins) and src['actual_regular_files']==6168 and src['appledouble_magic_count']==0,'Actual full6168 fileset/magic0')
 for n,h in pins.items():need(src['files'][n]['sha256']==h and not any(x.startswith('._') for x in Path(n).parts),'Each actual source bytes exact')
 negatives=0
 # Six meaningful pure outcome corruptions: object/array, missing/default/pagination and date types.
 samples=[([],{}),({'last_page':0},{'last_page':1}),({'list':[],'signInfo':{'status':1}},{'list':[],'signInfo':{}}),({'list':['2026-10-11']},{'list':'2026-10-11'}),({'data':[]},{'data':[{'user_id':999}]}),({'x':1},{'x':'1'})]
 for a,b in samples:
  try:same(a,b,'counterexample')
  except AssertionError:negatives+=1
  else:raise AssertionError('Corrupt source result accepted')
 need(negatives==6,'Six relevant pure outcome counterexamples')
 results[lane]={'actor':pre['actor'],'tenant':10001,'HTTP':10,'all_SQL':len(budget),'original_native_SQL':len(executed),'native_binding_pairs':pairs,'DML':0,'DDL':0,'outcomes':outcomes,'counterexamples':negatives}
f=load(HERE/'pre-HTTP-freeze.json')
for n,h in f['files'].items():need(hashlib.sha256((HERE/n).read_bytes()).hexdigest()==h,'Original before-HTTP tool/oracle/source freeze')
print(json.dumps({'checks':checks,'lanes':results,'SQL':0,'HTTP':0,'limitations':['8 empty/default DB reads only; no nonempty data invented','Index literal404 noDB','Weight actual tool metadata scope failure; not business/DB defect','Old0.01/0.02 and original seals untouched']},ensure_ascii=False,indent=2))