"""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()

​