#!/usr/bin/env python3
from pathlib import Path
import os, subprocess, shutil, sys, uuid, json, time, hashlib, re

ROOT=Path(__file__).resolve().parents[1]
SQL=ROOT/'sql'
EVID=ROOT/'evidence'/'runtime'
STATUS=ROOT/'evidence'/'runtime-status.json'
dsn=os.getenv('MAKNOW_DATABASE_URL')
confirm=os.getenv('MAKNOW_DANGEROUS_LAB_CONFIRM')

# Authoring/build mode never fabricates database evidence.
if not dsn or confirm!='DAT004_LOCAL_DISPOSABLE':
    print(json.dumps({'status':'NOT_RUN','reason':'Set MAKNOW_DATABASE_URL and MAKNOW_DANGEROUS_LAB_CONFIRM=DAT004_LOCAL_DISPOSABLE explicitly.'}))
    sys.exit(3)
if not shutil.which('psql'):
    print(json.dumps({'status':'NOT_RUN','reason':'psql not found'}))
    sys.exit(3)

# Never mix stale evidence from a previous run with a new run.
if EVID.exists(): shutil.rmtree(EVID)
EVID.mkdir(parents=True,exist_ok=True)

schema='maknow_dat004_'+uuid.uuid4().hex[:12]
base=['psql',dsn,'-X','-v','ON_ERROR_STOP=1','-q','-A','-t','-F','|']
input_files=[
 'schema.sql','seed.sql','blocking_t1.sql','blocking_t2.sql','deadlock_t1.sql','deadlock_t2.sql',
 'ordered_t1.sql','ordered_t2.sql','nowait_holder.sql','nowait_contender.sql','skip_locked_holder.sql',
 'skip_locked_consumer.sql','optimistic_current.sql','optimistic_stale.sql'
]

def sha256(path: Path) -> str:
    return hashlib.sha256(path.read_bytes()).hexdigest()

def call(args,env=None,timeout=30):
    return subprocess.run(args,text=True,capture_output=True,env=env,timeout=timeout)

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 env_for_schema():
    e=os.environ.copy(); e['PGOPTIONS']=f'-c search_path={schema},pg_catalog -c deadlock_timeout=100ms'; return e

def query(sql):
    p=call(base+['-c',sql],env=env_for_schema())
    if p.returncode: raise RuntimeError(p.stderr.strip() or f'query rc={p.returncode}')
    return p.stdout.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',encoding='utf-8')
    (EVID/f'{tag}.elapsed_seconds.txt').write_text(f'{elapsed:.6f}\n',encoding='utf-8')

def run_file(name,tag,allowed=(0,)):
    started=time.monotonic(); p=call(base+['-f',str(SQL/name)],env=env_for_schema()); elapsed=time.monotonic()-started
    write_probe(tag,p,elapsed)
    if p.returncode not in allowed: raise RuntimeError(f'{name} rc={p.returncode}: {p.stderr.strip()}')
    return {'returncode':p.returncode,'stdout':p.stdout,'stderr':p.stderr,'elapsed_seconds':elapsed}

def pair(a,b,tag,delay=0.35,allowed_a=(0,),allowed_b=(0,)):
    ea=env_for_schema(); eb=env_for_schema(); start=time.monotonic()
    pa=subprocess.Popen(base+['-f',str(SQL/a)],stdout=subprocess.PIPE,stderr=subprocess.PIPE,text=True,env=ea)
    time.sleep(delay)
    b_start=time.monotonic(); pb=subprocess.Popen(base+['-f',str(SQL/b)],stdout=subprocess.PIPE,stderr=subprocess.PIPE,text=True,env=eb)
    try:
        oa,erra=pa.communicate(timeout=25)
        ob,errb=pb.communicate(timeout=25)
    except subprocess.TimeoutExpired:
        for proc in (pa,pb):
            if proc.poll() is None: proc.kill()
        oa,erra=pa.communicate(); ob,errb=pb.communicate()
        raise RuntimeError(f'{tag} timed out; child processes killed')
    total=time.monotonic()-start; b_elapsed=time.monotonic()-b_start
    for suffix,val in [('t1.stdout',oa),('t1.stderr',erra),('t2.stdout',ob),('t2.stderr',errb)]:
        (EVID/f'{tag}.{suffix}.txt').write_text(val,encoding='utf-8')
    meta={'t1':pa.returncode,'t2':pb.returncode,'total_elapsed_seconds':total,'t2_elapsed_seconds':b_elapsed,'launch_delay_seconds':delay}
    (EVID/f'{tag}.meta.json').write_text(json.dumps(meta,indent=2)+'\n',encoding='utf-8')
    if pa.returncode not in allowed_a or pb.returncode not in allowed_b:
        raise RuntimeError(f'{tag} unexpected rc: {pa.returncode},{pb.returncode}')
    return {**meta,'t1_stderr':erra,'t2_stderr':errb,'t1_stdout':oa,'t2_stdout':ob}

def nonempty_rows(stdout: str):
    return [x.strip() for x in stdout.splitlines() if x.strip()]

