"""Offline real zero-boundary HTTP/SQL/CAS verifier; no DB or network writes."""
import base64,copy,hashlib,importlib.util,json,re
from pathlib import Path
from oracle import oracle
HERE=Path(__file__).resolve().parent
spec=importlib.util.spec_from_file_location('native_verifier',HERE.parent/'verify.py');native=importlib.util.module_from_spec(spec);spec.loader.exec_module(native)
PLAN=json.loads((HERE/'log-http-plan.json').read_text());check=native.check;exact=native.exact
def load(lane,name):return json.loads((HERE/'private'/lane/name).read_text())
def run():
    result=dict(HTTP=0,full_DTO={},logs={},CAS={},sentinel_deltas={},negative_cases=0);wires={}
    for lane in ['B','C']:
        origin=load(lane,'original100.json');d=load(lane,'dictionary.json');expected=oracle(origin,d);exact(expected,load(lane,'expected-before-zero-http.json'),'immutable preHTTP zero oracle')
        zero=load(lane,'batch-before10.json');want=copy.deepcopy(origin);want['dishes'][0]['raw_flag_weight']='0.00' if lane=='B' else 0.0;actual=copy.deepcopy(zero);del actual['sentinel'];del want['sentinel'];exact(actual,want,'zero only raw_flag_weight; natural sentinel separately recorded')
        prior=load(lane,'logs-before10.json');check(len(prior)==148,'148 prior logs');original=copy.deepcopy(prior);attempts=load(lane,'http-attempts-zero4.json');check(len(attempts)==4,'four real HTTP');result['HTTP']+=len(attempts);wires[lane]={}
        for attempt,case in zip(attempts,PLAN['cases']):
            check(attempt['label']==case['label'] and attempt['complete'],'complete ordered HTTP');w=load(lane,'wire-'+case['label']+'.json');wires[lane][case['label']]=w
            check(base64.b64decode(attempt['body_base64']).decode()==w['body'] and hashlib.sha256(w['body'].encode()).hexdigest()==w['body_sha256'],'actual client/wire/SHA')
            action=case['path'].split('/')[-1];check(w['status']==expected[action]['status'],'original full status '+action);exact(json.loads(w['body']),expected[action]['body'],lane+'/'+action+' zero full DTO/error contract')
            exact(w['actual_client_headers'],{'Accept':'application/json, text/plain, */*','Cache-Control':'max-age=28880'},'actual proved headers')
            check(w['source_unchanged'] and w['source_before']==w['source_after'],'own fullzero protected percase');check(w['old_logs_unchanged'] and w['logs_before']==prior and w['logs_after']==prior+w['new_logs'],'all prior logs exact')
            native.native_logs(w,case,lane,prior);prior=w['logs_after']
            model=w['reflection']['app\\model\\DateMenu'];check(model['sha256']=='56281c991c500b92c7b72660873abf52bdf0e66b7111dfe330f2d6c0ee0a2be4','actual loaded final model562')
            check(model['file'] in w['included_files'] and '/tmp/cdx-onto-nutrition10-zero4-20261011/source/application/model/DateMenu.php'==model['file'],'actual source loaded in isolated runtime')
            check(type(w['source_before']['dishes'][0]['raw_flag_weight']) is (str if lane=='B' else float),'native physical DEC return type recorded')
        check(len(prior)==166 and prior[:148]==original,'148+18 unique physical logs');check(len({r['id'] for r in prior})==166 and len({r['uuid'] for r in prior})==166,'all166 IDs and GUIDs unique')
        z=load(lane,'cas-zero.json');restore=load(lane,'cas-restore-retry1.json' if lane=='B' else 'cas-restore.json')
        for journal,values in [(z,['0.00','100.00']),(restore,['100.00','0.00'])]:
            check(journal['committed'] and journal['logs_unchanged'],'real commit/old logs protected');check(journal['triggers']==[],'no hidden dishes triggers');check(len(journal['events'])==2,'lock SELECT + one CAS')
            lock,event=journal['events'];check(lock['queryString']=='SELECT * FROM ydy_dishes WHERE uuid=? AND create_by=? FOR UPDATE' and lock['binds']==['C8EAA110-0002-4000-8000-202610110002','CDXNR11A'] and lock['rows']==journal['before']['dishes'],'exact locked full original record')
            sql='UPDATE ydy_dishes SET raw_flag_weight=?'+(', update_time=update_time' if lane=='B' else '')+' WHERE uuid=? AND create_by=? AND raw_flag_weight=?'
            check(event['queryString']==sql and event['binds']==[values[0],'C8EAA110-0002-4000-8000-202610110002','CDXNR11A',values[1]] and event['rowCount']==1 and event['success'] and event['errorInfo'][0]=='00000','exact CAS native SQL/binds/affected1')
            before=copy.deepcopy(journal['before']);after=copy.deepcopy(journal['after']);before['dishes'][0]['raw_flag_weight']=after['dishes'][0]['raw_flag_weight'];del before['sentinel'];del after['sentinel'];exact(after,before,'only one physical field changed')
        if lane=='B':
            fail=load(lane,'cas-restore.json');check(fail['committed'] is False and fail['rollback_verified'] is False and fail['events'][1]['rowCount']==1,'original conservative sentinel rollback raw preserved')
            diag=load(lane,'restore-diagnostic.json');check(diag['own_full_zero_unchanged'] and diag['logs_exact_last_wire'] and list(diag['sentinel_deltas'])==['menu_recommend'],'independent own-only rollback verified; external sentinel unknown')
            check(diag['current_logs']==prior,'failed CAS did not alter166 logs');result['CAS'][lane]=dict(attempts=3,commit=2,rollback=1,natural_delta_tables=list(diag['sentinel_deltas']))
        else:result['CAS'][lane]=dict(attempts=2,commit=2,rollback=0)
        restored=load(lane,'restored100.json');a=copy.deepcopy(restored);b=copy.deepcopy(origin);del a['sentinel'];del b['sentinel'];exact(a,b,'full original100 restored')
        final=load(lane,'final-zero4.json');check(final['source_restored100'] and final['old148_unchanged'] and final['logs']==prior and final['log_count']==166,'final fullsource/log conservation')
        resource=load(lane,'resource-check.json');check(all(resource[k] for k in ['own_worker_absent','own_port_closed','original_parent_present','request_absent','cache_absent']) and resource['AppleDouble']==[],'exact own resources cleared')
        provenance=load(lane,'source-provenance.json');check(provenance['source_php_files']==958 and provenance['source_lock_sha256']=='57d57588410be71682c8597d092b125cb26cf2b3d489b6ce7f1a136c56bad96b','APP958 full source pin')
        result['sentinel_deltas'][lane]={k:dict(before=v,after=restored['sentinel'][k]) for k,v in origin['sentinel'].items() if restored['sentinel'][k]!=v};result['logs'][lane]=dict(old=148,new=18,final=166);result['full_DTO'][lane]={a:dict(status=e['status'],code_type=type(e['body']['code']).__name__) for a,e in expected.items()}
    for case in PLAN['cases']:check(wires['B'][case['label']]['body']==wires['C'][case['label']]['body'] and wires['B'][case['label']]['status']==wires['C'][case['label']]['status'],'all four original full raw response parity')
    error={'code':'2','message':'Inf and NaN cannot be JSON encoded','data':[]}
    for bad in [{'code':2,'message':error['message'],'data':[]},{**error,'data':{}},{**error,'message':''},{**error,'unknown':1}]:
        try:exact(bad,error,'counterexample')
        except AssertionError:result['negative_cases']+=1
        else:raise AssertionError('Verifier accepted mutation')
    result['assertions']=native.COUNT;result['status']='Four consumers real zero parity proven; two original stage NaN errors retained; physical DM type restoration verified; full100 restored'
    (HERE/'verification-zero4.json').write_text(json.dumps(result,ensure_ascii=False,indent=2)+'\n');print(json.dumps(result,ensure_ascii=False))
if __name__=='__main__':run()