"""Illustrative offline operation profile. No authorization, network, or execution.""" import json, math, re, sys from datetime import datetime, timezone from pathlib import Path SCHEMA = json.loads(r'''{"$schema":"https://json-schema.org/draft/2020-12/schema","$id":"urn:zeroexternal:operation:schema:v1","title":"ZeroExternal operation profile v1 (illustrative proposal)","$comment":"Shape validation is not authorization. Runtime validators additionally enforce depth <= 12, nodes <= 512 and cumulative UTF-8 strings/keys <= 16384 bytes. The schema does not enforce those aggregate limits, Unicode scalar validity or calendar semantics by itself. Use the SDK conformance validator. The source pattern is a conservative absolute HTTPS/URN URI subset. UTC timestamps allow seconds 00-59 and 1-3 fractional digits; no leap seconds. All time formats must be actively asserted by external validators.","type":"object","additionalProperties":false,"required":["specversion","id","source","type","subject","time","datacontenttype","data"],"properties":{"specversion":{"const":"1.0"},"id":{"type":"string","minLength":1,"maxLength":36,"pattern":"^[0-9a-f]{8}-[0-9a-f]{4}-[1-8][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$"},"source":{"type":"string","minLength":1,"maxLength":256,"pattern":"^(urn:[A-Za-z0-9][A-Za-z0-9-]{0,31}:[A-Za-z0-9][A-Za-z0-9._~:/-]{0,255}|https://[A-Za-z0-9]([A-Za-z0-9-]*[A-Za-z0-9])?(\\.[A-Za-z0-9]([A-Za-z0-9-]*[A-Za-z0-9])?)*(/[A-Za-z0-9._~:/-]*)?)$","format":"uri"},"type":{"const":"ai.zeroexternal.operation.requested.v1"},"subject":{"type":"string","minLength":1,"maxLength":128,"pattern":"^[A-Za-z0-9][A-Za-z0-9._:/-]{0,127}$"},"time":{"type":"string","minLength":1,"maxLength":24,"pattern":"^[0-9]{4}-(0[1-9]|1[0-2])-(0[1-9]|[12][0-9]|3[01])T([01][0-9]|2[0-3]):[0-5][0-9]:[0-5][0-9](\\.[0-9]{1,3})?Z$","format":"date-time"},"datacontenttype":{"const":"application/json"},"data":{"type":"object","additionalProperties":false,"required":["profile","operation_id","scenario_id","actor","intent","capability_ref","budget","idempotency_key","evidence_required","acceptance","settlement"],"properties":{"profile":{"const":"zeroexternal.operation.v1"},"operation_id":{"type":"string","minLength":1,"maxLength":36,"pattern":"^[0-9a-f]{8}-[0-9a-f]{4}-[1-8][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$"},"scenario_id":{"enum":["science","agents","people","robotics","payments","supply-chain","healthcare","manufacturing","agriculture","real-estate","creative","enterprise","custom"]},"actor":{"type":"object","additionalProperties":false,"required":["id","kind","owner_id"],"properties":{"id":{"type":"string","minLength":1,"maxLength":128,"pattern":"^[A-Za-z0-9][A-Za-z0-9._:/-]{0,127}$"},"kind":{"enum":["agent","person","robot","service","organization"]},"owner_id":{"type":"string","minLength":1,"maxLength":128,"pattern":"^[A-Za-z0-9][A-Za-z0-9._:/-]{0,127}$"}}},"intent":{"type":"object","additionalProperties":false,"required":["action","target","expected_revision","parameters"],"properties":{"action":{"type":"string","minLength":1,"maxLength":128,"pattern":"^[A-Za-z0-9][A-Za-z0-9._:/-]{0,127}$"},"target":{"type":"string","minLength":1,"maxLength":128,"pattern":"^[A-Za-z0-9][A-Za-z0-9._:/-]{0,127}$"},"expected_revision":{"type":"integer","minimum":0,"maximum":1000000000},"parameters":{"type":"object","maxProperties":64,"propertyNames":{"type":"string","minLength":1,"maxLength":64,"not":{"enum":["__proto__","prototype","constructor"]}},"additionalProperties":{"$ref":"#/$defs/parameter"}}}},"capability_ref":{"type":"string","minLength":1,"maxLength":128,"pattern":"^[A-Za-z0-9][A-Za-z0-9._:/-]{0,127}$"},"budget":{"type":"object","additionalProperties":false,"required":["unit","maximum"],"properties":{"unit":{"type":"string","minLength":1,"maxLength":128,"pattern":"^[A-Za-z0-9][A-Za-z0-9._:/-]{0,127}$"},"maximum":{"type":"integer","minimum":0,"maximum":1000000000}}},"idempotency_key":{"type":"string","minLength":1,"maxLength":128,"pattern":"^[A-Za-z0-9][A-Za-z0-9._:/-]{0,127}$"},"evidence_required":{"type":"array","minItems":1,"maxItems":6,"uniqueItems":true,"items":{"enum":["artifact_digest","test_report","review_record","execution_trace","measurement","source_reference"]}},"acceptance":{"type":"object","additionalProperties":false,"required":["reviewer_id","criteria"],"properties":{"reviewer_id":{"type":"string","minLength":1,"maxLength":128,"pattern":"^[A-Za-z0-9][A-Za-z0-9._:/-]{0,127}$"},"criteria":{"type":"array","minItems":1,"maxItems":8,"uniqueItems":true,"items":{"type":"string","minLength":1,"maxLength":256}}}},"settlement":{"type":"object","additionalProperties":false,"required":["mode","automatic"],"properties":{"mode":{"enum":["none","separate_acceptance"]},"automatic":{"const":false}}}}}},"$defs":{"parameter":{"anyOf":[{"type":"null"},{"type":"boolean"},{"type":"number","minimum":-9007199254740991,"maximum":9007199254740991},{"type":"string","maxLength":2048},{"type":"array","maxItems":64,"items":{"$ref":"#/$defs/parameter"}},{"type":"object","maxProperties":64,"propertyNames":{"type":"string","minLength":1,"maxLength":64,"not":{"enum":["__proto__","prototype","constructor"]}},"additionalProperties":{"$ref":"#/$defs/parameter"}}]}}}''') C = json.loads(r'''{"profile":"zeroexternal.operation.v1","scenarios":["science","agents","people","robotics","payments","supply-chain","healthcare","manufacturing","agriculture","real-estate","creative","enterprise","custom"],"evidence":["artifact_digest","test_report","review_record","execution_trace","measurement","source_reference"],"ident":"^[A-Za-z0-9][A-Za-z0-9._:/-]{0,127}$","uuid":"^[0-9a-f]{8}-[0-9a-f]{4}-[1-8][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$","time":"^[0-9]{4}-(0[1-9]|1[0-2])-(0[1-9]|[12][0-9]|3[01])T([01][0-9]|2[0-3]):[0-5][0-9]:[0-5][0-9](\\.[0-9]{1,3})?Z$","source":"^(urn:[A-Za-z0-9][A-Za-z0-9-]{0,31}:[A-Za-z0-9][A-Za-z0-9._~:/-]{0,255}|https://[A-Za-z0-9]([A-Za-z0-9-]*[A-Za-z0-9])?(\\.[A-Za-z0-9]([A-Za-z0-9-]*[A-Za-z0-9])?)*(/[A-Za-z0-9._~:/-]*)?)$"}''') PROFILE = C['profile'] SCENARIOS = tuple(C['scenarios']) EVIDENCE_TYPES = tuple(C['evidence']) LIMITS = {'maxDepth':12,'maxNodes':512,'maxStringBytes':16384,'maxWireBytes':32768} FORBIDDEN = {'__proto__','constructor','prototype'} def pointer(p,k): return p+'/'+str(k).replace('~','~0').replace('/','~1') def issue(path,code): return {'path':path,'code':code} def timestamp(s): if not isinstance(s,str) or not re.fullmatch(C['time'],s) or s.startswith('0000-'): return None try: return datetime.fromisoformat(s.replace('Z','+00:00')).timestamp()*1000 except ValueError: return None def bounded(value): nodes=0 nbytes=0 stack=set() def string(v,p): nonlocal nbytes if any(0xD800<=ord(c)<=0xDFFF for c in v): return issue(p,'unicode') nbytes+=len(v.encode('utf-8')) if len(v)>2048 or nbytes>LIMITS['maxStringBytes']: return issue(p,'resource_limit') def visit(v,p,depth): nonlocal nodes nodes+=1 if nodes>LIMITS['maxNodes'] or depth>LIMITS['maxDepth']: return issue(p,'resource_limit') if v is None or type(v) is bool: return None if type(v) is str: return string(v,p) if type(v) in (int,float): return None if abs(v)<=9007199254740991 and math.isfinite(v) else issue(p,'number_range') if type(v) not in (dict,list): return issue(p,'json_type') if id(v) in stack: return issue(p,'cycle') if len(v)>64: return issue(p,'resource_limit') if type(v) is dict and any(type(k) is not str for k in v): return issue(p,'json_type') stack.add(id(v)) keys=sorted(v) if type(v) is dict else sorted(range(len(v)),key=str) for k in keys: if type(v) is dict: if not k or len(k)>64 or k in FORBIDDEN: return issue(pointer(p,k),'property_name') e=string(k,pointer(p,k)) if e: return e e=visit(v[k],pointer(p,k),depth+1) if e: return e stack.remove(id(v)) return visit(value,'',0) def matches(v,t): if t=='object': return type(v) is dict if t=='array': return type(v) is list if t=='number': return type(v) in (int,float) if t=='integer': return type(v) in (int,float) and v==int(v) if t=='null': return v is None if t=='boolean': return type(v) is bool if t=='string': return type(v) is str return False def equal(a,b): return type(a) is type(b) and a==b def validate_with(value,schema): early=bounded(value) if early: return [early] errors=[] def walk(v,s,p): if '$ref' in s: s=SCHEMA['$defs'][s['$ref'].split('/')[-1]] if 'anyOf' in s: option=next((o for o in s['anyOf'] if matches(v,o['type'])),None) if option is None: errors.append(issue(p,'type')); return return walk(v,option,p) if 'type' in s and not matches(v,s['type']): errors.append(issue(p,'type')); return if 'const' in s and not equal(v,s['const']): errors.append(issue(p,'const')) if 'enum' in s and not any(equal(v,x) for x in s['enum']): errors.append(issue(p,'enum')) if type(v) is str: if ('minLength' in s and len(v)s['maxLength']): errors.append(issue(p,'length')) if 'pattern' in s and re.fullmatch(s['pattern'],v) is None: errors.append(issue(p,'pattern')) if s.get('format')=='date-time' and timestamp(v) is None: errors.append(issue(p,'date_time')) if type(v) in (int,float) and (('minimum' in s and vs['maximum'])): errors.append(issue(p,'range')) if type(v) is list: if ('minItems' in s and len(v)s['maxItems']): errors.append(issue(p,'items')) if s.get('uniqueItems') and len({json.dumps(x,sort_keys=True) for x in v})!=len(v): errors.append(issue(p,'unique')) if 'items' in s: for i,x in enumerate(v): walk(x,s['items'],pointer(p,i)) if type(v) is dict: for k in s.get('required',[]): if k not in v: errors.append(issue(pointer(p,k),'required')) if 'maxProperties' in s and len(v)>s['maxProperties']: errors.append(issue(p,'properties')) for k in sorted(v): if k in s.get('properties',{}): walk(v[k],s['properties'][k],pointer(p,k)) elif s.get('additionalProperties') is False: errors.append(issue(pointer(p,k),'unknown_field')) elif type(s.get('additionalProperties')) is dict: walk(v[k],s['additionalProperties'],pointer(p,k)) walk(value,schema,'') return sorted(errors,key=lambda e:(e['path'],e['code'])) def validateOperation(value): errors=validate_with(value,SCHEMA) return {'valid':not errors,'errors':errors,'authorization':'not_evaluated'} def exact(properties): return {'type':'object','required':list(properties),'properties':properties,'additionalProperties':False} IDENT=SCHEMA['properties']['data']['properties']['capability_ref'] D=SCHEMA['properties']['data']['properties'] POLICY_SCHEMA=exact({'reference':IDENT,'actor_id':IDENT,'action':IDENT,'target':IDENT,'expected_revision':D['intent']['properties']['expected_revision'],'expires_at':SCHEMA['properties']['time'],'budget':exact({'unit':IDENT,'limit':D['budget']['properties']['maximum']}),'scenario_id':{'enum':C['scenarios']}}) def evaluatePolicy(operation,policy,now): def result(reasons): return {'decision':'deny' if reasons else 'allow','reasons':reasons,'simulation_only':True,'authoritative':False} if not validateOperation(operation)['valid']: return result(['contract_invalid']) if validate_with(policy,POLICY_SCHEMA): return result(['policy_invalid']) clock=timestamp(now) if clock is None: return result(['clock_invalid']) d=operation['data']; reasons=[] checks=[(d['capability_ref'],policy['reference'],'capability_reference_mismatch'),(d['actor']['id'],policy['actor_id'],'actor_mismatch'),(d['intent']['action'],policy['action'],'action_mismatch'),(d['intent']['target'],policy['target'],'target_mismatch'),(d['intent']['expected_revision'],policy['expected_revision'],'revision_mismatch'),(d['scenario_id'],policy['scenario_id'],'scenario_mismatch'),(d['budget']['unit'],policy['budget']['unit'],'budget_unit_mismatch')] for a,b,reason in checks: if a!=b: reasons.append(reason) if clock>=timestamp(policy['expires_at']): reasons.append('policy_expired') if d['budget']['maximum']>policy['budget']['limit']: reasons.append('budget_exceeded') return result(reasons) def parseOperationJSON(text): def invalid(code): return {'operation':None,'validation':{'valid':False,'errors':[issue('',code)],'authorization':'not_evaluated'}} if type(text) is not str: return invalid('wire_limit') try: if len(text.encode('utf-8'))>LIMITS['maxWireBytes']: return invalid('wire_limit') except UnicodeEncodeError: return invalid('invalid_json') try: def unique_pairs(pairs): result={} for k,v in pairs: if k in result: raise ValueError('duplicate_json_key') result[k]=v return result operation=json.loads(text,object_pairs_hook=unique_pairs,parse_constant=lambda _: (_ for _ in ()).throw(ValueError())) return {'operation':operation,'validation':validateOperation(operation)} except (ValueError,RecursionError) as error: return invalid('duplicate_json_key' if str(error)=='duplicate_json_key' else 'invalid_json') ACTIONS = json.loads(r'''{"science":"propose_experiment","agents":"propose_delegation","people":"propose_review","robotics":"propose_route","payments":"propose_invoice","supply-chain":"propose_shipment","healthcare":"propose_care_review","manufacturing":"propose_work_order","agriculture":"propose_irrigation_plan","real-estate":"propose_inspection","creative":"propose_asset","enterprise":"propose_procurement","custom":"propose_task"}''') def makeExample(scenario='science'): if scenario not in SCENARIOS: raise ValueError('Unknown scenario') suffix=str(SCENARIOS.index(scenario)+1).zfill(12) return {'specversion':'1.0','id':f'a1700000-0000-4000-8000-{suffix}','source':'urn:zeroexternal:illustrative-producer','type':'ai.zeroexternal.operation.requested.v1','subject':f'{scenario}/task-001','time':'2026-09-26T12:00:00Z','datacontenttype':'application/json','data':{'profile':PROFILE,'operation_id':f'b2700000-0000-4000-8000-{suffix}','scenario_id':scenario,'actor':{'id':'actor:example','kind':'robot' if scenario=='robotics' else 'person' if scenario=='people' else 'agent','owner_id':'owner:example'},'intent':{'action':ACTIONS[scenario],'target':f'{scenario}/task-001','expected_revision':1,'parameters':{'proposal':'illustrative','domain_adapter_required':True}},'capability_ref':'capability:example-reference','budget':{'unit':'review_units','maximum':10},'idempotency_key':f'{scenario}:task-001:proposal-1','evidence_required':['artifact_digest','review_record'],'acceptance':{'reviewer_id':'reviewer:example','criteria':['Domain adapter validates all parameters.','Named reviewer separately accepts the evidence.']},'settlement':{'mode':'separate_acceptance','automatic':False}}} def makeExamplePolicy(operation=None): d=(operation if operation is not None else makeExample())['data'] return {'reference':d['capability_ref'],'actor_id':d['actor']['id'],'action':d['intent']['action'],'target':d['intent']['target'],'expected_revision':d['intent']['expected_revision'],'expires_at':'2026-09-26T13:00:00Z','budget':{'unit':d['budget']['unit'],'limit':10},'scenario_id':d['scenario_id']} # Python-style aliases have the same contract. validate_operation=validateOperation evaluate_policy=evaluatePolicy def main(): if len(sys.argv)!=3 or sys.argv[1]!='check': print('Usage: python operation_profile.py check operation.json',file=sys.stderr); return 2 try: with open(sys.argv[2],'rb') as f: raw=f.read(LIMITS['maxWireBytes']+1) if len(raw)>LIMITS['maxWireBytes']: result={'valid':False,'errors':[issue('','wire_limit')],'authorization':'not_evaluated'} else: result=parseOperationJSON(raw.decode('utf-8'))['validation'] except (OSError,UnicodeDecodeError): result={'valid':False,'errors':[issue('','input_error')],'authorization':'not_evaluated'} print(json.dumps(result,indent=2)); return 0 if result['valid'] else 1 if __name__=='__main__': sys.exit(main())