"""Offline verification only: never performs HTTP, SQL, cache writes or replay."""
from pathlib import Path
import json, hashlib, re, copy
from datetime import datetime, timezone, timedelta
from decimal import Decimal

HERE=Path(__file__).resolve().parent
RAW=Path('/tmp/cdx-ai-core-scan-pay-20261011')
checks=0
def require(ok, text):
    global checks
    checks+=1
    if not ok: raise AssertionError(text)
def load(p): return json.loads(p.read_text())
def lines(p): return [json.loads(v) for v in p.read_text().splitlines()]
def value_type(v):
    return 'NULL' if v is None else 'boolean' if isinstance(v,bool) else 'integer' if isinstance(v,int) else 'double' if isinstance(v,float) else 'string' if isinstance(v,str) else 'array'
def decimal(v):
    require(isinstance(v,str) and re.fullmatch(r'-?\d+(?:\.\d{1,2})?',v) is not None,'Server monetary text, no float epsilon')
    return Decimal(v)

pins=load(HERE/'source-frozen-sha.json')['files']
require(len(pins)==6168,'Whole frozen source count')
results={}
for lane, http_count, inserts, updates, success in [('B',10,8,6,'DIRECT_SUCCESS_RETRY2'),('C',9,7,6,'DIRECT_SUCCESS_RETRY1')]:
    p=RAW/f'private-{lane}-final'
    prepared=load(p/'prepared.json'); final=load(p/'persistence-final.json'); financial=load(p/'physical-money-final.json')
    events=lines(p/'business-native.jsonl'); clients=lines(p/'client-results.jsonl')
    require(len(clients)==http_count,f'{lane} actual client HTTP count')
    dml=[e for e in events if e['layer']=='actual PDOStatement.execute' and e['operation'] in ['INSERT','UPDATE']]
    require(sum(e['operation']=='INSERT' for e in dml)==inserts and sum(e['operation']=='UPDATE' for e in dml)==updates,f'{lane} exact actual DML attempts including rollback')
    for e in dml: require(e['success'] is True and e['rowCount']==1,'Every actual DML affected exactly one')
    require(not any(e.get('operation') in ['DELETE','DDL'] for e in events),'No recorded destructive SQL')
    # Original framework PARAM_FLOAT 21 becomes the original native PARAM_STR 2.
    plans={}; pair_count=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
        require(key in plans,'Native execute has original framework plan')
        for k,v in plans[key].items() if isinstance(plans[key],dict) else enumerate(plans[key]):
            param=str(k+1) if isinstance(k,int) else ':'+k
            if isinstance(v,list):
                val,typ=v
                if typ==1 and val=='': val=0
                elif typ==21:
                    if isinstance(val,str): val=float(val)
                    typ=2
            else: val,typ=v,2
            b=e['bindValue'][param]
            require(type(b['value']) is type(val) and b['value']==val and b['pdo_type']==typ and b['php_type']==value_type(val),'Full exact original/native bind value, PHP type and PDO type')
            pair_count+=1
    own=final['own_fullrows']; ids=prepared['ids']
    require(len(prepared['rows'])==7 and len(prepared['writes'])==7,'Exactly seven committed fixture rows')
    plan=load(RAW/'typed-prepare-plan.json')
    schema=load(p/f'preflight-{lane}.json')['schema']; fields=0; defaults=0
    tz=timezone(timedelta(hours=8))
    start=datetime.fromisoformat(load(p/'prepare-attempt.json')['started_at'])
    end=datetime.fromisoformat(prepared['finished_at'])
    for entry in plan['rows']:
        row=prepared['rows'][entry['name']]; db='AI' if entry['connection']=='mysql' else 'ZHCT'; table=entry['table'][7 if db=='AI' else 4:]
        names=[c['column_name'].lower() for c in schema[db][table]['columns']]
        require(set(names)==set(row),'Every physical fixture column retained including omitted defaults')
        require(Decimal(str(row[entry['pk']]))==ids[entry['name']],'Physical generated identity')
        for key,(typ,v) in entry['values'].items():
            if typ=='ref': v=ids[v]
            if typ in ['int','ref']: require(not isinstance(row[key],bool) and Decimal(str(row[key]))==v,'Exact supplied numeric/ref field')
            elif table=='user' and db=='AI' and key in ['balance','gift_balance','pay_money']: require(Decimal(str(row[key]))==Decimal(v),'Original fixture declared DECIMAL value retained')
            elif lane=='C' and db=='ZHCT' and table in ['staff','user'] and key=='create_time': require(row[key]==v+'.000000','Only documented explicit zero-fraction CREATE_TIME representation')
            else: require(type(row[key]) is str and row[key]==v,'Exact supplied string field')
            fields+=1
        for column in schema[db][table]['columns']:
            key=column['column_name'].lower()
            if key in entry['values'] or key==entry['pk']: continue
            actual=row[key]; default=column.get('column_default',column.get('data_default')); physical_type=column['data_type'].lower()
            if default is None or str(default).upper()=='NULL': require(actual is None,'Omitted original NULL default')
            elif str(default).upper() in ['CURRENT_TIMESTAMP','CURRENT_TIMESTAMP()','SYSDATE','CURRENT_DATE']:
                require(isinstance(actual,str),'Native default clock string retained')
                instant=datetime.fromisoformat(actual).replace(tzinfo=tz)
                require(start-timedelta(seconds=5)<=instant<=end+timedelta(seconds=5),'Original default server clock within actual preparation 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',physical_type):
                    require(actual is not None and not isinstance(actual,bool) and Decimal(str(actual))==Decimal(literal),'Omitted physical numeric default exact, no epsilon')
                elif physical_type=='datetime':
                    target=literal+'.000000' if lane=='C' and column.get('data_scale')==6 and '.' not in literal else literal
                    require(type(actual) is str and actual==target,'Omitted declared DATETIME default exact native precision; raw unchanged')
                else: require(type(actual) is str and actual==literal,'Omitted original text/date default exact bytes')
            defaults+=1
    require(final['old_fullrows']==load(p/'guard.json')['old_protection'],'All original protected full rows unchanged')
    require(final['actual_paid_once'] is True,'Exactly one actual committed direct payment')
    orders=own['meal_order']; direct=[o for o in orders if o['remark']=='CDXCORE11S1_DIRECT']; pending=[o for o in orders if int(o['source'])==8]
    require(len(orders)==2 and len(direct)==1 and len(pending)==1,'One coffee order and one direct paid order')
    o=direct[0]; s=own['account_settlement']; logs=own['user_balance_log']
    require(int(o['pay_status'])==20 and int(o['order_status'])==30 and int(o['user_id'])==ids['STORE_U1'],'Paid direct ownership/status')
    require(len(s)==len(logs)==1,'Exactly one committed settlement/log')
    require(int(s[0]['settlement_id'])==int(o['account_settlement_id']) and int(logs[0]['order_id'])==int(o['id']) and s[0]['biz_order_no']==o['order_no'],'Order/settlement/log identity chain')
    require(s[0]['settlement_no']=='CS:'+o['order_no'] and s[0]['biz_type']=='consume' and int(s[0]['status'])==20,'Original settlement key/status')
    require(s[0]['deduction_order_snapshot']==o['deduction_order_snapshot']=='["gift","subsidy","cash"]','Original cash/gift/subsidy order snapshot')
    money=financial['whole_declared_DECIMAL_columns']
    user=[r for r in money['AI.user']['rows'] if int(r['user_id'])==ids['AI_U1']][0]
    for key,v in [('balance_text','0.99'),('gift_balance_text','0.00'),('pay_money_text','0.01')]: require(decimal(user[key])==Decimal(v),'Exact final physical wallet CAST text')
    settlement=money['AI.account_settlement']['rows'][0]; log=money['AI.user_balance_log']['rows'][0]
    for key,v in [('amount_text','0.01'),('cash_amount_text','0.01'),('subsidy_amount_text','0.00'),('gift_amount_text','0.00')]: require(decimal(settlement[key])==Decimal(v),'Exact physical settlement money')
    require(decimal(log['money_text'])==Decimal('-0.01') and decimal(log['balance_after_text'])==Decimal('0.99'),'Exact complete balance-log money')
    order_money=[r for r in money['ZHCT.meal_order']['rows'] if int(r['id'])==int(o['id'])][0]
    for field in ['total_price','ori_total_price','order_price','pay_price','cash']:
        require(decimal(order_money[field+'_text'])==Decimal('0.01'),'Original order 1-cent amount')
    for field in ['gift','subsidy','remain_subsidy','remain_gift']: require(decimal(order_money[field+'_text'])==0,'No excluded funding')
    for field in ['remain_balance','remain_cash']: require(decimal(order_money[field+'_text'])==Decimal('0.99'),'Remaining 99 cents')
    for key in ['initial_menu','meal_order_pickup','staff_department','one_card_account_map','user_subsidy','user_gift_log']: require(own[key]==[],'Excluded own side-effect surface empty')
    committed=[e['connection'] for e in events if e['scenario']==success and e['layer']=='actual PDO.commit']
    require(committed==['ZHCT','AI'],'Original Store inner commit before independent AI outer commit; not XA')
    for wfile in p.glob('wire-*.json'):
        if wfile.name.startswith('wire-attempt-'): continue
        w=load(wfile)
        require(w['original_before']==w['original_after'],'Each captured HTTP original protection')
        for name,h in w['included_source_sha256'].items(): require(pins[name]==h,'Actual included source equals frozen current source')
    require(load(p/'worker-stopped.json')['port_closed'] is True,'Own worker stopped and exact port closed')
    require(load(p/'cleanup.json')['tokens']=={'U1':'revoked','U2':'revoked'},'Both original own tokens revoked')
    require(len(load(p/'cache-cleanup.json')['own_cache_files_removed'])==3,'Only own remaining cache files removed')
    # Whole response contract where actual success body exists; never fill C lost body.
    prelabel='PREPAY_FIRST_RETRY' if lane=='B' else 'PREPAY_FIRST_RETRY1'
    response=json.loads(load(p/f'wire-{prelabel}.json')['raw_response_body'])
    require(list(response)==['status','message','data','trace_id'] and response['status']==200 and response['message']=='订单创建成功','Complete actual prepay success envelope')
    require(list(response['data'])==['orderId','isPay'] and type(response['data']['orderId']) is int and response['data']['isPay']==1,'Complete original prepay data/types')
    repeat=json.loads(load(p/'wire-PREPAY_REPEAT.json')['raw_response_body'])
    require(response['data']==repeat['data'],'Repeat returns same actual prepay ID with zero DML')
    for label,message,status in [('PREPAY_OTHER','预扣款订单已被其他用户绑定',500),('PREPAY_ZERO','缺少预扣款订单参数',500),('DIRECT_ZERO','支付金额必须大于0',500),('DIRECT_INSUFFICIENT','余额不足，请充值后再尝试！',8001)]:
        w=load(p/f'wire-{label}.json'); r=json.loads(w['raw_response_body'])
        require(list(r)==['message','status','data','trace_id'] and r['message']==message and type(r['status']) is int and r['status']==status and r['data']==[],'Complete original negative envelope/types')
        require(w['own_before']==w['own_after'],'Negative own full business rows unchanged')
    if lane=='B':
        r=json.loads(load(p/f'wire-{success}.json')['raw_response_body']); dto=r['data']
        require(list(r)==['status','message','data','trace_id'] and r['status']==200 and r['message']=='支付成功','Actual complete B direct success envelope')
        keys=['orderId','order_no','pay_price','pay_time','restaurant_id','meal_date','meal_times','restaurant_method','pickup_no','remain_balance','cash','subsidy','gift','discount_amount','management_fee','marketing_strategy_id']
        require(list(dto)==keys,'All sixteen success fields and original order retained')
        for k in ['orderId','restaurant_id','meal_times','restaurant_method','marketing_strategy_id']: require(type(dto[k]) is int,'Exact response integer field')
        for k in set(keys)-{'orderId','restaurant_id','meal_times','restaurant_method','marketing_strategy_id'}: require(type(dto[k]) is str,'Exact response string field')
        require(dto['orderId']==int(o['id']) and dto['order_no']==o['order_no'] and dto['pay_price']==dto['cash']=='0.01' and dto['remain_balance']=='0.99','Success DTO actual physical linkage/money')
        require(dto['subsidy']==dto['gift']==dto['discount_amount']==dto['management_fee']=='0.00' and dto['pickup_no']=='' and dto['marketing_strategy_id']==0,'All default/zero data fields')
        require(load(p/'run-direct-last-complete.json')['results']['DIRECT_REPEAT'].get('NOT_RUN') is not None,'B missed repeat window not replayed')
    else:
        require(not (p/f'wire-{success}.json').exists() and not (p/f'response-before-observers-{success}.json').exists(),'C actual success DTO genuinely missing, not reconstructed')
        require(load(p/'C-insufficient-attempt.json')['repeat_NOT_RUN'] is not None,'C successful body not replayed after observer error')
    results[lane]={'HTTP':http_count,'business_native_INSERT_attempts':inserts,'business_native_UPDATE_attempts':updates,'exact_original_native_binding_pairs':pair_count,'fixture_explicit_fields':fields,'fixture_omitted_defaults':defaults,'direct_paid_order':int(o['id']),'scope_status':'Partial; C direct success DTO limited; repeat NOT_RUN'}

