from pathlib import Path
import json,re,hashlib,datetime,collections,sys
from decimal import Decimal
H=Path(__file__).resolve().parent;R=Path('/tmp/cdx-ai-scale-owned-flow-20261011');N=0
def j(p):return json.loads(p.read_text())
def lines(p):return list(map(json.loads,p.read_text().splitlines()))
def ck(v,m):
 global N
 N+=1
 assert v,m
def mutable_removed(rows):
 rows=json.loads(json.dumps(rows))
 for r in rows['staff']:
  for k in ('weight','update_time'):r.pop(k)
 for r in rows['user_weight']:
  for k in ('weight','height','bmi','update_time'):r.pop(k)
 return rows
def physical(row,want,schema,lane,clock):
 cols={c['column_name'].lower():c for c in schema['columns']};ck(set(row)==set(cols),'Complete physical fullrow columns')
 for k,c in cols.items():
  v=row[k];typ=c['data_type'].lower()
  if k in want:
   x=want[k]
   if x=='@identity':ck(isinstance(v,int) and v>0,k+' real native identity')
   elif x=='@clock':
    if 'int'in typ:ck(isinstance(v,int) and clock[0]-2<=v<=clock[1]+2,k+' INT time')
    else:
     ck(isinstance(v,str) and re.fullmatch(r'2026-10-11 \d\d:\d\d:\d\d'+(r'\.\d{6}' if lane=='C' else ''),v)!=None,k+' native time precision')
     stamp=datetime.datetime.fromisoformat(v).replace(tzinfo=datetime.timezone(datetime.timedelta(hours=8))).timestamp();ck(clock[0]-2<=stamp<=clock[1]+2,k+' source current time')
   elif x=='@measured':ck(v==('2026-10-11 08:30:00.000000' if lane=='C' else '2026-10-11 08:30:00'),k+' source fixed measured date preserve precision')
   elif typ=='decimal':ck(Decimal(str(v))==Decimal(str(x)),k+' DEC value')
   else:ck(type(v)==type(x) and v==x,k+' literal PHP/native full value')
  else:
   d=c.get('column_default',c.get('data_default'))
   if d is None:ck(v is None,k+' NULL default')
   elif 'int'in typ:ck(isinstance(v,int) and v==int(d),k+' INT default')
   elif typ=='decimal':ck(Decimal(str(v))==Decimal(str(d)),k+' DEC default')
   else:ck(v==str(d).strip("'"),k+' string default')
