"""Independent original-table/source oracle. Pure offline, no SQL/HTTP/mutations."""
from pathlib import Path
from decimal import Decimal
from datetime import datetime, timedelta
import json, hashlib, re, copy

HERE=Path(__file__).resolve().parent
RAW=Path('/tmp/cdx-ai-core-direct002-20261011')
checks=0
def need(ok,label):
    global checks
    checks+=1
    if not ok:raise AssertionError(label)
def load(p):return json.loads(p.read_text())
def lines(p):return [json.loads(s) for s in p.read_text().splitlines()]
def decimal(v):
    need(type(v) is str and re.fullmatch(r'-?\d+(?:\.\d{1,2})?',v) is not None,'Exact server monetary CAST text')
    return Decimal(v)
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'
def validate_outcome(w,financial,ids):
    before=w['own_before'];after=w['own_after']
    need(w['original_before']==w['original_after'],'Original13 fullrows exactly conserved')
    for k in ['AI_U2','STORE_S1','STORE_U1','STORE_U2','STORE_R1','STORE_P1','initial_menu','meal_order_pickup','staff_department','one_card_account_map','user_subsidy','user_gift_log']:
        need(before[k]==after[k],'Protected identity/prepay/excluded surface '+k)
    new={}
    for table,pk in [('meal_order','id'),('account_settlement','settlement_id'),('user_balance_log','log_id')]:
        need(len(after[table])==len(before[table])+1,'Exactly one committed new row '+table)
        for row in before[table]:need([r for r in after[table] if r[pk]==row[pk]]==[row],'Old0.01/prepay fullrow '+table)
        new[table]=[r for r in after[table] if r[pk] not in {x[pk] for x in before[table]}][0]
    o=new['meal_order'];s=new['account_settlement'];log=new['user_balance_log']
    need(o['remark']=='CDXCORE11S1_DIRECT_002' and int(o['source'])==4 and int(o['pay_status'])==20 and int(o['order_status'])==30,'New exact payment scope/status')
    need(int(o['user_id'])==ids['STORE_U1'] and int(o['restaurant_id'])==ids['STORE_R1'],'New exact real ownership')
    need(int(s['settlement_id'])==int(o['account_settlement_id']) and int(log['order_id'])==int(o['id']),'Exact generated settlement/order/log linkage')
    need(s['biz_order_no']==o['order_no'] and s['settlement_no']=='CS:'+o['order_no'] and s['biz_type']=='consume' and int(s['status'])==20,'Exact settlement key/status')
    need(int(s['user_id'])==int(log['user_id'])==ids['AI_U1'] and int(s['store_id'])==int(log['store_id'])==2147001111,'Ledger tenant/actor')
    need(s['deduction_order_snapshot']==o['deduction_order_snapshot']=='["gift","subsidy","cash"]','Source deduction snapshot')
    need(log['describe']=='用户消费：'+o['order_no'] and int(log['scene'])==20,'Original balance-log semantics')
    need(int(o['print_status'])==0 and int(o['restaurant_method'])==1 and after['meal_order_pickup']==[],'Dine-in/no pickup/print')
    au=copy.deepcopy(after['AI_U1'][0]);bu=copy.deepcopy(before['AI_U1'][0])
    for k in ['balance','gift_balance','pay_money']:au.pop(k);bu.pop(k)
    need(au==bu,'AI user every other physical column unchanged')
    money=financial['whole_declared_DECIMAL_columns']
    def byid(table,pk,id):
        rows=[r for r in money[table]['rows'] if int(r[pk])==int(id)];need(len(rows)==1,'Exact CAST identity');return rows[0]
    u=byid('AI.user','user_id',ids['AI_U1'])
    for k,v in [('balance_text','0.97'),('gift_balance_text','0.00'),('pay_money_text','0.03')]:need(decimal(u[k])==Decimal(v),'Wallet money '+k)
    sm=byid('AI.account_settlement','settlement_id',s['settlement_id']);lm=byid('AI.user_balance_log','log_id',log['log_id']);om=byid('ZHCT.meal_order','id',o['id'])
    for k,v in [('amount','0.02'),('cash_amount','0.02'),('gift_amount','0.00'),('subsidy_amount','0.00')]:need(decimal(sm[k+'_text'])==Decimal(v),'Settlement money '+k)
    for k,v in [('money','-0.02'),('balance_after','0.97')]:need(decimal(lm[k+'_text'])==Decimal(v),'Log money '+k)
    for k in ['total_price','ori_total_price','order_price','pay_price','cash']:need(decimal(om[k+'_text'])==Decimal('0.02'),'New order amount '+k)
    for k in ['subsidy','gift','remain_subsidy','remain_gift','discount_amount','management_fee']:need(decimal(om[k+'_text'])==0,'Excluded money '+k)
    for k in ['remain_balance','remain_cash']:need(decimal(om[k+'_text'])==Decimal('0.97'),'Order remaining money '+k)
    r=json.loads(w['response']['raw_response_body']);dto=r['data']
    need(list(r)==['status','message','data','trace_id'] and type(r['status']) is int and r['status']==200 and r['message']=='支付成功','Complete original envelope/order/types')
    need(type(r['trace_id']) is str and re.fullmatch(r'[0-9a-f]{13}',r['trace_id']) is not None,'Original uniqid trace type/format')
    expected={'orderId':int(o['id']),'order_no':o['order_no'],'pay_price':'0.02','pay_time':o['pay_time'].split('.')[0],'restaurant_id':ids['STORE_R1'],'meal_date':o['meal_date'],'meal_times':int(o['meal_times']),'restaurant_method':1,'pickup_no':'','remain_balance':'0.97','cash':'0.02','subsidy':'0.00','gift':'0.00','discount_amount':'0.00','management_fee':'0.00','marketing_strategy_id':0}
    need(list(dto)==list(expected),'All sixteen original DTO fields/order')
    for k,v in expected.items():need(type(dto[k]) is type(v) and dto[k]==v,'Independent original table/source DTO '+k)
    return new