summary={
 'schema':'DAT004_RUNTIME_STATUS_v2',
 'status':'RUNNING',
 'generated_schema':schema,
 'postgresql_runtime_evidence_claimed':False,
 'fabricated_runtime_evidence':False,
 'input_sha256':{name:sha256(SQL/name) for name in input_files},
 'model_contract_sha256':sha256(ROOT/'model-contract.json'),
 'cleanup_status':'NOT_ATTEMPTED',
 'gates':{},
 'probes':{},
}
run_error=None
created=False
try:
    admin(f'CREATE SCHEMA {schema}'); created=True
    version_num=int(admin('SHOW server_version_num'))
    summary['postgresql_version_num']=version_num
    summary['gates']['postgresql_major_18']=180000 <= version_num < 190000
    if not summary['gates']['postgresql_major_18']: raise RuntimeError(f'PostgreSQL 18 required, got server_version_num={version_num}')

    run_file('schema.sql','setup_schema'); run_file('seed.sql','setup_seed')

    blocking=pair('blocking_t1.sql','blocking_t2.sql','blocking')
    blocking_ok=(blocking['t1']==0 and blocking['t2']==0 and blocking['t2_elapsed_seconds']>=1.0)
    summary['probes']['blocking']={'t1':blocking['t1'],'t2':blocking['t2'],'t2_elapsed_seconds':blocking['t2_elapsed_seconds']}
    summary['gates']['blocking_wait_observed']=blocking_ok
    if not blocking_ok: raise RuntimeError('blocking probe did not demonstrate an observable wait')

    run_file('seed.sql','reset_after_blocking')
    dead=pair('deadlock_t1.sql','deadlock_t2.sql','deadlock',allowed_a=(0,3),allowed_b=(0,3))
    allerr=dead['t1_stderr']+dead['t2_stderr']; dead_failed=[x for x in [dead['t1'],dead['t2']] if x!=0]
    dead_ok=(sorted([dead['t1'],dead['t2']])==[0,3] and '40P01' in allerr)
    summary['probes']['deadlock']={'t1':dead['t1'],'t2':dead['t2'],'saw_40P01':'40P01' in allerr}
    summary['gates']['deadlock_exactly_one_victim_40P01']=dead_ok
    if not dead_ok: raise RuntimeError('deadlock probe did not produce exactly one 40P01 victim')

    run_file('seed.sql','reset_after_deadlock')
    ordered=pair('ordered_t1.sql','ordered_t2.sql','ordered')
    ordered_ok=(ordered['t1']==0 and ordered['t2']==0 and '40P01' not in (ordered['t1_stderr']+ordered['t2_stderr']))
    summary['probes']['ordered']={'t1':ordered['t1'],'t2':ordered['t2'],'saw_40P01':'40P01' in (ordered['t1_stderr']+ordered['t2_stderr'])}
    summary['gates']['ordered_pair_no_targeted_deadlock']=ordered_ok
    if not ordered_ok: raise RuntimeError('ordered probe did not complete cleanly')

    run_file('seed.sql','reset_before_nowait')
    nw=pair('nowait_holder.sql','nowait_contender.sql','nowait',allowed_a=(0,),allowed_b=(3,))
    nw_ok=(nw['t1']==0 and nw['t2']==3 and '55P03' in nw['t2_stderr'])
    summary['probes']['nowait']={'holder_rc':nw['t1'],'contender_rc':nw['t2'],'saw_55P03':'55P03' in nw['t2_stderr']}
    summary['gates']['nowait_exact_55P03']=nw_ok
    if not nw_ok: raise RuntimeError('NOWAIT probe did not produce 55P03')

    run_file('seed.sql','reset_before_skip_locked')
    sk=pair('skip_locked_holder.sql','skip_locked_consumer.sql','skip_locked')
    rows=nonempty_rows(sk['t2_stdout']); sk_ok=(sk['t1']==0 and sk['t2']==0 and 'QUEUE_CONSUMER|2' in rows and not any(x=='QUEUE_CONSUMER|1' for x in rows))
    summary['probes']['skip_locked']={'holder_rc':sk['t1'],'consumer_rc':sk['t2'],'consumer_rows':rows}
    summary['gates']['skip_locked_omits_locked_job1']=sk_ok
    if not sk_ok: raise RuntimeError('SKIP LOCKED probe did not return the next eligible unlocked job')

    run_file('seed.sql','reset_before_optimistic')
    cur=run_file('optimistic_current.sql','optimistic_current'); stale=run_file('optimistic_stale.sql','optimistic_stale')
    current_rows=nonempty_rows(cur['stdout']); stale_rows=nonempty_rows(stale['stdout']); final=query("SELECT note||'|'||version FROM reservations WHERE reservation_id=1")
    cur_ok=(current_rows==['1|8']); stale_ok=(stale_rows==[]); final_ok=(final=='current-write|8')
    summary['probes']['optimistic']={'current_rows':current_rows,'stale_rows':stale_rows,'final_state':final}
    summary['gates']['optimistic_current_one_row']=cur_ok
    summary['gates']['optimistic_stale_zero_rows']=stale_ok
    summary['gates']['optimistic_final_state_unchanged_by_stale']=final_ok
    if not (cur_ok and stale_ok and final_ok): raise RuntimeError('optimistic compare-and-swap bounded observables failed')

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']=all(summary['gates'].values()) and run_error is None
if summary['all_runtime_gates_pass']:
    summary['status']='REAL_RUNTIME_EVIDENCE_CAPTURED_CONTRACT_PASS'
    summary['postgresql_runtime_evidence_claimed']=True
else:
    summary['status']='REAL_RUNTIME_CONTRACT_FAIL'
    summary['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())
STATUS.write_text(json.dumps(summary,indent=2,sort_keys=True)+'\n',encoding='utf-8')
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)