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.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793import copyimport hashlibimport inspectimport jsonimport pathlibimport randomimport unittest
try: from decisions import d001_runtime as dexcept ImportError: # Direct execution from decisions/. import d001_runtime as d
def ref(t, byte): width = 32 if t in {"PayloadId","ValueId","OccurrenceId","RevisionId"} else 16 return {"$kind":"ref","type":t,"hex":bytes([byte]).hex()*width}def lit(x): return {"op":"literal","value":x}def ep(source, value): return {"source":source,"value":value}
def profile(recipe="opaque"): ids = { "ActorId":d.IdRule("opaque","ActorId"), "EvidenceId":d.IdRule("opaque","EvidenceId"), "NodeId":d.IdRule("opaque","NodeId"), "SnapshotId":d.IdRule("opaque","SnapshotId"), "PinId":d.IdRule("opaque","PinId"), "NamespaceId":d.IdRule("opaque","NamespaceId"), "AliasId":d.IdRule("opaque","AliasId"), "LocatorRecordId":d.IdRule("opaque","LocatorRecordId"), "CommitmentId":d.IdRule("opaque","CommitmentId"), "PayloadId":d.IdRule("derived","PayloadId","payload.domain","payload.schema",fields=("schema_ref","codec_ref","body_bytes"),width=32), "ValueId":d.IdRule("derived","ValueId","value.domain","value.schema",fields=("subject_ref","payload_ref","tombstone"),width=32), "OccurrenceId":d.IdRule("derived","OccurrenceId","occ.domain","occ.schema",fields=("value_ref","predecessor_refs","actor_ref","nonce"),width=32), "RevisionId":d.IdRule("derived","RevisionId","rev.domain","rev.schema",fields=("subject_ref","payload_ref","predecessor_refs","actor_ref","nonce","tombstone"),width=32), } rr = { "Actor":{"id": {"ref":"ActorId"}, "name":"text"}, "Evidence":{"id":{"ref":"EvidenceId"},"data":"bytes"}, "Commitment":{"id":{"ref":"CommitmentId"},"actor_ref":{"ref":"ActorId"}}, "Namespace":{"id":{"ref":"NamespaceId"},"name":"text"}, "AliasVersion":{"id":{"ref":"AliasId"},"namespace_ref":{"ref":"NamespaceId"},"label":"text","target_ref":{"optional":{"ref":"NodeId"}},"predecessor_ref":{"optional":{"ref":"AliasId"}},"instance_order":"uint64","tombstone":"bool","expires_at":{"optional":"uint64"}}, "Node":{"id":{"ref":"NodeId"},"label":"text"}, "PayloadEnvelope":{"id":{"ref":"PayloadId"},"schema_ref":"text","codec_ref":"text","body_bytes":"bytes","body_digest":"digest32","body_length":"uint64"}, "PayloadBody":{"envelope_ref":{"ref":"PayloadId"},"body":"bytes"}, "Value":{"id":{"ref":"ValueId"},"subject_ref":{"ref":"NodeId"},"payload_ref":{"ref":"PayloadId"},"tombstone":"bool"}, "Occurrence":{"id":{"ref":"OccurrenceId"},"value_ref":{"ref":"ValueId"},"predecessor_refs":{"array":{"ref":"OccurrenceId"}},"actor_ref":{"ref":"ActorId"},"commitment_ref":{"ref":"CommitmentId"},"nonce":"text","authorization_evidence_ref":{"ref":"EvidenceId"},"instance_order":"uint64"}, "Revision":{"id":{"ref":"RevisionId"},"subject_ref":{"ref":"NodeId"},"payload_ref":{"ref":"PayloadId"},"predecessor_refs":{"array":{"ref":"RevisionId"}},"actor_ref":{"ref":"ActorId"},"nonce":"text","tombstone":"bool","instance_order":"uint64"}, "Snapshot":{"id":{"ref":"SnapshotId"},"label":"text"}, "Pin":{"id":{"ref":"PinId"},"target_ref":{"ref":["NodeId","PayloadId"]},"expires_at":"uint64","active":"bool"}, "Lifecycle":{"target_ref":{"ref":"NodeId"},"state":"text","tombstone":"bool","expires_at":{"optional":"uint64"},"instance_order":"uint64"}, "LocatorVersion":{"id":{"ref":"LocatorRecordId"},"target_ref":{"ref":"ActorId"},"endpoint":"text","available":"bool","tombstone":"bool","expires_at":{"optional":"uint64"},"predecessor_ref":{"optional":{"ref":"LocatorRecordId"}},"instance_order":"uint64"}, "SlotDef":{"slot":"slot","owner_ref":{"ref":"ActorId"},"kind":"text"}, "Cell":{"subject_ref":{"ref":"NodeId"},"payload_ref":{"ref":"PayloadId"},"counter":"uint64"}, "Content":{"id":{"ref":"PayloadId"},"typed":"bytes"}, } rules = {k:d.RecordRule(v, id_field=("id" if "id" in v else None), id_type=(v.get("id",{}).get("ref") if isinstance(v.get("id"),dict) else None)) for k,v in rr.items()} rules["Occurrence"] = d.RecordRule(rr["Occurrence"], id_type="OccurrenceId", instance_order_field="instance_order", predecessor_fields=("predecessor_refs",)) rules["Revision"] = d.RecordRule(rr["Revision"], id_type="RevisionId", instance_order_field="instance_order", predecessor_fields=("predecessor_refs",)) rules["AliasVersion"] = d.RecordRule(rr["AliasVersion"], id_type="AliasId", instance_order_field="instance_order", predecessor_fields=("predecessor_ref",)) rules["LocatorVersion"] = d.RecordRule(rr["LocatorVersion"], id_type="LocatorRecordId", instance_order_field="instance_order", predecessor_fields=("predecessor_ref",)) rules["SlotDef"] = d.RecordRule(rr["SlotDef"], id_field="slot") rules["PayloadBody"] = d.RecordRule(rr["PayloadBody"], id_field=None) rules["Lifecycle"] = d.RecordRule(rr["Lifecycle"], id_field=None, instance_order_field="instance_order") edges = { "node-payload":d.EdgeRule(frozenset({"NodeId"}),frozenset({"PayloadId"}),"identity-strong"), "node-node":d.EdgeRule(frozenset({"NodeId"}),frozenset({"NodeId"}),"audit-summary"), "occ-value":d.EdgeRule(frozenset({"OccurrenceId"}),frozenset({"ValueId"}),"identity-strong"), } availability=d.AvailabilityRole("PayloadEnvelope","PayloadId","PayloadBody","envelope_ref","body","body_digest","body_length","body_bytes") pin=d.PinRole("Pin","PinId","target_ref","active","expires_at") return d.ProfileSemantics("test-profile","fixture",ids,rules,edges, binding_kinds=frozenset({"state-head","active","root-policy","endpoint-occupant"}), state_recipe={"kind":recipe}, provenance_types=frozenset({"OccurrenceId","RevisionId"}), availability_role=availability, pin_role=pin, versioned_reference_roles=(d.VersionedReferenceRole("AliasVersion","AliasId","namespace_ref","label","target_ref","predecessor_ref","instance_order","tombstone","expires_at",("NamespaceId",),("NodeId",)),), provenance_roles=(d.ProvenanceRole("Occurrence","OccurrenceId","actor_ref","commitment_ref","authorization_evidence_ref","predecessor_refs",("ActorId",),("CommitmentId",),("EvidenceId",),("Actor",),("Commitment",),("Evidence",),("OccurrenceId",),("Occurrence",)),))
def action(step, op, args, fault=None): return {"action_version":"D001-ACTION-3","step_id":step,"op":op,"args":args,"fault":fault}
def opaque_args(rt, typ, symbol, fields, out=None, edges=None): return {"record_type":rt,"fields":fields,"id":{"kind":"opaque","id_type":typ,"domain_id":"main","symbol":symbol},"edges":edges or [],"out_ref":out}
def keyed_args(rt, key, fields, out=None): return {"record_type":rt,"fields":fields,"id":{"kind":"keyed","key":key},"edges":[],"out_ref":out}def derived_args(rt, typ, fields, out=None, edges=None): return {"record_type":rt,"fields":fields,"id":{"kind":"derived","id_type":typ},"edges":edges or [],"out_ref":out}
class RuntimeTest(unittest.TestCase): def setUp(self): self.p=profile(); self.r=d.Runtime(self.p); self.n=0 def go(self, op, args, expect=None, fault=None): self.n+=1; event=self.r.apply(action("s"+str(self.n),op,args,fault=fault)) if expect is not None: self.assertEqual(event["error"]["code"],expect) return event def put_actor(self, symbol="a", out="actor"): self.go("record.put",opaque_args("Actor","ActorId",symbol,{"id":ref("ActorId",0),"name":symbol},out)) # Runtime computes key independently; align id field to resulting ref is not required by generic schema. return self.r.store["registers"][out] def put_node(self, symbol="n", out="node"): self.go("record.put",opaque_args("Node","NodeId",symbol,{"id":ref("NodeId",0),"label":symbol},out)); return self.r.store["registers"][out] def put_evidence(self): self.go("record.put",opaque_args("Evidence","EvidenceId","e",{"id":ref("EvidenceId",0),"data":d.bytes_value(b"e")},"evidence")); return self.r.store["registers"]["evidence"] def put_payload(self, body=b"abc", out="payload", fault=None, expect=None, schema="s"): fields={"id":{"$kind":"ref","type":"PayloadId","hex":"00"*32},"schema_ref":schema,"codec_ref":"raw","body_bytes":d.bytes_value(body),"body_digest":d.bytes_value(hashlib.sha256(body).digest(),"digest32"),"body_length":len(body)} return self.go("record.put",derived_args("PayloadEnvelope","PayloadId",fields,out),expect,fault)
def test_record_put_bind_and_errors_no_commit(self): n=self.put_node(); self.assertEqual(self.r.query({"op":"reg","name":"node"})["value"],n) ev=self.go("record.put",opaque_args("Node","NodeId","n",{"id":ref("NodeId",0),"label":"n"}),"ALREADY_EXISTS") self.assertEqual(ev["status"],"NO_COMMIT"); self.assertEqual(ev["before_digest"],ev["after_digest"]) ev=self.go("record.put",opaque_args("Node","NodeId","n",{"id":ref("NodeId",0),"label":"changed"}),"DUPLICATE_ID") self.assertEqual(ev["status"],"NO_COMMIT") self.go("index.set",{"binding_kind":"state-head","key":n,"value":n,"expected_version":None,"out_version":"v"}) self.assertEqual(self.r.store["registers"]["v"],1) self.go("index.set",{"binding_kind":"state-head","key":n,"value":n,"expected_version":0,"out_version":None},"CONTEXT_MISMATCH") self.go("index.set",{"binding_kind":"bad","key":n,"value":n,"expected_version":None,"out_version":None},"UNSUPPORTED_BINDING") self.go("index.set",{"binding_kind":"state-head","key":ref("NodeId",9),"value":n,"expected_version":None,"out_version":None},"UNKNOWN_REFERENCE") self.go("record.put",opaque_args("Bad","NodeId","x",{},None),"UNSUPPORTED_RECORD") bad=opaque_args("Node","PayloadId","x",{"id":ref("NodeId",0),"label":"x"}) self.go("record.put",bad,"TYPE_MISMATCH") bad=opaque_args("Node","NodeId","x",{"id":ref("NodeId",0)}) self.go("record.put",bad,"MALFORMED")
def test_derived_collision_cross_type_and_fatal(self): self.put_payload(b"one"); target=self.r.store["registers"]["payload"] f={"kind":"HASH_RESULT_OVERRIDE","phase":"after-preimage-before-typed-lookup","raw_digest_hex":target["hex"]} # Same typed hash, unequal preimage is fatal and atomic; noncanonical fault outputs never persist. ev=self.put_payload(b"two",out=None,fault=f,expect="FATAL_INTEGRITY_FAILURE") self.assertEqual(ev["status"],"FATAL"); self.assertEqual(ev["before_digest"],ev["after_digest"]) # Equal raw bytes under a different concrete type remain a distinct typed reference. q=self.r.query({"op":"resolve","request":{"mode":"exact","reference":{"$kind":"ref","type":"NodeId","hex":target["hex"]},"context":None,"clock":0,"expected_types":["NodeId"]}}) self.assertEqual(q["status"],"MALFORMED")
def test_predecessor_edges_and_order_errors(self): a=self.put_actor(); e=self.put_evidence(); n=self.put_node(); self.put_payload(); p=self.r.store["registers"]["payload"] self.go("record.put",opaque_args("Commitment","CommitmentId","c",{"id":ref("CommitmentId",0),"actor_ref":a},"commit")); commit=self.r.store["registers"]["commit"] vf={"id":ref("ValueId",0),"subject_ref":n,"payload_ref":p,"tombstone":False} self.go("record.put",derived_args("Value","ValueId",vf,"val")); v=self.r.store["registers"]["val"] base={"id":ref("OccurrenceId",0),"value_ref":v,"predecessor_refs":[],"actor_ref":a,"commitment_ref":commit,"nonce":"1","authorization_evidence_ref":e,"instance_order":1} self.go("record.put",derived_args("Occurrence","OccurrenceId",base,"o1",[{"kind":"occ-value","from":ep("self",None),"to":ep("field","value_ref")}])) o1=self.r.store["registers"]["o1"] nxt={**base,"predecessor_refs":[o1],"nonce":"2","instance_order":2} self.go("record.put",derived_args("Occurrence","OccurrenceId",nxt,"o2")) bad={**base,"predecessor_refs":[o1],"nonce":"3","instance_order":1} self.go("record.put",derived_args("Occurrence","OccurrenceId",bad),"ORDER_VIOLATION") bad={**base,"predecessor_refs":[ref("OccurrenceId",7)],"nonce":"4","instance_order":3} self.go("record.put",derived_args("Occurrence","OccurrenceId",bad),"UNKNOWN_REFERENCE") bad={**base,"predecessor_refs":[n],"nonce":"5","instance_order":3} self.go("record.put",derived_args("Occurrence","OccurrenceId",bad),"TYPE_MISMATCH")
def test_availability_body_collection_restore_and_errors(self): self.put_payload(); p=self.r.store["registers"]["payload"] ev=self.go("availability.set",{"envelope_ref":p,"state":"AVAILABLE","body":d.bytes_value(b"abc"),"reason":None}) self.assertTrue(ev["output"]["verified"]) self.go("availability.set",{"envelope_ref":p,"state":"COLLECTED","body":None,"reason":None}) q=self.resolve(p,"payload-body",["PayloadId"]); self.assertEqual(q["status"],"CONTENT_COLLECTED") self.go("availability.set",{"envelope_ref":p,"state":"AVAILABLE","body":d.bytes_value(b"abc"),"reason":None}) self.assertEqual(self.resolve(p,"payload-body",["PayloadId"])["status"],"RESOLVED") self.go("availability.set",{"envelope_ref":p,"state":"AVAILABLE","body":d.bytes_value(b"bad"),"reason":None},"VERIFICATION_FAILED") self.go("availability.set",{"envelope_ref":ref("PayloadId",9),"state":"COLLECTED","body":None,"reason":None},"UNKNOWN") self.go("availability.set",{"envelope_ref":p,"state":"NOPE","body":None,"reason":None},"MALFORMED")
def test_mark_gated_collection_mid_epoch_resolution(self): self.put_payload(); payload=self.r.store["registers"]["payload"] self.go("availability.set",{"envelope_ref":payload,"state":"AVAILABLE","body":d.bytes_value(b"abc"),"reason":None}) self.go("roots.freeze",{"epoch":0,"root_refs":[],"pin_refs":[],"clock":0}) self.go("reachability.mark",{"epoch":0,"classes":["body-retaining"],"out":"mark"}) collected=self.go("availability.collect-if-unreachable",{"envelope_ref":payload,"epoch":0,"expected_reachable":False}); self.assertEqual(collected["output"]["status"],"COLLECTED") self.assertEqual(self.resolve(payload,"exact",["PayloadId"])["status"],"RESOLVED") self.assertEqual(self.resolve(payload,"payload-body",["PayloadId"])["status"],"CONTENT_COLLECTED") mismatch=self.go("availability.set",{"envelope_ref":payload,"state":"AVAILABLE","body":d.bytes_value(b"wrong"),"reason":None},"VERIFICATION_FAILED"); self.assertEqual(mismatch["before_digest"],mismatch["after_digest"]) restored=self.go("availability.set",{"envelope_ref":payload,"state":"AVAILABLE","body":d.bytes_value(b"abc"),"reason":None}); self.assertTrue(restored["output"]["verified"]); self.assertEqual(self.resolve(payload,"payload-body",["PayloadId"])["status"],"RESOLVED") print("PASS E15-mark-gated-body-collection")
def test_snapshot_binding_only_and_nonreuse(self): n=self.put_node(); self.go("index.set",{"binding_kind":"state-head","key":n,"value":n,"expected_version":None,"out_version":None}) snap=opaque_args("Snapshot","SnapshotId","snap",{"id":ref("SnapshotId",0),"label":"x"},"snap") self.go("snapshot.capture",{"name":"old","snapshot_record":snap,"binding_kinds":["state-head"],"root_context":{"x":1}}) n2=self.put_node("n2","n2"); self.go("index.set",{"binding_kind":"state-head","key":n,"value":n2,"expected_version":1,"out_version":None}) count=len(self.r.store["records"]); used=copy.deepcopy(self.r.store["allocations"]) self.go("snapshot.restore",{"name":"old"}) b=self.r.query({"op":"binding","kind":"state-head","key":lit(n)})["value"] self.assertEqual(b["value"],n); self.assertEqual(len(self.r.store["records"]),count); self.assertEqual(self.r.store["allocations"],used) self.go("record.put",opaque_args("Node","NodeId","n2",{"id":ref("NodeId",0),"label":"n2"}),"ALREADY_EXISTS") self.go("snapshot.restore",{"name":"missing"},"UNKNOWN") self.r.store["snapshots"]["old"]["root_context"]={"x":2} self.go("snapshot.restore",{"name":"old"},"VERIFICATION_FAILED") self.r.store["snapshots"]["old"]["root_context"]={"x":1} self.go("snapshot.capture",{"name":"old","snapshot_record":snap,"binding_kinds":["state-head"],"root_context":{}},"DUPLICATE_ID")
def test_roots_mark_fault_and_errors(self): n1=self.put_node("a","a"); n2=self.put_node("b","b"); n3=self.put_node("c","c") # Add a typed edge by creating a fresh node self and literal target. args=opaque_args("Node","NodeId","d",{"id":ref("NodeId",0),"label":"d"},"d",[{"kind":"node-node","from":ep("self",None),"to":ep("literal",n2)}]) self.go("record.put",args); root=self.r.store["registers"]["d"] pin=opaque_args("Pin","PinId","pin",{"id":ref("PinId",0),"target_ref":n1,"expires_at":10,"active":True},"pin") self.go("record.put",pin); pref=self.r.store["registers"]["pin"] self.go("roots.freeze",{"epoch":0,"root_refs":[root],"pin_refs":[pref],"clock":5}) fault={"kind":"CONCURRENT_ROOT_APPEND","phase":"after-root-copy-before-first-edge","epoch":1,"root_ref":n3} ev=self.go("reachability.mark",{"epoch":0,"classes":["audit-summary"],"out":"mark"},fault=fault) self.assertIn(d.typed_key(n2),ev["output"]["reachable"]); self.assertNotIn(d.typed_key(n3),ev["output"]["reachable"]) self.assertEqual(self.r.store["roots"]["pending"]["1"],[n3]) self.go("roots.freeze",{"epoch":0,"root_refs":[],"pin_refs":[],"clock":5},"CONTEXT_MISMATCH") self.go("roots.freeze",{"epoch":1,"root_refs":[ref("NodeId",9)],"pin_refs":[],"clock":5},"UNKNOWN_REFERENCE") self.go("reachability.mark",{"epoch":9,"classes":[],"out":None},"UNKNOWN") self.go("reachability.mark",{"epoch":0,"classes":[],"out":None},"CONTEXT_MISMATCH")
def resolve(self, reference, mode="exact", expected=None, context=None, clock=0): return self.r.query({"op":"resolve","request":{"mode":mode,"reference":reference,"context":context,"clock":clock,"expected_types":expected or [reference["type"]]}})
def test_resolver_statuses_and_precedence(self): n=self.put_node(); a=self.put_actor() self.assertEqual(self.resolve(n)["status"],"RESOLVED") self.assertEqual(self.r.query({"op":"resolve","request":{"bad":1}})["status"],"MALFORMED") self.assertEqual(self.resolve(n,expected=["ActorId"])["status"],"TYPE_MISMATCH") ctx={"binding_kind":"state-head","key":n,"expected_value":a} self.assertEqual(self.resolve(n,context=ctx)["status"],"CONTEXT_MISMATCH") self.assertEqual(self.resolve(ref("NodeId",8))["status"],"UNKNOWN") # Lifecycle-shaped exact records provide tombstone/expiry precedence. t=keyed_args("Lifecycle",{"parts":["t"]},{"target_ref":n,"state":"x","tombstone":True,"expires_at":0,"instance_order":1},"life") self.go("record.put",t); lr=self.r.store["registers"]["life"] # locator-current is target-native and accepts only an exact current-locator context. def locator(step,tomb=False,expires=None,available=False): args=opaque_args("LocatorVersion","LocatorRecordId",step,{"id":ref("LocatorRecordId",0),"target_ref":a,"endpoint":step,"available":available,"tombstone":tomb,"expires_at":expires,"predecessor_ref":None,"instance_order":len([x for x in self.r.store["records"].values() if x["record_type"]=="LocatorVersion"])+1},step); self.go("record.put",args); return self.r.store["registers"][step] l=locator("loc",tomb=True,expires=0); self.go("index.set",{"binding_kind":"endpoint-occupant","key":a,"value":l,"expected_version":None,"out_version":None}); ctx={"binding_kind":"endpoint-occupant","key":a,"expected_value":l} self.assertEqual(self.resolve(a,"locator-current",["ActorId"],ctx,5)["status"],"TOMBSTONED") a2=self.put_actor("b","actor2"); l2=locator("loc2",expires=0); self.go("index.set",{"binding_kind":"endpoint-occupant","key":a,"value":l2,"expected_version":1,"out_version":None}); ctx2={"binding_kind":"endpoint-occupant","key":a,"expected_value":l2} self.assertEqual(self.resolve(a,"locator-current",["ActorId"],ctx2,5)["status"],"EXPIRED") l3=locator("loc3"); self.go("index.set",{"binding_kind":"endpoint-occupant","key":a,"value":l3,"expected_version":2,"out_version":None}); ctx3={"binding_kind":"endpoint-occupant","key":a,"expected_value":l3} self.assertEqual(self.resolve(a,"locator-current",["ActorId"],ctx3)["status"],"UNAVAILABLE") self.assertEqual(self.resolve(a,"locator-current",["ActorId"],{"binding_kind":"endpoint-occupant","key":a2,"expected_value":l3})["status"],"CONTEXT_MISMATCH") self.assertEqual(self.resolve(a,"locator-current",["NodeId"],ctx3)["status"],"TYPE_MISMATCH") self.assertEqual(self.resolve(ref("ActorId",9),"locator-current",["ActorId"],{"binding_kind":"endpoint-occupant","key":ref("ActorId",9),"expected_value":l3})["status"],"CONTEXT_MISMATCH") self.put_payload(); p=self.r.store["registers"]["payload"] self.go("availability.set",{"envelope_ref":p,"state":"COLLECTED","body":None,"reason":None}) self.assertEqual(self.resolve(p,"payload-body",["PayloadId"])["status"],"CONTENT_COLLECTED") # Malformed outranks all latent conditions; context outranks unknown. malformed={"$kind":"ref","type":"NodeId","hex":"zz"} self.assertEqual(self.resolve(malformed,context=ctx)["status"],"MALFORMED") self.assertEqual(self.resolve(ref("NodeId",7),context=ctx)["status"],"CONTEXT_MISMATCH")
def test_every_query_ast_node(self): a=self.put_actor(); e=self.put_evidence(); n=self.put_node(); self.put_payload(); p=self.r.store["registers"]["payload"] self.go("index.set",{"binding_kind":"state-head","key":n,"value":p,"expected_version":None,"out_version":None}) # record, field, binding, trace rec={"op":"record","type":"Node","key":lit(n)} self.assertEqual(self.r.query({"op":"field","input":rec,"path":"/fields/label"})["value"],"n") self.assertEqual(self.r.query({"op":"binding","kind":"state-head","key":lit(n)})["value"]["value"],p) self.assertEqual(self.r.query({"op":"trace","select":{"step_id":"s1","op":None,"status":None},"project":"event","cardinality":"one"})["value"]["step_id"],"s1") # edge, equal, components self.go("record.put",opaque_args("Node","NodeId","edge",{"id":ref("NodeId",0),"label":"e"},"edge",[{"kind":"node-node","from":ep("self",None),"to":ep("literal",n)}])) en=self.r.store["registers"]["edge"] self.assertTrue(self.r.query({"op":"edge","kind":"node-node","from":lit(en),"to":lit(n)})["value"]["exists"]) self.assertTrue(self.r.query({"op":"equal","left":lit(n),"right":lit(copy.deepcopy(n))})["value"]["equal"]) cmp=self.r.query({"op":"compare-components","left":lit({"a":1}),"right":lit({"b":1}),"components":[{"name":"x","left_path":"/a","right_path":"/b"}]}) self.assertTrue(cmp["value"]["all_equal"]) # provenance self.go("record.put",opaque_args("Commitment","CommitmentId","query-c",{"id":ref("CommitmentId",0),"actor_ref":a},"query-c")); commit=self.r.store["registers"]["query-c"] vf={"id":ref("ValueId",0),"subject_ref":n,"payload_ref":p,"tombstone":False}; self.go("record.put",derived_args("Value","ValueId",vf,"v")); v=self.r.store["registers"]["v"] of={"id":ref("OccurrenceId",0),"value_ref":v,"predecessor_refs":[],"actor_ref":a,"commitment_ref":commit,"nonce":"x","authorization_evidence_ref":e,"instance_order":1} self.go("record.put",derived_args("Occurrence","OccurrenceId",of,"o")); o=self.r.store["registers"]["o"] self.assertTrue(self.r.query({"op":"verify-provenance","claim":lit(o)})["value"]["verified"]) self.assertEqual(self.r.query({"op":"verify-provenance","claim":lit(p)})["status"],"TYPE_MISMATCH") self.assertTrue(self.r.query({"op":"verify-body","envelope_ref":lit(p),"body":lit(d.bytes_value(b"abc"))})["value"]["verified"]) # canonical ID, serialize, reconstruct, relation-set, resolve cid=self.r.query({"op":"canonical-id","id_type":"ValueId","fields":lit({"subject_ref":n,"payload_ref":p,"tombstone":False}),"rule_override":None}) self.assertEqual(cid["value"]["ref"],v) ser=self.r.query({"op":"serialize-store"})["value"] rebuilt=self.r.query({"op":"reconstruct","bytes":lit(ser["bytes"]),"mode":"fresh-store"}) self.assertEqual(rebuilt["status"],"OK"); self.assertIn(d.typed_key(v),rebuilt["value"]["derived"]) rel=self.r.query({"op":"relation-set","kinds":["node-node"]}); self.assertEqual(len(rel["value"]),1) records=self.r.query({"op":"records","types":["Node"]}); self.assertEqual(len(records["value"]),2) section=self.r.query({"op":"store-section","section":"bindings"}); self.assertEqual(len(section["value"]),1) version=self.r.query({"op":"version-select","record_type":"Occurrence","match":lit({"actor_ref":a}),"as_of":lit(1)}) self.assertEqual(version["value"]["key"],o) self.assertEqual(self.resolve(n)["status"],"RESOLVED")
def test_serialization_golden_strict_reload_atomic_fresh(self): empty=d.empty_store(self.p) raw=d.serialize_store(empty) expected=(json.dumps({k:v for k,v in empty.items() if k not in ("registers","trace")},ensure_ascii=False,sort_keys=True,separators=(",",":"))+"\n").encode() self.assertEqual(raw,expected); self.assertTrue(raw.endswith(b"\n")) for bad in [b'{"a":1,"a":2}\n', '{"é":1}\n'.encode(), b'{"x":1.0}\n', b'{"x":18446744073709551616}\n', b'{}']: with self.assertRaises(d.D001Error): d.strict_json_loads(bad,require_lf=True) with self.assertRaises(d.D001Error): d.canonical_json({"$kind":"set","items":[1,1]}) with self.assertRaises(d.D001Error): d.canonical_json({"$kind":"map","entries":[["x",1],["x",2]]}) with self.assertRaises(d.D001Error): d.validate_ref({"$kind":"ref","type":"X","hex":"AA"}) n=self.put_node(); ser=d.serialize_store(self.r.store); fresh=d.reconstruct_store(ser,self.p,fresh=True) self.assertEqual(d.serialize_store(fresh),ser) self.r.store["registers"]["bytes"]=d.bytes_value(ser) self.go("store.reload",{"source_register":"bytes","destination":"current"}) noncanonical=ser.replace(b'":',b'": ',1); self.r.store["registers"]["noncanonical"]=d.bytes_value(noncanonical) self.go("store.reload",{"source_register":"noncanonical","destination":"current"},"MALFORMED") self.assertTrue(any(rec["key"]==n for rec in self.r.store["records"].values())) # Corrupt derived data and prove no replacement. self.put_payload(); good=d.serialize_store(self.r.store); image=d.strict_json_loads(good,require_lf=True) prec=next(x for x in image["records"].values() if x["record_type"]=="PayloadEnvelope"); prec["fields"]["schema_ref"]="corrupt" corrupt=d.canonical_json(image,lf=True); self.r.store["registers"]["bad"]=d.bytes_value(corrupt) ev=self.go("store.reload",{"source_register":"bad","destination":"current"},"VERIFICATION_FAILED") self.assertEqual(ev["before_digest"],ev["after_digest"]) self.go("store.reload",{"source_register":"missing","destination":"current"},"MALFORMED")
def test_cve1_vectors_and_wrappers(self): self.assertEqual(d.cve1(("text","é")),d.cve1(("text","é"))) self.assertNotEqual(d.cve1(("ref",ref("ActorId",1))),d.cve1(("ref",ref("NodeId",1)))) self.assertEqual(d.cve1(("uint64",1)).hex(),"04"+"00"*7+"01") with self.assertRaises(d.D001Error): d.cve1(("set",[("text","x"),("text","x")])) slot={"$kind":"slot","owner_actor_ref":ref("ActorId",1),"slot_type":"state","slot_name":"main"} self.assertEqual(d.validate_slot(slot),slot)
def test_authenticated_trace_effects(self): self.put_node(); event=self.r.store["trace"][0] exact={"op":"trace-effects","event":{"op":"trace","select":{"step_id":event["step_id"],"op":event["op"],"status":"COMMITTED"},"project":"event","cardinality":"one"}} result=self.r.query(exact); self.assertEqual(result["status"],"OK"); self.assertEqual(tuple(result["value"]),("step_id","status","read_count","write_count","record_writes","edge_writes","binding_writes","allocation_writes","writes")) forged=copy.deepcopy(event) variants=[{"op":"trace-effects","event":lit(forged)}] for mutate in ( lambda x:x.update(extra=True),lambda x:x.pop("output"),lambda x:x.update(status=True), lambda x:x["writes"].__setitem__(0,"/"+x["writes"][0]),lambda x:x["writes"].__setitem__(0,x["writes"][0].replace("/","//",1)), lambda x:x["writes"].__setitem__(0,"")): bad=copy.deepcopy(forged); mutate(bad); self.r.store["trace"][0]=bad; variants.append(exact) self.assertNotEqual(self.r.query(exact)["status"],"OK") self.r.store["trace"][0]=copy.deepcopy(event) self.assertEqual(self.r.query(variants[0])["status"],"MALFORMED") print("PASS authenticated-trace-effects")
def test_profile_recipe_smoke_declarative(self): recipes=["opaque","value+occurrence","occurrence-revision","mutable-cell","coalesced-content","slot"] for recipe in recipes: p=profile(recipe); r=d.Runtime(p) self.assertEqual(r.profile.state_recipe["kind"],recipe) self.assertEqual(r.store["profile_name"],"test-profile") # Structural slot insertion. a=self.put_actor(); slot={"$kind":"slot","owner_actor_ref":a,"slot_type":"state","slot_name":"main"} args={"record_type":"SlotDef","fields":{"slot":slot,"owner_ref":a,"kind":"state"},"id":{"kind":"structural","field":"slot"},"edges":[],"out_ref":"slot"} self.go("record.put",args); self.assertEqual(self.r.store["registers"]["slot"],slot) # Mutable cell and coalesced content are keyed/derived generic recipes, not dispatch. n=self.put_node("cell-node","cn"); self.put_payload(out="cp"); p=self.r.store["registers"]["cp"] self.go("record.put",keyed_args("Cell",n,{"subject_ref":n,"payload_ref":p,"counter":1})) self.go("record.put",derived_args("Content","PayloadId",{"id":ref("PayloadId",0),"typed":d.bytes_value(b"x")}),"MALFORMED")
def test_assertion_blind_source_guard_and_vocabularies(self): src=pathlib.Path(d.__file__).read_text() forbidden=["CASE_"+"ASSERTIONS","FAIL_"+"MATRIX","expected_"+"vectors","fixture_expected_"+"observations","expected_"+"outcome","unavailable_"+"roles","capability_"+"graph"] forbidden += ["E"+f"{i:02d}" for i in range(1,19)] self.assertFalse([x for x in forbidden if x in src]) self.assertEqual(len(d.OPS),9); self.assertIn("availability.collect-if-unreachable",d.OPS); self.assertEqual(len(d.FAULTS),2) self.assertEqual(d.QUERY_OPS,{"literal","reg","record","field","binding","trace","resolve","edge","equal","compare-components","verify-provenance","verify-body","canonical-id","canonical-value","typed-content-id","trace-effects","trace-write-excludes","serialize-store","reconstruct","relation-set","records","store-section","version-select","object-has-key","contains","typed-key","map-get","canonical-json","canonical-json-raw","canonical-json-normalized","identity-mapping","persisted-rebind","typed-keys","list-select","record-count","list-count"})
class RuntimeConformanceTest(unittest.TestCase): def make_payload_store(self): p=profile(); r=d.Runtime(p); body=b"abc" fields={"id":ref("PayloadId",0),"schema_ref":"s","codec_ref":"raw","body_bytes":d.bytes_value(body),"body_digest":d.bytes_value(hashlib.sha256(body).digest(),"digest32"),"body_length":len(body)} r.apply(action("p","record.put",derived_args("PayloadEnvelope","PayloadId",fields,"payload"))) return p,r
def reject_image(self,p,image): with self.assertRaises(d.D001Error): d.reconstruct_store(d.canonical_json(image,lf=True),p)
def test_reload_ledger_identity_reconciliation(self): p,r=self.make_payload_store(); image=d.strict_json_loads(d.serialize_store(r.store),require_lf=True) record_key=next(k for k,v in image["records"].items() if v["record_type"]=="PayloadEnvelope") raw=image["records"][record_key]["key"]["hex"] mutations=[] x=copy.deepcopy(image); del x["allocations"]["PayloadId"]; mutations.append(x) x=copy.deepcopy(image); x["allocations"]["Extra"]={"id_type":"Extra","kind":"opaque","seed":"fixture","next_order":0,"used":{}}; mutations.append(x) x=copy.deepcopy(image); del x["allocations"]["PayloadId"]["used"][raw]; x["allocations"]["PayloadId"]["next_order"]=0; mutations.append(x) x=copy.deepcopy(image); x["allocations"]["PayloadId"]["used"]["00"*32]={"first_order":1,"preimage_hex":"00","domain_id":None,"symbol":None}; x["allocations"]["PayloadId"]["next_order"]=2; mutations.append(x) x=copy.deepcopy(image); x["records"][record_key]["identity"]["kind"]="opaque"; mutations.append(x) x=copy.deepcopy(image); x["records"][record_key]["identity"]["preimage_hex"]="00"; mutations.append(x) x=copy.deepcopy(image); x["records"][record_key]["fields"]["schema_ref"]="changed"; mutations.append(x) for mutation in mutations: self.reject_image(p,mutation) print("PASS reload-ledger-identity-reconciliation")
def diamond(self,seed): p=profile(); r=d.Runtime(p); refs={} for sym in ("target","a","b"): r.apply(action(sym,"record.put",opaque_args("Node","NodeId",sym,{"id":ref("NodeId",0),"label":sym},sym))); refs[sym]=r.store["registers"][sym] edges=[{"kind":"node-node","from":ep("self",None),"to":ep("literal",refs["a"])},{"kind":"node-node","from":ep("self",None),"to":ep("literal",refs["b"])}] random.Random(seed).shuffle(edges) r.apply(action("root","record.put",opaque_args("Node","NodeId","root",{"id":ref("NodeId",0),"label":"root"},"root",edges))); root=r.store["registers"]["root"] # Add both diamond arms to the same target with insertion order varied by seed. additions=[("ea",refs["a"]),("eb",refs["b"])] random.Random(seed+10).shuffle(additions) for step,source in additions: key="node-node|"+d.typed_key(source)+"|"+d.typed_key(refs["target"]) r.store["edges"][key]={"kind":"node-node","from":source,"to":refs["target"],"retention_class":"audit-summary"} d.validate_store(r.store,p) r.apply(action("freeze","roots.freeze",{"epoch":0,"root_refs":[root],"pin_refs":[],"clock":0})) raw=d.serialize_store(r.store); a=d.Runtime(p,r.store); b=d.Runtime(p,d.reconstruct_store(raw,p)) ma=a.apply(action("ma","reachability.mark",{"epoch":0,"classes":["audit-summary"],"out":None}))["output"] mb=b.apply(action("mb","reachability.mark",{"epoch":0,"classes":["audit-summary"],"out":None}))["output"] self.assertEqual(ma,mb); return ma
def test_deterministic_reachability_provenance(self): marks=[self.diamond(seed) for seed in range(12)] self.assertTrue(all(x==marks[0] for x in marks)); self.assertEqual(len(marks[0]["reasons"]),4); self.assertEqual(sum(x["edge_kind"]=="frozen-root" for x in marks[0]["reasons"]),1) print("PASS deterministic-reachability-provenance")
def test_immutable_body_and_trace(self): p,r=self.make_payload_store(); payload=r.store["registers"]["payload"] args={"envelope_ref":payload,"state":"AVAILABLE","body":d.bytes_value(b"abc"),"reason":None} first=r.apply(action("a1","availability.set",args)); second=r.apply(action("a2","availability.set",args)) body_path="records/PayloadBody|"+d.typed_key(payload) self.assertIn(body_path,first["writes"]); self.assertNotIn(body_path,second["writes"]) r.store["records"][body_path.removeprefix("records/")]["fields"]["body"]=d.bytes_value(b"evil") before=d.store_digest(r.store) ev=r.apply(action("a3","availability.set",args)); self.assertEqual(ev["error"]["code"],"DUPLICATE_ID"); self.assertEqual(ev["before_digest"],ev["after_digest"]); self.assertEqual(before,d.store_digest(r.store)) print("PASS immutable-body-insertion"); print("PASS exact-trace-read-write-sets")
def test_persisted_sections_and_projection(self): p,r=self.make_payload_store(); image=d.strict_json_loads(d.serialize_store(r.store),require_lf=True) corrupt=[] x=copy.deepcopy(image); x["lifecycle"]={"x":[1]}; corrupt.append(x) x=copy.deepcopy(image); x["roots"]["pending"]={"junk":[123]}; corrupt.append(x) x=copy.deepcopy(image); x["availability"]={"x":{"envelope_ref":ref("PayloadId",9),"body_state":"AVAILABLE","body_record_key":None,"unavailable_reason":None}}; corrupt.append(x) x=copy.deepcopy(image); x["bindings"]={"bad":{"binding_kind":"state-head","key":"x","value":"x","version":1}}; corrupt.append(x) for mutation in corrupt: self.reject_image(p,mutation) ids=dict(p.id_rules); records=dict(p.record_rules); records["Lifecycle"]=d.RecordRule(records["Lifecycle"].fields,id_field=None,instance_order_field="instance_order",projection="life") proj={"life":{"kind":"lifecycle","target_field":"target_ref","state_field":"state","tombstone_field":"tombstone","expires_field":"expires_at"}} p2=d.ProfileSemantics(p.name,p.fixture_id,ids,records,dict(p.edge_rules),binding_kinds=p.binding_kinds,state_recipe=dict(p.state_recipe),projections=proj,provenance_types=p.provenance_types,availability_role=p.availability_role,pin_role=p.pin_role) rr=d.Runtime(p2); rr.apply(action("n","record.put",opaque_args("Node","NodeId","n",{"id":ref("NodeId",0),"label":"n"},"n"))); n=rr.store["registers"]["n"] rr.apply(action("l","record.put",keyed_args("Lifecycle",{"life":1},{"target_ref":n,"state":"live","tombstone":False,"expires_at":None,"instance_order":1}))) self.assertEqual(rr.store["lifecycle"][d.typed_key(n)]["state"],"live") self.assertFalse(hasattr(p2,"projection_hooks")) print("PASS complete-persisted-section-validation"); print("PASS declarative-projection-boundary")
def test_fault_and_closed_public_surface(self): p=profile() for op in d.OPS: r=d.Runtime(p) fault={"kind":"HASH_RESULT_OVERRIDE","phase":"after-preimage-before-typed-lookup","raw_digest_hex":"00"*32} if op not in ("record.put","snapshot.capture"): self.assertEqual(r.apply(action("x",op,{},fault=fault))["error"]["code"],"MALFORMED") r=d.Runtime(p) bad_actions=[action("a","record.put",{"record_type":"SlotDef","fields":{"slot":{"$kind":"slot","owner_actor_ref":ref("ActorId",1),"slot_type":"s","slot_name":"n"},"owner_ref":ref("ActorId",1),"kind":"x"},"id":{"kind":"structural","field":"missing"},"edges":[],"out_ref":None}), action("b","index.set",{"binding_kind":1,"key":"x","value":"x","expected_version":None,"out_version":None})] for act in bad_actions: self.assertEqual(r.apply(act)["status"],"NO_COMMIT") malformed_queries=[{"op":"record","type":1,"key":lit("x")},{"op":"binding","kind":1,"key":lit("x")},{"op":"edge","kind":1,"from":lit("x"),"to":lit("x")},{"op":"canonical-id","id_type":"PayloadId","fields":lit({}),"rule_override":[]}] for expr in malformed_queries: self.assertEqual(r.query(expr)["status"],"MALFORMED") print("PASS fault-op-phase-compatibility"); print("PASS closed-public-error-surface")
def test_closed_wrapper_resolver_pointer_grammar(self): for value in ({"$kind":"bogus","x":1},{"$kind":1}): with self.assertRaises(d.D001Error): d.canonical_json(value) r=d.Runtime(profile()) bad_requests=[{"mode":"garbage","reference":ref("NodeId",1),"context":None,"clock":0,"expected_types":["NodeId"]}, {"mode":"exact","reference":ref("NodeId",1),"context":None,"clock":0,"expected_types":[[]]}, {"mode":"exact","reference":ref("NodeId",1),"context":{"binding_kind":1,"key":"x","expected_value":"x"},"clock":0,"expected_types":["NodeId"]}] for request in bad_requests: self.assertEqual(r.query({"op":"resolve","request":request})["status"],"MALFORMED") for path in ("/~a","/~","/~2"): self.assertEqual(r.query({"op":"field","input":lit({"~a":1}),"path":path})["status"],"MALFORMED") print("PASS closed-wrapper-resolver-pointer-grammar")
def test_cve1_golden_vectors(self): vectors=[(("absent",None),"00"),(("null",None),"01"),(("bool",False),"02"),(("bool",True),"03"), (("uint64",0),"04"+"00"*8),(("uint64",(1<<64)-1),"04"+"ff"*8), (("bytes",b"A"),"05"+"00000001"+"41"),(("text","é"),"06"+"00000002"+"c3a9"), (("array",[("null",None)]),"07"+"00000001"+"01"), (("map",[(("text","k"),("bool",True))]),"08"+"00000001"+"06000000016b03"), (("set",[("bool",True),("bool",False)]),"09"+"00000002"+"0203")] for value,golden in vectors: self.assertEqual(d.cve1(value).hex(),golden) rv=ref("ActorId",1); expected="0a"+d.cve1(("text","ActorId")).hex()+d.cve1(("bytes",bytes.fromhex(rv["hex"]))).hex() self.assertEqual(d.cve1(("ref",rv)).hex(),expected) with self.assertRaises(d.D001Error): d._u32_frame_length(1<<32,"test") print("PASS CVE1-complete-golden-vectors")
def test_profile_semantics_eager_closure(self): base=profile() bad=[lambda: d.ProfileSemantics("p","f",{"X":d.IdRule("opaque","X",width=0)}, {}, {}), lambda: d.ProfileSemantics("p","f",dict(base.id_rules), {"Bad":d.RecordRule({"x":{"ref":"Missing"}},id_field=None)}, {}), lambda: d.ProfileSemantics("p","f",dict(base.id_rules), {"Bad":d.RecordRule({"x":"text"},id_field=None,projection="missing")}, {})] for make in bad: with self.assertRaises(d.D001Error): make() print("PASS profile-semantics-eager-closure")
def test_transition_graph_exact_schema_activation(self): # Reusing D-001 record/edge names is legal for a generic profile; only the complete exact schema activates graph validation. base=profile(); ids=dict(base.id_rules) for tag in ("GenerationId","SelectionId","WorldId","RootValueId","ParentId","PolicyId"): ids[tag]=d.IdRule("opaque",tag) records=dict(base.record_rules) records["Generation"]=d.RecordRule({"id":{"ref":"GenerationId"},"sequence":"uint64"},id_type="GenerationId") records["Selection"]=d.RecordRule({"id":{"ref":"SelectionId"},"sequence":"uint64"},id_type="SelectionId") edges=dict(base.edge_rules) for kind in ("selection-generation","selection-world","selection-root","selection-parent","selection-policy","generation-world","generation-root","generation-parent","generation-predecessor","selection-predecessor"): edges[kind]=d.EdgeRule(frozenset({"GenerationId"}),frozenset({"GenerationId"}),"audit-summary") generic=d.ProfileSemantics("generic-same-names","fixture",ids,records,edges) runtime=d.Runtime(generic); event=runtime.apply(action("g","record.put",opaque_args("Generation","GenerationId","g",{"id":ref("GenerationId",0),"sequence":1},"g"))) self.assertEqual((event["status"],event["error"]),("COMMITTED",None)) self.assertEqual(runtime.store["registers"]["g"]["type"],"GenerationId") print("PASS transition-graph-exact-schema-activation")
class RuntimeRepairGateTest(unittest.TestCase): def reject(self,p,image): with self.assertRaises(d.D001Error): d.reconstruct_store(d.canonical_json(image,lf=True),p) def payload(self): p=profile(); r=d.Runtime(p); body=b"abc"; fields={"id":ref("PayloadId",0),"schema_ref":"s","codec_ref":"raw","body_bytes":d.bytes_value(body),"body_digest":d.bytes_value(hashlib.sha256(body).digest(),"digest32"),"body_length":3} r.apply(action("p","record.put",derived_args("PayloadEnvelope","PayloadId",fields,"p"))); return p,r,r.store["registers"]["p"] def life(self): b=profile(); records=dict(b.record_rules); records["Lifecycle"]=d.RecordRule(records["Lifecycle"].fields,id_field=None,instance_order_field="instance_order",projection="life"); projections={"life":{"kind":"lifecycle","target_field":"target_ref","state_field":"state","tombstone_field":"tombstone","expires_field":"expires_at"}} p=d.ProfileSemantics(b.name,b.fixture_id,dict(b.id_rules),records,dict(b.edge_rules),binding_kinds=b.binding_kinds,state_recipe=dict(b.state_recipe),projections=projections,provenance_types=b.provenance_types,availability_role=b.availability_role,pin_role=b.pin_role); r=d.Runtime(p); r.apply(action("n","record.put",opaque_args("Node","NodeId","n",{"id":ref("NodeId",0),"label":"n"},"n"))); return p,r,r.store["registers"]["n"] def test_body_and_role(self): p,r,pay=self.payload(); r.apply(action("a","availability.set",{"envelope_ref":pay,"state":"AVAILABLE","body":d.bytes_value(b"abc"),"reason":None})); image=d.strict_json_loads(d.serialize_store(r.store),require_lf=True); key=p.availability_role.body_record_type+"|"+d.typed_key(pay); bad=copy.deepcopy(image); bad["records"][key]["fields"][p.availability_role.body_field]=d.bytes_value(b"evil"); self.reject(p,bad) nrt=d.Runtime(p); nrt.apply(action("n","record.put",opaque_args("Node","NodeId","n",{"id":ref("NodeId",0),"label":"n"},"n"))); node=nrt.store["registers"]["n"]; self.assertEqual(nrt.apply(action("x","availability.set",{"envelope_ref":node,"state":"COLLECTED","body":None,"reason":None}))["status"],"NO_COMMIT"); bad=copy.deepcopy(image); bad["availability"]={d.typed_key(node):{"envelope_ref":node,"body_state":"COLLECTED","body_record_key":None,"unavailable_reason":None}}; self.reject(p,bad) print("PASS persisted-body-envelope-reconciliation"); print("PASS availability-envelope-role-closure") def test_opaque_evidence(self): p=profile(); r=d.Runtime(p); r.apply(action("n","record.put",opaque_args("Node","NodeId","sym",{"id":ref("NodeId",0),"label":"n"},None))); image=d.strict_json_loads(d.serialize_store(r.store),require_lf=True); rk=next(k for k,v in image["records"].items() if v["record_type"]=="Node"); raw=image["records"][rk]["key"]["hex"] for place,field in ((image["records"][rk]["identity"],"domain_id"),(image["records"][rk]["identity"],"symbol"),(image["allocations"]["NodeId"]["used"][raw],"domain_id"),(image["allocations"]["NodeId"]["used"][raw],"symbol")): bad=copy.deepcopy(image); target=bad["records"][rk]["identity"] if place is image["records"][rk]["identity"] else bad["allocations"]["NodeId"]["used"][raw]; target[field]="other"; self.reject(p,bad) print("PASS opaque-domain-symbol-reconciliation") def test_lifecycle(self): p,r,node=self.life(); put=lambda step,key,state,tomb,exp,order:r.apply(action(step,"record.put",keyed_args("Lifecycle",key,{"target_ref":node,"state":state,"tombstone":tomb,"expires_at":exp,"instance_order":order}))) put("l1","l1","old",False,1,1); put("l2","l2","new",True,None,2); req={"mode":"exact","reference":node,"context":None,"clock":0,"expected_types":["NodeId"]}; self.assertEqual(r.query({"op":"resolve","request":req})["status"],"TOMBSTONED"); image=d.strict_json_loads(d.serialize_store(r.store),require_lf=True); bad=copy.deepcopy(image); bad["lifecycle"]={}; self.reject(p,bad); bad=copy.deepcopy(image); bad["lifecycle"][d.typed_key(node)]["state"]="forged"; self.reject(p,bad); ev=r.apply(action("eq","record.put",keyed_args("Lifecycle","eq",{"target_ref":node,"state":"x","tombstone":False,"expires_at":None,"instance_order":2}))); self.assertEqual(ev["error"]["code"],"ORDER_VIOLATION"); self.assertEqual(ev["before_digest"],ev["after_digest"]) p2,r2,n2=self.life(); r2.apply(action("e","record.put",keyed_args("Lifecycle","e",{"target_ref":n2,"state":"live","tombstone":False,"expires_at":2,"instance_order":1}))); req.update(reference=n2,clock=3); self.assertEqual(r2.query({"op":"resolve","request":req})["status"],"EXPIRED"); print("PASS lifecycle-record-projection-reconciliation") def test_mark(self): p=profile(); r=d.Runtime(p); r.apply(action("n","record.put",opaque_args("Node","NodeId","n",{"id":ref("NodeId",0),"label":"n"},"n"))); node=r.store["registers"]["n"]; r.apply(action("f","roots.freeze",{"epoch":0,"root_refs":[node],"pin_refs":[],"clock":0})); r.apply(action("m","reachability.mark",{"epoch":0,"classes":["audit-summary"],"out":None})); image=d.strict_json_loads(d.serialize_store(r.store),require_lf=True) for field,value in (("reachable",["forged"]),("reasons",[{"target":"forged","source":"forged","edge_kind":"node-node","retention_class":"audit-summary"}]),("deferred_to_epoch",1)): bad=copy.deepcopy(image); bad["roots"]["frozen"]["0"]["mark"][field]=value; self.reject(p,bad) print("PASS persisted-mark-recomputation") def test_immutable_profile_and_widths(self): ids={"X":d.IdRule("opaque","X")}; fields={"id":{"ref":"X"}}; records={"XRec":d.RecordRule(fields,id_type="X")}; state={"kind":"x"}; p=d.ProfileSemantics("p","f",ids,records,{},state_recipe=state); ids["Y"]=d.IdRule("opaque","Y"); fields["extra"]="text"; state["kind"]="evil"; self.assertNotIn("Y",p.id_rules); self.assertNotIn("extra",p.record_rules["XRec"].fields); self.assertEqual(p.state_recipe["kind"],"x"); self.assertRaises((TypeError,AttributeError),lambda:p.id_rules.update({"Y":d.IdRule("opaque","Y")})); self.assertRaises((TypeError,AttributeError),lambda:p.record_rules["XRec"].fields["id"].update({"ref":"Y"})); self.assertRaises(AttributeError,setattr,p,"name","evil") rt=d.Runtime(profile()) for typ in ("NodeId","PayloadId"): for hx in ("","00"): req={"mode":"exact","reference":{"$kind":"ref","type":typ,"hex":hx},"context":None,"clock":0,"expected_types":[typ]}; self.assertEqual(rt.query({"op":"resolve","request":req})["status"],"MALFORMED") print("PASS immutable-profile-semantics"); print("PASS profile-aware-reference-widths") def test_generated_malformed_shape_matrix(self): bads=(None,0,True,"x",[],{"unexpected":1}) for op in d.OPS: for i,bad in enumerate(bads): self.assertEqual(d.Runtime(profile()).apply(action(op+str(i),op,bad))["status"],"NO_COMMIT") for op in d.QUERY_OPS - {"serialize-store"}: self.assertEqual(d.Runtime(profile()).query({"op":op})["status"],"MALFORMED")
class RuntimeFourthRepairTest(unittest.TestCase): def reject(self, p, image): with self.assertRaises(d.D001Error): d.reconstruct_store(d.canonical_json(image, lf=True), p)
def populated(self): p=profile(); r=d.Runtime(p); body=b"abc" fields={"id":ref("PayloadId",0),"schema_ref":"s","codec_ref":"raw","body_bytes":d.bytes_value(body),"body_digest":d.bytes_value(hashlib.sha256(body).digest(),"digest32"),"body_length":len(body)} r.apply(action("p","record.put",derived_args("PayloadEnvelope","PayloadId",fields,"payload"))); payload=r.store["registers"]["payload"] r.apply(action("a","availability.set",{"envelope_ref":payload,"state":"AVAILABLE","body":d.bytes_value(body),"reason":None})) return p,r,payload
def test_reload_prefers_authoritative_envelope_over_keyed_body(self): p,r,payload=self.populated(); loaded=d.reconstruct_store(d.serialize_store(r.store),p) q=d.Runtime(p,loaded).query({"op":"resolve","request":{"mode":"payload-body","reference":payload,"context":None,"clock":0,"expected_types":["PayloadId"]}}) self.assertEqual(q["status"],"RESOLVED") print("PASS reload-authoritative-ref-selection")
def test_outcome_blind_action_errors(self): p=profile(); r=d.Runtime(p) bad=action("bad","record.put",opaque_args("Missing","NodeId","x",{},None)) event=r.apply(bad) self.assertEqual((event["status"],event["error"]["code"]),("NO_COMMIT","UNSUPPORTED_RECORD")) # The neutral runner continues after ordinary errors. event2=r.apply(action("good","record.put",opaque_args("Node","NodeId","n",{"id":ref("NodeId",0),"label":"n"},"n"))) self.assertEqual(event2["status"],"COMMITTED") legacy=copy.deepcopy(action("legacy","record.put",{})); legacy["expect_error"]="MALFORMED" rejected=r.apply(legacy); self.assertEqual(rejected["error"]["code"],"MALFORMED") original=r._put try: def fatal(*_): raise d.D001Error("FATAL_INTEGRITY_FAILURE","probe",fatal=True) r._put=fatal; fatal_event=r.apply(action("fatal","record.put",{})); self.assertEqual(fatal_event["status"],"FATAL") def unexpected(*_): raise RuntimeError("probe") r._put=unexpected; internal=r.apply(action("internal","record.put",{})); self.assertEqual((internal["status"],internal["error"]["code"]),("NO_COMMIT","INTERNAL_ERROR")) finally: r._put=original self.assertEqual([x["step_id"] for x in r.store["trace"]],["bad","good","legacy","fatal","internal"]) print("PASS outcome-blind-action-errors")
def test_authoritative_reference_uniqueness(self): p,r,payload=self.populated(); image=d.strict_json_loads(d.serialize_store(r.store),require_lf=True) matches=[(k,v) for k,v in image["records"].items() if d.typed_key(v["key"])==d.typed_key(payload)] self.assertEqual(len(matches),2) # one authoritative envelope + one managed keyed body for ordered in (matches,list(reversed(matches))): x=copy.deepcopy(image); rest=[(k,v) for k,v in x["records"].items() if d.typed_key(v["key"])!=d.typed_key(payload)] x["records"]={k:v for k,v in rest+ordered} loaded=d.reconstruct_store(d.canonical_json(x,lf=True),p) self.assertEqual(d._find_ref(loaded,payload,p)[1]["record_type"],p.availability_role.envelope_record_type) # Add a second authoritative record type for PayloadId; both orders reject. fields=dict(p.record_rules["PayloadEnvelope"].fields) records=dict(p.record_rules); records["OtherEnvelope"]=d.RecordRule(fields,id_type="PayloadId") p2=d.ProfileSemantics(p.name,p.fixture_id,dict(p.id_rules),records,dict(p.edge_rules),binding_kinds=p.binding_kinds,state_recipe=dict(p.state_recipe),projections={k:dict(v) for k,v in p.projections.items()},provenance_types=p.provenance_types,availability_role=p.availability_role,pin_role=p.pin_role) auth=next(v for v in image["records"].values() if v["record_type"]=="PayloadEnvelope") dup=copy.deepcopy(auth); dup["record_type"]="OtherEnvelope"; dupkey="OtherEnvelope|"+d.typed_key(payload) for first in (False,True): x=copy.deepcopy(image); items=list(x["records"].items()); pair=(dupkey,copy.deepcopy(dup)); items=([pair]+items if first else items+[pair]); x["records"]={k:v for k,v in items}; x["registers"]={}; x["trace"]=[] with self.assertRaises(d.D001Error) as cm: d.validate_store(x,p2) self.assertEqual(cm.exception.code,"VERIFICATION_FAILED") # No authoritative match has a unique keyed fallback; two fallbacks are ambiguous. bare=d.empty_store(p); body=copy.deepcopy(next(v for v in image["records"].values() if v["record_type"]=="PayloadBody")); bare["records"]={"one":body} self.assertEqual(d._find_ref(bare,payload,p)[1]["record_type"],"PayloadBody") bare["records"]["two"]=copy.deepcopy(body) with self.assertRaises(d.D001Error): d._find_ref(bare,payload,p) self.assertIsNone(d._find_ref(d.empty_store(p),payload,p)) print("PASS authoritative-reference-uniqueness")
def test_envelope_body_bidirectional_integrity(self): p,r,payload=self.populated(); image=d.strict_json_loads(d.serialize_store(r.store),require_lf=True) erkey=next(k for k,v in image["records"].items() if v["record_type"]=="PayloadEnvelope") bkey=p.availability_role.body_record_type+"|"+d.typed_key(payload) mutations=[] for field,value in (("body_digest",d.bytes_value(b"x"*32,"digest32")),("body_length",999),("body_bytes",d.bytes_value(b"evil"))): x=copy.deepcopy(image); x["records"][erkey]["fields"][field]=value; mutations.append(x) x=copy.deepcopy(image); del x["availability"][d.typed_key(payload)]; mutations.append(x) x=copy.deepcopy(image); rec=x["records"].pop(bkey); x["records"][bkey+"x"]=rec; x["availability"][d.typed_key(payload)]["body_record_key"]=bkey+"x"; mutations.append(x) x=copy.deepcopy(image); other=copy.deepcopy(x["availability"][d.typed_key(payload)]); x["availability"]["duplicate"]=other; mutations.append(x) for mutation in mutations: self.reject(p,mutation) orphan=keyed_args("PayloadBody","orphan",{"envelope_ref":payload,"body":d.bytes_value(b"abc")}) self.assertEqual(r.apply(action("orphan","record.put",orphan))["error"]["code"],"TYPE_MISMATCH") print("PASS envelope-body-bidirectional-integrity")
def test_frozen_pin_source_reconciliation(self): p=profile(); r=d.Runtime(p) for sym in ("a","b"): r.apply(action(sym,"record.put",opaque_args("Node","NodeId",sym,{"id":ref("NodeId",0),"label":sym},sym))) a,b=r.store["registers"]["a"],r.store["registers"]["b"] r.apply(action("pin","record.put",opaque_args("Pin","PinId","pin",{"id":ref("PinId",0),"target_ref":a,"expires_at":10,"active":True},"pin"))); pin=r.store["registers"]["pin"] r.apply(action("freeze","roots.freeze",{"epoch":0,"root_refs":[],"pin_refs":[pin],"clock":5})); r.apply(action("mark","reachability.mark",{"epoch":0,"classes":[],"out":None})) image=d.strict_json_loads(d.serialize_store(r.store),require_lf=True); frozen=image["roots"]["frozen"]["0"] x=copy.deepcopy(image); x["roots"]["frozen"]["0"]["pins"][0]["target_ref"]=b; x["roots"]["frozen"]["0"]["mark"]["reachable"]=[d.typed_key(b)]; self.reject(p,x) x=copy.deepcopy(image); x["roots"]["frozen"]["0"]["pins"][0]["expires_at"]=11; self.reject(p,x) source_key=next(k for k,v in image["records"].items() if v["key"]==pin) x=copy.deepcopy(image); x["records"][source_key]["fields"]["active"]=False; self.reject(p,x) x=copy.deepcopy(image); x["records"][source_key]["record_type"]="Node"; self.reject(p,x) self.assertEqual(frozen["pins"][0]["target_ref"],a) print("PASS frozen-pin-source-reconciliation")
def test_all_query_reference_width_boundaries(self): p=profile(); r=d.Runtime(p); r.apply(action("n","record.put",opaque_args("Node","NodeId","n",{"id":ref("NodeId",0),"label":"n"},"n"))); n=r.store["registers"]["n"] r.apply(action("i","index.set",{"binding_kind":"state-head","key":n,"value":n,"expected_version":None,"out_version":None})) valid={"NodeId":n,"OccurrenceId":ref("OccurrenceId",1),"PayloadId":ref("PayloadId",1)} def variants(t): width=p.id_rules[t].width return [("short",{"$kind":"ref","type":t,"hex":"00"}), ("long",{"$kind":"ref","type":t,"hex":"00"*(width+1)}), ("unknown",{"$kind":"ref","type":"MissingId","hex":"00"*width}), ("valid",valid[t])] consumers=[ ("record","NodeId",lambda x:{"op":"record","type":"Node","key":lit(x)}), ("binding","NodeId",lambda x:{"op":"binding","kind":"state-head","key":lit(x)}), ("edge-from","NodeId",lambda x:{"op":"edge","kind":"node-node","from":lit(x),"to":lit(n)}), ("edge-to","NodeId",lambda x:{"op":"edge","kind":"node-node","from":lit(n),"to":lit(x)}), ("provenance","OccurrenceId",lambda x:{"op":"verify-provenance","claim":lit(x)}), ("verify-body","PayloadId",lambda x:{"op":"verify-body","envelope_ref":lit(x),"body":lit(d.bytes_value(b"abc"))}), ("canonical-id","NodeId",lambda x:{"op":"canonical-id","id_type":"ValueId","fields":lit({"subject_ref":x,"payload_ref":ref("PayloadId",1),"tombstone":False}),"rule_override":None}), ] for name,typ,make in consumers: for label,value in variants(typ): result=r.query(make(value)) if label != "valid": self.assertIn(result["status"],("MALFORMED","TYPE_MISMATCH"),(name,label,result)) else: self.assertNotIn(result["status"],("MALFORMED","TYPE_MISMATCH"),(name,label,result)) short=variants("NodeId")[0][1] transitions=[ action("keyed","record.put",keyed_args("Cell",short,{"subject_ref":n,"payload_ref":ref("PayloadId",1),"counter":1})), action("edge","record.put",opaque_args("Node","NodeId","edge",{"id":ref("NodeId",0),"label":"e"},None,[{"kind":"node-node","from":ep("self",None),"to":ep("literal",short)}])), action("index","index.set",{"binding_kind":"state-head","key":short,"value":n,"expected_version":None,"out_version":None}), action("snap","snapshot.capture",{"name":"bad","snapshot_record":opaque_args("Snapshot","SnapshotId","bad",{"id":ref("SnapshotId",0),"label":"bad"}),"binding_kinds":[],"root_context":{"ref":short}}), action("freeze","roots.freeze",{"epoch":0,"root_refs":[short],"pin_refs":[],"clock":0}), ] for transition in transitions: self.assertEqual(d.Runtime(p).apply(transition)["status"],"NO_COMMIT") request={"mode":"exact","reference":n,"context":{"binding_kind":"state-head","key":short,"expected_value":n},"clock":0,"expected_types":["NodeId"]} self.assertEqual(r.query({"op":"resolve","request":request})["status"],"MALFORMED") forged=copy.deepcopy(r.store); forged["registers"]["bad"]=short with self.assertRaises(d.D001Error): d.validate_store(forged,p) print("PASS all-query-reference-width-boundaries")
def test_query_native_exception_closure(self): p=profile(); r=d.Runtime(p); r.apply(action("n","record.put",opaque_args("Node","NodeId","n",{"id":ref("NodeId",0),"label":"n"},"n"))); n=r.store["registers"]["n"] wrong=r.query({"op":"verify-body","envelope_ref":lit(n),"body":lit(d.bytes_value(b"x"))}); self.assertEqual(wrong["status"],"TYPE_MISMATCH") base=profile(); no_role=d.ProfileSemantics(base.name,base.fixture_id,dict(base.id_rules),dict(base.record_rules),dict(base.edge_rules),binding_kinds=base.binding_kinds,state_recipe=dict(base.state_recipe),provenance_types=base.provenance_types,pin_role=base.pin_role) self.assertEqual(d.Runtime(no_role).query({"op":"verify-body","envelope_ref":lit(ref("PayloadId",1)),"body":lit(d.bytes_value(b"x"))})["status"],"TYPE_MISMATCH") original=r._eval; r._eval=lambda expr: (_ for _ in ()).throw(KeyError("hostile")) try: result=r.query({"op":"literal","value":1}) finally: r._eval=original self.assertEqual(result,{"status":"INTERNAL_ERROR","value":None,"error":"KeyError"}) print("PASS query-native-exception-closure")
class RuntimeUnit21Test(unittest.TestCase): def put(self,r,step,args): event=r.apply(action(step,"record.put",args)); self.assertEqual(event["status"],"COMMITTED",event); return event["output"]["ref"] def alias_query(self,r,mode,ns,label,version=None,clock=0): reference={"namespace_ref":ns,"label":label} if version is not None: reference["version_ref"]=version return r.query({"op":"resolve","request":{"mode":mode,"reference":reference,"context":None,"clock":clock,"expected_types":["AliasId"]}}) def test_generic_versioned_reference_resolution(self): p=profile(); r=d.Runtime(p); ns=self.put(r,"ns",opaque_args("Namespace","NamespaceId","ns",{"id":ref("NamespaceId",0),"name":"main"})); a=self.put(r,"a",opaque_args("Node","NodeId","a",{"id":ref("NodeId",0),"label":"a"})); b=self.put(r,"b",opaque_args("Node","NodeId","b",{"id":ref("NodeId",0),"label":"b"})) def av(step,target,pred,order,label="x",tomb=False,expires=None): return self.put(r,step,opaque_args("AliasVersion","AliasId",step,{"id":ref("AliasId",0),"namespace_ref":ns,"label":label,"target_ref":target,"predecessor_ref":pred,"instance_order":order,"tombstone":tomb,"expires_at":expires})) v1=av("v1",a,None,1); v2=av("v2",b,v1,2); tomb=av("tomb",None,v2,3,tomb=True); v4=av("v4",a,tomb,4) q=self.alias_query(r,"alias-current",ns,"x"); self.assertEqual((q["status"],q["matched_version_ref"],q["denotation_ref"]),("RESOLVED",v4,a)) self.assertEqual(self.alias_query(r,"alias-as-of",ns,"x",v1)["denotation_ref"],a); self.assertEqual(self.alias_query(r,"alias-as-of",ns,"x",v2)["denotation_ref"],b); self.assertEqual(self.alias_query(r,"alias-as-of",ns,"x",tomb)["status"],"TOMBSTONED") expired=av("expired",b,None,5,"expired",False,2); self.assertEqual(self.alias_query(r,"alias-as-of",ns,"expired",expired,3)["status"],"EXPIRED") av("fork1",a,None,6,"fork"); av("fork2",b,None,7,"fork"); self.assertEqual(self.alias_query(r,"alias-current",ns,"fork")["status"],"CONTEXT_MISMATCH") exact=r.query({"op":"resolve","request":{"mode":"exact","reference":v1,"context":None,"clock":99,"expected_types":["AliasId"]}}); self.assertEqual(exact["denotation_ref"],v1) print("PASS generic-versioned-reference-resolution") def test_structured_provenance_verification(self): p=profile(); r=d.Runtime(p); actor=self.put(r,"actor",opaque_args("Actor","ActorId","a",{"id":ref("ActorId",0),"name":"a"})); actor2=self.put(r,"actor2",opaque_args("Actor","ActorId","b",{"id":ref("ActorId",0),"name":"b"})); evidence=self.put(r,"e",opaque_args("Evidence","EvidenceId","e",{"id":ref("EvidenceId",0),"data":d.bytes_value(b"e")})); commit=self.put(r,"c",opaque_args("Commitment","CommitmentId","c",{"id":ref("CommitmentId",0),"actor_ref":actor})); node=self.put(r,"n",opaque_args("Node","NodeId","n",{"id":ref("NodeId",0),"label":"n"})) payload=self.put(r,"p",derived_args("PayloadEnvelope","PayloadId",{"id":ref("PayloadId",0),"schema_ref":"s","codec_ref":"raw","body_bytes":d.bytes_value(b"x"),"body_digest":d.bytes_value(hashlib.sha256(b"x").digest(),"digest32"),"body_length":1})); value=self.put(r,"v",derived_args("Value","ValueId",{"id":ref("ValueId",0),"subject_ref":node,"payload_ref":payload,"tombstone":False})); occurrence=self.put(r,"o",derived_args("Occurrence","OccurrenceId",{"id":ref("OccurrenceId",0),"value_ref":value,"predecessor_refs":[],"actor_ref":actor,"commitment_ref":commit,"nonce":"n","authorization_evidence_ref":evidence,"instance_order":1})) claim={"claimed_occurrence_ref":occurrence,"actor_ref":actor,"commitment_ref":commit,"authorization_evidence_ref":evidence,"causal_inputs":[]}; query=lambda c:r.query({"op":"verify-provenance","claim":lit(c)}) self.assertTrue(query(occurrence)["value"]["verified"]); self.assertTrue(query(claim)["value"]["verified"]) copied=dict(claim); copied["claimed_occurrence_ref"]=ref("OccurrenceId",9); self.assertEqual(query(copied)["value"]["match_status"],"UNKNOWN_MATCH") mismatch=dict(claim); mismatch["actor_ref"]=actor2; self.assertEqual(query(mismatch)["value"]["match_status"],"FIELD_MISMATCH") wrong=dict(claim); wrong["claimed_occurrence_ref"]=node; self.assertEqual(query(wrong)["status"],"TYPE_MISMATCH") malformed=dict(claim); malformed["causal_inputs"]="bad"; self.assertEqual(query(malformed)["status"],"MALFORMED") print("PASS structured-provenance-verification") def test_mark_gated_body_collection(self): p=profile(); r=d.Runtime(p); body=b"abc"; payload=self.put(r,"p",derived_args("PayloadEnvelope","PayloadId",{"id":ref("PayloadId",0),"schema_ref":"s","codec_ref":"raw","body_bytes":d.bytes_value(body),"body_digest":d.bytes_value(hashlib.sha256(body).digest(),"digest32"),"body_length":3})); self.assertEqual(r.apply(action("available","availability.set",{"envelope_ref":payload,"state":"AVAILABLE","body":d.bytes_value(body),"reason":None}))["status"],"COMMITTED"); pin=self.put(r,"pin",opaque_args("Pin","PinId","pin",{"id":ref("PinId",0),"target_ref":payload,"expires_at":0,"active":True})) r.apply(action("freeze0","roots.freeze",{"epoch":0,"root_refs":[],"pin_refs":[pin],"clock":0})); r.apply(action("mark0","reachability.mark",{"epoch":0,"classes":["body-retaining"],"out":None})); self.assertEqual(r.apply(action("reachable","availability.collect-if-unreachable",{"envelope_ref":payload,"epoch":0,"expected_reachable":True}))["error"]["code"],"CONTEXT_MISMATCH") deferred=self.put(r,"deferred",opaque_args("Node","NodeId","deferred",{"id":ref("NodeId",0),"label":"deferred"})); r.apply(action("freeze1","roots.freeze",{"epoch":1,"root_refs":[],"pin_refs":[pin],"clock":1})); self.assertEqual(r.apply(action("incomplete","availability.collect-if-unreachable",{"envelope_ref":payload,"epoch":1,"expected_reachable":False}))["error"]["code"],"CONTEXT_MISMATCH") fault={"kind":"CONCURRENT_ROOT_APPEND","phase":"after-root-copy-before-first-edge","epoch":2,"root_ref":deferred}; r.apply(action("mark1","reachability.mark",{"epoch":1,"classes":["body-retaining"],"out":None},fault)); self.assertEqual(r.apply(action("wrong","availability.collect-if-unreachable",{"envelope_ref":payload,"epoch":1,"expected_reachable":True}))["error"]["code"],"CONTEXT_MISMATCH"); event=r.apply(action("collect","availability.collect-if-unreachable",{"envelope_ref":payload,"epoch":1,"expected_reachable":False})); self.assertEqual(event["output"]["status"],"COLLECTED") self.assertIsNotNone(d._find_ref(r.store,payload,p)); self.assertEqual(r.query({"op":"resolve","request":{"mode":"payload-body","reference":payload,"context":None,"clock":1,"expected_types":["PayloadId"]}})["status"],"CONTENT_COLLECTED"); self.assertIn("2",r.store["roots"]["pending"]); self.assertTrue(any(x["record_type"]=="Pin" for x in r.store["records"].values())); self.assertEqual(r.apply(action("twice","availability.collect-if-unreachable",{"envelope_ref":payload,"epoch":1,"expected_reachable":False}))["error"]["code"],"CONTEXT_MISMATCH") loaded=d.reconstruct_store(d.serialize_store(r.store),p); self.assertEqual(loaded["availability"][d.typed_key(payload)]["body_state"],"COLLECTED") print("PASS mark-gated-body-collection")
def test_alias_context_precedence(self): p=profile(); r=d.Runtime(p); ns=self.put(r,"ns",opaque_args("Namespace","NamespaceId","ns",{"id":ref("NamespaceId",0),"name":"main"})); target=self.put(r,"target",opaque_args("Node","NodeId","target",{"id":ref("NodeId",0),"label":"target"})) def av(step,label,pred=None,tomb=False): return self.put(r,step,opaque_args("AliasVersion","AliasId",step,{"id":ref("AliasId",0),"namespace_ref":ns,"label":label,"target_ref":None if tomb else target,"predecessor_ref":pred,"instance_order":len([x for x in r.store["records"].values() if x["record_type"]=="AliasVersion"])+1,"tombstone":tomb,"expires_at":None})) v=av("v","x"); tomb=av("tomb","dead",None,True); av("f1","fork"); av("f2","fork") r.apply(action("bind","index.set",{"binding_kind":"state-head","key":ns,"value":target,"expected_version":None,"out_version":None})) def query(mode,label,context,version=None): reference={"namespace_ref":ns,"label":label}; if version is not None: reference["version_ref"]=version return r.query({"op":"resolve","request":{"mode":mode,"reference":reference,"context":context,"clock":0,"expected_types":["AliasId"]}}) valid={"binding_kind":"state-head","key":ns,"expected_value":target}; absent={"binding_kind":"state-head","key":target,"expected_value":target}; wrong={"binding_kind":"state-head","key":ns,"expected_value":ns} self.assertEqual(query("alias-current","x",valid)["status"],"RESOLVED"); self.assertEqual(query("alias-as-of","x",valid,v)["status"],"RESOLVED") for context in (absent,wrong): for mode,version in (("alias-current",None),("alias-as-of",v)): self.assertEqual(query(mode,"x",context,version)["status"],"CONTEXT_MISMATCH") malformed={"binding_kind":"missing","key":ns,"expected_value":target} self.assertEqual(query("alias-current","fork",malformed)["status"],"MALFORMED") self.assertEqual(query("alias-current","fork",absent)["status"],"CONTEXT_MISMATCH") self.assertEqual(query("alias-current","missing",absent)["status"],"CONTEXT_MISMATCH") self.assertEqual(query("alias-current","dead",absent)["status"],"CONTEXT_MISMATCH") self.assertEqual(query("alias-current","dead",valid)["status"],"TOMBSTONED") print("PASS alias-context-precedence")
def test_alias_type_denotation_closure(self): p=profile(); r=d.Runtime(p); ns=self.put(r,"ns",opaque_args("Namespace","NamespaceId","ns",{"id":ref("NamespaceId",0),"name":"main"})); target=self.put(r,"target",opaque_args("Node","NodeId","target",{"id":ref("NodeId",0),"label":"target"})) v=self.put(r,"v",opaque_args("AliasVersion","AliasId","v",{"id":ref("AliasId",0),"namespace_ref":ns,"label":"x","target_ref":target,"predecessor_ref":None,"instance_order":1,"tombstone":False,"expires_at":None})) def q(mode,reference,expected=("AliasId",)): return r.query({"op":"resolve","request":{"mode":mode,"reference":reference,"context":None,"clock":0,"expected_types":list(expected)}}) resolved=q("alias-current",{"namespace_ref":ns,"label":"x"}); self.assertEqual(resolved,{"status":"RESOLVED","value":{"denotation_ref":target,"availability":None},"denotation_ref":target,"matched_version_ref":v}) self.assertEqual(q("alias-as-of",{"namespace_ref":ns,"label":"x","version_ref":target})["status"],"TYPE_MISMATCH") self.assertEqual(q("alias-current",{"namespace_ref":target,"label":"x"})["status"],"TYPE_MISMATCH") short=copy.deepcopy(v); short["hex"]="00"; self.assertEqual(q("alias-as-of",{"namespace_ref":ns,"label":"x","version_ref":short})["status"],"MALFORMED") forged=copy.deepcopy(r.store); rec=forged["records"]["AliasVersion|"+d.typed_key(v)]; dangling=ref("NodeId",9); rec["fields"]["target_ref"]=dangling clean=d.Runtime(p); clean.store=forged unknown=clean.query({"op":"resolve","request":{"mode":"alias-current","reference":{"namespace_ref":ns,"label":"x"},"context":None,"clock":0,"expected_types":["AliasId"]}}) self.assertEqual((unknown["status"],unknown["matched_version_ref"],unknown["denotation_ref"]),("UNKNOWN",v,None)) print("PASS alias-type-denotation-closure")
def test_grounded_structured_provenance(self): p=profile(); r=d.Runtime(p); actor=self.put(r,"actor",opaque_args("Actor","ActorId","a",{"id":ref("ActorId",0),"name":"a"})); evidence=self.put(r,"e",opaque_args("Evidence","EvidenceId","e",{"id":ref("EvidenceId",0),"data":d.bytes_value(b"e")})); commit=self.put(r,"c",opaque_args("Commitment","CommitmentId","c",{"id":ref("CommitmentId",0),"actor_ref":actor})); node=self.put(r,"n",opaque_args("Node","NodeId","n",{"id":ref("NodeId",0),"label":"n"})); payload=self.put(r,"p",derived_args("PayloadEnvelope","PayloadId",{"id":ref("PayloadId",0),"schema_ref":"s","codec_ref":"raw","body_bytes":d.bytes_value(b"x"),"body_digest":d.bytes_value(hashlib.sha256(b"x").digest(),"digest32"),"body_length":1})); value=self.put(r,"v",derived_args("Value","ValueId",{"id":ref("ValueId",0),"subject_ref":node,"payload_ref":payload,"tombstone":False})) fields={"id":ref("OccurrenceId",0),"value_ref":value,"predecessor_refs":[],"actor_ref":actor,"commitment_ref":commit,"nonce":"n","authorization_evidence_ref":evidence,"instance_order":1}; occurrence=self.put(r,"o",derived_args("Occurrence","OccurrenceId",fields)); claim={"claimed_occurrence_ref":occurrence,"actor_ref":actor,"commitment_ref":commit,"authorization_evidence_ref":evidence,"causal_inputs":[]} for c in (occurrence,claim): self.assertEqual(r.query({"op":"verify-provenance","claim":lit(c)})["value"]["match_status"],"MATCH") absent=copy.deepcopy(fields); absent["authorization_evidence_ref"]=ref("EvidenceId",9); absent["nonce"]="dangling"; self.assertEqual(r.apply(action("bad","record.put",derived_args("Occurrence","OccurrenceId",absent)))["error"]["code"],"UNKNOWN_REFERENCE") forged=copy.deepcopy(r.store); forged["records"]["Occurrence|"+d.typed_key(occurrence)]["fields"]["authorization_evidence_ref"]=ref("EvidenceId",9) with self.assertRaises(d.D001Error): d.validate_store(forged,p) collision=copy.deepcopy(r.store); fake=copy.deepcopy(collision["records"]["Evidence|"+d.typed_key(evidence)]); fake["record_type"]="Cell"; fake["identity"]={"kind":"keyed","domain_id":None,"symbol":None,"preimage_hex":None}; fake["fields"]={"subject_ref":node,"payload_ref":payload,"counter":1}; collision["records"].pop("Evidence|"+d.typed_key(evidence)); collision["records"]["Cell|"+d.typed_key(evidence)]=fake with self.assertRaises(d.D001Error): d.validate_store(collision,p) loaded=d.reconstruct_store(d.serialize_store(r.store),p); self.assertEqual(d.Runtime(p,loaded).query({"op":"verify-provenance","claim":lit(claim)})["value"]["match_status"],"MATCH") # Well-formed same-role forgeries reach matching and directly return non-MATCH. actor2=self.put(r,"actor2",opaque_args("Actor","ActorId","b",{"id":ref("ActorId",0),"name":"b"})) evidence2=self.put(r,"e2",opaque_args("Evidence","EvidenceId","e2",{"id":ref("EvidenceId",0),"data":d.bytes_value(b"e2")})) commit2=self.put(r,"c2",opaque_args("Commitment","CommitmentId","c2",{"id":ref("CommitmentId",0),"actor_ref":actor2})) for field,value in (("actor_ref",actor2),("commitment_ref",commit2),("authorization_evidence_ref",evidence2)): wrong=copy.deepcopy(claim); wrong[field]=value; result=r.query({"op":"verify-provenance","claim":lit(wrong)}); self.assertEqual((result["status"],result["value"]["match_status"]),("OK","FIELD_MISMATCH")) wrong_causal=copy.deepcopy(claim); wrong_causal["causal_inputs"]=[occurrence]; self.assertEqual(r.query({"op":"verify-provenance","claim":lit(wrong_causal)})["value"]["match_status"],"FIELD_MISMATCH") dangling=copy.deepcopy(claim); dangling["claimed_occurrence_ref"]=ref("OccurrenceId",9); self.assertEqual(r.query({"op":"verify-provenance","claim":lit(dangling)})["value"]["match_status"],"UNKNOWN_MATCH") cross=copy.deepcopy(claim); cross["claimed_occurrence_ref"]={"$kind":"ref","type":"RevisionId","hex":occurrence["hex"]}; cross_result=r.query({"op":"verify-provenance","claim":lit(cross)}); self.assertNotEqual((cross_result.get("value") or {}).get("match_status"),"MATCH") revision=self.put(r,"revision",derived_args("Revision","RevisionId",{"id":ref("RevisionId",0),"subject_ref":node,"payload_ref":payload,"predecessor_refs":[],"actor_ref":actor,"nonce":"roleless","tombstone":False,"instance_order":1})) roleless_result=r.query({"op":"verify-provenance","claim":lit(revision)}); self.assertEqual((roleless_result["status"],roleless_result["value"]["match_status"]),("OK","UNSUPPORTED")) print("PASS grounded-structured-provenance") print("PASS E12-grounded-cross-profile-provenance")
def test_version_chain_key_continuity(self): p=profile(); r=d.Runtime(p); ns1=self.put(r,"ns1",opaque_args("Namespace","NamespaceId","ns1",{"id":ref("NamespaceId",0),"name":"one"})); ns2=self.put(r,"ns2",opaque_args("Namespace","NamespaceId","ns2",{"id":ref("NamespaceId",0),"name":"two"})); target=self.put(r,"target",opaque_args("Node","NodeId","target",{"id":ref("NodeId",0),"label":"target"})); base=self.put(r,"base",opaque_args("AliasVersion","AliasId","base",{"id":ref("AliasId",0),"namespace_ref":ns1,"label":"x","target_ref":target,"predecessor_ref":None,"instance_order":1,"tombstone":False,"expires_at":None})) def attempt(step,ns,label,order): return r.apply(action(step,"record.put",opaque_args("AliasVersion","AliasId",step,{"id":ref("AliasId",0),"namespace_ref":ns,"label":label,"target_ref":target,"predecessor_ref":base,"instance_order":order,"tombstone":False,"expires_at":None}))) self.assertEqual(attempt("cross-label",ns1,"y",2)["error"]["code"],"CONTEXT_MISMATCH"); self.assertEqual(attempt("cross-ns",ns2,"x",3)["error"]["code"],"CONTEXT_MISMATCH") forged=copy.deepcopy(r.store); event=attempt("good",ns1,"x",4); self.assertEqual(event["status"],"COMMITTED"); good=event["output"]["ref"]; forged=copy.deepcopy(r.store); forged["records"]["AliasVersion|"+d.typed_key(good)]["fields"]["label"]="wrong" with self.assertRaises(d.D001Error): d.validate_store(forged,p) print("PASS version-chain-key-continuity")
if __name__ == "__main__": unittest.main(verbosity=2)