summary={};cross={};responses={}
plan=j(H/'execution-plan.json')
for lane in ('B','C'):
 p=R/f'private-{lane}-final';a=j(p/'owned-actor.json');pre=j(p/'preflight-state.json');final=j(p/'final-capture.json');native=lines(p/'business-native.jsonl');budget=lines(p/'sql-budget.jsonl');schema=j(p/'store-schema.json');AI_schema=j(p/'schema.json');base=j(p/'scale-baseline.json')
 ck([x['n'] for x in budget]==list(range(1,len(budget)+1)) and len(budget)<=1000,'Unified monotonic1000');ck(final['SQL']==len(budget),'Exact allSQL total');ck(j(p/'preflight-ready.json')['SQL']<=180,'BeforeAuth source/constructor180')
 warm=j(p/'constructor-warm.json')['connections']
 for db,alias,port,dbschema in [('AI','mysql','5238','AIZHCT_APP'),('ZHCT','mysql_zhct','5237','ZHCT_APP')]:
  c=warm[db];ck(c['alias']==alias and c['PDO']=='OwnedPDO','Original App alias+nativePDO')
  if lane=='C':ck(c['driver']=='dm' and c['connector']=='CoreDm' and c['source_connector']=='app\\common\\database\\Dm' and c['host']=='127.0.0.1' and c['port']==port and c['actual_schema'][0]['schema_name']==dbschema,'True C both original Dm+dmPDO+exact live schema/port')
  else:ck(c['driver']=='mysql' and c['connector']=='CoreMysql' and c['actual_schema'][0]['version'].startswith('5.6.'),'True B two source MySQL56')
 ck(j(p/'run-complete.json')['actual_HTTP']==4 and len(list(p.glob('client-attempt-*.json')))==4 and len(list(p.glob('request-*.json')))==4,'Exactly4HTTP/no replay')
 ck(not list(p.glob('*error-*.json')) and not any('error'in x for x in native),'No swallowed native/observer/harness errors')
 ck(not (p/'prepare-attempt.json').exists(),'Zero prepare writes')
 allwrites=[x for x in native if x['layer']=='actual PDOStatement.execute' and x.get('operation')in('INSERT','UPDATE')];ck(dict(collections.Counter((x['connection'],x['operation']) for x in allwrites))=={('ZHCT','INSERT'):10,('AI','UPDATE'):4},'Exactly Store10I/AI4U')
 ck(all(x['success'] and x['rowCount']==1 for x in allwrites),'Each native DML affected1')
 prior_plans={}
 for x in native:
  key=(x['scenario'],x['connection'],x.get('queryString'))
  if x['layer']=='original framework bind plan':prior_plans[key]=x['bind_plan']
  elif x['layer']=='actual PDOStatement.execute':
   ck(key in prior_plans,'Native original bindplan paired');bp=prior_plans[key];ck(len(bp)==len(x['bindValue']),'Every native binding paired')
   for k,pair in (enumerate(bp) if isinstance(bp,list) else bp.items()):
    v,t=pair if isinstance(pair,list) else (pair,2)
    if t==21 and isinstance(v,str):v=float(v)
    if t==1 and v=='':v=0
    b=x['bindValue'][str(k+1) if isinstance(k,int) else ':'+k];ck(type(b['value'])==type(v) and b['value']==v and b['pdo_type']==(2 if t==21 else t) and b['ok'],'Raw original PHP/PDO binding incl actual FLOAT21 conversion')
 responses[lane]={}
 for n,label in [(1,'W1_WELLAND_ADD'),(2,'W2_YUNKANGBAO_ADD')]:
  w1=n==1;after=j(p/f'after-{label}.json');request=next(c for c in plan['cases'] if c['label']==label);form=json.loads(request['body_raw'])['form'];clockrecord=j(p/f'response-{label}.json');clock=(clockrecord['business_start'],clockrecord['business_end']);vendor='welland' if w1 else 'yunkangbao';scope=after['scale'];session=next(s for s in scope['sessions'] if s['vendor']==vendor);sid=session['session_id']
  ck(len(scope['sessions'])==n and len(scope['raw'])==n and len(scope['records'])==3*n,'Exact 5newrows per writer')
  ck(after['Store_committed'] and after['AI_committed'],'Both real commits observed')
  commits=[x for x in native if x['scenario']==label and x['layer']=='actual PDO.commit'];ck([x['connection'] for x in commits]==['ZHCT','AI'] and all(x['success'] for x in commits),'Original Store commit then AI commit nonXA')
  aiwrite=min(x['server_microtime'] for x in allwrites if x['scenario']==label and x['connection']=='AI');ck(aiwrite>commits[0]['server_microtime'],'AI DML only after actual Store commit')
  want={'session_id':'@identity','ai_user_id':a['user_id'],'event_type':'measurement','start_time':'@clock' if w1 else '@measured','end_time':'@clock' if w1 else '@measured','source_platform':'direct_device','vendor':vendor,'device_type':'body_scale','device_model':form['device_model'],'raw_hash':None,'business_context':'health','is_valid':1,'created_at':'@clock'}
  physical(session,want,schema['measurement_session'],lane,clock)
  raw=next(r for r in scope['raw'] if r['session_id']==sid);literal=j(p/'expected-literals.json')[label];ck(json.loads(literal['raw_json_bytes'])==form,'Original form types/values in expected raw');physical(raw,{'id':'@identity','session_id':sid,'data':literal['raw_json_bytes'],'created_at':'@clock'},schema['measurement_session_raw'],lane,clock)
  records=[r for r in scope['records'] if r['session_id']==sid];ck([r['metric_code'] for r in records]==['weight','height','bmi'],'Original metric loop/array order')
  for r in records:
   code=r['metric_code'];value=str(form[code]);want={'record_id':'@identity','session_id':sid,'ai_user_id':a['user_id'],'metric_code':code,'value':value,'unit':{'weight':'kg','height':'cm','bmi':None}[code],'segment':None,'segment_key':'','data_source_type':'algorithm','algorithm_vendor':vendor,'algorithm_version':'CDXSCALE11S1','source_platform':'direct_device' if w1 else 'vendor_api','start_time':'@clock' if w1 else '@measured','end_time':'@clock' if w1 else '@measured','created_at':'@clock'}
   physical(r,want,schema['health_record'],lane,clock)
   typ=next(c['data_type'].lower() for c in schema['health_record']['columns'] if c['column_name'].lower()=='value');ck(typ=='varchar' and isinstance(r['value'],str),'Actual health.value VARCHAR, not fake DEC')
  ai=after['AI'];ck(mutable_removed(ai)==mutable_removed(base),'Every original own9 except exact fields unchanged');weight=ai['user_weight'][0];ck(weight['id']==(670 if lane=='B' else 1988) and weight['weight']==str(form['weight']) and weight['height']=='181.25' and weight['bmi']==str(form['bmi']),'Original same-day VARCHAR update three values')
  ck(Decimal(str(ai['staff'][0]['weight']))==Decimal(str(form['weight'])) and Decimal(str(ai['staff'][0]['height']))==Decimal('180.25'),'Staff DEC weight only, height unchanged')
  ck(ai['user_health_metric_record']==base['user_health_metric_record'],'Manual AI health2 unchanged - no MetricProfileSync')
  ck(after['AI_decimal_text']['staff'][0]['weight_text']==str(form['weight']) and after['AI_decimal_text']['staff'][0]['height_text']=='180.25','Staff actual DB DEC CAST texts')
  texts={r['record_id']:r['value_text'] for r in after['Store_value_text']};ck(all(texts[r['record_id']]==r['value'] for r in scope['records']),'Actual VARCHAR value CAST bytes preserved')
 for case in plan['cases']:
  label=case['label'];r=j(p/f'response-{label}.json');client=j(p/f'client-{label}.json');d=json.loads(r['raw_response_body']);responses[lane][label]=d
  ck(r['http_status']==client['http_status']==200 and r['raw_response_body']==client['body_raw'],'Original full client/wire exact');ck(r['business_end']<=r['captured_microtime']<=client['end'],'Original immutable response before send')
  ck(list(d)==['status','message','data','trace_id'] and d['status']==200 and re.fullmatch('[0-9a-f]{13}',d['trace_id'])!=None,'Full envelope/type/order')
  if label.startswith('W'):ck(d['message']==case['expected_message'] and d['data']==[],'Source add message and array envelope')
  else:
   w1=label.startswith('R1');vendor='welland' if w1 else 'yunkangbao';snap=j(p/f'read-scale-{label}.json');s=next(s for s in snap['sessions'] if s['vendor']==vendor);form=j(p/'expected-literals.json')['W1_WELLAND_ADD' if w1 else 'W2_YUNKANGBAO_ADD']['form'];want={'session_id':s['session_id'],'measured_at':s['start_time'],'vendor':vendor,'device_type':'body_scale','algorithm_vendor':vendor,'algorithm_version':'CDXSCALE11S1','summary':{code:{'value':form[code],'unit':{'weight':'kg','height':'cm','bmi':None}[code]} for code in ['weight','height','bmi']},'segments':[],'raw':form,'bmi_cn':'标准'}
   ck(d['message']=='ok' and json.dumps(d['data'],ensure_ascii=False,separators=(',',':'))==json.dumps(want,ensure_ascii=False,separators=(',',':')),'Independent whole10fields/source typed nested DTO')
  ck(j(p/f'oracle-{label}.json')['full_typed_ordered_DTO'],'Frozen oracle complete');inc=j(p/f'included-{label}.json');pins=j(H/'source-frozen-sha.json')['files'];ck(all(pins.get(k)==v for k,v in inc['files'].items()),'All actual include source SHA')
  cap=(100 if lane=='B' else 180) if label.startswith('W') else (60 if lane=='B' else 120);ck(inc['native_count']<=cap,'Per request exact native cap');ck(j(p/f'request-{label}.json')['case']==case,'Literal request types/method/bytes')
 for name in ['serve','userapi']:
  captures=lines(p/f'{name}-captures.jsonl');ck(len(captures)==2 and all(x['owned_parameters_valid'] for x in captures),'Only two exact own boundary captures')
  ck(all(not x['real_queue_executed' if name=='serve' else 'real_http_executed'] for x in captures),'No real outside effects')
  if name=='userapi':ck(all(x['isolated_result']['is_success']==0 and x['isolated_result']['zhct_user_id']==0 for x in captures),'Explicit failure not fake Store sync success')
 ck(final['all_protection_match'] and final['protection']==pre['protection'] and final['unrelated_hashes']==pre['unrelated_hashes'],'All old/funds/ownprotected/disease and nonownStore3hash exact')
 ck(not any('ydy_prepay_order' in x['queryString'].lower() for x in lines(p/'independent-native.jsonl')),'Root-released Prepay raw not queried/blocked')
 ck(j(p/'cleanup.json')['tokens']['U1']=='revoked' and j(p/'worker-stopped.json')['port_closed'] and len(j(p/'cache-cleanup.json')['own_cache_files_removed'])==2,'Token1/cache2/PID/ports cleared')
 fs=j(p/'source-actual-fileset.json');ck(fs['actual_regular_files']==6168 and fs['appledouble_magic_count']==0 and set(fs['files'])==set(pins) and all(v['sha256']==pins[k] for k,v in fs['files'].items()),'Actual full source6168 allbytes noAppleDouble');ck(j(p/'helper-actual-fileset.json')['files']==j(H/'helper-sha.json'),'Actual all23helper set/SHA')
 summary[lane]={'HTTP':4,'prepare':0,'Store_INSERT':10,'AI_UPDATE':4,'DELETE':0,'DDL':0,'SQL':len(budget),'preflight_SQL':j(p/'preflight-ready.json')['SQL'],'session_ids':[s['session_id'] for s in final['scale']['sessions']],'new_Store_rows':10,'native_errors':0,'both_original_driver_proof':True,'source_DTO_pass':4,'old_protection':True,'resources_cleaned':True}