def verify_lane(lane):
    p=RAW/f'private-{lane}-final';w=load(p/'wire.json');money=load(p/'physical-money-final.json');prepared=load(p/'prepared.json');ids=prepared['ids'];new=validate_outcome(w,money,ids)
    response=load(p/'response-original.json');client=load(p/'client-result.json')
    need(response['raw_response_body']==client['body_raw']==w['response']['raw_response_body'],'Original pre-observer body equals actual client bytes')
    need(response['http_status']==client['http_status']==200,'Original and client HTTP200')
    need(not (p/'observer-error.json').exists() and not (p/'harness-error.json').exists(),'No observer or harness error')
    need(load(p/'preflight-budget.json')['total']<=160,'Actual bounded preflight SQL')
    expected_counts={'INSERT.ydy_meal_order':1,'UPDATE.yoshop_user':2,'INSERT.yoshop_user_balance_log':1,'INSERT.yoshop_account_settlement':1,'UPDATE.ydy_meal_order':1}
    need(w['native_DML_counts']==expected_counts,'Exact3I3U native per-table counts')
    events=lines(p/'business-native.jsonl');formal=[e for e in events if e['scenario']=='DIRECT_002'];actual=[e for e in formal if e['layer']=='actual PDOStatement.execute']
    need(len(actual)==w['business_native_SQL']<=({'B':80,'C':160}[lane]),'Exact formal native SQL cap')
    dml=[e for e in actual if e['operation'] in ['INSERT','UPDATE']]
    need(len(dml)==6 and sum(e['operation']=='INSERT' for e in dml)==3,'Exactly six native DML')
    for e in dml:need(e['success'] is True and e['rowCount']==1,'Every DML affected1')
    need([e['connection'] for e in formal if e['layer']=='actual PDO.commit']==['ZHCT','AI'],'Original Store commit then AI commit; nonXA')
    need(not any(e['layer']=='actual PDO.rollBack' for e in formal),'No payment rollback/retry')
    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,'Every native execute original framework plan')
        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),'Exact every framework/native value/PHPtype/PDOtype pair');pairs+=1
    pins=load(HERE/'source-frozen-sha.json')['files']
    for included in [w['included_source_sha256'],load(p/'included-before-send.json')]:
        for n,h in included.items():need(n in pins and pins[n]==h,'Every actualincluded source pin')
    need(load(p/'persistence-final.json')['own_fullrows']==w['own_after'],'Independent final snapshot unchanged after wire')
    # Every new physical field, original supplied value, omitted default, and timestamp remains visible.
    schema=load(p/f'preflight-{lane}.json')['schema'];written={};timefacts=[]
    for e in [e for e in formal if e['layer']=='original framework bind plan' and e['queryString'].lstrip().startswith(('INSERT','UPDATE'))]:
        sql=re.sub(r'[`"]([a-zA-Z_][a-zA-Z0-9_]*)[`"]',r'\1',e['queryString']);table=re.search(r'^(?:INSERT INTO|UPDATE) (\w+)',sql).group(1)
        if table=='yoshop_user':continue
        values={}
        if sql.startswith('INSERT') and ' SET ' not in sql:
            m=re.search(r'\(([^)]+)\)\s*VALUES\s*\(([^)]+)\)',sql);tokens=zip(map(str.strip,m[1].split(',')),map(str.strip,m[2].split(',')))
        else:tokens=re.findall(r'(\w+)\s*=\s*(:\w+)',sql.split(' WHERE ')[0])
        for field,token in tokens:values[field]=e['bind_plan'][token.lstrip(':')][0]
        written.setdefault(table,{}).update(values)
    fields=0;defaults=0
    for short,table,db,pk in [('meal_order','ydy_meal_order','ZHCT','id'),('account_settlement','yoshop_account_settlement','AI','settlement_id'),('user_balance_log','yoshop_user_balance_log','AI','log_id')]:
        row=new[short];cols=schema[db][short]['columns'];need(set(row)=={c['column_name'].lower() for c in cols},'All physical new-row columns retained')
        for c in cols:
            k=c['column_name'].lower();v=row[k];typ=c['data_type'].lower()
            if k==pk:need(int(v)>0,'Actual generated physical identity');continue
            if k in written[table]:
                want=written[table][k]
                if re.search('int|decimal|numeric|number|float|double|real|bit',typ):need(v is not None and Decimal(str(v))==Decimal(str(want)),'Exact physical supplied numeric '+table+'.'+k)
                elif typ=='datetime':
                    target=str(want)
                    if '.' in target:
                        dt=datetime.fromisoformat(target)
                        if lane=='B':target=(dt+timedelta(microseconds=500000)).replace(microsecond=0).strftime('%Y-%m-%d %H:%M:%S')
                    elif lane=='C' and c.get('data_scale')==6:target+='.'+'0'*6
                    need(type(v) is str and v==target,'Exact physical DATETIME native precision/rounding '+table+'.'+k)
                    timefacts.append({'table':table,'field':k,'original_bind':want,'physical':v,'scale':c.get('data_scale',0)})
                else:need(type(v) is type(want) and v==want,'Exact physical supplied bytes '+table+'.'+k)
                fields+=1
            else:
                default=c.get('column_default',c.get('data_default'))
                if default is None or str(default).upper()=='NULL':need(v is None,'Omitted NULL default '+table+'.'+k)
                elif str(default).upper() in ['CURRENT_TIMESTAMP','CURRENT_TIMESTAMP()','SYSDATE','CURRENT_DATE']:
                    dt=datetime.fromisoformat(v);start=datetime.fromtimestamp(w['response']['business_start']);end=datetime.fromtimestamp(w['response']['business_end']);need(start-timedelta(seconds=5)<=dt<=end+timedelta(seconds=5),'Default server clock within business interval')
                else:
                    literal=str(default)
                    if literal.startswith("'") and literal.endswith("'"):literal=literal[1:-1].replace("''", "'")
                    if re.search('int|decimal|numeric|number|double|float|real|bit',typ):need(v is not None and Decimal(str(v))==Decimal(literal),'Omitted numeric default '+table+'.'+k)
                    elif typ=='datetime':need(v==(literal+'.000000' if lane=='C' and c.get('data_scale')==6 and '.' not in literal else literal),'Omitted DATETIME native default '+table+'.'+k)
                    else:need(type(v) is str and v==literal,'Omitted text/date default bytes '+table+'.'+k)
                defaults+=1
    need(load(p/'cleanup.json')['tokens']=={'U1':'revoked'},'Only new own U1 token revoked')
    need(load(p/'worker-stopped.json')['port_closed'] is True,'Own worker/port cleaned')
    negatives=0
    corruptions=[lambda x:x['own_after']['account_settlement'].clear(),lambda x:x['own_after']['meal_order_pickup'].append({'id':1}),lambda x:x['own_after']['AI_U2'][0].update(balance=4)]
    for change in corruptions:
        x=copy.deepcopy(w);change(x)
        try:validate_outcome(x,money,ids)
        except (AssertionError,IndexError):negatives+=1
        else:raise AssertionError('Corrupted fullrow outcome accepted')
    for key in json.loads(w['response']['raw_response_body'])['data']:
        x=copy.deepcopy(w);r=json.loads(x['response']['raw_response_body']);r['data'][key]=False;x['response']['raw_response_body']=json.dumps(r)
        try:validate_outcome(x,money,ids)
        except AssertionError:negatives+=1
        else:raise AssertionError('Corrupted DTO field accepted '+key)
    need(negatives==19,'All independent ledger/DTO counterexamples rejected')
    return {'HTTP':1,'INSERT':3,'UPDATE':3,'native_SQL':len(actual),'binding_pairs':pairs,'physical_supplied_fields':fields,'physical_omitted_defaults':defaults,'full_DTO_data_fields':16,'order':int(new['meal_order']['id']),'settlement':int(new['account_settlement']['settlement_id']),'log':int(new['user_balance_log']['log_id']),'counterexamples':negatives,'timefacts':timefacts}

if __name__=='__main__':
    results={lane:verify_lane(lane) for lane in ['B','C']}
    print(json.dumps({'checks':checks,'lanes':results,'SQL':0,'HTTP':0,'limitations':['Original C0.01 response remains missing','repeat/Redis/onecard/external/concurrency/XA not tested','Staff7/PrepayONUPDATE/unique P0 remain open']},ensure_ascii=False,indent=2))