#!/usr/bin/env python3
"""Independent journal verifier for approved half ties. No DB, network or writes."""
import copy,hashlib,json
from pathlib import Path
R=Path(__file__).resolve().parent
PB=(R.parent/"primary-source-derivation-20261011/half-tie-probe-proposal.json").read_bytes()
assert hashlib.sha256(PB).hexdigest()=="169ada31e30e80ab5dea1059f8de512c47cff634df14ed5194f999a0cf7b78c3"
P=json.loads(PB);N=0
def ck(ok,msg):
 global N
 N+=1
 if not ok:raise AssertionError(msg)
def ev(es,name):return [e["data"] for e in es if e["event"]==name]
def load(lane,sha):
 raw=(R/"private-evidence"/(lane+"-run.jsonl")).read_bytes()
 ck(hashlib.sha256(raw).hexdigest()==sha,"Raw unchanged "+lane)
 es=[json.loads(x) for x in raw.splitlines()]
 ck([e["seq"] for e in es]==list(range(1,len(es)+1)),"Complete event sequence")
 return es
B=load("B","ef1f7d1b0b3eacd4390209b14c977e169356ec7b10315ec4a85822454800c098")
C=load("C","d88c49742383de7b62546aaf5189ede4b1099d7c715a4c9224e76acba187a680")
def verify(es,lane):
 s=P["lanes"][lane]
 expected=[(c["case_id"],c["DML"]) for c in s["point_cases"]]+[(c["case_id"],d) for c in s["aggregate_cases"] for d in c["DML"]]
 cases=[c["case_id"] for c in s["point_cases"]+s["aggregate_cases"]]
 ck([d["case_id"] for d in ev(es,"case-start")]==cases,"All52 ordered cases")
 actual=ev(es,"DML-start");ck(len(actual)==56,"Exactly56 INSERT")
 for a,(case,d) in zip(actual,expected):
  ck(a["case_id"]==case and a["sql"]==d["sql"],"Every exact original INSERT SQL/case/order")
  ck([v["original"] for v in a["parameters"]]==d["bindings"],"Every original bind/type/lexeme unchanged")
  ck(a["original_nutrient_values"]==d["original_nutrient_values"],"Original values")
  for v in a["parameters"]:
   o=v["original"];ck(v["runtime"]["value"]==o["original_value"] and v["runtime"]["php_type"]==o["php_type"] and v["pdo_type_code"]==o["PDO_type_code"],"Runtime input never converted/rounded")
   ck(not isinstance(o["original_value"],str) or o["original_value"].isascii(),"ASCII string bound")
 ck(len(ev(es,"DML-success"))==56 and not ev(es,"DML-failure"),"Actual INSERT outcomes")
 ck(len(ev(es,"rollback-success"))==52 and [x["case_id"] for x in ev(es,"case-complete")]==cases,"52 rollback/completions")
 ck([x["sql"] for x in ev(es,"DDL-success")]==[s["templates"]["create"],s["templates"]["drop"]],"Single original CREATE and DROP")
 ck(not ev(es,"read-failure") and not ev(es,"prepare-only-failure") and not ev(es,"stopped"),"No hidden SQL/tool failure")
 prepares=ev(es,"prepare-only-start")
 ck(len(prepares)==(9 if lane=="B" else 8) and len(ev(es,"prepare-only-success"))==len(prepares),"Read templates compile before INSERT")
 first_dml=next(i for i,e in enumerate(es) if e["event"]=="DML-start")
 ck(all(i<first_dml for i,e in enumerate(es) if e["event"]=="prepare-only-success"),"Prepares precede all DML")
 ck(ev(es,"cleanup-complete")==[{"fresh_object_count":0}],"Fresh independent object0")
 own=ev(es,"cleanup-ownership-gate")
 ck(len(own)==1 and own[0]["independent_zero"] and own[0]["actual_names_pinned"],"Independent0+ownership/fingerprint")
 # Compare all actual natural arrays, not the generated boolean alone.
 sqls=[q["sql"] for q in s["natural_queries"]]
 rows=[x["rows"] for x in ev(es,"read-success") if x["sql"] in sqls]
 ck(len(rows)==20 and rows[:10]==rows[10:],"All natural source/metadata arrays equal")
 ck(ev(es,"natural-before")[0]["sha256"]==ev(es,"natural-after")[0]["sha256"],"Actual natural digest same")
 # Actual own schema at CREATE plus both cleanup connections must match.
 meta_sqls=[q["sql"] for q in s["schema_queries"]];active=False;current=[];cycles=[]
 for e in es:
  if e["event"]=="DDL-success":active=e["data"]["sql"].startswith("CREATE ")
  if active and e["event"]=="read-success":
   d=e["data"]
   if d["sql"]==meta_sqls[0]:current=[d["rows"]]
   elif current and len(current)<len(meta_sqls) and d["sql"]==meta_sqls[len(current)]:current.append(d["rows"])
   if len(current)==len(meta_sqls):cycles.append(current);current=[]
 ck(len(cycles)==3 and cycles[0]==cycles[1]==cycles[2],"Full raw definition/PK/index/FK/trigger fingerprint stable")
 ck(cycles[0]==ev(es,"created-schema-gate")[0]["metadata"],"Explicit original CREATE metadata same")
 zero=[x for x in ev(es,"read-success") if x["sql"]==s["templates"]["count"]]
 ck(len(zero)==54 and all(x["rows"][0]["n"]==0 for x in zero),"Every rollback+both cleanup counts0")
 native=[x for x in ev(es,"read-success") if x["sql"]==s["templates"]["native_point"]]
 pair=[x for x in ev(es,"read-success") if x["sql"]==s["templates"]["native_rows"]]
 ck(len(native)==48 and all(len(x["rows"])==1 and x["rows"][0]["id"]==1 and x["rows"][0]["marker"]==s["marker"] for x in native),"Own48 one-row observations")
 ck(len(pair)==4 and all([r["id"] for r in x["rows"]]==[1,2] and all(r["marker"]==s["marker"] for r in x["rows"]) for x in pair),"Own4 pair observations, max2")
 aggregates=[x for x in ev(es,"read-success") if x["sql"]==s["templates"]["aggregate_native"]]
 ck(len(aggregates)==4 and all(x["rows"][0]["n"]==2 for x in aggregates),"All four real nonempty aggregates")
 total=ev(es,"complete");ck(len(total)==1,"One completion");t=total[0]
 ck((t["cases"],t["DDL"],t["INSERT_attempts"],t["rollbacks"])==(52,2,56,52),"Exact final mutation/rollback counts")
 probe=sum(x["data"]["bucket"]=="probe" for x in es if x["event"]=="read-start")+len(prepares)
 natural=sum(x["data"]["bucket"]=="natural" for x in es if x["event"]=="read-start")
 ck((probe,natural)==(t["probe_queries"],t["natural_queries"]) and probe<=360 and natural<=24,"Independent total attempt budgets")
 if lane=="B":
  cfg=ev(es,"B-effective-config")[0];ck(cfg=={"real_loader_constants":True,"effective_charset":"utf8","effective_port":3306,"params_keys":[],"has_initSQL":False,"four_keys_match_local":True,"local_has_charset":False,"local_has_hostport":False},"Actual effective App config, never direct local")
  identity=next(x["rows"][0] for x in ev(es,"read-success") if x["sql"].startswith("SELECT VERSION()"))
  ck(identity["cs_client"]==identity["cs_connection"]=="utf8","Actual session UTF8 from effective config")
  w=ev(es,"warnings");wr=[x["rows"] for x in ev(es,"read-success") if x["sql"]=="SHOW WARNINGS"]
  ck(len(w)==len(wr)==56 and [x["rows"] for x in w]==wr,"Every original warning record retained")
 else:ck(not ev(es,"warnings") and t["C_warning_status"]=="unknown","C warning API unknown not zero")
