File size: 2,938 Bytes
2e4432d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 | """Pure deterministic control/verification. No trained model or teaching oracle here.
These functions supply exact execution, not learned state prediction."""
from .common import ERRORS,validate_step
from .emitter import emit
from icl.runtime import checked_binary,checked_unary,ICLError
from icl.api import compile_source,run_ocir
def validate_plan(p):
if type(p)!=dict or set(p)!={'initial','memory','steps'}:raise ValueError('PLAN_SCHEMA')
if type(p['initial'])!=int or not -100000<=p['initial']<=100000:raise ValueError('INITIAL_RANGE')
if p['memory'] is not None and (type(p['memory'])!=int or not -100000<=p['memory']<=100000):raise ValueError('MEMORY_RANGE')
if type(p['steps'])!=list or not 1<=len(p['steps'])<=32:raise ValueError('STEP_LIMIT')
for s in p['steps']:validate_step(s)
def protocol_guard(action,s,memory,journal_permission):
"""Trusted wrapper check; NEVER modifies a prediction or grants capability."""
f=s['flags']
if action=='TRANSLATE':
for field,error in [('observer_match','OBSERVER_MISMATCH'),('frame_match','FRAME_MISMATCH'),('unit_match','UNIT_MISMATCH')]:
if not f[field]:return error
if action=='RESTORE':
if not f['witness_present']:return 'NON_IDENTIFIABLE'
if not f['witness_valid']:return 'WITNESS_CORRUPT'
if action=='RECALL' and memory is None:return 'MEMORY_MISSING'
if action=='JOURNAL' and (journal_permission is not True or f['journal_allowed'] is not True):return 'EFFECT_DENIED'
return None
def advance_runtime(action,x,a,memory):
"""Normative runtime state transition using checked ICL primitives, not labels."""
if action in ('ADD','TRANSLATE','RECORD_ADD','LIST_ADD'):x=checked_binary('+',x,a)
elif action=='SUB':x=checked_binary('-',x,a)
elif action=='MUL':x=checked_binary('*',x,a)
elif action=='MAX':x=max(x,a)
elif action=='MIN':x=min(x,a)
elif action=='ABS':x=checked_unary('-',x) if x<0 else x
elif action=='NEGATE':x=checked_unary('-',x)
elif action=='REPEAT':
for _ in range(3):x=checked_binary('+',x,a)
elif action=='STORE':memory=x
elif action=='RECALL':
if memory is None:raise ValueError('MEMORY_MISSING')
x=memory
elif action not in ('RESTORE','JOURNAL'):raise ValueError('ACTION_UNKNOWN')
return x,memory
def vm_verify(plan,pred,journal_permission=False,engine='vm'):
if pred['guard_rejection'] is not None:return {'executed':False,'reason':'INDEPENDENT_GUARD_REJECTED','raw_prediction_preserved':True}
actions=pred['actions']+[ERRORS[0]]*(len(plan['steps'])-len(pred['actions']))
src=emit(plan,actions);ocir=compile_source(src)
out=run_ocir(ocir,args=[plan['initial'],plan['memory'] or 0],engine=engine,policy={'journal':{'allow':journal_permission is True}})
return {'executed':True,'source_program':src,'ocir':ocir,'result':out,'runtime_consistency':out['value']==pred['composed_result']}
|