#!/usr/bin/env python3
from __future__ import annotations
from pathlib import Path
from urllib.parse import urlparse
import hashlib,json,os,re,secrets,shutil,subprocess,sys

ROOT=Path(__file__).resolve().parents[2]
LEARNER=ROOT/'learner'
RUNTIME_DIR=LEARNER/'evidence'/'runtime'
STATUS_PATH=LEARNER/'evidence'/'runtime-status.json'
POLICY_PATH=LEARNER/'toolchain-policy.json'
TODO_RE=re.compile(r'__TODO_[A-Z0-9_]+__')
REQUIRED_MARKERS=[
 'DAT009_EVIDENCE_EXACT','DAT009_EVIDENCE_ANN','DAT009_EVIDENCE_LEXICAL','DAT009_EVIDENCE_FUSION','DAT009_EVIDENCE_FILTER','DAT009_EVIDENCE_ACCEPTANCE',
 'DAT009_PROBE_EXACT_VS_ANN_RECALL','DAT009_PROBE_FILTER_CANDIDATE_LOSS','DAT009_PROBE_EMBEDDING_VERSION_BREAK','DAT009_PROBE_NAIVE_SCORE_ADDITION']
METRIC_MAP={
 'cosine':('<=>','vector_cosine_ops'),
 'l2':('<->','vector_l2_ops'),
 'inner_product':('<#>','vector_ip_ops')
}
ANN_PARAM={'hnsw':'hnsw.ef_search','ivfflat':'ivfflat.probes'}
class SafetyError(RuntimeError): pass
class SubmissionError(RuntimeError): pass

def sha256(p:Path)->str: return hashlib.sha256(p.read_bytes()).hexdigest()
def load_policy(): return json.loads(POLICY_PATH.read_text())
def write_status(o:dict): STATUS_PATH.write_text(json.dumps(o,indent=2)+"\n")
def base_status(status,reason,**extra):
    o={'schema':'DAT009_RUNTIME_STATUS_v2','status':status,'runtime_claimable':False,
       'actual_postgresql_runtime_evidence_claimed':False,'actual_exact_ann_lexical_fusion_latency_evidence_claimed':False,
       'postgresql_version_num':None,'pgvector_version':None,'database_identity':None,'host_class':None,'generated_schema':None,
       'input_sha256':{},'runtime_receipt':None,'cleanup_status':'NOT_RUN','reason':reason}
    o.update(extra); return o

def validate_target(url:str,confirmation:str,policy:dict|None=None)->dict:
    policy=policy or load_policy()
    if confirmation!='YES': raise SafetyError('explicit disposable confirmation missing')
    u=urlparse(url)
    if u.scheme not in {'postgresql','postgres'}: raise SafetyError('unsupported database URL scheme')
    host=(u.hostname or '').lower()
    if host not in set(policy['allowed_hosts']): raise SafetyError('database host is not loopback-only')
    db=(u.path or '').lstrip('/').split('?',1)[0]
    pref=policy['database_name_prefix']
    if not db.startswith(pref) or db==pref: raise SafetyError('database name is outside disposable prefix contract')
    if db.lower() in {'postgres','template0','template1'}: raise SafetyError('system/default database forbidden')
    return {'host_class':'LOOPBACK_ONLY','database_identity':db}

def unresolved_submission(paths):
    bad=[]
    for p in paths:
        if not p.exists(): bad.append('MISSING:'+p.name); continue
        hits=TODO_RE.findall(p.read_text())
        if hits: bad.append('UNRESOLVED:'+p.name+':'+','.join(sorted(set(hits))))
    return bad