for label in ['R1_WELLAND_BY_DATE','R2_YUNKANGBAO_BY_DATE']:
 b=responses['B'][label]['data']['measured_at'];c=responses['C'][label]['data']['measured_at'];ck(re.fullmatch(r'2026-10-11 \d\d:\d\d:\d\d',b)!=None and re.fullmatch(r'2026-10-11 \d\d:\d\d:\d\d\.000000',c)!=None,'Observed cross-library native date shape difference retained')
 cross[label]={'B_measured_at':b,'C_measured_at':c,'cross_contract':'FAIL_DATE_STRING_SHAPE_SECONDS_VS_6FRACTION','source_per_side':'PASS','normalization_applied':False}
ck(cross['R2_YUNKANGBAO_BY_DATE']['B_measured_at']=='2026-10-11 08:30:00' and cross['R2_YUNKANGBAO_BY_DATE']['C_measured_at']=='2026-10-11 08:30:00.000000','Same fixed input genuine date contract red, not natural sample drift')
out={'status':'PARTIAL_REAL_DB_WRITES_PASS_TWO_READER_DATE_SHAPE_RED','checks':N,'lanes':summary,'cross_date_contract':cross,'scope':'Explicit UserApi failure/Serve capture only; true Store user/sync still separately pending; VARCHAR health values not DEC, CLOB raw genuine full bytes'}
if '--record'in sys.argv:
 ck(not (H.parent/'manifest.json').exists(),'No sealed overwrite');(H/'audit-result.json').write_text(json.dumps(out,ensure_ascii=False,indent=2)+'\n')
print(json.dumps(out,ensure_ascii=False))