# In-memory outcome counterexamples; never edit/replay source evidence.
def ledger_guard(data):
    own=data['own_fullrows']; o=[x for x in own['meal_order'] if x['remark']=='CDXCORE11S1_DIRECT']
    assert len(o)==1 and int(o[0]['pay_status'])==20
    assert len(own['account_settlement'])==len(own['user_balance_log'])==1
    assert int(own['user_balance_log'][0]['order_id'])==int(o[0]['id'])
    assert own['account_settlement'][0]['deduction_order_snapshot']=='["gift","subsidy","cash"]'
    assert own['meal_order_pickup']==[]
base=load(RAW/'private-B-final/persistence-final.json'); ledger_guard(base); negatives=0
for field,bad in [('pay_status',10),('id',999999)]:
    x=copy.deepcopy(base); [r for r in x['own_fullrows']['meal_order'] if r['remark']=='CDXCORE11S1_DIRECT'][0][field]=bad
    try: ledger_guard(x)
    except AssertionError: negatives+=1
    else: raise AssertionError('Corrupt order accepted')
for mutate in [lambda x:x['own_fullrows']['account_settlement'].clear(),lambda x:x['own_fullrows']['user_balance_log'][0].update(order_id=999999),lambda x:x['own_fullrows']['account_settlement'][0].update(deduction_order_snapshot='["cash","gift","subsidy"]'),lambda x:x['own_fullrows']['meal_order_pickup'].append({'id':1})]:
    x=copy.deepcopy(base);mutate(x)
    try: ledger_guard(x)
    except AssertionError: negatives+=1
    else: raise AssertionError('Corrupted ledger/side-effect accepted')
require(negatives==6,'Six offline outcome counterexamples rejected')
print(json.dumps({'checks':checks,'outcome_counterexamples':negatives,'lanes':results,'SQL':0,'HTTP':0,'limitations':['C original 0.01 success DTO not captured','5-second direct repeat both NOT_RUN','B own prepay observed cron cancellation, exact historical/current fullrows preserved','Staff seven columns and settlement rules P0 remain open']},ensure_ascii=False,indent=2))