#!/usr/bin/env python3
from pathlib import Path
import os,sys,subprocess,uuid,json,hashlib,shutil,time,re
ROOT=Path(__file__).resolve().parents[1]; SQL=ROOT/'sql'; EVID=ROOT/'evidence'/'runtime'; STATUS=ROOT/'evidence'/'runtime-status.json'
dsn=os.environ.get('DAT005_DATABASE_URL',''); confirm=os.environ.get('DAT005_CONFIRM_DISPOSABLE','')

def write_status(obj): STATUS.write_text(json.dumps(obj,indent=2,sort_keys=True)+'\n',encoding='utf-8')
def sha(path): return hashlib.sha256(path.read_bytes()).hexdigest()
if confirm!='DAT005_LOCAL_DISPOSABLE':
    write_status({'schema':'DAT005_RUNTIME_STATUS_v2','status':'NOT_RUN_CONFIRMATION_REQUIRED','actual_postgresql_runtime_evidence_claimed':False,'fabricated_runtime_evidence':False}); print('Refusing: set DAT005_CONFIRM_DISPOSABLE=DAT005_LOCAL_DISPOSABLE',file=sys.stderr); sys.exit(2)
if not dsn or shutil.which('psql') is None:
    write_status({'schema':'DAT005_RUNTIME_STATUS_v2','status':'NOT_RUN_MISSING_REAL_POSTGRESQL_RUNTIME','actual_postgresql_runtime_evidence_claimed':False,'fabricated_runtime_evidence':False}); print('Real PostgreSQL/psql unavailable.',file=sys.stderr); sys.exit(3)
if EVID.exists(): shutil.rmtree(EVID)
EVID.mkdir(parents=True,exist_ok=True)
schema='maknow_dat005_'+uuid.uuid4().hex[:12]
base=['psql',dsn,'-X','-v','ON_ERROR_STOP=1','-v','VERBOSITY=verbose','-q','-A','-t','-F','|']
input_files=['schema.sql','seed.sql','baseline_old_reader.sql','baseline_old_writer.sql','expand_add_phone_e164.sql','early_not_null_failure.sql','backfill.sql','verify_backfill.sql','new_reader.sql','dual_writer.sql','resolve_target_null_fixture.sql','enforce_not_null.sql','old_writer_after_enforce.sql','drop_legacy_bad.sql','contract_remove_legacy.sql','lock_holder.sql','alter_lock_contender.sql']

def env_schema():
    e=os.environ.copy(); e['PGOPTIONS']=f'-c search_path={schema},pg_catalog -c statement_timeout=10000'; return e

def call(args,env=None,timeout=20,input_text=None): return subprocess.run(args,text=True,capture_output=True,env=env,timeout=timeout,input=input_text)
def admin(sql):
    p=call(base+['-c',sql]);
    if p.returncode: raise RuntimeError(p.stderr.strip() or f'admin rc={p.returncode}')
    return p.stdout.strip()
def query(sql):
    p=call(base+['-c',sql],env=env_schema());
    if p.returncode: raise RuntimeError(p.stderr.strip() or f'query rc={p.returncode}')
    return p.stdout.strip()
def nonempty(s):return [x.strip() for x in s.splitlines() if x.strip()]
def write_probe(tag,p,elapsed):
    (EVID/f'{tag}.stdout.txt').write_text(p.stdout,encoding='utf-8'); (EVID/f'{tag}.stderr.txt').write_text(p.stderr,encoding='utf-8'); (EVID/f'{tag}.returncode.txt').write_text(str(p.returncode)+'\n'); (EVID/f'{tag}.elapsed_seconds.txt').write_text(f'{elapsed:.6f}\n')
def run_file(name,tag,expect_ok=True,timeout=20):
    st=time.monotonic(); p=call(base+['-f',str(SQL/name)],env=env_schema(),timeout=timeout); elapsed=time.monotonic()-st; write_probe(tag,p,elapsed)
    ok=(p.returncode==0)==expect_ok
    return {'returncode':p.returncode,'stdout':p.stdout,'stderr':p.stderr,'elapsed_seconds':elapsed,'pass':ok}
def run_text(text,tag,expect_ok=True,timeout=20):
    st=time.monotonic(); p=call(base,env=env_schema(),timeout=timeout,input_text=text); elapsed=time.monotonic()-st; write_probe(tag,p,elapsed)
    ok=(p.returncode==0)==expect_ok
    return {'returncode':p.returncode,'stdout':p.stdout,'stderr':p.stderr,'elapsed_seconds':elapsed,'pass':ok}
