from pathlib import Path
import json,hashlib,re,datetime,collections,sys
from decimal import Decimal,ROUND_HALF_UP
H=Path(__file__).resolve().parent;R=Path('/tmp/cdx-ai-weight-owned-flow-20261011');N=0
def check(x,label):
 global N
 N+=1
 assert x,label
def j(p):return json.loads(p.read_text())
def lines(p):return [json.loads(x) for x in p.read_text().splitlines()]
def physical(row,expected,schema,lane,clockrange):
 cols={c['column_name'].lower():c for c in schema['columns']}
 check(set(row)==set(cols),'full physical column set')
 for k,c in cols.items():
  v=row[k];typ=c['data_type'].lower()
  if k in expected:
   want=expected[k]
   if want is None:check(v is None,k+' explicit native NULL')
   elif want=='@identity':check(isinstance(v,int) and v>0,k+' generated')
   elif want=='@time':
    if 'int' in typ:check(isinstance(v,int) and clockrange[0]-2<=v<=clockrange[1]+2,k+' int time')
    else:
     check(isinstance(v,str) and re.fullmatch(r'\d{4}-\d\d-\d\d \d\d:\d\d:\d\d'+(r'\.\d{6}' if lane=='C' else ''),v)!=None,k+' physical precision')
     ts=datetime.datetime.fromisoformat(v).replace(tzinfo=datetime.timezone(datetime.timedelta(hours=8))).timestamp();check(clockrange[0]-2<=ts<=clockrange[1]+2,k+' source clock')
   elif typ=='decimal':check(Decimal(str(v))==Decimal(str(want)),k+' DEC value')
   else:check(type(v)==type(want) and v==want,k+' literal type/value')
  else:
   default=c.get('column_default',c.get('data_default'))
   if default is None:check(v is None,k+' NULL default')
   elif str(default).upper()=='CURRENT_TIMESTAMP':physical({k:v},{k:'@time'},{'columns':[c]},lane,clockrange)
   elif 'int' in typ:check(isinstance(v,int) and v==int(default),k+' INT default')
   elif typ=='decimal':check(Decimal(str(v))==Decimal(str(default)),k+' DEC default')
   else:check(v==str(default).strip("'"),k+' string default')