def parse_submission_contract(text:str)->dict:
    def grab(pattern,label):
        m=re.search(pattern,text,re.M)
        if not m: raise SubmissionError('missing '+label)
        return tuple(x.strip().lower() for x in m.groups())
    metric=grab(r'^--\s*DAT009_METRIC_PAIR:\s*([^|\n]+)\|([^|\n]+)\|([^\n]+)$','metric pair')
    fam=grab(r'^--\s*DAT009_ANN_FAMILY:\s*([^\n]+)$','ANN family')[0]
    param=grab(r'^--\s*DAT009_ANN_PARAMETER:\s*([^=\n]+)=([^\n]+)$','ANN parameter')
    fusion=grab(r'^--\s*DAT009_FUSION:\s*([^|\n]+)\|([^\n]+)$','fusion declaration')
    if metric[0] not in METRIC_MAP or metric[1:]!=METRIC_MAP[metric[0]]: raise SubmissionError('metric/operator/operator-class mismatch')
    if fam not in ANN_PARAM: raise SubmissionError('unsupported ANN family')
    if param[0]!=ANN_PARAM[fam]: raise SubmissionError('ANN parameter does not match ANN family')
    if not fusion[0] or not fusion[1]: raise SubmissionError('fusion method/parameter must be explicit')
    if re.search(r'lexical_score\s*\+\s*(distance|similarity)|(distance|similarity)\s*\+\s*lexical_score',text,re.I):
        raise SubmissionError('naive raw lexical/vector score addition is forbidden')
    return {'metric':metric[0],'distance_operator':metric[1],'operator_class':metric[2],'ann_family':fam,'ann_parameter_name':param[0],'fusion_method':fusion[0]}

def run_cmd(cmd,env,timeout=20): return subprocess.run(cmd,stdout=subprocess.PIPE,stderr=subprocess.PIPE,text=True,env=env,timeout=timeout)