def pair(holder,contender,tag,delay=.35):
    ea=env_schema(); eb=env_schema(); start=time.monotonic(); hp=subprocess.Popen(base+['-f',str(SQL/holder)],stdout=subprocess.PIPE,stderr=subprocess.PIPE,text=True,env=ea); time.sleep(delay); cstart=time.monotonic(); cp=subprocess.Popen(base+['-f',str(SQL/contender)],stdout=subprocess.PIPE,stderr=subprocess.PIPE,text=True,env=eb)
    try:
        co,ce=cp.communicate(timeout=8); ho,he=hp.communicate(timeout=8)
    except subprocess.TimeoutExpired:
        for proc in (hp,cp):
            if proc.poll() is None: proc.kill()
        co,ce=cp.communicate(); ho,he=hp.communicate(); raise RuntimeError(f'{tag} timeout; child processes killed')
    c_elapsed=time.monotonic()-cstart; total=time.monotonic()-start
    for n,v in [('holder.stdout',ho),('holder.stderr',he),('contender.stdout',co),('contender.stderr',ce)]: (EVID/f'{tag}.{n}.txt').write_text(v)
    (EVID/f'{tag}.meta.json').write_text(json.dumps({'holder_rc':hp.returncode,'contender_rc':cp.returncode,'contender_elapsed_seconds':c_elapsed,'total_elapsed_seconds':total},indent=2)+'\n')
    return {'holder_rc':hp.returncode,'contender_rc':cp.returncode,'holder_stdout':ho,'holder_stderr':he,'contender_stdout':co,'contender_stderr':ce,'contender_elapsed_seconds':c_elapsed}

