#!/usr/bin/env python3
from pathlib import Path
import hashlib,json,os,re,secrets,shutil,subprocess,sys,time
root=Path(__file__).resolve().parents[1]
status_path=root/'evidence/runtime-status.json'; ev=root/'evidence/runtime'; ev.mkdir(parents=True,exist_ok=True)
psql=shutil.which('psql'); url=os.environ.get('MAKNOW_DATABASE_URL'); confirm=os.environ.get('MAKNOW_DAT003_DISPOSABLE_CONFIRM')
SQL_FILES=['schema.sql','seed.sql','read_committed_t1.sql','read_committed_t2.sql','repeatable_read_t1.sql','repeatable_read_t2.sql','serializable_t1.sql','serializable_t2.sql','read_uncommitted_probe.sql','schedule_notes.sql']
def h(p): return hashlib.sha256(Path(p).read_bytes()).hexdigest()
def hashes():
    d={f'sql/{n}':h(root/'sql'/n) for n in SQL_FILES}
    d['model-contract.json']=h(root/'model-contract.json')
    return d
def emit(obj): status_path.write_text(json.dumps(obj,indent=2)+'\n'); print(json.dumps(obj,indent=2))
def base(st):
    return {'schema':'DAT003_RUNTIME_STATUS_v2','status':st,'postgresql_version':None,'fabricated_runtime_evidence':False,'required_for_learner_completion':True,'isolation':'EPHEMERAL_RANDOM_SCHEMA','generated_schema':None,'schedule_contract_pass':False,'whole_transaction_retry_pass':False,'post_retry_invariant_pass':False,'cleanup':'NOT_STARTED','input_sha256':hashes(),'evidence_files':[]}
if not psql or not url:
    out=base('NOT_RUN_MISSING_REAL_POSTGRESQL_RUNTIME'); emit(out); raise SystemExit(3)
if confirm!='YES':
    out=base('NOT_RUN_DISPOSABLE_CONFIRMATION_REQUIRED'); emit(out); raise SystemExit(4)
schema='maknow_dat003_'+secrets.token_hex(6)
if not re.fullmatch(r'maknow_dat003_[0-9a-f]{12}',schema): raise SystemExit('UNSAFE_GENERATED_SCHEMA')

def proc_sql(sql,tuples=False):
    a=[psql,'-X',url,'-v','ON_ERROR_STOP=1']
    if tuples: a += ['-At','-F','|','-P','null=__NULL__']
    a += ['-c',sql]
    return subprocess.run(a,text=True,capture_output=True)
def script_env():
    e=os.environ.copy(); old=e.get('PGOPTIONS','').strip(); scoped=f'-c search_path={schema}'
    e['PGOPTIONS']=(old+' '+scoped).strip(); return e
def runfile(name):
    return subprocess.run([psql,'-X',url,'-v','ON_ERROR_STOP=1','-f',str(root/'sql'/name)],text=True,capture_output=True,env=script_env())
def require(p,label):
    if p.returncode!=0: raise RuntimeError(label+': '+p.stderr.strip())
    return p
def reset():
    for n in ['schema.sql','seed.sql']: require(runfile(n),n.upper())
def pair(a,b,label,delay=.25):
    p1=subprocess.Popen([psql,'-X',url,'-v','ON_ERROR_STOP=1','-f',str(root/'sql'/a)],text=True,stdout=subprocess.PIPE,stderr=subprocess.PIPE,env=script_env())
    time.sleep(delay)
    p2=subprocess.Popen([psql,'-X',url,'-v','ON_ERROR_STOP=1','-f',str(root/'sql'/b)],text=True,stdout=subprocess.PIPE,stderr=subprocess.PIPE,env=script_env())
    o1,e1=p1.communicate(); o2,e2=p2.communicate()
    for suffix,data in [('t1.stdout.txt',o1),('t1.stderr.txt',e1),('t2.stdout.txt',o2),('t2.stderr.txt',e2)]: (ev/f'{label}-{suffix}').write_text(data)
    return {'t1':{'rc':p1.returncode,'out':o1,'err':e1},'t2':{'rc':p2.returncode,'out':o2,'err':e2}}
def line_value(text,marker):
    for line in text.splitlines():
        if line.startswith(marker+'|'):
            try: return int(line.split('|',1)[1])
            except: return None
    return None
