"""Public fixed-suite dynamics evaluation v2; numerical agreement, not a generalization certificate."""
import json, os, sys, pathlib, subprocess, signal, resource, time, tempfile
import importlib.metadata
import numpy as np
from physics import make_instance, gate
CHILD = r'''
import json,sys,importlib.util,resource,time
method, batch, output = sys.argv[1:]
try:
    spec=importlib.util.spec_from_file_location('method',method)
    mod=importlib.util.module_from_spec(spec); spec.loader.exec_module(mod)
    p=getattr(mod,'PRIORS',None)
    if not isinstance(p,dict) or not isinstance(p.get('family'),str) or not p['family'].strip() or not isinstance(p.get('parameters'),dict) or not isinstance(p.get('notes'),str):
        raise ValueError('PRIORS must contain nonempty family:string, parameters:dict and notes:string')
    json.dumps(p,allow_nan=False)
    results=[]
    for inst in json.load(open(batch)):
        t=time.monotonic()
        import numpy as np
        y=np.asarray(mod.simulate(inst,inst['times']),dtype=float).tolist()
        results.append({'trace':y,'seconds':time.monotonic()-t})
    json.dump({'ok':True,'priors':p,'results':results,'peak_rss_kib':resource.getrusage(resource.RUSAGE_SELF).ru_maxrss},open(output,'w'),allow_nan=False)
except BaseException as e:
    json.dump({'ok':False,'error':type(e).__name__+': '+str(e)[:500]},open(output,'w'))
'''
def run_batch(method, instances, seconds, policy, scratch, label):
    batch=scratch/(label+'.json'); output=scratch/(label+'-out.json')
    batch.write_text(json.dumps(instances)); child=scratch/'child.py'; child.write_text(CHILD)
    env={'PATH':os.environ.get('PATH',''),'HOME':str(scratch),'PYTHONDONTWRITEBYTECODE':'1'}
    env.update({k:str(policy['child_threads']) for k in ('OMP_NUM_THREADS','OPENBLAS_NUM_THREADS','MKL_NUM_THREADS','NUMBA_NUM_THREADS')})
    def limits():
        cap=policy['child_memory_mb']*1024**2; resource.setrlimit(resource.RLIMIT_AS,(cap,cap))
    start=time.monotonic()
    proc=subprocess.Popen([sys.executable,'-I',str(child),str(method),str(batch),str(output)],cwd=scratch,env=env,preexec_fn=limits,start_new_session=True,stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL)
    timeout=False
    try: proc.wait(timeout=seconds)
    except subprocess.TimeoutExpired: timeout=True
    finally:
        try: os.killpg(proc.pid,signal.SIGKILL)
        except ProcessLookupError: pass
        proc.wait()
    wall=time.monotonic()-start
    if timeout: return {'ok':False,'error':f'{label} exceeded its full {seconds}s budget','wall_seconds':wall}
    try: result=json.loads(output.read_text())
    except Exception: result={'ok':False,'error':'child produced no valid JSON'}
    result['wall_seconds']=wall
    return result

def main():
    inp=pathlib.Path(os.environ.get('TFP_INPUT','/input')); out=pathlib.Path(os.environ.get('TFP_OUTPUT','/output'))
    manifest=json.loads((inp/'manifest.json').read_text())
    refs={e['key']:pathlib.Path(e['path']) for e in manifest['reference']}
    method=next(pathlib.Path(e['path']) for e in manifest['artifacts'] if e['role']=='method')
    fam=json.loads(refs['family'].read_text()); policy=json.loads(refs['spec'].read_text()); truth=json.loads(refs['traces'].read_text())
    ev={'family':fam['family'],'suite_version':fam['suite']['version'],'instance_errors':[], 'accuracy_probe_error':None,
        'runtime':{'python':sys.version.split()[0],**{x:importlib.metadata.version(x) for x in ('numpy','scipy','quimb','physics-tenpy')}},
        'reference_validation':truth['validation'],'resource_limits':policy,'budgets':fam['budget_seconds']}
    def finish(verdict,reason,score=None):
        me,parent,uid=os.getpid(),os.getppid(),os.getuid()
        for name in os.listdir('/proc'):
            if not name.isdigit() or int(name) in (me,parent): continue
            try:
                status=pathlib.Path('/proc')/name/'status'
                if any(line.startswith('Uid:') and int(line.split()[1])==uid for line in status.read_text().splitlines()): os.kill(int(name),signal.SIGKILL)
            except (OSError,ValueError): pass
        (out/'evidence').mkdir(parents=True,exist_ok=True)
        (out/'evidence/summary.json').write_text(json.dumps(ev,allow_nan=False,indent=1))
        result={'result':verdict,'reason':reason}
        if score is not None: result['score']=score
        (out/'result.json').write_text(json.dumps(result,allow_nan=False)); print(json.dumps(result),flush=True)
    if len(method.read_bytes())>policy['max_source_bytes']: return finish('refuted','Source exceeds byte limit')
    seeds=fam['suite']['seeds']
    if len(truth['cases'])!=len(seeds) or truth['version']!=fam['suite']['version']: return finish('inconclusive','Reference suite mismatch')
    for seed,case in zip(seeds,truth['cases']):
        if case['seed']!=seed or case['instance']!=make_instance(fam,fam['exact']['size'],seed): return finish('inconclusive','Reference instance mismatch')
    probe=truth['accuracy_probe']
    if probe['instance']!=make_instance(fam,fam['accuracy_probe']['size'],fam['accuracy_probe']['seed']): return finish('inconclusive','Probe mismatch')
    scale=make_instance(fam,fam['scaling']['size'],seeds[0]+100000)
    groups=[('scaling',[scale],None),('exact_suite',[c['instance'] for c in truth['cases']],[c['trace'] for c in truth['cases']]),('accuracy_probe',[probe['instance']],[probe['trace']])]
    with tempfile.TemporaryDirectory(prefix='dynamics-v2-',dir=out) as temp:
        scratch=pathlib.Path(temp)
        for label,instances,expected in groups:
            result=run_batch(method,instances,fam['budget_seconds'][label],policy,scratch,label)
            ev[label]={'wall_seconds':result['wall_seconds'],'peak_rss_kib':result.get('peak_rss_kib')}
            if not result.get('ok'): return finish('refuted',result.get('error','Invalid method'))
            if 'priors' in ev and ev['priors']!=result['priors']: return finish('refuted','PRIORS changed between evaluation batches')
            ev['priors']=result['priors']
            if len(result['results'])!=len(instances): return finish('refuted','Wrong number of output traces')
            for i,(inst,row) in enumerate(zip(instances,result['results'])):
                reason=gate(row['trace'],inst,policy['bounds_tolerance'],policy['conservation_tolerance'],label)
                if reason: return finish('refuted',reason)
                if expected is not None:
                    err=float(np.max(np.abs(np.array(row['trace'])-np.array(expected[i]))))
                    if label=='exact_suite': ev['instance_errors'].append(err)
                    else: ev['accuracy_probe_error']=err
        ev['worst_instance_error']=max(ev['instance_errors']); ev['mean_instance_max_abs_error']=float(np.mean(ev['instance_errors']))
        finish('verified','Public-suite numerical error measured; larger-size run passed sanity gates. Probe error is reported separately; generalization and source honesty are not certified.',ev['mean_instance_max_abs_error'])
if __name__=='__main__': main()