summary={'schema':'DAT005_RUNTIME_STATUS_v2','status':'RUNNING','generated_schema':schema,'actual_postgresql_runtime_evidence_claimed':False,'fabricated_runtime_evidence':False,'input_sha256':{n:sha(SQL/n) for n in input_files},'model_contract_sha256':sha(ROOT/'model-contract.json'),'cleanup_status':'NOT_ATTEMPTED','gates':{},'probes':{}}
write_status(summary)
created=False; run_error=None
try:
    admin(f'CREATE SCHEMA {schema}'); created=True
    vn=int(admin('SHOW server_version_num')); summary['postgresql_version_num']=vn; summary['gates']['postgresql_major_18']=180000<=vn<190000
    if not summary['gates']['postgresql_major_18']: raise RuntimeError(f'PostgreSQL 18 required, got {vn}')
    for f,t in [('schema.sql','setup_schema'),('seed.sql','setup_seed')]:
        r=run_file(f,t); assert r['pass'],f
    brr=run_file('baseline_old_reader.sql','baseline_old_reader'); brw=run_file('baseline_old_writer.sql','baseline_old_writer')
    summary['gates']['baseline_old_contract']=brr['pass'] and brw['pass'] and query("SELECT count(*) FROM customer_contact WHERE customer_id=10 AND phone_legacy='555-000-0010'")=='1'
    if not summary['gates']['baseline_old_contract']: raise RuntimeError('baseline old contract not observed')
    ex=run_file('expand_add_phone_e164.sql','expand'); summary['gates']['expand_applied']=ex['pass'] and query("SELECT count(*) FROM information_schema.columns WHERE table_schema=current_schema() AND table_name='customer_contact' AND column_name='phone_e164'")=='1'
    if not summary['gates']['expand_applied']: raise RuntimeError('expand not observed')
    early=run_file('early_not_null_failure.sql','early_not_null',expect_ok=False); eall=early['stderr']+early['stdout']; summary['gates']['early_not_null_expected_failure_23502']=early['pass'] and '23502' in eall
    if not summary['gates']['early_not_null_expected_failure_23502']: raise RuntimeError('early NOT NULL did not fail with 23502')
    bf=run_file('backfill.sql','backfill'); ver=run_file('verify_backfill.sql','verify_before_resolution'); rows=nonempty(ver['stdout']); expected='5|4|0|1|0'; summary['probes']['verify_before_resolution']={'rows':rows,'expected':expected}; summary['gates']['target_null_rows_before_resolution']=bf['pass'] and ver['pass'] and rows==[expected]
    if not summary['gates']['target_null_rows_before_resolution']: raise RuntimeError('backfill readiness counts do not match target invariant')
    nr=run_file('new_reader.sql','new_reader'); dw=run_file('dual_writer.sql','dual_writer'); summary['gates']['overlap_old_new_contract']=nr['pass'] and dw['pass'] and query("SELECT phone_legacy||'|'||phone_e164 FROM customer_contact WHERE customer_id=11")=='555-000-0011|+15550000011'
    if not summary['gates']['overlap_old_new_contract']: raise RuntimeError('old/new overlap evidence failed')
    rr=run_file('resolve_target_null_fixture.sql','resolve_target_null_fixture'); ver2=run_file('verify_backfill.sql','verify_after_resolution'); rows2=nonempty(ver2['stdout']); expected2='6|5|0|0|0'; summary['probes']['verify_after_resolution']={'rows':rows2,'expected':expected2}; summary['gates']['target_null_rows_after_resolution']=rr['pass'] and ver2['pass'] and rows2==[expected2]
    if not summary['gates']['target_null_rows_after_resolution']: raise RuntimeError('explicit fixture resolution did not close target-null gate')
    enf=run_file('enforce_not_null.sql','enforce'); summary['gates']['not_null_enforced']=enf['pass'] and query("SELECT is_nullable FROM information_schema.columns WHERE table_schema=current_schema() AND table_name='customer_contact' AND column_name='phone_e164'")=='NO'
    if not summary['gates']['not_null_enforced']: raise RuntimeError('NOT NULL not enforced')
    late=run_file('old_writer_after_enforce.sql','old_writer_after_enforce',expect_ok=False); lall=late['stderr']+late['stdout']; summary['gates']['old_writer_after_enforce_23502']=late['pass'] and '23502' in lall
    if not summary['gates']['old_writer_after_enforce_23502']: raise RuntimeError('late old writer did not expose expected compatibility break')
    # Removal negative probe is transaction-scoped; connection abort/close rolls back the DROP.
    removal="BEGIN;\nALTER TABLE customer_contact DROP COLUMN phone_legacy;\nSELECT customer_id, phone_legacy FROM customer_contact LIMIT 1;\nROLLBACK;\n"
    dr=run_text(removal,'drop_legacy_old_reader_negative',expect_ok=False); dall=dr['stderr']+dr['stdout']; summary['gates']['old_reader_break_after_drop_42703']=dr['pass'] and '42703' in dall
    if not summary['gates']['old_reader_break_after_drop_42703']: raise RuntimeError('drop/removal probe did not prove old-reader break')
    lk=pair('lock_holder.sql','alter_lock_contender.sql','alter_lock'); lerr=lk['contender_stderr']+lk['contender_stdout']; summary['probes']['alter_lock']={'holder_rc':lk['holder_rc'],'contender_rc':lk['contender_rc'],'contender_elapsed_seconds':lk['contender_elapsed_seconds'],'saw_55P03':'55P03' in lerr}; summary['gates']['lock_timeout_55P03']=lk['holder_rc']==0 and lk['contender_rc']!=0 and '55P03' in lerr and 0.5<=lk['contender_elapsed_seconds']<3.0
    if not summary['gates']['lock_timeout_55P03']: raise RuntimeError('ALTER lock probe did not produce bounded 55P03 lock timeout')
except Exception as ex:
    run_error=f'{type(ex).__name__}: {ex}'
finally:
    if created:
        try: admin(f'DROP SCHEMA IF EXISTS {schema} CASCADE'); summary['cleanup_status']='PASS'
        except Exception as ex: summary['cleanup_status']='FAIL'; summary['cleanup_error']=f'{type(ex).__name__}: {ex}'
    else: summary['cleanup_status']='NOT_CREATED'
summary['gates']['cleanup_pass']=summary['cleanup_status']=='PASS'
summary['all_runtime_gates_pass']=bool(summary['gates']) and all(summary['gates'].values()) and run_error is None
if summary['all_runtime_gates_pass']:
    summary['status']='REAL_RUNTIME_EVIDENCE_CAPTURED_CONTRACT_PASS'; summary['actual_postgresql_runtime_evidence_claimed']=True
else:
    summary['status']='REAL_RUNTIME_CONTRACT_FAIL'; summary['actual_postgresql_runtime_evidence_claimed']=False; summary['error']=run_error
summary['evidence_files']=sorted(str(p.relative_to(ROOT)) for p in EVID.rglob('*') if p.is_file())
write_status(summary); print(json.dumps({'status':summary['status'],'schema':schema,'all_runtime_gates_pass':summary['all_runtime_gates_pass'],'cleanup_status':summary['cleanup_status']})); sys.exit(0 if summary['all_runtime_gates_pass'] else 2)