Blame for source/icl_native/offline_infer.py

"""Standard-library inference with real trained weights evaluated by the ICL VM.
No Torch, NumPy, network API, label oracle or training object is needed.
The native model selects finite semantic opcodes, not unrestricted source text.
"""
from __future__ import annotations
import argparse,copy,json,math,sys
from pathlib import Path
from .common import ACTIONS,ERRORS,FEATURE_NAMES,features,strict_load,dump,sha,canonical
from .execution import validate_plan,protocol_guard,advance_runtime,vm_verify
from icl.api import run_ocir,loads_ocir,SCHEMA_VERSION,SCHEMA_SHA256
from icl.runtime import ICLError
import icl
SHAPES={'w1':[30,32],'b1':[1,32],'w2':[32,32],'b2':[1,32],'wo':[32,22],'bo':[1,22]}
class OfflineModel:
 def __init__(self,path:Path):
  if not path.is_dir():raise ValueError('MODEL_NOT_LOADED')
  man=strict_load(path/'manifest.json',200000)
  if man.get('schema')!='ICL_R5_OFFLINE_BUNDLE/1' or set(man.get('files',{}))!={'weights.json','compiled_model.ocir','model.icl','model_config.json'}:raise ValueError('BUNDLE_SCHEMA')
  for n,m in man['files'].items():
   p=path/n
   if p.is_symlink() or not p.is_file() or p.stat().st_size!=m['bytes'] or p.stat().st_size>2_000_000 or sha(p)!=m['sha256']:raise ValueError('MODEL_INTEGRITY')
  core=Path(icl.__file__).parent
  if set(man['runtime_file_hashes'])!={'__init__.py','__main__.py','api.py','checker.py','cli.py','frontend.py','interpreter.py','legacy_afc.py','runtime.py','vm.py','OCIR_SCHEMA.json','TOKEN_VOCAB.json'}:raise ValueError('RUNTIME_LOCK_SCHEMA')
  if any(sha(core/n)!=h for n,h in man['runtime_file_hashes'].items()):raise ValueError('RUNTIME_INCOMPATIBLE')
  cfg=strict_load(path/'model_config.json');self.cfg=cfg
  if cfg['actions']!=ACTIONS or cfg['feature_names']!=FEATURE_NAMES or cfg['schema_version']!=SCHEMA_VERSION or cfg['schema_sha256']!=SCHEMA_SHA256:raise ValueError('MODEL_CONTRACT')
  self.weights=strict_load(path/'weights.json',500000)
  if set(self.weights)!=set(SHAPES):raise ValueError('WEIGHT_KEYS')
  for k,v in self.weights.items():
   if set(v)!={'shape','data'} or v['shape']!=SHAPES[k] or len(v['data'])!=math.prod(SHAPES[k]) or any(type(x) not in (float,int) or not math.isfinite(x) for x in v['data']):raise ValueError('WEIGHT_SHAPE_OR_FINITE')
  self.graph=loads_ocir((path/'compiled_model.ocir').read_text())
 def logits(self,row):
  if len(row)!=30 or any(not math.isfinite(v) for v in row):raise ValueError('FEATURE_INPUT')
  out=run_ocir(self.graph,args=[{'shape':[1,30],'data':row}]+[self.weights[k] for k in SHAPES],engine='vm')
  return out['value']['data']
 def predict(self,plan,journal_permission=False):
  validate_plan(plan);x=plan['initial'];mem=plan['memory'];actions=[];conf=[];guard=None;result=None;attempts=0;allow=0
  for j,s0 in enumerate(plan['steps']):
   s=copy.deepcopy(s0);s['flags']['journal_allowed']=journal_permission is True and s['flags']['journal_allowed'] is True
   logits=self.logits(features(s,x,mem is not None));m=max(logits);den=sum(math.exp(v-m) for v in logits);a=max(range(len(logits)),key=lambda k:logits[k]);act=ACTIONS[a]
   actions.append(act);conf.append(math.exp(logits[a]-m)/den)
   if act in ERRORS:result={'tag':'Err','error':act};break
   if act=='JOURNAL':attempts+=1
   err=protocol_guard(act,s0,mem,journal_permission)
   if err:guard={'step':j,'reason':err};result={'tag':'Err','error':err};break
   try:x,mem=advance_runtime(act,x,s0['arg'],mem)
   except (ICLError,OverflowError,ValueError) as e:
    reason=getattr(e,'code',str(e));guard={'step':j,'reason':reason};result={'tag':'Err','error':reason};break
   if act=='JOURNAL':allow+=1
  return {'actions':actions,'confidence':min(conf) if conf else 0.0,'composed_result':result or {'tag':'Ok','value':x},'guard_rejection':guard,'journal_attempts':attempts,'journal_allow_decisions':allow,'execution_scope':'host simulation until vm_verify executes the candidate'}
def main(argv=None):
 p=argparse.ArgumentParser();p.add_argument('--model',type=Path,required=True);p.add_argument('--request',type=Path,required=True);p.add_argument('--output',type=Path);p.add_argument('--permit-local-journal',action='store_true');a=p.parse_args(argv)
 try:
  m=OfflineModel(a.model);req=strict_load(a.request,100000);pred=m.predict(req,a.permit_local_journal);v=vm_verify(req,pred,a.permit_local_journal)
  answer={'schema':'ICL_R5_PRODUCT_RESULT/1','model_seed':m.cfg['seed'],'proposal':pred,'verification':v,'scope':'finite structured intent -> learned opcode -> deterministic ICL code and VM','learned':'opcode and finite error-class decision','not_learned':'source syntax, composition control, exact arithmetic, authoritative permission, persistent memory','confidence_is_uncalibrated':True,'tensor_backend':'ICL VM pure Python float64','third_party_weights_used':False}
  if a.output:dump(a.output,answer)
  print(canonical(answer));return 0
 except Exception as e:
  print(canonical({'status':'ERROR','error_type':type(e).__name__,'reason':str(e)}),file=sys.stderr);return 2
if __name__=='__main__':raise SystemExit(main())