def has_40001(x): return '40001' in (x.get('err','')+'\n'+x.get('out','')) or 'serialization failure' in (x.get('err','')+'\n'+x.get('out','')).lower() or 'could not serialize access' in (x.get('err','')+'\n'+x.get('out','')).lower()
status=base('RUNTIME_FAILED'); status['generated_schema']=schema; created=False
try:
    require(proc_sql(f'CREATE SCHEMA "{schema}";'),'CREATE_SCHEMA'); created=True
    ver=require(proc_sql('SHOW server_version;',True),'VERSION'); status['postgresql_version']=ver.stdout.strip().splitlines()[-1] if ver.stdout.strip() else None
    # Read Committed: both sessions succeed and T1 observes 1 then 0.
    reset(); rc=pair('read_committed_t1.sql','read_committed_t2.sql','read-committed')
    rc_ok=(rc['t1']['rc']==0 and rc['t2']['rc']==0 and line_value(rc['t1']['out'],'T1_RC_FIRST')==1 and line_value(rc['t1']['out'],'T1_RC_SECOND')==0)
    # Repeatable Read: T1 snapshot stays 1 and its conflicting write is rejected.
    reset(); rr=pair('repeatable_read_t1.sql','repeatable_read_t2.sql','repeatable-read')
    rr_ok=(rr['t1']['rc']!=0 and rr['t2']['rc']==0 and line_value(rr['t1']['out'],'T1_RR_FIRST')==1 and line_value(rr['t1']['out'],'T1_RR_SECOND')==1 and has_40001(rr['t1']))
    # Serializable: exactly one original transaction fails; retry that whole transaction from BEGIN.
    reset(); ser=pair('serializable_t1.sql','serializable_t2.sql','serializable')
    failed=[k for k in ('t1','t2') if ser[k]['rc']!=0]; succeeded=[k for k in ('t1','t2') if ser[k]['rc']==0]
    ser_initial_ok=(len(failed)==1 and len(succeeded)==1 and has_40001(ser[failed[0]]))
    retry_name='serializable_t1.sql' if failed==['t1'] else ('serializable_t2.sql' if failed==['t2'] else None)
    retry=None
    if retry_name:
        retry=runfile(retry_name); (ev/'serializable-retry.stdout.txt').write_text(retry.stdout); (ev/'serializable-retry.stderr.txt').write_text(retry.stderr)
    retry_ok=bool(retry_name and retry and retry.returncode==0)
    inv=require(subprocess.run([psql,'-X',url,'-At','-F','|','-v','ON_ERROR_STOP=1','-c',f'SET search_path TO "{schema}"; SELECT i.capacity, count(r.reservation_id) FILTER (WHERE r.active), (count(r.reservation_id) FILTER (WHERE r.active) <= i.capacity) FROM inventory i LEFT JOIN reservations r ON r.item_id=i.item_id WHERE i.item_id=1 GROUP BY i.capacity;'],text=True,capture_output=True),'INVARIANT_CHECK')
    (ev/'post-retry-invariant.txt').write_text(inv.stdout)
    vals=inv.stdout.strip().split('|'); invariant_ok=(len(vals)==3 and vals[0]=='1' and vals[1]=='1' and vals[2].lower() in {'t','true'})
    ru=runfile('read_uncommitted_probe.sql'); (ev/'read-uncommitted.stdout.txt').write_text(ru.stdout); (ev/'read-uncommitted.stderr.txt').write_text(ru.stderr)
    ru_ok=ru.returncode==0
    status['pair_returncodes']={'read_committed':[rc['t1']['rc'],rc['t2']['rc']],'repeatable_read':[rr['t1']['rc'],rr['t2']['rc']],'serializable':[ser['t1']['rc'],ser['t2']['rc']]}
    status['schedule_checks']={'read_committed':rc_ok,'repeatable_read':rr_ok,'serializable_initial_conflict':ser_initial_ok,'read_uncommitted_probe':ru_ok}
    status['schedule_contract_pass']=all(status['schedule_checks'].values())
    status['retry_session']=failed[0] if len(failed)==1 else None
    status['whole_transaction_retry_pass']=retry_ok
    status['post_retry_invariant']={'capacity':vals[0] if len(vals)>0 else None,'active_reservations':vals[1] if len(vals)>1 else None,'holds':invariant_ok}
    status['post_retry_invariant_pass']=invariant_ok
    if status['schedule_contract_pass'] and retry_ok and invariant_ok:
        status['status']='REAL_RUNTIME_EVIDENCE_CAPTURED_CONTRACT_PASS'
    else:
        status['status']='RUNTIME_OBSERVATION_CONTRACT_FAILED'
    status['evidence_files']=sorted('runtime/'+p.name for p in ev.iterdir() if p.is_file())
except Exception as e:
    status['error_class']=type(e).__name__; status['error']=str(e)[:1200]
finally:
    if created:
        cp=proc_sql(f'DROP SCHEMA IF EXISTS "{schema}" CASCADE;'); status['cleanup']='PASS' if cp.returncode==0 else 'FAIL'
    else: status['cleanup']='NOT_NEEDED'
    emit(status)
if status['status']!='REAL_RUNTIME_EVIDENCE_CAPTURED_CONTRACT_PASS' or status['cleanup']!='PASS': raise SystemExit(7)