Something went wrong. Try again.
Architecture research toward an agent-native operating system; rigorous, pre-prototype, and intentionally self-aware.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249"""Dependency-free, candidate-neutral D-002 scope interpreter and selector."""import hashlibimport jsonimport unicodedata
U64_MAX=(1<<64)-1FACT_TYPES=("slot_binding","revocation_fact","budget_ledger_fact","inbox_delivery_occurrence","inbox_processing_projection","execution_occurrence","resource_acquisition_fact","resource_binding_plan","observation_ingress","observation_interpretation","effect_intent","external_occurrence_evidence","settled_effect_fact","commitment_identity","commitment_progress_projection")M_WORLD=frozenset({"slot_binding","inbox_processing_projection","resource_binding_plan","observation_interpretation","effect_intent","commitment_progress_projection"})EXPECTED_SCOPE_VECTORS={ "G":{fact:"GLOBAL" for fact in FACT_TYPES}, "W":{fact:"WORLD" for fact in FACT_TYPES}, "M":{fact:("WORLD" if fact in M_WORLD else "GLOBAL") for fact in FACT_TYPES},}TRACE_IDS=tuple("S%02d"%i for i in range(1,11))INVARIANT_IDS=tuple("I%d"%i for i in range(1,8))TRACE_INVARIANTS={ "S01":["I1","I1","I5"],"S02":["I2","I2","I5"],"S03":["I3","I3","I3","I3","I5"], "S04":["I4","I4","I5","I5","I5"],"S05":["I3","I3","I4","I5","I5","I1"], "S06":["I3","I3","I4","I5","I5","I5"],"S07":["I4","I5","I5","I5","I4","I7"], "S08":["I4","I4","I4","I5"],"S09":["I4","I3","I4","I5","I5","I5","I5"], "S10":["I6","I6","I6","I6","I6","I6","I6","I2","I3","I4","I5","I5"]}TRACE_PREDICATES={ "S01":["same","values_equal","refs_exact"],"S02":["field_equal","field_equal","refs_exact"], "S03":["field_equal","field_equal","field_equal","field_equal","refs_exact"], "S04":["count","different","refs_exact","refs_exact","refs_exact"], "S05":["field_equal","count","count","refs_exact","refs_exact","count"], "S06":["field_equal","field_equal","count","refs_exact","refs_exact","refs_exact"], "S07":["count","refs_exact","refs_exact","refs_exact","different","field_equal"], "S08":["count","count","count","refs_exact"], "S09":["count","count","different","refs_exact","refs_exact","refs_exact","refs_exact"], "S10":["values_equal","values_equal","refs_exact","field_equal","count","values_equal","field_equal","same","same","same","refs_exact","refs_exact"]}SETTLED_EFFECT_NOTE="Conditional GLOBAL classification only; existence and lifecycle are owned by D-004/D-005 and no settlement transition is exercised."OPS={"fork","append","read","discard","allocate","spend","acquire","authority_use","promote"}PREDICATES={"same","different","count","values_equal","field_equal","refs_exact","refs_contain"}
def _json_value(v): if isinstance(v,str): if unicodedata.normalize("NFC",v)!=v: raise ValueError("all strings and keys must be NFC") elif v is None or isinstance(v,bool): pass elif isinstance(v,int): if isinstance(v,bool) or not 0<=v<=U64_MAX: raise ValueError("integers must be unsigned u64") elif isinstance(v,float): raise ValueError("floats are forbidden") elif isinstance(v,list): for x in v:_json_value(x) elif isinstance(v,dict): for k,x in v.items(): if not isinstance(k,str) or unicodedata.normalize("NFC",k)!=k: raise ValueError("all strings and keys must be NFC") _json_value(x) else: raise ValueError("non-JSON value")
def canonical_bytes(value): """Canonical NFC UTF-8 JSON; keys sort by UTF-8 bytes; compact; one LF.""" _json_value(value) def order(v): if isinstance(v,dict): return {k:order(v[k]) for k in sorted(v,key=lambda k:k.encode("utf-8"))} if isinstance(v,list): return [order(x) for x in v] return v return (json.dumps(order(value),ensure_ascii=False,separators=(",",":"),allow_nan=False)+"\n").encode("utf-8")
def parse_canonical(data): if not isinstance(data,bytes): raise ValueError("manifest must be supplied as bytes") try: text=data.decode("utf-8") value=json.loads(text,parse_float=lambda _:(_ for _ in ()).throw(ValueError("floats forbidden")),parse_constant=lambda _:(_ for _ in ()).throw(ValueError("constants forbidden"))) except (UnicodeError,json.JSONDecodeError) as e: raise ValueError("invalid UTF-8 JSON") from e if canonical_bytes(value)!=data: raise ValueError("non-canonical manifest bytes") return value
def _keys(obj,required,optional=()): keys=set(obj) if isinstance(obj,dict) else set() if not isinstance(obj,dict) or not set(required)<=keys or not keys<=set(required)|set(optional): raise ValueError("schema closure violation")
def _u64(v): return isinstance(v,int) and not isinstance(v,bool) and 0<=v<=U64_MAX
def _text(v,label): if not isinstance(v,str) or not v: raise ValueError("invalid "+label)
def _refs(v): if not isinstance(v,dict) or not v: raise ValueError("invalid refs") for key,value in v.items(): _text(key,"ref key"); _text(value,"ref value")
def validate_manifest(m): _json_value(m); _keys(m,{"schema_version","decision","metadata","candidates","fact_types","invariants","traces"}) if not _u64(m["schema_version"]) or m["schema_version"]!=2: raise ValueError("unsupported schema version") _keys(m["decision"],{"id","status"}) if m["decision"]!={"id":"D-002","status":"OPEN"}: raise ValueError("decision identity/status mismatch") _keys(m["metadata"],{"artifact_id","invalidates","invalidation_reason","settled_effect_fact"}) expected_metadata={"artifact_id":"d002-scope-manifest-v2","invalidates":"d002-scope-manifest-v1","invalidation_reason":"Pre-execution adversarial review found that v1 validation accepted freshly rehashed candidate-vector swaps; v2 binds the exact preregistered G/W/M vectors.","settled_effect_fact":SETTLED_EFFECT_NOTE} if m["metadata"]!=expected_metadata: raise ValueError("metadata mismatch") if m["fact_types"]!=list(FACT_TYPES): raise ValueError("fact types mismatch") if not isinstance(m["candidates"],list) or [x.get("id") for x in m["candidates"]]!=["G","W","M"]: raise ValueError("candidate coverage mismatch") for candidate in m["candidates"]: _keys(candidate,{"id","scope_vector"}); _text(candidate["id"],"candidate id") if not isinstance(candidate["scope_vector"],dict) or set(candidate["scope_vector"])!=set(FACT_TYPES) or any(not isinstance(x,str) or x not in ("GLOBAL","WORLD") for x in candidate["scope_vector"].values()): raise ValueError("invalid scope vector") if candidate["scope_vector"]!=EXPECTED_SCOPE_VECTORS[candidate["id"]]: raise ValueError("candidate scope vector differs from preregistered contract") if not isinstance(m["invariants"],list) or [x.get("id") for x in m["invariants"]]!=list(INVARIANT_IDS): raise ValueError("invariant coverage mismatch") for inv in m["invariants"]: _keys(inv,{"id","statement"}) if not isinstance(inv["statement"],str) or not inv["statement"]: raise ValueError("invalid invariant") if not isinstance(m["traces"],list) or [x.get("id") for x in m["traces"]]!=list(TRACE_IDS): raise ValueError("trace coverage mismatch") op_fields={ "fork":({"id","op","parent","world"},set()),"append":({"id","op","fact_type","name","refs","value","world"},set()), "read":({"id","op","fact_type","world"},{"scope"}),"discard":({"id","op","world"},set()), "allocate":({"id","op","counter","amount","refs"},set()),"spend":({"id","op","counter","amount","name","refs","world"},set()), "acquire":({"id","op","resource","name","refs","world"},set()),"authority_use":({"id","op","grant","world"},{"granted"}), "promote":({"id","op","names","world"},set())} pred_fields={"same":{"invariant","predicate","left","right"},"different":{"invariant","predicate","left","right"},"count":{"invariant","predicate","observation","count"},"values_equal":{"invariant","predicate","observation","values"},"field_equal":{"invariant","predicate","observation","field","value"},"refs_exact":{"invariant","predicate","observation","refs"},"refs_contain":{"invariant","predicate","observation","refs"}} for trace in m["traces"]: _keys(trace,{"id","operations","assertions"}) if not isinstance(trace["operations"],list) or not isinstance(trace["assertions"],list) or not trace["operations"] or not trace["assertions"]: raise ValueError("empty trace") ids=[] for operation in trace["operations"]: kind=operation.get("op") if isinstance(operation,dict) else None if kind not in OPS: raise ValueError("unsupported operation") req,opt=op_fields[kind]; _keys(operation,req,opt); _text(operation["id"],"operation id"); _text(kind,"operation") ; ids.append(operation["id"]) for field in ("parent","world","fact_type","name","counter","resource","grant"): if field in operation: _text(operation[field],field) if "refs" in operation: _refs(operation["refs"]) if "scope" in operation and operation["scope"] not in ("GLOBAL","WORLD"): raise ValueError("invalid scope") if "names" in operation: if not isinstance(operation["names"],list) or not operation["names"]: raise ValueError("invalid names") for name in operation["names"]: _text(name,"name") if len(operation["names"])!=len(set(operation["names"])): raise ValueError("duplicate Promotion name") if "granted" in operation and not isinstance(operation["granted"],bool): raise ValueError("invalid granted") if "fact_type" in operation and operation["fact_type"] not in FACT_TYPES+("*",): raise ValueError("unknown fact type") if "amount" in operation and not _u64(operation["amount"]): raise ValueError("operation integer must be unsigned u64") if len(ids)!=len(set(ids)): raise ValueError("duplicate operation id") named=set(ids) invariant_order=[]; predicate_order=[] for a in trace["assertions"]: pred=a.get("predicate") if isinstance(a,dict) else None if pred not in PREDICATES: raise ValueError("unsupported predicate") _keys(a,pred_fields[pred]); _text(a["invariant"],"invariant"); _text(pred,"predicate") if a["invariant"] not in INVARIANT_IDS: raise ValueError("unknown invariant") invariant_order.append(a["invariant"]); predicate_order.append(pred) if "count" in a and not _u64(a["count"]): raise ValueError("assertion integer must be unsigned u64") if pred=="field_equal": _text(a["field"],"field") for field in ("values","refs"): if field in a and not isinstance(a[field],list): raise ValueError("invalid assertion list") if "refs" in a: for ref in a["refs"]: _refs(ref) for field in ("observation","left","right"): if field in a: _text(a[field],field) if a[field].split(".",1)[0] not in named: raise ValueError("assertion references unknown observation") if invariant_order!=TRACE_INVARIANTS[trace["id"]] or predicate_order!=TRACE_PREDICATES[trace["id"]]: raise ValueError("trace assertion contract mismatch") return m
def load_manifest(data): return validate_manifest(parse_canonical(data))
def _visible(state,world,fact_type,scope=None): records=[] if scope in (None,"GLOBAL"): records.extend(state["global"]) if scope in (None,"WORLD"): if world not in state["worlds"]: raise ValueError("unknown/discarded World") records.extend(state["worlds"][world]) return [r for r in records if fact_type=="*" or r["fact_type"]==fact_type]
def _view(records): return {"count":len(records),"records":records,"refs":[r["refs"] for r in records],"values":[r["value"] for r in records],"record_scopes":[r["scope"] for r in records]}
def _append(state,vector,fact_type,name,refs,value,world): if world not in state["worlds"]: raise ValueError("unknown/discarded World") record={"fact_type":fact_type,"name":name,"refs":refs,"scope":vector[fact_type],"value":value} (state["global"] if record["scope"]=="GLOBAL" else state["worlds"][world]).append(record) return record
def _resolve(observations,path): bits=path.split("."); value=observations[bits[0]] for bit in bits[1:]: value=value[bit] return value
def _assert(a,obs): p=a["predicate"] if p=="same": actual={"left":_resolve(obs,a["left"]),"right":_resolve(obs,a["right"])}; passed=actual["left"]==actual["right"] elif p=="different": actual={"left":_resolve(obs,a["left"]),"right":_resolve(obs,a["right"])}; passed=actual["left"]!=actual["right"] elif p=="count": actual=obs[a["observation"]]["count"]; passed=actual==a["count"] elif p=="values_equal": actual=obs[a["observation"]]["values"]; passed=actual==a["values"] elif p=="field_equal": actual=obs[a["observation"]][a["field"]]; passed=actual==a["value"] elif p=="refs_exact": actual=obs[a["observation"]]["refs"]; passed=actual==a["refs"] else: actual=obs[a["observation"]]["refs"]; passed=all(x in actual for x in a["refs"]) return {"evidence":{"actual":actual,"predicate":p},"invariant":a["invariant"],"pass":passed}
def execute_trace(trace,scope_vector): """Execute one declarative trace using only its supplied scope vector.""" if set(scope_vector)!=set(FACT_TYPES): raise ValueError("incomplete scope vector") state={"global":[],"worlds":{"W0":[]},"allocations":{}} observations={}; emitted=[] for x in trace["operations"]: kind=x["op"] if kind=="fork": if x["parent"] not in state["worlds"] or x["world"] in state["worlds"]: raise ValueError("invalid fork") state["worlds"][x["world"]]=list(state["worlds"][x["parent"]]); out={"parent":x["parent"],"world":x["world"]} elif kind=="discard": if x["world"]=="W0" or x["world"] not in state["worlds"]: raise ValueError("invalid discard") del state["worlds"][x["world"]]; out={"discarded":x["world"]} elif kind=="append": out={"record":_append(state,scope_vector,x["fact_type"],x["name"],x["refs"],x["value"],x["world"])} elif kind=="read": out=_view(_visible(state,x["world"],x["fact_type"],x.get("scope"))) elif kind=="allocate": if x["counter"] in state["allocations"]: raise ValueError("counter already allocated") state["allocations"][x["counter"]]=x["amount"]; out={"allocated":x["amount"],"counter":x["counter"],"refs":x["refs"]} elif kind=="spend": prior=_visible(state,x["world"],"budget_ledger_fact"); spent=sum(r["value"]["amount"] for r in prior if r["value"]["accepted"] and r["value"]["counter"]==x["counter"]) accepted=x["counter"] in state["allocations"] and x["amount"]<=state["allocations"][x["counter"]]-spent value={"accepted":accepted,"amount":x["amount"],"counter":x["counter"]} if accepted:_append(state,scope_vector,"budget_ledger_fact",x["name"],x["refs"],value,x["world"]) out={"accepted":accepted,"spent":spent+(x["amount"] if accepted else 0)} elif kind=="acquire": accepted=not any(r["value"]==x["resource"] for r in _visible(state,x["world"],"resource_acquisition_fact")) if accepted:_append(state,scope_vector,"resource_acquisition_fact",x["name"],x["refs"],x["resource"],x["world"]) out={"accepted":accepted,"resource":x["resource"]} elif kind=="authority_use": revoked=any(r["value"]==x["grant"] for r in _visible(state,x["world"],"revocation_fact")); out={"allowed":bool(x.get("granted",False)) and not revoked,"revoked":revoked} else: if len(x["names"])!=len(set(x["names"])): raise ValueError("duplicate Promotion name") source=_visible(state,x["world"],"*"); matches={name:[r for r in source if r["name"]==name] for name in x["names"]} if any(len(records)!=1 for records in matches.values()): raise ValueError("Promotion names must identify exactly one visible fact") chosen=[matches[name][0] for name in x["names"]] if any(r["scope"]!="WORLD" for r in chosen): out={"accepted":False,"names":x["names"],"promoted":0,"reason":"selected fact is not WORLD"} else: existing={r["name"] for r in state["worlds"]["W0"]} if any(r["name"] in existing for r in chosen): raise ValueError("Promotion name is ambiguous with canonical state") for r in chosen: state["worlds"]["W0"].append(dict(r)); existing.add(r["name"]) out={"accepted":True,"names":x["names"],"promoted":len(chosen)} observations[x["id"]]=out; emitted.append({"id":x["id"],"observation":out,"op":kind}) evidence=[_assert(a,observations) for a in trace["assertions"]] return {"assertions":evidence,"operations":emitted,"pass":all(x["pass"] for x in evidence),"trace_id":trace["id"]}
def execute_manifest(manifest): validate_manifest(manifest) return {c["id"]:[execute_trace(t,c["scope_vector"]) for t in manifest["traces"]] for c in manifest["candidates"]}
def select(manifest_bytes,identity,results): """Hash-bind canonical input, recompute it, and accept only byte-exact JSON results.""" try: _keys(identity,{"byte_length","sha256"}) if not _u64(identity["byte_length"]) or not isinstance(identity["sha256"],str): raise ValueError("invalid manifest identity") if identity["byte_length"]!=len(manifest_bytes) or identity["sha256"]!=hashlib.sha256(manifest_bytes).hexdigest(): raise ValueError("manifest identity mismatch") manifest=load_manifest(manifest_bytes) expected=execute_manifest(manifest) _json_value(results) if results!=expected or canonical_bytes(results)!=canonical_bytes(expected): raise ValueError("caller results differ from deterministic execution") passing=[cid for cid,runs in expected.items() if all(r["pass"] for r in runs)] return {"decision_status":"SELECTED" if len(passing)==1 else "OPEN","selected_candidate":passing[0] if len(passing)==1 else None} except Exception as e: return {"blocked":True,"defect":str(e),"decision_status":"OPEN","selected_candidate":None}