def main():
    policy=load_policy(); url=os.environ.get(policy['database_url_env'],''); confirm='YES' if os.environ.get('MAKNOW_DAT009_CONFIRM_DISPOSABLE')=='YES' else ''
    try: target=validate_target(url,confirm,policy)
    except Exception as e:
        write_status(base_status('REFUSED_BY_SAFETY_GATE',str(e))); print('REFUSED:',e); return 2
    psql=shutil.which('psql')
    if not psql:
        write_status(base_status('NOT_RUN_MISSING_PSQL','psql binary not available',**target)); print('NOT RUN: psql binary not available'); return 3
    submissions=[LEARNER/'submission/03_hybrid_workbench.sql',LEARNER/'submission/04_failure_probes.sql']
    bad=unresolved_submission(submissions)
    if bad:
        write_status(base_status('REFUSED_INCOMPLETE_SUBMISSION','; '.join(bad),**target)); print('REFUSED: incomplete submission'); return 4
    try: sub_contract=parse_submission_contract(submissions[0].read_text())
    except Exception as e:
        write_status(base_status('REFUSED_SUBMISSION_CONTRACT',str(e),**target)); print('REFUSED:',e); return 5
    if RUNTIME_DIR.exists(): shutil.rmtree(RUNTIME_DIR)
    RUNTIME_DIR.mkdir(parents=True)
    env=os.environ.copy(); env['PGCONNECT_TIMEOUT']=str(policy['connect_timeout_seconds']); env['PGOPTIONS']=f"-c statement_timeout={policy['statement_timeout_ms']} -c lock_timeout=3000"
    probe_sql="SELECT current_database() || '|' || current_setting('server_version_num') || '|' || COALESCE((SELECT extversion FROM pg_extension WHERE extname='vector'),'MISSING');"
    probe=run_cmd([psql,'-X','-At',url,'-v','ON_ERROR_STOP=1','-c',probe_sql],env)
    if probe.returncode:
        write_status(base_status('RUN_FAILED_NONCLAIMABLE','server/vector identity probe failed',**target)); return 6
    line=probe.stdout.strip().splitlines()[-1] if probe.stdout.strip() else ''
    parts=line.split('|')
    if len(parts)!=3:
        write_status(base_status('RUN_FAILED_NONCLAIMABLE','server/vector identity probe malformed',**target)); return 7
    db,version_num,vector_ver=parts
    if db!=target['database_identity'] or not version_num.startswith('18') or vector_ver!=policy['pgvector_version']:
        write_status(base_status('REFUSED_VERSION_OR_IDENTITY_MISMATCH','database identity, PostgreSQL major, or pgvector version mismatch',**target,postgresql_version_num=version_num,pgvector_version=vector_ver)); return 8
    schema='maknow_dat009_'+secrets.token_hex(6)
    inputs=[LEARNER/'sql/01_schema.sql',LEARNER/'sql/02_seed.sql',*submissions]
    hashes={p.relative_to(ROOT).as_posix():sha256(p) for p in inputs}
    combined="\\set ON_ERROR_STOP on\n"+f'CREATE SCHEMA "{schema}";\nSET search_path TO "{schema}", public;\n'
    for p in inputs: combined += f"\n-- BEGIN {p.name}\n"+p.read_text()+f"\n-- END {p.name}\n"
    runtime_sql=RUNTIME_DIR/'runtime.sql'; runtime_sql.write_text(combined)
    out=RUNTIME_DIR/'postgresql.stdout.txt'; err=RUNTIME_DIR/'postgresql.stderr.txt'; exec_result=None; cleanup=None
    try:
        exec_result=run_cmd([psql,'-X',url,'-v','ON_ERROR_STOP=1','-f',str(runtime_sql)],env,timeout=40)
        out.write_text(exec_result.stdout); err.write_text(exec_result.stderr)
    except subprocess.TimeoutExpired as e:
        out.write_text(e.stdout or ''); err.write_text((e.stderr or '')+'\nTIMEOUT\n')
    finally:
        cleanup=run_cmd([psql,'-X',url,'-v','ON_ERROR_STOP=1','-c',f'DROP SCHEMA IF EXISTS "{schema}" CASCADE;'],env,timeout=15)
    stdout=out.read_text(); markers={m:(m in stdout) for m in REQUIRED_MARKERS}; markers_ok=all(markers.values())
    execution_ok=exec_result is not None and exec_result.returncode==0; cleanup_ok=cleanup.returncode==0
    receipt={'schema':'DAT009_RUNTIME_RECEIPT_v1','postgresql_version_num':version_num,'pgvector_version':vector_ver,'database_identity':db,'host_class':target['host_class'],'generated_schema':schema,'submission_contract':sub_contract,'input_sha256':hashes,'execution_returncode':None if exec_result is None else exec_result.returncode,'required_markers':markers,'cleanup_returncode':cleanup.returncode,'cleanup_status':'PASS' if cleanup_ok else 'FAIL','runtime_sql_sha256':sha256(runtime_sql),'stdout_sha256':sha256(out),'stderr_sha256':sha256(err)}
    rp=RUNTIME_DIR/'runner-receipt.json'; rp.write_text(json.dumps(receipt,indent=2)+"\n")
    claimable=execution_ok and cleanup_ok and markers_ok
    st=base_status('RAN_REAL_POSTGRESQL' if claimable else 'RUN_FAILED_NONCLAIMABLE','all runtime gates passed' if claimable else 'execution, evidence-marker, or cleanup gate failed',postgresql_version_num=version_num,pgvector_version=vector_ver,database_identity=db,host_class=target['host_class'],generated_schema=schema,input_sha256=hashes,runtime_receipt='learner/evidence/runtime/runner-receipt.json',cleanup_status='PASS' if cleanup_ok else 'FAIL')
    st['runtime_claimable']=claimable; st['actual_postgresql_runtime_evidence_claimed']=claimable; st['actual_exact_ann_lexical_fusion_latency_evidence_claimed']=claimable
    write_status(st); print('PASS REAL POSTGRESQL RUNTIME' if claimable else 'NONCLAIMABLE RUNTIME FAILURE'); return 0 if claimable else 9
if __name__=='__main__': raise SystemExit(main())