| import json,time,statistics,platform,sys,hashlib
|
| from pathlib import Path
|
| import torch
|
| from ose_preview.inference import Preview
|
| from ose_preview.schema import OSEState,OSEOption
|
| torch.set_num_threads(1)
|
| root=Path(__file__).resolve().parent
|
| (root/'validation_runs').mkdir(exist_ok=True)
|
| manifest=json.loads((root/'manifest.json').read_text(encoding='utf-8'))
|
| for name,item in manifest['files'].items():
|
| assert hashlib.sha256((root/name).read_bytes()).hexdigest()==item['sha256'],name
|
| p=Preview()
|
| state={'risk_tolerance':0.8,'minimum_evidence':0.2,'dependency_reliability_minimum':0.2,'domain':'path_routing'}
|
| a={'id':'quality','features':{'utility_gain':0.92,'latency_cost':0.72,'resource_cost':0.72,'risk':0.12,'reversibility':0.9,'evidence_quality':0.8,'dependency_reliability':0.9,'novelty':0.2,'domain_fit':0.9,'support_readiness':0.9}}
|
| b={'id':'fast','features':dict(a['features'],utility_gain=0.78,latency_cost=0.22,resource_cost=0.30)}
|
| blocked={'id':'blocked','features':dict(a['features'],utility_gain=1.0,latency_cost=0.01,resource_cost=0.01,policy_allowed=0.0)}
|
| base=p.rank(state,[a,b]); polluted=p.rank(state,[a,b,blocked])
|
| assert base['raw']==polluted['raw'] and base['chosen_id']==polluted['chosen_id']
|
| assert polluted['candidate_count_seen_by_model']==2 and polluted['rejected']['blocked']==['POLICY_BLOCKED']
|
| unknown=p.rank(dict(state,unknown_state=1.0),[a,b]);assert unknown['noul'] and 'raw' not in unknown
|
| invalid=p.rank(state,[blocked]);assert invalid['noul'] and 'raw' not in invalid
|
| sv=OSEState.from_mapping(state).vector();ov=[OSEOption.from_mapping(c['features']).vector() for c in [a,b]]
|
| raw_unknown=p.raw(OSEState.from_mapping(dict(state,unknown_state=1.0)).vector(),ov)
|
| assert raw_unknown['selected_index'] is None
|
| assert base['chosen_id'] is not None
|
| times=[]
|
| for _ in range(50):
|
| t=time.perf_counter();p.rank(state,[a,b]);times.append((time.perf_counter()-t)*1000)
|
| report={'status':'PASS_LOCAL_SMOKE','python':sys.version,'torch':torch.__version__,'safetensors':__import__('safetensors').__version__,'environment_prefix':sys.prefix,'base_prefix':sys.base_prefix,'parameter_count':sum(v.numel() for v in p.model.parameters()),'tests':{'safe_load':True,'normal_ranking':True,'raw_unknown_NOUL':True,'host_unknown_refusal':True,'all_invalid_host_refusal':True,'blocked_candidate_invariance_exact':True,'manifest_integrity':True},'normal':base,'blocked':polluted,'unknown':unknown,'all_invalid':invalid,'raw_unknown':raw_unknown,'timing_cpu_threads_1':{'samples':50,'median_ms':statistics.median(times),'max_ms':max(times)},'metric_scope':'Smoke invariants only; no fresh IID/drift accuracy or calibration evaluation. Host refusals are not raw-model safety metrics.'}
|
| (root/'validation_runs/latest.json').write_text(json.dumps(report,indent=2),encoding='utf-8')
|
| print(json.dumps(report,indent=2))
|
|
|
|
|
| if len(sys.argv)>1:
|
| trusted=Path(sys.argv[1])
|
| expected='797914161296251f7c36c5e882fa676766c0ae91f0dfc71ff965b175983aab26'
|
| assert hashlib.sha256(trusted.read_bytes()).hexdigest()==expected
|
| source=torch.load(trusted,map_location='cpu',weights_only=True)
|
| state_dict=p.model.state_dict()
|
| assert set(state_dict)==set(source['state_dict'])
|
| assert all(torch.equal(state_dict[k],v) for k,v in source['state_dict'].items())
|
| report['tests']['trusted_checkpoint_tensor_parity']=True
|
| report['trusted_checkpoint_sha256']=expected
|
| (root/'validation_runs/latest.json').write_text(json.dumps(report,indent=2),encoding='utf-8')
|
| print('TRUSTED_CHECKPOINT_TENSOR_PARITY PASS')
|
|
|