verify(B,"B");verify(C,"C");baseline=N
tests=[
 ("round neighboring input",lambda es:ev(es,"DML-start")[8]["parameters"][2]["original"].update(original_value="0.005")),
 ("convert host type",lambda es:ev(es,"DML-start")[0]["parameters"][2]["runtime"].update(php_type="double")),
 ("wrong PDO type",lambda es:ev(es,"DML-start")[0]["parameters"][2].update(pdo_type_code=1)),
 ("business target",lambda es:ev(es,"DML-start")[0].update(sql="INSERT INTO ydy_nutrition VALUES(?,?,?,?)")),
 ("case omitted",lambda es:es.remove(next(e for e in es if e["event"]=="case-start"))),
 ("rollback omitted",lambda es:es.remove(next(e for e in es if e["event"]=="rollback-success"))),
 ("warning hidden",lambda es:es.remove(next(e for e in es if e["event"]=="warnings"))),
 ("false utf8mb4",lambda es:ev(es,"B-effective-config")[0].update(effective_charset="utf8mb4")),
 ("cleanup missing",lambda es:es.remove(next(e for e in es if e["event"]=="cleanup-complete"))),
 ("bad aggregate count",lambda es:next(x["rows"][0] for x in ev(es,"read-success") if x["sql"]==P["lanes"]["B"]["templates"]["aggregate_native"]).update(n=1)),
]
rejected=[]
for name,mutate in tests:
 es=copy.deepcopy(B);mutate(es)
 try:verify(es,"B")
 except (AssertionError,KeyError):rejected.append(name)
 else:raise AssertionError("Mutation survived "+name)
print(json.dumps({"status":"ACTUAL_HALF_TIE_EVIDENCE_VERIFIED_NOT_A_FIX","baseline_assertions":baseline,
 "negative_mutations_rejected":rejected,"B":ev(B,"complete")[0],"C":ev(C,"complete")[0],
 "actual_effective_B_UTF8_verified":True,"all_original56_bindings_preserved_each_lane":True,
 "new_verifier_DB_operations":0},ensure_ascii=False,indent=2))