summary={}
for lane in ('B','C'):
 p=R/f'private-{lane}-final';s=j(p/'schema.json');a=j(p/'owned-actor.json');pre=j(p/'prepared-owned-rows.json');final=j(p/'final-capture.json');rows=final['rows'];w1=j(p/'after-W1_MANUAL.json');w2=j(p/'after-W2_GOAL.json');native=lines(p/'business-native.jsonl');budget=lines(p/'sql-budget.jsonl')
 check([x['n'] for x in budget]==list(range(1,len(budget)+1)),'Unified monotonic no reset budget');check(len(budget)<=1000,'Unified cap')
 before_prepare=next(x['n'] for x in budget if x['kind']=='prepare.INSERT')-1
 check(final['SQL']==len(budget),'Final SQL exact');check(before_prepare<=180,'Actual all failed/recovered preflight and prewarm before prepare cap')
 check(j(p/'run-complete.json')['actual_HTTP']==8,'Exactly8HTTP');check(len(list(p.glob('request-*.json')))==8 and len(list(p.glob('client-attempt-*.json')))==8,'No replay')
 check(not list(p.glob('*error-*.json')) and not any('error' in x for x in native),'No swallowed original DB/gate error')
 plans={}
 for event in native:
  key=(event['scenario'],event['connection'],event.get('queryString'))
  if event['layer']=='original framework bind plan':plans[key]=event['bind_plan']
  elif event['layer']=='actual PDOStatement.execute':
   check(key in plans,'Original framework plan precedes actual execute');plan=plans[key];actual=event['bindValue'];check(len(plan)==len(actual),'Every native bind covered')
   for name,pair in (enumerate(plan) if isinstance(plan,list) else plan.items()):
    nativekey=str(name+1) if isinstance(name,int) else ':'+name;b=actual[nativekey]
    value,kind=pair if isinstance(pair,list) else (pair,2)
    if kind==21:value=float(value)
    check(b['pdo_type']==(2 if kind==21 else kind) and type(b['value'])==type(value) and b['value']==value and b['ok'],'Exact original PHP value/type and native PDO flag including FLOAT21→2')
 prep=lines(p/'prepare-native.jsonl');check([x.get('operation') for x in prep if 'operation'in x]==['begin','commit'],'Real preparation transaction');check(len([x for x in prep if 'queryString'in x])==2 and all(x['affected']==1 for x in prep if 'queryString'in x),'Exactly2prepare I')
 for label,expected in [('W1_MANUAL',{'INSERT':6,'UPDATE':1}),('W2_GOAL',{'INSERT':1,'UPDATE':2})]:
  actual=[x for x in native if x['scenario']==label and x['layer']=='actual PDOStatement.execute' and x.get('operation') in ('INSERT','UPDATE')]
  check(dict(collections.Counter(x['operation'] for x in actual))==expected,label+' formal counts');check(all(x['rowCount']==1 and x['success'] for x in actual),label+' every affected1')
  tx=[x for x in native if x['scenario']==label and x['layer'].startswith('actual PDO.') and 'Transaction' in x['layer'] or x['scenario']==label and x['layer']=='actual PDO.commit']
  check(any(x['layer']=='actual PDO.commit' and x['success'] for x in tx),label+' commit')
 w1commit=next(x for x in native if x['scenario']=='W1_MANUAL' and x['layer']=='actual PDO.commit')['server_microtime']
 healthsql=[x for x in native if x['scenario']=='W1_MANUAL' and x['layer']=='actual PDOStatement.execute' and x.get('operation')=='INSERT' and 'user_health_metric_record' in x['queryString']]
 check(len(healthsql)==2 and all(x['server_microtime']>w1commit for x in healthsql),'Two real postcommit health I')
 check(w1['rows']['user_health_metric_record']==w2['rows']['user_health_metric_record']==rows['user_health_metric_record'],'W2 health not resynced / full rows unchanged')
 check(pre['rows']['user']==w1['rows']['user']==w2['rows']['user']==rows['user'],'Owned user never mutated')
 check(final['unrelated_hashes_match'] and final['old_protection_match'],'Old13/funds/prepay/unrelated hash')
 check(final['unrelated_fulltable_hashes']==j(p/'preflight-state.json')['unrelated_fulltable_hashes'],'Each9table unrelated hash exact')
 check(sum(map(len,rows.values()))==9 and all(len(rows[t])==(0 if t=='user_course' else 2 if t=='user_health_metric_record' else 1) for t in rows),'Exactly9marked rows/course0')
 at=datetime.datetime.fromisoformat(j(p/'prepare-attempt.json')['at']).timestamp();end=max(j(x)['business_end'] for x in p.glob('response-*.json'));rng=(at,end)
 base={'user_id':a['user_id'],'store_id':10001,'nick_name':a['marker'],'mobile':'00000009992','gender':1,'avatar_id':0,'platform':'api','business':'zhct','channel':'10000','is_delete':0,'balance':'0.00','gift_balance':'0.00','pay_money':'0.00'}
 physical(rows['user'][0],base,s['user'],lane,rng)
 staff={'id':a['staff_id'],'user_id':a['user_id'],'store_id':10001,'name':a['marker'],'phone':'00000009992','gender':1,'birthday':'2000-01-01','height':'180.25','weight':'72.50','is_delete':0,'create_by':'CDXWGT11','create_time':'@time','update_time':'@time'}
 physical(rows['staff'][0],staff,s['staff'],lane,rng)
 for stage,weight,height,bmi in [(w1,'72.34','181.25','22.02'),(w2,'72.5','180.25','22.31')]:
  physical(stage['rows']['user_weight'][0],{'id':'@identity','user_id':a['user_id'],'store_id':10001,'date':1791648000,'weight':weight,'height':height,'bmi':bmi,'create_time':'@time','update_time':'@time'},s['user_weight'],lane,rng)
  check((Decimal(weight)/(Decimal(height)/100)**2).quantize(Decimal('.01'),ROUND_HALF_UP)==Decimal(bmi),'Independent exactdecimal BMI vector')
 physical(rows['user_weight_status'][0],{'id':'@identity','user_id':a['user_id'],'store_id':10001,'start_date':1791648000,'initial_weight':'72.5','target_weight':'71.50','adjust_speed':2,'status':3,'end_date':1792771200,'create_time':'@time','update_time':'@time'},s['user_weight_status'],lane,rng)
 physical(rows['user_bind'][0],{'bind_id':'@identity','business':'zhct','zhct_user_id':0,'user_id':a['user_id'],'store_id':10001,'create_time':'@time','update_time':'@time'},s['user_bind'],lane,rng)
 for t in ('user_score','user_score_log'):
  expected={'id':'@identity','user_id':a['user_id'],'store_id':10001,'type':1,'score':1,'create_time':'@time','update_time':'@time'}
  if t.endswith('log'):expected.update(meal_times=0,course='',date='2026-10-11')
  physical(rows[t][0],expected,s[t],lane,rng)
 for h in rows['user_health_metric_record']:
  isw=h['metric_code']=='weight';expected={'id':'@identity','user_id':a['user_id'],'store_id':10001,'metric_code':'weight' if isw else 'bmi','value':'72.34' if isw else '22.02','secondary_value':None,'unit':'kg' if isw else '','measure_date':'2026-10-11','measure_time':h['measure_time'],'measure_site':'','source_type':1,'source_key':'user-weight:'+str(rows['user_weight'][0]['id']),'evaluation_code':'recorded' if isw else 'normal','evaluation_name':'已记录' if isw else '正常','evaluation_tone':'recorded' if isw else 'normal','disease_type':0,'rule_version':'v1','is_delete':0,'create_time':'@time','update_time':'@time'}
  physical(h,expected,s['user_health_metric_record'],lane,rng);check(re.fullmatch(r'\d\d:\d\d:\d\d',h['measure_time'])!=None,'Actual measure time format')
 casts=final['decimal_text'];check(casts['staff'][0]['weight_text']=='72.50' and casts['staff'][0]['height_text']=='180.25','Actual DEC scale');check([x['value_text'] for x in casts['user_health_metric_record']]==['72.3400','22.0200'],'Actual health DEC4 scale')
 for f in p.glob('response-*.json'):
  label=f.stem[9:];r=j(f);client=j(p/f'client-{label}.json');o=j(p/f'oracle-{label}.json');d=json.loads(r['raw_response_body'])
  check(r['http_status']==client['http_status']==200 and r['raw_response_body']==client['body_raw'],'Exact saved wire client bytes/status')
  check(r['business_end']<=r['captured_microtime']<=client['end'],'Save original response before returned wire')
  check(list(d)==['status','message','data','trace_id'] and d['status']==200 and d['message']=='success' and re.fullmatch('[0-9a-f]{13}',d['trace_id'])!=None,'Full envelope')
  check(d['data']==o['expected_data'] and o['full_typed_ordered_DTO'],'Full independently projected DTO')
  check(json.dumps(d['data'],ensure_ascii=False,separators=(',',':'))==json.dumps(o['expected_data'],ensure_ascii=False,separators=(',',':')),'Nested key order/type JSON exact')
  inc=j(p/f'included-{label}.json');pins=j(H/'source-frozen-sha.json')['files'];check(all(pins.get(k)==v for k,v in inc['files'].items()),'All actual included source pins')
  cap=(100 if lane=='B' else 180) if label.startswith('W') else (50 if lane=='B' else 90);check(inc['native_count']<=cap,'Per request native cap')
 for name in ('serve','userapi'):
  cap=lines(p/f'{name}-captures.jsonl');check(len(cap)==2 and [x['scenario'] for x in cap]==['W1_MANUAL','W2_GOAL'] and all(x['owned_parameters_valid'] for x in cap),'Exact two boundary captures '+name)
  check(all(not x['real_queue_executed' if name=='serve' else 'real_http_executed'] for x in cap),'No true boundary side effect '+name)
  if name=='userapi':check(all(x['isolated_result']['is_success']==0 and x['isolated_result']['zhct_user_id']==0 for x in cap),'Explicit failure no fake Store success')
 check(j(p/'cleanup.json')['tokens']['U1']=='revoked' and j(p/'worker-stopped.json')['port_closed'],'One token revoked/worker port stopped');check(len(j(p/'cache-cleanup.json')['own_cache_files_removed'])==2,'Only own2cache cleaned')
 source=j(p/'source-actual-fileset.json');check(source['actual_regular_files']==6168 and source['appledouble_magic_count']==0,'Actual6168fileset magic0');check(set(source['files'])==set(pins),'Whole source fileset identical');check(all(v['sha256']==pins[k] for k,v in source['files'].items()),'Whole source6168SHAs')
 check(j(p/'helper-actual-fileset.json')['files']==j(H/'helper-final-sha.json'),'Actual whole23helper set/SHA')
 summary[lane]={'actor':a,'HTTP':8,'prepare_INSERT':2,'formal_INSERT':7,'formal_UPDATE':3,'total_SQL':len(budget),'preflight_SQL':before_prepare,'marked_rows':9,'DTO_full_pass':8,'native_errors':0,'old_and_unrelated_protection':True,'resources_cleaned':True}
check(hashlib.sha256((H.parent/'machine-plan.json').read_bytes()).hexdigest()=='b525956eb8ae884230c984082e20c87d67e1f1da244a688d89b553044e192d29','Approved source plan immutable')
out={'status':'PASS_AI_LOCAL_OWNED_WEIGHT_NONEMPTY_STORE_LOOPBACK_PENDING','checks':N,'lanes':summary,'limits':['UserApi failure isolated only; real Store /api/user/sync positive remains pending','Health API does not support weight/bmi; sidecar native verification passed','B initial preflight memory failure retained; actual SQL counted; no HTTP/DML at failure','Known physical DEC retrieval/type and DATETIME0/6 differences retained']}
if '--record' in sys.argv:
 check(not (H/'manifest.json').exists() and not (H.parent/'manifest.json').exists(),'No write to sealed evidence')
 (H/'audit-result.json').write_text(json.dumps(out,ensure_ascii=False,indent=2)+'\n')
print(json.dumps(out,ensure_ascii=False))