"""Generic, assertion-blind D-001 experiment substrate. This module implements D001-STORE-2 and D001-Q-2. Concrete identity profiles are data supplied as ProfileSemantics; this module never dispatches on profile or fixture case names. """ from __future__ import annotations import copy import hashlib import json import struct import unicodedata from dataclasses import dataclass, field from types import MappingProxyType from typing import Any, Mapping U64_MAX = (1 << 64) - 1 STORE_VERSION = "D001-STORE-2" QUERY_VERSION = "D001-Q-2" RESOLVER_ORDER = ( "MALFORMED", "TYPE_MISMATCH", "CONTEXT_MISMATCH", "UNKNOWN", "TOMBSTONED", "EXPIRED", "CONTENT_COLLECTED", "UNAVAILABLE", "RESOLVED", ) OPS = frozenset({ "record.put", "index.set", "availability.set", "availability.collect-if-unreachable", "snapshot.capture", "snapshot.restore", "roots.freeze", "reachability.mark", "store.reload", }) FAULTS = frozenset({"HASH_RESULT_OVERRIDE", "CONCURRENT_ROOT_APPEND"}) FAULT_COMPATIBILITY = { "record.put": {("HASH_RESULT_OVERRIDE", "after-preimage-before-typed-lookup")}, "snapshot.capture": {("HASH_RESULT_OVERRIDE", "after-preimage-before-typed-lookup")}, "reachability.mark": {("CONCURRENT_ROOT_APPEND", "after-root-copy-before-first-edge")}, } RESOLVER_MODES = frozenset({"exact", "alias-current", "alias-as-of", "locator-current", "payload-body"}) RETENTION_CLASSES = frozenset({"identity-strong", "body-retaining", "audit-summary", "operational"}) WRAPPER_KINDS = frozenset({"bytes", "digest32", "ref", "slot", "set", "map"}) QUERY_OPS = frozenset({ "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 D001Error(Exception): """Closed runtime failure carrying a stable code and optional detail.""" def __init__(self, code: str, detail: Any = None, *, fatal: bool = False): super().__init__(code) self.code, self.detail, self.fatal = code, detail, fatal def as_json(self) -> dict[str, Any]: return {"code": self.code, "detail": self.detail} def _fail(code: str, detail: Any = None, *, fatal: bool = False) -> None: raise D001Error(code, detail, fatal=fatal) def _nfc(s: Any, label: str = "text") -> str: if type(s) is not str or unicodedata.normalize("NFC", s) != s: _fail("MALFORMED", f"{label} must be NFC text") return s def _u64(v: Any, label: str = "uint64") -> int: if type(v) is not int or not 0 <= v <= U64_MAX: _fail("MALFORMED", f"{label} must be uint64") return v def _exact(obj: Any, keys: set[str], label: str) -> dict[str, Any]: if type(obj) is not dict or set(obj) != keys: _fail("MALFORMED", f"{label} fields") return obj def _hex(s: Any, label: str, length: int | None = None) -> bytes: if type(s) is not str or s.lower() != s or len(s) % 2: _fail("MALFORMED", f"{label} lowercase even hex") try: raw = bytes.fromhex(s) except ValueError: _fail("MALFORMED", f"{label} hex") if length is not None and len(raw) != length: _fail("MALFORMED", f"{label} length") return raw def validate_ref(v: Any, allowed: set[str] | None = None) -> dict[str, Any]: _exact(v, {"$kind", "type", "hex"}, "ref") if v["$kind"] != "ref": _fail("MALFORMED", "ref kind") _nfc(v["type"], "ref type") _hex(v["hex"], "ref") if allowed is not None and v["type"] not in allowed: _fail("TYPE_MISMATCH", {"actual": v["type"], "allowed": sorted(allowed)}) return copy.deepcopy(v) def validate_profile_ref(v: Any, profile: "ProfileSemantics", allowed: set[str] | None = None) -> dict[str, Any]: """Validate a semantic reference, including its profile-declared exact width.""" ref = validate_ref(v, allowed) rule = profile.id_rules.get(ref["type"]) if rule is None: _fail("TYPE_MISMATCH", {"actual": ref["type"], "allowed": sorted(profile.id_rules)}) _hex(ref["hex"], "ref", rule.width) return ref def validate_slot(v: Any, profile: "ProfileSemantics" | None = None) -> dict[str, Any]: _exact(v, {"$kind", "owner_actor_ref", "slot_type", "slot_name"}, "slot") if v["$kind"] != "slot": _fail("MALFORMED", "slot kind") (validate_profile_ref(v["owner_actor_ref"], profile) if profile is not None else validate_ref(v["owner_actor_ref"])) _nfc(v["slot_type"], "slot type") _nfc(v["slot_name"], "slot name") return copy.deepcopy(v) def validate_semantic_endpoint(v: Any, profile: "ProfileSemantics") -> dict[str, Any]: if type(v) is dict and v.get("$kind") == "slot": return validate_slot(v, profile) return validate_profile_ref(v, profile) def typed_key(v: Any) -> str: if type(v) is dict and v.get("$kind") == "ref": validate_ref(v) return "ref|" + v["type"] + "|" + v["hex"] if type(v) is dict and v.get("$kind") == "slot": validate_slot(v) return "slot|" + canonical_json(v).decode("utf-8") return "key|" + canonical_json(v).decode("utf-8") def bytes_value(raw: bytes, kind: str = "bytes") -> dict[str, str]: if kind not in ("bytes", "digest32") or (kind == "digest32" and len(raw) != 32): _fail("MALFORMED", "binary kind/length") return {"$kind": kind, "hex": raw.hex()} def binary_raw(v: Any, expected: str | None = None) -> bytes: _exact(v, {"$kind", "hex"}, "binary") if v["$kind"] not in ("bytes", "digest32") or expected and v["$kind"] != expected: _fail("MALFORMED", "binary kind") return _hex(v["hex"], v["$kind"], 32 if v["$kind"] == "digest32" else None) def _validate_json(v: Any, path: str = "$") -> Any: if v is None or type(v) is bool: return v if type(v) is int: return _u64(v, path) if type(v) is float: _fail("MALFORMED", f"float at {path}") if type(v) is str: return _nfc(v, path) if type(v) is list: return [_validate_json(x, f"{path}[]") for x in v] if type(v) is not dict: _fail("MALFORMED", f"non-JSON at {path}") if "$kind" in v and v.get("$kind") not in WRAPPER_KINDS: _fail("MALFORMED", f"reserved wrapper kind at {path}") if v.get("$kind") in ("bytes", "digest32"): binary_raw(v) return copy.deepcopy(v) if v.get("$kind") == "ref": return validate_ref(v) if v.get("$kind") == "slot": return validate_slot(v) if v.get("$kind") == "set": _exact(v, {"$kind", "items"}, "set") if type(v["items"]) is not list: _fail("MALFORMED", "set items") items = [_validate_json(x, path + ".items") for x in v["items"]] enc = [cve1_from_json(x) for x in items] if len(set(enc)) != len(enc): _fail("MALFORMED", "duplicate set element") return {"$kind": "set", "items": [x for _, x in sorted(zip(enc, items), key=lambda p: p[0])]} if v.get("$kind") == "map": _exact(v, {"$kind", "entries"}, "map") if type(v["entries"]) is not list: _fail("MALFORMED", "map entries") pairs = [] for p in v["entries"]: if type(p) is not list or len(p) != 2: _fail("MALFORMED", "map pair") k, val = _validate_json(p[0], path + ".key"), _validate_json(p[1], path + ".value") pairs.append((cve1_from_json(k), [k, val])) if len({x[0] for x in pairs}) != len(pairs): _fail("MALFORMED", "duplicate map key") return {"$kind": "map", "entries": [p for _, p in sorted(pairs, key=lambda x: x[0])]} out, normalized = {}, set() for k, val in v.items(): nk = _nfc(k, path + ".key") if nk in normalized: _fail("MALFORMED", "NFC key collision") normalized.add(nk) out[nk] = _validate_json(val, path + "." + nk) return out def canonical_json(v: Any, *, lf: bool = False) -> bytes: clean = _validate_json(v) text = json.dumps(clean, ensure_ascii=False, sort_keys=True, separators=(",", ":")) return (text + ("\n" if lf else "")).encode("utf-8") def strict_json_loads(raw: bytes, *, require_lf: bool = False) -> Any: if type(raw) is not bytes: _fail("MALFORMED", "bytes required") if require_lf and (not raw.endswith(b"\n") or raw.endswith(b"\n\n")): _fail("MALFORMED", "exact terminal LF") try: text = raw.decode("utf-8") except UnicodeDecodeError: _fail("MALFORMED", "UTF-8") def pairs(ps: list[tuple[str, Any]]) -> dict[str, Any]: out, seen = {}, set() for k, v in ps: nk = unicodedata.normalize("NFC", k) if k != nk or nk in seen: _fail("MALFORMED", "duplicate or non-NFC key") seen.add(nk) out[k] = v return out try: val = json.loads(text, object_pairs_hook=pairs, parse_float=lambda _: _fail("MALFORMED", "float"), parse_int=lambda s: _u64(int(s))) except D001Error: raise except (ValueError, json.JSONDecodeError): _fail("MALFORMED", "JSON") return _validate_json(val) # CVE1 local exact codec. def _frame(tag: int, payload: bytes = b"") -> bytes: return bytes([tag]) + payload def _u32_frame_length(v: int, label: str) -> bytes: if type(v) is not int or not 0 <= v <= 0xffffffff: _fail("MALFORMED", f"{label} exceeds uint32") return struct.pack(">I", v) def cve1(v: Any) -> bytes: """Encode explicit canonical values represented as (kind, value) tuples.""" if type(v) is not tuple or len(v) != 2: _fail("MALFORMED", "explicit CVE1 value") kind, x = v if kind == "absent": return _frame(0) if kind == "null": return _frame(1) if kind == "bool": if type(x) is not bool: _fail("MALFORMED", "CVE1 bool") return _frame(3 if x else 2) if kind == "uint64": return _frame(4, struct.pack(">Q", _u64(x))) if kind in ("bytes", "text"): raw = x if kind == "bytes" else _nfc(x).encode("utf-8") if type(raw) is not bytes: _fail("MALFORMED", "CVE1 bytes") return _frame(5 if kind == "bytes" else 6, _u32_frame_length(len(raw), "CVE1 scalar length") + raw) if kind in ("array", "set"): if type(x) not in (list, tuple): _fail("MALFORMED", "CVE1 compound") parts = [cve1(q) for q in x] if kind == "set": parts.sort() if len(parts) != len(set(parts)): _fail("MALFORMED", "duplicate CVE1 set") return _frame(7 if kind == "array" else 9, _u32_frame_length(len(parts), "CVE1 compound count") + b"".join(parts)) if kind == "map": if type(x) not in (list, tuple): _fail("MALFORMED", "CVE1 map") pairs = sorted((cve1(k), cve1(val)) for k, val in x) if len(pairs) != len({k for k, _ in pairs}): _fail("MALFORMED", "duplicate CVE1 map key") return _frame(8, _u32_frame_length(len(pairs), "CVE1 map count") + b"".join(k + val for k, val in pairs)) if kind == "ref": validate_ref(x) return _frame(10, cve1(("text", x["type"])) + cve1(("bytes", bytes.fromhex(x["hex"])))) if kind == "slot": if type(x) is not dict: _fail("MALFORMED","CVE1 slot") return _frame(11, cve1(("ref",x["owner_actor_ref"])) + cve1(("text",x["slot_type"])) + cve1(("text",x["slot_name"]))) _fail("MALFORMED", "unknown CVE1 kind") def cve1_from_json(v: Any) -> bytes: v = _validate_json(v) if v is None: return cve1(("null", None)) if type(v) is bool: return cve1(("bool", v)) if type(v) is int: return cve1(("uint64", v)) if type(v) is str: return cve1(("text", v)) if type(v) is list: return cve1(("array", [("bytes", cve1_from_json(x)) for x in v])) if v.get("$kind") in ("bytes", "digest32"): return cve1(("bytes", binary_raw(v))) if v.get("$kind") == "ref": return cve1(("ref", v)) if v.get("$kind") == "set": return cve1(("set", [("bytes", cve1_from_json(x)) for x in v["items"]])) if v.get("$kind") == "map": return cve1(("map", [(("bytes", cve1_from_json(k)), ("bytes", cve1_from_json(x))) for k, x in v["entries"]])) return cve1(("map", [(("text", k), ("bytes", cve1_from_json(x))) for k, x in v.items()])) def canonical_value(value: Any, schema: Any, profile: "ProfileSemantics") -> tuple[Any, bytes]: """Convert one serialized semantic value to CVE1 under a closed schema descriptor.""" if type(schema) is not dict or len(schema) != 1: _fail("MALFORMED","canonical schema") if "optional" in schema: inner=schema["optional"] if value == {"$kind":"absent"}: return value,cve1(("absent",None)) return canonical_value(value,inner,profile) if value == {"$kind":"absent"}: _fail("MALFORMED","absent requires optional schema") if "array" in schema: if type(value) is not list: _fail("MALFORMED","canonical array") items=[canonical_value(x,schema["array"],profile) for x in value] return [x[0] for x in items],cve1(("array",[("bytes",x[1]) for x in items])) if "set" in schema: _exact(value,{"$kind","items"},"canonical set") if value["$kind"]!="set" or type(value["items"]) is not list: _fail("MALFORMED","canonical set") items=[canonical_value(x,schema["set"],profile) for x in value["items"]]; encoded=[x[1] for x in items] if len(encoded)!=len(set(encoded)): _fail("MALFORMED","duplicate canonical set") ordered=sorted(items,key=lambda x:x[1]); return {"$kind":"set","items":[x[0] for x in ordered]},cve1(("set",[("bytes",x[1]) for x in ordered])) if "map" in schema: desc=schema["map"] if type(desc) is not dict or set(desc)!={"key","value"}: _fail("MALFORMED","canonical map schema") _exact(value,{"$kind","entries"},"canonical map") if value["$kind"]!="map" or type(value["entries"]) is not list: _fail("MALFORMED","canonical map") pairs=[] for pair in value["entries"]: if type(pair) is not list or len(pair)!=2: _fail("MALFORMED","canonical map pair") k=canonical_value(pair[0],desc["key"],profile); v=canonical_value(pair[1],desc["value"],profile); pairs.append((k,v)) if len({k[1] for k,_ in pairs})!=len(pairs): _fail("MALFORMED","duplicate canonical map key") pairs.sort(key=lambda x:x[0][1]); normalized={"$kind":"map","entries":[[k[0],v[0]] for k,v in pairs]} return normalized,cve1(("map",[(("bytes",k[1]),("bytes",v[1])) for k,v in pairs])) if "type" not in schema or type(schema["type"]) is not str: _fail("MALFORMED","canonical scalar schema") typ=_nfc(schema["type"],"canonical type") if typ=="null": if value is not None: _fail("MALFORMED","canonical null") return None,cve1(("null",None)) if typ=="bool": if type(value) is not bool: _fail("MALFORMED","canonical bool") return value,cve1(("bool",value)) if typ=="uint64": return _u64(value),cve1(("uint64",value)) if typ=="text": return _nfc(value,"canonical text"),cve1(("text",value)) if typ in ("bytes","digest32"): raw=binary_raw(value,typ); return copy.deepcopy(value),cve1(("bytes",raw)) if typ=="slot": slot=validate_slot(value,profile); return slot,cve1(("slot",slot)) allowed=set(profile.unions.get(typ,{typ})) if typ in profile.id_rules or typ in profile.unions: ref=validate_profile_ref(value,profile,allowed); return ref,cve1(("ref",ref)) _fail("MALFORMED","unknown canonical schema type") @dataclass(frozen=True) class IdRule: kind: str type_tag: str domain: str = "" schema: str = "" codec: str = "CVE1" version: int = 1 fields: tuple[str, ...] = () width: int = 16 @dataclass(frozen=True) class RecordRule: fields: Mapping[str, Any] id_field: str | None = "id" id_type: str | None = None instance_order_field: str | None = None predecessor_fields: tuple[str, ...] = () projection: str | None = None @dataclass(frozen=True) class EdgeRule: from_types: frozenset[str] to_types: frozenset[str] retention_class: str def _deep_freeze(value: Any) -> Any: if isinstance(value, Mapping): return MappingProxyType({k: _deep_freeze(v) for k, v in value.items()}) if type(value) is list: return tuple(_deep_freeze(v) for v in value) if type(value) is tuple: return tuple(_deep_freeze(v) for v in value) if type(value) is set: return frozenset(_deep_freeze(v) for v in value) return value def _deep_copy_plain(value: Any) -> Any: if isinstance(value, Mapping): return {k: _deep_copy_plain(v) for k, v in value.items()} if type(value) in (list, tuple): return [_deep_copy_plain(v) for v in value] if type(value) in (set, frozenset): return {_deep_copy_plain(v) for v in value} return copy.deepcopy(value) @dataclass(frozen=True) class AvailabilityRole: envelope_record_type: str envelope_id_type: str body_record_type: str envelope_ref_field: str body_field: str digest_field: str length_field: str envelope_body_field: str @dataclass(frozen=True) class PinRole: pin_record_type: str pin_id_type: str target_field: str active_field: str expires_field: str @dataclass(frozen=True) class VersionedReferenceRole: record_type: str id_type: str namespace_field: str label_field: str target_field: str predecessor_field: str order_field: str tombstone_field: str expires_field: str namespace_types: tuple[str, ...] target_types: tuple[str, ...] @dataclass(frozen=True) class ProvenanceRole: record_type: str occurrence_id_type: str actor_field: str commitment_field: str authorization_evidence_field: str causal_inputs_field: str | None actor_types: tuple[str, ...] commitment_types: tuple[str, ...] evidence_types: tuple[str, ...] actor_record_types: tuple[str, ...] commitment_record_types: tuple[str, ...] evidence_record_types: tuple[str, ...] causal_types: tuple[str, ...] causal_record_types: tuple[str, ...] @dataclass(frozen=True) class TypedContentRole: state_id_type: str subject_union: str payload_id_type: str derivation_kind: str width: int codec: str version: int fields: tuple[str, ...] type_tag_descriptor: str canonical_bytes_descriptor: str constant_type_tag: str applicability_domain: str applicability_schema: str canonical_value_fields: tuple[str, ...] canonical_encoding: str result_fields: tuple[str, ...] @dataclass(frozen=True) class ProfileSemantics: name: str fixture_id: str id_rules: Mapping[str, IdRule] record_rules: Mapping[str, RecordRule] edge_rules: Mapping[str, EdgeRule] unions: Mapping[str, frozenset[str]] = field(default_factory=dict) binding_kinds: frozenset[str] = frozenset() state_recipe: Mapping[str, Any] = field(default_factory=dict) projections: Mapping[str, Mapping[str, Any]] = field(default_factory=dict) provenance_types: frozenset[str] = frozenset() availability_role: AvailabilityRole | None = None pin_role: PinRole | None = None versioned_reference_roles: tuple[VersionedReferenceRole, ...] = () provenance_roles: tuple[ProvenanceRole, ...] = () typed_content_role: TypedContentRole | None = None def __post_init__(self) -> None: _nfc(self.name, "profile name"); _nfc(self.fixture_id, "fixture id") if not all(type(x) is dict for x in (self.id_rules, self.record_rules, self.edge_rules, self.unions, self.state_recipe, self.projections)): _fail("MALFORMED", "profile mappings") for tag, rule in self.id_rules.items(): _nfc(tag, "ID tag") if type(rule) is not IdRule or tag != rule.type_tag or rule.kind not in ("opaque", "derived", "structural"): _fail("MALFORMED", "ID rule") _nfc(rule.type_tag, "ID type"); _nfc(rule.domain, "ID domain"); _nfc(rule.schema, "ID schema"); _nfc(rule.codec, "ID codec") _u64(rule.version, "ID version") if type(rule.width) is not int or not 1 <= rule.width <= 64: _fail("MALFORMED", "ID width") if type(rule.fields) is not tuple or len(rule.fields) != len(set(rule.fields)): _fail("MALFORMED", "ID fields") for f in rule.fields: _nfc(f, "ID field") if rule.kind == "derived" and (rule.width != 32 or not rule.domain or not rule.schema or rule.codec != "CVE1" or not rule.fields): _fail("MALFORMED", "derived ID rule") if rule.kind != "derived" and rule.fields: _fail("MALFORMED", "non-derived ID fields") structural_types = {"StateSlotRef"} for union, members in self.unions.items(): _nfc(union, "union"); if type(members) is not frozenset or not members or not members <= set(self.id_rules) | structural_types: _fail("MALFORMED", "union") def descriptor_ok(desc: Any) -> None: if desc in ("any", "text", "uint64", "bool", "bytes", "digest32", "slot"): return if not isinstance(desc, Mapping) or len(desc) != 1: _fail("MALFORMED", "field descriptor") kind, inner = next(iter(desc.items())) if kind in ("optional", "array", "set"): descriptor_ok(inner); return if kind == "ref": values = inner if type(inner) in (list, tuple, set, frozenset) else self.unions.get(inner, {inner}) if not values or any(type(x) is not str or x not in set(self.id_rules) | structural_types for x in values): _fail("MALFORMED", "ref descriptor") return _fail("MALFORMED", "field descriptor") for rt, rr in self.record_rules.items(): _nfc(rt, "record type") if type(rr) is not RecordRule or not isinstance(rr.fields, Mapping): _fail("MALFORMED", "record rule") for f, desc in rr.fields.items(): _nfc(f, "record field"); descriptor_ok(desc) if rr.id_field is not None and rr.id_field not in rr.fields: _fail("MALFORMED", "record ID field") if rr.id_type is not None and rr.id_type not in self.id_rules: _fail("MALFORMED", "record ID type") if rr.instance_order_field is not None and (rr.instance_order_field not in rr.fields or rr.fields[rr.instance_order_field] != "uint64"): _fail("MALFORMED", "instance order field") if type(rr.predecessor_fields) is not tuple or any(x not in rr.fields for x in rr.predecessor_fields): _fail("MALFORMED", "predecessor fields") if rr.projection is not None and rr.projection not in self.projections: _fail("MALFORMED", "projection reference") if rr.id_type is not None and rr.id_field is not None: desc = rr.fields[rr.id_field] if not isinstance(desc, Mapping) or desc.get("ref") != rr.id_type: _fail("MALFORMED", "record ID descriptor") for kind, edge in self.edge_rules.items(): _nfc(kind, "edge kind") if type(edge) is not EdgeRule or edge.retention_class not in RETENTION_CLASSES or not edge.from_types or not edge.to_types: _fail("MALFORMED", "edge rule") if not edge.from_types <= set(self.id_rules) | {"StateSlotRef"} or not edge.to_types <= set(self.id_rules) | {"StateSlotRef"}: _fail("MALFORMED", "edge types") if type(self.binding_kinds) is not frozenset or any(type(x) is not str or not _nfc(x) for x in self.binding_kinds): _fail("MALFORMED", "binding kinds") if type(self.provenance_types) is not frozenset or not self.provenance_types <= set(self.id_rules): _fail("MALFORMED", "provenance types") _validate_json(dict(self.state_recipe)) for name, spec in self.projections.items(): _nfc(name, "projection name"); _exact(spec, {"kind", "target_field", "state_field", "tombstone_field", "expires_field"}, "projection") if spec["kind"] != "lifecycle": _fail("MALFORMED", "projection kind") for field_name in ("target_field", "state_field", "tombstone_field", "expires_field"): _nfc(spec[field_name], "projection field") for rr in self.record_rules.values(): if rr.projection is None: continue spec = self.projections[rr.projection] expected = (("target_field", "ref"), ("state_field", "text"), ("tombstone_field", "bool"), ("expires_field", {"optional": "uint64"})) for spec_key, descriptor in expected: field_name = spec[spec_key] if field_name not in rr.fields: _fail("MALFORMED", "projection field missing") actual = rr.fields[field_name] if descriptor == "ref": if not isinstance(actual, Mapping) or set(actual) != {"ref"}: _fail("MALFORMED", "projection target descriptor") elif actual != descriptor: _fail("MALFORMED", "projection field descriptor") if rr.instance_order_field is None: _fail("MALFORMED", "projection instance order required") role = self.availability_role if role is not None: if type(role) is not AvailabilityRole: _fail("MALFORMED", "availability role") for value in (role.envelope_record_type, role.envelope_id_type, role.body_record_type, role.envelope_ref_field, role.body_field, role.digest_field, role.length_field, role.envelope_body_field): _nfc(value, "availability role") if role.envelope_id_type not in self.id_rules or role.envelope_record_type not in self.record_rules or role.body_record_type not in self.record_rules: _fail("MALFORMED", "availability role type") envelope = self.record_rules[role.envelope_record_type] body = self.record_rules[role.body_record_type] required_envelope = {role.digest_field: "digest32", role.length_field: "uint64", role.envelope_body_field: "bytes"} if envelope.id_type != role.envelope_id_type or any(envelope.fields.get(k) != v for k, v in required_envelope.items()): _fail("MALFORMED", "availability envelope descriptors") if body.id_field is not None or dict(body.fields) != {role.envelope_ref_field: {"ref": role.envelope_id_type}, role.body_field: "bytes"}: _fail("MALFORMED", "availability body descriptors") pin = self.pin_role if pin is not None: if type(pin) is not PinRole: _fail("MALFORMED", "pin role") for value in (pin.pin_record_type, pin.pin_id_type, pin.target_field, pin.active_field, pin.expires_field): _nfc(value, "pin role") if pin.pin_record_type not in self.record_rules or pin.pin_id_type not in self.id_rules: _fail("MALFORMED", "pin role type") record = self.record_rules[pin.pin_record_type] if record.id_type != pin.pin_id_type or record.fields.get(pin.active_field) != "bool" or record.fields.get(pin.expires_field) != "uint64": _fail("MALFORMED", "pin role descriptors") target = record.fields.get(pin.target_field) if not isinstance(target, Mapping) or set(target) != {"ref"}: _fail("MALFORMED", "pin target descriptor") if type(self.versioned_reference_roles) is not tuple or type(self.provenance_roles) is not tuple: _fail("MALFORMED", "semantic roles") typed = self.typed_content_role if typed is not None: if type(typed) is not TypedContentRole: _fail("MALFORMED", "typed content role") text_values = (typed.state_id_type, typed.subject_union, typed.payload_id_type, typed.derivation_kind, typed.codec, typed.type_tag_descriptor, typed.canonical_bytes_descriptor, typed.constant_type_tag, typed.applicability_domain, typed.applicability_schema, typed.canonical_encoding) for value in text_values: _nfc(value, "typed content role") _u64(typed.version, "typed content version") if type(typed.width) is not int or typed.width != 32: _fail("MALFORMED", "typed content width") if (typed.state_id_type,typed.subject_union,typed.payload_id_type)!=("ContentId","SubjectRef","PayloadId"): _fail("MALFORMED", "typed content semantic types") exact_fields=("type_tag","canonical_bytes") exact_value_fields=("subject_ref","payload_ref") exact_result=("ref","canonical_bytes","preimage_hex","subject_ref","payload_ref") if typed.derivation_kind!="derived" or typed.codec!="CVE1" or typed.fields!=exact_fields or typed.type_tag_descriptor!="NFC-text" or typed.canonical_bytes_descriptor!="bytes" or typed.constant_type_tag!="state" or typed.canonical_value_fields!=exact_value_fields or typed.canonical_encoding!="canonical-json-utf8" or typed.result_fields!=exact_result: _fail("MALFORMED", "typed content exact contract") rule=self.id_rules.get(typed.state_id_type) if rule is None or (rule.kind,rule.width,rule.codec,rule.version,rule.fields,rule.domain,rule.schema)!=(typed.derivation_kind,typed.width,typed.codec,typed.version,typed.fields,typed.applicability_domain,typed.applicability_schema): _fail("MALFORMED", "typed content derivation closure") subjects=self.unions.get(typed.subject_union) payload=self.id_rules.get(typed.payload_id_type) if subjects is None or payload is None: _fail("MALFORMED", "typed content operand types") carriers=[rr for rr in self.record_rules.values() if rr.id_type==typed.state_id_type and rr.fields.get("type_tag")=="text" and rr.fields.get("canonical_bytes")=="bytes"] if not carriers: _fail("MALFORMED", "typed content state schema applicability") def declared_ref_types(desc: Any, *, optional: bool, label: str) -> tuple[str, ...]: if optional: if not isinstance(desc, Mapping) or set(desc) != {"optional"}: _fail("MALFORMED", label) desc = desc["optional"] if not isinstance(desc, Mapping) or set(desc) != {"ref"}: _fail("MALFORMED", label) inner = desc["ref"] values = inner if type(inner) in (list, tuple, set, frozenset) else self.unions.get(inner, {inner}) if not values or any(type(x) is not str for x in values): _fail("MALFORMED", label) return tuple(sorted(values)) def declared_records(types: tuple[str, ...], records: tuple[str, ...], label: str) -> None: if type(records) is not tuple or not records or len(records) != len(set(records)): _fail("MALFORMED", label) covered = set() for record_type in records: _nfc(record_type, label); record = self.record_rules.get(record_type) if record is None or record.id_type not in types: _fail("MALFORMED", label) covered.add(record.id_type) if covered != set(types): _fail("MALFORMED", label) version_modes = set() for version_role in self.versioned_reference_roles: if type(version_role) is not VersionedReferenceRole: _fail("MALFORMED", "versioned reference role") for value in (version_role.record_type, version_role.id_type, version_role.namespace_field, version_role.label_field, version_role.target_field, version_role.predecessor_field, version_role.order_field, version_role.tombstone_field, version_role.expires_field): _nfc(value, "versioned reference role") fields = (version_role.namespace_field, version_role.label_field, version_role.target_field, version_role.predecessor_field, version_role.order_field, version_role.tombstone_field, version_role.expires_field) if len(fields) != len(set(fields)): _fail("MALFORMED", "versioned role distinct fields") if version_role.id_type in version_modes: _fail("MALFORMED", "duplicate versioned role") version_modes.add(version_role.id_type) rr = self.record_rules.get(version_role.record_type) if rr is None or rr.id_type != version_role.id_type or rr.instance_order_field != version_role.order_field or version_role.predecessor_field not in rr.predecessor_fields: _fail("MALFORMED", "versioned role identity/order") expected = {version_role.label_field:"text", version_role.order_field:"uint64", version_role.tombstone_field:"bool", version_role.expires_field:{"optional":"uint64"}} if any(rr.fields.get(k) != v for k,v in expected.items()): _fail("MALFORMED", "versioned role descriptors") namespace_types = declared_ref_types(rr.fields.get(version_role.namespace_field), optional=False, label="versioned namespace descriptor") target_types = declared_ref_types(rr.fields.get(version_role.target_field), optional=True, label="versioned target descriptor") predecessor_types = declared_ref_types(rr.fields.get(version_role.predecessor_field), optional=True, label="versioned predecessor descriptor") if type(version_role.namespace_types) is not tuple or tuple(sorted(version_role.namespace_types)) != namespace_types or len(set(version_role.namespace_types)) != len(version_role.namespace_types): _fail("MALFORMED", "versioned namespace types") if type(version_role.target_types) is not tuple or tuple(sorted(version_role.target_types)) != target_types or len(set(version_role.target_types)) != len(version_role.target_types): _fail("MALFORMED", "versioned target types") if predecessor_types != (version_role.id_type,): _fail("MALFORMED", "versioned predecessor type") provenance_ids = set() for provenance_role in self.provenance_roles: if type(provenance_role) is not ProvenanceRole: _fail("MALFORMED", "provenance role") for value in (provenance_role.record_type, provenance_role.occurrence_id_type, provenance_role.actor_field, provenance_role.commitment_field, provenance_role.authorization_evidence_field): _nfc(value, "provenance role") if provenance_role.causal_inputs_field is not None: _nfc(provenance_role.causal_inputs_field, "provenance role") fields = (provenance_role.actor_field, provenance_role.commitment_field, provenance_role.authorization_evidence_field) + (() if provenance_role.causal_inputs_field is None else (provenance_role.causal_inputs_field,)) if len(fields) != len(set(fields)): _fail("MALFORMED", "provenance role distinct fields") if provenance_role.occurrence_id_type in provenance_ids: _fail("MALFORMED", "duplicate provenance role") provenance_ids.add(provenance_role.occurrence_id_type) rr=self.record_rules.get(provenance_role.record_type) if rr is None or rr.id_type != provenance_role.occurrence_id_type or provenance_role.occurrence_id_type not in self.provenance_types: _fail("MALFORMED", "provenance role identity") actual_types = [] for field_name, label in ((provenance_role.actor_field,"actor"),(provenance_role.commitment_field,"commitment"),(provenance_role.authorization_evidence_field,"evidence")): actual_types.append(declared_ref_types(rr.fields.get(field_name), optional=False, label="provenance "+label+" descriptor")) causal_actual = () if provenance_role.causal_inputs_field is not None: causal=rr.fields.get(provenance_role.causal_inputs_field) if not isinstance(causal, Mapping) or set(causal) != {"array"}: _fail("MALFORMED", "provenance causal descriptor") causal_actual=declared_ref_types(causal["array"], optional=False, label="provenance causal descriptor") declared_types = (provenance_role.actor_types, provenance_role.commitment_types, provenance_role.evidence_types) for declared, actual, label in zip(declared_types, actual_types, ("actor","commitment","evidence")): if type(declared) is not tuple or tuple(sorted(declared)) != actual or len(set(declared)) != len(declared): _fail("MALFORMED", "provenance "+label+" types") if provenance_role.causal_inputs_field is None: if provenance_role.causal_types != () or provenance_role.causal_record_types != (): _fail("MALFORMED", "null provenance causal declarations") else: if type(provenance_role.causal_types) is not tuple or tuple(sorted(provenance_role.causal_types)) != causal_actual or len(set(provenance_role.causal_types)) != len(provenance_role.causal_types): _fail("MALFORMED", "provenance causal types") declared_records(provenance_role.causal_types, provenance_role.causal_record_types, "provenance causal records") declared_records(provenance_role.actor_types, provenance_role.actor_record_types, "provenance actor records") declared_records(provenance_role.commitment_types, provenance_role.commitment_record_types, "provenance commitment records") declared_records(provenance_role.evidence_types, provenance_role.evidence_record_types, "provenance evidence records") # Freeze all caller-owned mappings and nested descriptors after eager validation. frozen_records = {} for key, rr in self.record_rules.items(): frozen_records[key] = RecordRule(_deep_freeze(_deep_copy_plain(rr.fields)), rr.id_field, rr.id_type, rr.instance_order_field, rr.predecessor_fields, rr.projection) object.__setattr__(self, "id_rules", MappingProxyType(copy.deepcopy(dict(self.id_rules)))) object.__setattr__(self, "record_rules", MappingProxyType(frozen_records)) object.__setattr__(self, "edge_rules", MappingProxyType(copy.deepcopy(dict(self.edge_rules)))) object.__setattr__(self, "unions", MappingProxyType(copy.deepcopy(dict(self.unions)))) object.__setattr__(self, "state_recipe", _deep_freeze(copy.deepcopy(dict(self.state_recipe)))) object.__setattr__(self, "projections", _deep_freeze({k: copy.deepcopy(v) for k, v in self.projections.items()})) def _trace_path(value: Any) -> str: path=_nfc(value,"trace path") if not path or path.startswith("/") or path.endswith("/") or "//" in path: _fail("MALFORMED","noncanonical trace path") parts=path.split("/") if any(not part or part in (".","..") for part in parts): _fail("MALFORMED","trace path segment") return path def _validate_trace_event(event: Any) -> dict[str, Any]: keys={"seq","step_id","op","status","reads","writes","output","error","before_digest","after_digest"} _exact(event,keys,"trace event") _u64(event["seq"],"trace sequence"); _nfc(event["step_id"],"trace step"); _nfc(event["op"],"trace op") if event["status"] not in ("COMMITTED","NO_COMMIT","FATAL"): _fail("MALFORMED","trace status") for name in ("reads","writes"): if type(event[name]) is not list: _fail("MALFORMED","trace paths") for path in event[name]: _trace_path(path) _validate_json(event["output"]) if event["error"] is not None: _exact(event["error"],{"code","detail"},"trace error"); _nfc(event["error"]["code"],"trace error code"); _validate_json(event["error"]["detail"]) if (event["status"]=="COMMITTED") != (event["error"] is None): _fail("MALFORMED","trace status/error") _hex(event["before_digest"],"trace before digest",32); _hex(event["after_digest"],"trace after digest",32) return event def empty_store(profile: ProfileSemantics, clock: int = 0) -> dict[str, Any]: _u64(clock) allocations = {} for tag, rule in profile.id_rules.items(): if rule.kind in ("opaque", "derived"): allocations[tag] = {"id_type": tag, "kind": rule.kind, "seed": profile.fixture_id, "next_order": 0, "used": {}} return { "store_version": STORE_VERSION, "profile_name": profile.name, "clock": clock, "records": {}, "allocations": allocations, "bindings": {}, "edges": {}, "lifecycle": {}, "snapshots": {}, "override_provenance": {}, "roots": {"next_epoch": 0, "frozen": {}, "pending": {}}, "availability": {}, "registers": {}, "trace": [], } def _store_image(store: dict[str, Any]) -> dict[str, Any]: keys = {"store_version", "profile_name", "clock", "records", "allocations", "bindings", "edges", "lifecycle", "snapshots", "override_provenance", "roots", "availability"} return {k: copy.deepcopy(store[k]) for k in keys} def serialize_store(store: dict[str, Any]) -> bytes: return canonical_json(_store_image(store), lf=True) def store_digest(store: dict[str, Any]) -> str: return hashlib.sha256(serialize_store(store)).hexdigest() def _field_validate(value: Any, descriptor: Any, profile: ProfileSemantics) -> Any: if descriptor == "any": return _validate_json(value) if descriptor == "text": return _nfc(value) if descriptor == "uint64": return _u64(value) if descriptor == "bool": if type(value) is not bool: _fail("MALFORMED", "bool") return value if descriptor == "bytes": binary_raw(value, "bytes"); return copy.deepcopy(value) if descriptor == "digest32": binary_raw(value, "digest32"); return copy.deepcopy(value) if descriptor == "slot": return validate_slot(value, profile) if isinstance(descriptor, Mapping) and "optional" in descriptor: return None if value is None else _field_validate(value, descriptor["optional"], profile) if isinstance(descriptor, Mapping) and "array" in descriptor: if type(value) is not list: _fail("MALFORMED", "array") return [_field_validate(x, descriptor["array"], profile) for x in value] if isinstance(descriptor, Mapping) and "set" in descriptor: if type(value) is not dict or value.get("$kind") != "set": _fail("MALFORMED", "set") clean = _validate_json(value) for x in clean["items"]: _field_validate(x, descriptor["set"], profile) return clean if isinstance(descriptor, Mapping) and "ref" in descriptor: allowed = set(descriptor["ref"] if type(descriptor["ref"]) in (list, tuple, set, frozenset) else profile.unions.get(descriptor["ref"], {descriptor["ref"]})) if type(value) is dict and value.get("$kind") == "slot": if "StateSlotRef" not in allowed: _fail("TYPE_MISMATCH", {"actual": "StateSlotRef", "allowed": sorted(allowed)}) return validate_slot(value, profile) return validate_profile_ref(value, profile, allowed - {"StateSlotRef"}) _fail("MALFORMED", "unknown field descriptor") def _derive(rule: IdRule, fields: Mapping[str, Any], override: Mapping[str, Any] | None = None) -> tuple[dict[str, Any], bytes]: if type(fields) is not dict or set(fields) != set(rule.fields): _fail("MALFORMED", "derived fields") if override is not None: _exact(override, {"schema", "codec"}, "rule override") schema, codec = _nfc(override["schema"], "override schema"), _nfc(override["codec"], "override codec") else: schema, codec = rule.schema, rule.codec envelope = { "domain": rule.domain, "schema": schema, "codec": codec, "hash_algorithm": "sha2-256", "version": rule.version, "fields": {k: fields[k] for k in rule.fields}, } preimage = cve1_from_json(envelope) return {"$kind": "ref", "type": rule.type_tag, "hex": hashlib.sha256(preimage).hexdigest()}, preimage def _opaque_ref(profile: ProfileSemantics, rule: IdRule, domain_id: str, symbol: str) -> dict[str, Any]: raw = hashlib.sha256((profile.fixture_id + "/" + profile.name + "/" + domain_id + "/" + symbol).encode()).digest()[:rule.width] return {"$kind": "ref", "type": rule.type_tag, "hex": raw.hex()} def _lifecycle_projection(store: dict[str, Any], profile: ProfileSemantics) -> dict[str, Any]: heads: dict[str, tuple[int, dict[str, Any]]] = {} for rec in store["records"].values(): rr = profile.record_rules[rec["record_type"]] if rr.projection is None: continue spec = profile.projections[rr.projection] target = validate_semantic_endpoint(rec["fields"][spec["target_field"]], profile) order = rec["instance_order"] key = typed_key(target) old = heads.get(key) if old is not None and old[0] == order: _fail("ORDER_VIOLATION", "duplicate lifecycle target order") value = {"target_ref": target, "state": rec["fields"][spec["state_field"]], "tombstone": rec["fields"][spec["tombstone_field"]], "expires_at": rec["fields"][spec["expires_field"]]} if old is None or order > old[0]: heads[key] = (order, value) return {key: value for key, (_, value) in heads.items()} def _compute_mark(store: dict[str, Any], frozen: dict[str, Any], epoch: int, classes: list[str], deferred_to_epoch: int | None, deferred_root: dict[str, Any] | None) -> dict[str, Any]: work = list(frozen["root_refs"]) + [p["target_ref"] for p in frozen["pins"]] seen: dict[str, Any] = {} reasons: list[dict[str, Any]] = [ {"target": typed_key(x), "source": "root|epoch|" + str(epoch), "edge_kind": "frozen-root", "retention_class": "identity-strong"} for x in frozen["root_refs"] ] + [ {"target": typed_key(p["target_ref"]), "source": typed_key(p["pin_ref"]), "edge_kind": "active-audit-pin", "retention_class": "audit-summary"} for p in frozen["pins"] ] queue = [(x, None, None) for x in work] outgoing = sorted(store["edges"].values(), key=lambda e: (e["kind"], typed_key(e["from"]), typed_key(e["to"]), e["retention_class"])) while queue: node, source, edge = queue.pop(0); nk = typed_key(node) if nk in seen: continue seen[nk] = node if edge is not None: reasons.append({"target": nk, "source": typed_key(source), "edge_kind": edge["kind"], "retention_class": edge["retention_class"]}) for candidate in outgoing: if candidate["retention_class"] in classes and typed_key(candidate["from"]) == nk: queue.append((candidate["to"], node, candidate)) return {"epoch": epoch, "classes": list(classes), "reachable": sorted(seen), "reasons": sorted(reasons, key=lambda x: (x["target"], x["source"], x["edge_kind"])), "completed": True, "deferred_to_epoch": deferred_to_epoch, "deferred_root": deferred_root} def _record_key(record_type: str, key: Any) -> str: return record_type + "|" + typed_key(key) def _authoritative_record(rec: dict[str, Any], ref: dict[str, Any], profile: ProfileSemantics) -> bool: rr = profile.record_rules[rec["record_type"]] if ref.get("$kind") == "ref": return rr.id_type == ref["type"] if ref.get("$kind") == "slot": return rr.id_field is not None and rr.fields.get(rr.id_field) == "slot" return False def _freeze_receipt(epoch: int, clock: int, roots: list[Any], pins: list[Any], predecessor: str | None) -> str: payload = {"epoch": epoch, "frozen_at_clock": clock, "root_refs": roots, "pins": pins, "predecessor_receipt": predecessor, "latest_token": epoch} return hashlib.sha256(canonical_json(payload)).hexdigest() def _override_key(ref: dict[str, Any]) -> str: return typed_key(ref) def _is_cross_typed_hash_override(store: dict[str, Any], natural_ref: dict[str, Any], stored_ref: dict[str, Any], preimage_hex: str | None = None) -> bool: """Authorize only an exact persisted pre-freeze cross-domain experiment receipt.""" provenance = store.get("override_provenance", {}).get(_override_key(stored_ref)) if provenance is None or natural_ref == stored_ref: return False if provenance.get("forced_target_ref") != stored_ref or provenance.get("natural_ref") != natural_ref: return False if preimage_hex is not None and provenance.get("natural_preimage_hash") != hashlib.sha256(bytes.fromhex(preimage_hex)).hexdigest(): return False source_ref = provenance.get("source_ref") sources = [rec for rec in store["records"].values() if rec["key"] == source_ref and type(source_ref) is dict and rec["identity"]["kind"] in ("opaque","derived")] return bool(len(sources) == 1 and source_ref["type"] != stored_ref["type"] and source_ref["hex"] == stored_ref["hex"]) def _find_ref(store: dict[str, Any], ref: dict[str, Any], profile: ProfileSemantics | None = None) -> tuple[str, dict[str, Any]] | None: """Find one semantic record; ambiguity is never resolved by map order.""" tk = typed_key(ref) matches = [(k, rec) for k, rec in store["records"].items() if typed_key(rec["key"]) == tk] if not matches: return None if profile is not None: authoritative = [x for x in matches if _authoritative_record(x[1], ref, profile)] if len(authoritative) > 1: _fail("VERIFICATION_FAILED", {"ambiguous_authoritative_ref": tk}) if authoritative: return authoritative[0] # Keyed records (notably role-managed bodies) are a compatibility fallback only. if len(matches) > 1: _fail("VERIFICATION_FAILED", {"ambiguous_fallback_ref": tk}) return matches[0] def _require_role_ref(store: dict[str, Any], ref: Any, profile: ProfileSemantics, allowed_types: tuple[str, ...], allowed_records: tuple[str, ...], label: str) -> tuple[str, dict[str, Any]]: clean = validate_profile_ref(ref, profile, set(allowed_types)) found = _find_ref(store, clean, profile) if not found: _fail("UNKNOWN_REFERENCE", label) if found[1]["record_type"] not in allowed_records or not _authoritative_record(found[1], clean, profile): _fail("TYPE_MISMATCH", label) return found def _validate_version_predecessor(store: dict[str, Any], rec: dict[str, Any], role: VersionedReferenceRole, profile: ProfileSemantics) -> None: pred = rec["fields"][role.predecessor_field] if pred is None: return found = _find_ref(store, pred, profile) if not found: _fail("UNKNOWN_REFERENCE", role.predecessor_field) predecessor = found[1] if predecessor["record_type"] != role.record_type: _fail("TYPE_MISMATCH", role.predecessor_field) if (predecessor["fields"][role.namespace_field] != rec["fields"][role.namespace_field] or predecessor["fields"][role.label_field] != rec["fields"][role.label_field]): _fail("CONTEXT_MISMATCH", "version predecessor key") def _validate_provenance_grounding(store: dict[str, Any], rec: dict[str, Any], role: ProvenanceRole, profile: ProfileSemantics) -> None: fields = rec["fields"] _require_role_ref(store, fields[role.actor_field], profile, role.actor_types, role.actor_record_types, role.actor_field) _require_role_ref(store, fields[role.commitment_field], profile, role.commitment_types, role.commitment_record_types, role.commitment_field) _require_role_ref(store, fields[role.authorization_evidence_field], profile, role.evidence_types, role.evidence_record_types, role.authorization_evidence_field) if role.causal_inputs_field is not None: causal_inputs = fields[role.causal_inputs_field] if type(causal_inputs) is not list: _fail("MALFORMED", role.causal_inputs_field) seen = set() for causal_ref in causal_inputs: clean = validate_profile_ref(causal_ref, profile, set(role.causal_types)) key = typed_key(clean) if key in seen: _fail("MALFORMED", "duplicate causal input") seen.add(key) _require_role_ref(store, clean, profile, role.causal_types, role.causal_record_types, role.causal_inputs_field) def _require_edge(store: dict[str, Any], kind: str, source: Any, target: Any) -> None: key = kind + "|" + typed_key(source) + "|" + typed_key(target) if key not in store["edges"]: _fail("VERIFICATION_FAILED", "missing transition edge "+kind) def _validate_transition_graph(store: dict[str, Any], profile: ProfileSemantics) -> None: # This optional protected graph contract activates only for the complete exact schema. Generic profiles may reuse # record or edge names without inheriting D-001 transition semantics or reaching field-indexing exceptions. expected_records = { "Generation": RecordRule({"id":{"ref":"GenerationId"},"world_ref":{"ref":"WorldId"},"root_ref":{"ref":"RootValueId"},"parent_ref":{"ref":"ParentId"},"predecessor_generation_ref":{"optional":{"ref":"GenerationId"}},"transition_order":"uint64"}, "id", "GenerationId", "transition_order", ("predecessor_generation_ref",), None), "Selection": RecordRule({"id":{"ref":"SelectionId"},"world_ref":{"ref":"WorldId"},"root_ref":{"ref":"RootValueId"},"generation_ref":{"ref":"GenerationId"},"parent_ref":{"ref":"ParentId"},"policy_ref":{"ref":"PolicyId"},"predecessor_selection_ref":{"optional":{"ref":"SelectionId"}},"nonce":"text","kind":"text","transition_order":"uint64"}, "id", "SelectionId", "transition_order", ("predecessor_selection_ref",), None), } expected_edges = { "selection-generation": EdgeRule(frozenset({"SelectionId"}),frozenset({"GenerationId"}),"identity-strong"), "selection-world": EdgeRule(frozenset({"SelectionId"}),frozenset({"WorldId"}),"identity-strong"), "selection-root": EdgeRule(frozenset({"SelectionId"}),frozenset({"RootValueId"}),"identity-strong"), "selection-parent": EdgeRule(frozenset({"SelectionId"}),frozenset({"ParentId"}),"audit-summary"), "selection-policy": EdgeRule(frozenset({"SelectionId"}),frozenset({"PolicyId"}),"audit-summary"), "generation-world": EdgeRule(frozenset({"GenerationId"}),frozenset({"WorldId"}),"identity-strong"), "generation-root": EdgeRule(frozenset({"GenerationId"}),frozenset({"RootValueId"}),"identity-strong"), "generation-parent": EdgeRule(frozenset({"GenerationId"}),frozenset({"ParentId"}),"audit-summary"), "generation-predecessor": EdgeRule(frozenset({"GenerationId"}),frozenset({"GenerationId"}),"audit-summary"), "selection-predecessor": EdgeRule(frozenset({"SelectionId"}),frozenset({"SelectionId"}),"audit-summary"), } if any(profile.record_rules.get(name) != rule for name,rule in expected_records.items()): return if any(profile.edge_rules.get(name) != rule for name,rule in expected_edges.items()): return generations = {rec["fields"]["transition_order"]: rec for rec in store["records"].values() if rec["record_type"] == "Generation"} selections = {rec["fields"]["transition_order"]: rec for rec in store["records"].values() if rec["record_type"] == "Selection"} if not set(selections) <= set(generations) or set(generations) - set(selections) not in (set(), {len(generations)}): _fail("ORDER_VIOLATION", "generation/selection order domains") if set(generations) != set(range(1, len(generations)+1)) or set(selections) != set(range(1, len(selections)+1)): _fail("ORDER_VIOLATION", "transition order continuity") for order in sorted(selections): generation, selection = generations[order], selections[order] gf, sf = generation["fields"], selection["fields"] if sf["generation_ref"] != generation["key"] or any(sf[name] != gf[name] for name in ("world_ref","root_ref","parent_ref")): _fail("CONTEXT_MISMATCH", "generation/selection pair") expected_g = None if order == 1 else generations[order-1]["key"] expected_s = None if order == 1 else selections[order-1]["key"] if gf["predecessor_generation_ref"] != expected_g or sf["predecessor_selection_ref"] != expected_s: _fail("ORDER_VIOLATION", "transition predecessor chain") for kind, source, target in (("generation-world",generation["key"],gf["world_ref"]),("generation-root",generation["key"],gf["root_ref"]),("generation-parent",generation["key"],gf["parent_ref"]),("selection-generation",selection["key"],sf["generation_ref"]),("selection-world",selection["key"],sf["world_ref"]),("selection-root",selection["key"],sf["root_ref"]),("selection-parent",selection["key"],sf["parent_ref"]),("selection-policy",selection["key"],sf["policy_ref"])): _require_edge(store,kind,source,target) if order > 1: _require_edge(store,"generation-predecessor",generation["key"],expected_g) _require_edge(store,"selection-predecessor",selection["key"],expected_s) def _validate_occurrence_graph(store: dict[str, Any], profile: ProfileSemantics) -> None: expected = RecordRule({"id":{"ref":"OccurrenceId"},"revision_value_ref":{"ref":"RevisionValueId"},"predecessor_occurrence_refs":{"array":{"ref":"OccurrenceId"}},"actor_ref":{"ref":"ActorRef"},"commitment_ref":{"ref":"CommitmentId"},"causal_inputs":{"array":{"ref":"OccurrenceId"}},"attempt_ref":{"ref":"AttemptId"},"authorization_evidence_ref":{"ref":"AuthorizationEvidenceId"},"instance_order":"uint64"},"id","OccurrenceId","instance_order",("predecessor_occurrence_refs",),None) if profile.record_rules.get("Occurrence") != expected: return for rec in store["records"].values(): if rec["record_type"] != "Occurrence": continue _require_edge(store,"occurrence-value",rec["key"],rec["fields"]["revision_value_ref"]) _require_edge(store,"protected-state-attempt",rec["key"],rec["fields"]["attempt_ref"]) for predecessor in rec["fields"]["predecessor_occurrence_refs"]: _require_edge(store,"predecessor-revision",rec["key"],predecessor) def _type_of(endpoint: Any) -> str: if type(endpoint) is dict and endpoint.get("$kind") == "ref": return endpoint["type"] if type(endpoint) is dict and endpoint.get("$kind") == "slot": return "StateSlotRef" _fail("TYPE_MISMATCH", "edge endpoint") def _embedded_refs(v: Any): if type(v) is dict and v.get("$kind") == "ref": yield v elif type(v) is dict: for x in v.values(): yield from _embedded_refs(x) elif type(v) in (list, tuple): for x in v: yield from _embedded_refs(x) def _validate_consumed_refs(v: Any, profile: ProfileSemantics) -> None: """Validate refs/slots recursively at a semantic consumer, not in generic literals.""" if type(v) is dict and v.get("$kind") == "ref": validate_profile_ref(v, profile); return if type(v) is dict and v.get("$kind") == "slot": validate_slot(v, profile); return if type(v) is dict: for x in v.values(): _validate_consumed_refs(x, profile) elif type(v) in (list, tuple): for x in v: _validate_consumed_refs(x, profile) def _expected_identity(rr: RecordRule, key: Any, profile: ProfileSemantics) -> tuple[str, IdRule | None]: if rr.id_type is not None: rule = profile.id_rules[rr.id_type] return rule.kind, rule if type(key) is dict and key.get("$kind") == "slot": return "structural", None return "keyed", None def _validate_binding_entry(key: str, binding: Any, profile: ProfileSemantics) -> None: _exact(binding, {"binding_kind", "key", "value", "version"}, "binding") kind = _nfc(binding["binding_kind"], "binding kind") _validate_consumed_refs([binding["key"], binding["value"]], profile) if kind not in profile.binding_kinds or key != kind + "|" + typed_key(binding["key"]): _fail("MALFORMED", "binding key") _validate_json(binding["value"]); _u64(binding["version"]) def validate_store(store: Any, profile: ProfileSemantics) -> dict[str, Any]: required = {"store_version", "profile_name", "clock", "records", "allocations", "bindings", "edges", "lifecycle", "snapshots", "override_provenance", "roots", "availability", "registers", "trace"} _exact(store, required, "store") if store["store_version"] != STORE_VERSION or store["profile_name"] != profile.name: _fail("MALFORMED", "store identity") _u64(store["clock"]) for name in ("records", "allocations", "bindings", "edges", "lifecycle", "snapshots", "override_provenance", "availability", "registers"): if type(store[name]) is not dict: _fail("MALFORMED", name) if type(store["trace"]) is not list: _fail("MALFORMED", "trace") _validate_json(store["registers"]); _validate_json(store["trace"]) _validate_consumed_refs([store["registers"], store["trace"]], profile) allocation_tags = {tag for tag, rule in profile.id_rules.items() if rule.kind in ("opaque", "derived")} if set(store["allocations"]) != allocation_tags: _fail("VERIFICATION_FAILED", "allocation domain set") ledger_uses: dict[str, dict[str, Any]] = {} for tag, ledger in store["allocations"].items(): _exact(ledger, {"id_type", "kind", "seed", "next_order", "used"}, "allocation") rule = profile.id_rules[tag] if ledger["id_type"] != tag or ledger["kind"] != rule.kind or ledger["seed"] != profile.fixture_id: _fail("VERIFICATION_FAILED", "allocation rule") next_order = _u64(ledger["next_order"]) if type(ledger["used"]) is not dict or next_order != len(ledger["used"]): _fail("VERIFICATION_FAILED", "allocation continuity") orders = set() for raw_id, use in ledger["used"].items(): _hex(raw_id, "allocated ID", rule.width); _exact(use, {"first_order", "preimage_hex", "domain_id", "symbol"}, "allocation use") order = _u64(use["first_order"]) if order in orders or order >= next_order: _fail("VERIFICATION_FAILED", "allocation order") orders.add(order) if rule.kind == "derived": _hex(use["preimage_hex"], "preimage") if use["domain_id"] is not None or use["symbol"] is not None: _fail("VERIFICATION_FAILED", "derived allocation evidence") else: if use["preimage_hex"] is not None: _fail("VERIFICATION_FAILED", "opaque preimage") domain_id, symbol = _nfc(use["domain_id"], "opaque domain"), _nfc(use["symbol"], "opaque symbol") if _opaque_ref(profile, rule, domain_id, symbol)["hex"] != raw_id: _fail("VERIFICATION_FAILED", "opaque recipe") if orders != set(range(next_order)): _fail("VERIFICATION_FAILED", "allocation order gaps") ledger_uses[tag] = ledger["used"] orders: dict[tuple[str, int], str] = {}; observed: dict[str, set[str]] = {tag: set() for tag in allocation_tags} for key, rec in store["records"].items(): _exact(rec, {"record_type", "key", "fields", "identity", "instance_order"}, "record") rt = _nfc(rec["record_type"], "record type") if rt not in profile.record_rules: _fail("UNSUPPORTED_RECORD", rt) rr = profile.record_rules[rt] if key != _record_key(rt, rec["key"]): _fail("VERIFICATION_FAILED", "record key") if type(rec["fields"]) is not dict or set(rec["fields"]) != set(rr.fields): _fail("MALFORMED", "record fields") for f, desc in rr.fields.items(): _field_validate(rec["fields"][f], desc, profile) role = profile.availability_role if role is not None and rt == role.envelope_record_type: raw = binary_raw(rec["fields"][role.envelope_body_field], "bytes") if rec["fields"][role.length_field] != len(raw) or binary_raw(rec["fields"][role.digest_field], "digest32") != hashlib.sha256(raw).digest(): _fail("VERIFICATION_FAILED", "envelope body digest/length") _exact(rec["identity"], {"kind", "domain_id", "symbol", "preimage_hex"}, "identity") expected_kind, rule = _expected_identity(rr, rec["key"], profile) ident = rec["identity"] if ident["kind"] != expected_kind: _fail("VERIFICATION_FAILED", "identity kind") if rule is not None: ref = validate_profile_ref(rec["key"], profile, {rule.type_tag}) _hex(ref["hex"], "record ID", rule.width); observed[rule.type_tag].add(ref["hex"]) use = ledger_uses[rule.type_tag].get(ref["hex"]) if use is None: _fail("VERIFICATION_FAILED", "missing allocation use") if expected_kind == "derived": selected = {f: rec["fields"][f] for f in rule.fields} expected_ref, pre = _derive(rule, selected) cross_typed_override = _is_cross_typed_hash_override(store, expected_ref, ref, pre.hex()) if (expected_ref != ref and not cross_typed_override) or ident["domain_id"] is not None or ident["symbol"] is not None or ident["preimage_hex"] != pre.hex() or use["preimage_hex"] != pre.hex(): _fail("VERIFICATION_FAILED", "derived identity") else: if type(ident["domain_id"]) is not str or type(ident["symbol"]) is not str or ident["preimage_hex"] is not None: _fail("VERIFICATION_FAILED", "opaque identity") if ident["domain_id"] != use["domain_id"] or ident["symbol"] != use["symbol"] or _opaque_ref(profile, rule, ident["domain_id"], ident["symbol"]) != ref: _fail("VERIFICATION_FAILED", "opaque identity recipe") elif ident["domain_id"] is not None or ident["symbol"] is not None or ident["preimage_hex"] is not None: _fail("VERIFICATION_FAILED", "nonallocated identity") _u64(rec["instance_order"]) if rr.instance_order_field: order = rec["fields"][rr.instance_order_field] if order != rec["instance_order"] or (rt, order) in orders: _fail("ORDER_VIOLATION", "instance order") orders[(rt, order)] = key for pf in rr.predecessor_fields: for pred in rec["fields"][pf] if type(rec["fields"][pf]) is list else ([] if rec["fields"][pf] is None else [rec["fields"][pf]]): found = _find_ref(store, pred, profile) if not found: _fail("UNKNOWN_REFERENCE", pf) if found[1]["record_type"] != rt: _fail("TYPE_MISMATCH", pf) if found[1]["instance_order"] >= rec["instance_order"]: _fail("ORDER_VIOLATION", pf) for rec in store["records"].values(): for role in profile.versioned_reference_roles: if rec["record_type"] == role.record_type: _validate_version_predecessor(store, rec, role, profile) for role in profile.provenance_roles: if rec["record_type"] == role.record_type: _validate_provenance_grounding(store, rec, role, profile) for tag in allocation_tags: if observed[tag] != set(ledger_uses[tag]): _fail("VERIFICATION_FAILED", "orphan allocation use") for provenance_key, provenance in store["override_provenance"].items(): keys={"source_ref","forced_target_ref","raw_digest","natural_ref","natural_preimage_hash","fault_kind","fault_phase","pre_freeze_epoch","pre_freeze_token"} _exact(provenance, keys, "override provenance") source=validate_profile_ref(provenance["source_ref"],profile); target=validate_profile_ref(provenance["forced_target_ref"],profile); natural=validate_profile_ref(provenance["natural_ref"],profile) if provenance_key!=_override_key(target) or source["type"]==target["type"] or source["hex"]!=target["hex"] or binary_raw(provenance["raw_digest"],"digest32").hex()!=target["hex"]: _fail("VERIFICATION_FAILED","override provenance refs") if provenance["fault_kind"]!="HASH_RESULT_OVERRIDE" or provenance["fault_phase"]!="after-preimage-before-typed-lookup" or provenance["pre_freeze_epoch"]!=0 or provenance["pre_freeze_token"] is not None: _fail("VERIFICATION_FAILED","override provenance phase") source_found=_find_ref(store,source,profile); target_found=_find_ref(store,target,profile) if not source_found or not target_found or not _authoritative_record(source_found[1],source,profile) or not _authoritative_record(target_found[1],target,profile): _fail("VERIFICATION_FAILED","override provenance records") target_rule=profile.id_rules[target["type"]] if target_rule.kind!="derived": _fail("VERIFICATION_FAILED","override target kind") selected={f:target_found[1]["fields"][f] for f in target_rule.fields}; expected,pre=_derive(target_rule,selected) if natural!=expected or natural==target or provenance["natural_preimage_hash"]!=hashlib.sha256(pre).hexdigest() or target_found[1]["identity"]["preimage_hex"]!=pre.hex(): _fail("VERIFICATION_FAILED","override natural identity") use=store["allocations"][target["type"]]["used"].get(target["hex"]) if use is None or use["preimage_hex"]!=pre.hex(): _fail("VERIFICATION_FAILED","override allocation linkage") # Validate authoritative multiplicity independently of record map order. semantic_keys = {} for rec in store["records"].values(): key = rec["key"] if type(key) is dict and key.get("$kind") in ("ref", "slot") and _authoritative_record(rec, key, profile): tk = typed_key(key) if tk in semantic_keys: _fail("VERIFICATION_FAILED", {"ambiguous_authoritative_ref": tk}) semantic_keys[tk] = rec["record_type"] for key, binding in store["bindings"].items(): _validate_binding_entry(key, binding, profile) for ref in _embedded_refs([binding["key"], binding["value"]]): validate_profile_ref(ref, profile) if not _find_ref(store, ref, profile): _fail("UNKNOWN_REFERENCE", "binding reference") for key, lifecycle in store["lifecycle"].items(): _exact(lifecycle, {"target_ref", "state", "tombstone", "expires_at"}, "lifecycle") target = validate_semantic_endpoint(lifecycle["target_ref"], profile) if key != typed_key(target) or not _find_ref(store, target, profile): _fail("VERIFICATION_FAILED", "lifecycle target") _nfc(lifecycle["state"], "lifecycle state") if type(lifecycle["tombstone"]) is not bool: _fail("MALFORMED", "lifecycle tombstone") if lifecycle["expires_at"] is not None: _u64(lifecycle["expires_at"]) if store["lifecycle"] != _lifecycle_projection(store, profile): _fail("VERIFICATION_FAILED", "lifecycle projection") for k, edge in store["edges"].items(): _exact(edge, {"kind", "from", "to", "retention_class"}, "edge") kind = _nfc(edge["kind"], "edge kind") if kind not in profile.edge_rules: _fail("TYPE_MISMATCH", "edge kind") er = profile.edge_rules[kind] if edge["retention_class"] != er.retention_class or _type_of(edge["from"]) not in er.from_types or _type_of(edge["to"]) not in er.to_types: _fail("TYPE_MISMATCH", "edge") for endpoint in (edge["from"], edge["to"]): if type(endpoint) is dict and endpoint.get("$kind") == "ref": validate_profile_ref(endpoint, profile) elif type(endpoint) is dict and endpoint.get("$kind") == "slot": validate_slot(endpoint, profile) if k != kind + "|" + typed_key(edge["from"]) + "|" + typed_key(edge["to"]): _fail("VERIFICATION_FAILED", "edge key") if not _find_ref(store, edge["from"], profile) or not _find_ref(store, edge["to"], profile): _fail("UNKNOWN_REFERENCE", "edge endpoint") _validate_transition_graph(store, profile) _validate_occurrence_graph(store, profile) for name, snap in store["snapshots"].items(): _nfc(name, "snapshot name"); _exact(snap, {"snapshot_ref", "bindings", "binding_kinds", "root_context", "snapshot_digest", "store_digest_before"}, "snapshot") snapshot_ref = validate_profile_ref(snap["snapshot_ref"], profile) if not _find_ref(store, snapshot_ref, profile): _fail("UNKNOWN_REFERENCE", "snapshot ref") if type(snap["binding_kinds"]) is not list or len(snap["binding_kinds"]) != len(set(snap["binding_kinds"])) or any(x not in profile.binding_kinds for x in snap["binding_kinds"]): _fail("MALFORMED", "snapshot binding kinds") if type(snap["bindings"]) is not dict: _fail("MALFORMED", "snapshot bindings") for bkey, binding in snap["bindings"].items(): _validate_binding_entry(bkey, binding, profile) if binding["binding_kind"] not in snap["binding_kinds"]: _fail("VERIFICATION_FAILED", "snapshot binding selection") for ref in _embedded_refs([binding["key"], binding["value"]]): if not _find_ref(store, ref, profile): _fail("UNKNOWN_REFERENCE", "snapshot binding ref") _validate_json(snap["root_context"]); _validate_consumed_refs(snap["root_context"], profile) _hex(snap["snapshot_digest"], "snapshot digest", 32); _hex(snap["store_digest_before"], "store digest", 32) payload = {"bindings": snap["bindings"], "binding_kinds": snap["binding_kinds"], "root_context": snap["root_context"]} if hashlib.sha256(canonical_json(payload)).hexdigest() != snap["snapshot_digest"]: _fail("VERIFICATION_FAILED", "snapshot") _exact(store["roots"], {"next_epoch", "frozen", "pending"}, "roots"); next_epoch = _u64(store["roots"]["next_epoch"]) if type(store["roots"]["frozen"]) is not dict or type(store["roots"]["pending"]) is not dict: _fail("MALFORMED", "root maps") for epoch, pending in store["roots"]["pending"].items(): try: parsed = int(epoch) except (TypeError, ValueError): _fail("MALFORMED", "pending epoch") if str(_u64(parsed)) != epoch or parsed < next_epoch or type(pending) is not list: _fail("MALFORMED", "pending epoch") if len({typed_key(validate_profile_ref(ref, profile)) for ref in pending}) != len(pending): _fail("MALFORMED", "duplicate pending root") if any(not _find_ref(store, ref, profile) for ref in pending): _fail("UNKNOWN_REFERENCE", "pending root") for epoch, frozen in store["roots"]["frozen"].items(): try: parsed = int(epoch) except (TypeError, ValueError): _fail("MALFORMED", "epoch key") if str(_u64(parsed)) != epoch or parsed >= next_epoch: _fail("MALFORMED", "epoch key") _exact(frozen, {"epoch", "root_refs", "pins", "frozen_at_clock", "predecessor_receipt", "latest_token", "freeze_receipt", "mark"}, "frozen roots") if frozen["epoch"] != parsed or frozen["latest_token"] != parsed or type(frozen["root_refs"]) is not list or type(frozen["pins"]) is not list: _fail("MALFORMED", "frozen epoch") _u64(frozen["frozen_at_clock"]) predecessor = None if parsed == 0 else store["roots"]["frozen"].get(str(parsed-1),{}).get("freeze_receipt") if (parsed > 0 and predecessor is None) or frozen["predecessor_receipt"] != predecessor or frozen["freeze_receipt"] != _freeze_receipt(parsed,frozen["frozen_at_clock"],frozen["root_refs"],frozen["pins"],predecessor): _fail("VERIFICATION_FAILED","freeze receipt") root_keys = [typed_key(validate_profile_ref(ref, profile)) for ref in frozen["root_refs"]] if root_keys != sorted(set(root_keys)) or any(not _find_ref(store, ref, profile) for ref in frozen["root_refs"]): _fail("VERIFICATION_FAILED", "frozen roots") pin_keys = [] pin_role = profile.pin_role if frozen["pins"] and pin_role is None: _fail("VERIFICATION_FAILED", "pin role unsupported") for pin in frozen["pins"]: _exact(pin, {"pin_ref", "target_ref", "expires_at"}, "frozen pin") pin_ref = validate_profile_ref(pin["pin_ref"], profile, {pin_role.pin_id_type}) target_ref = validate_profile_ref(pin["target_ref"], profile) expires_at = _u64(pin["expires_at"]) source = _find_ref(store, pin_ref, profile) if not source or source[1]["record_type"] != pin_role.pin_record_type or not _find_ref(store, target_ref, profile): _fail("UNKNOWN_REFERENCE", "frozen pin") fields = source[1]["fields"] if (fields[pin_role.active_field] is not True or fields[pin_role.target_field] != target_ref or fields[pin_role.expires_field] != expires_at or expires_at < frozen["frozen_at_clock"]): _fail("VERIFICATION_FAILED", "frozen pin source") pin_keys.append(typed_key(pin_ref)) if pin_keys != sorted(set(pin_keys)): _fail("VERIFICATION_FAILED", "frozen pins") if frozen["mark"] is not None: mark = frozen["mark"]; _exact(mark, {"epoch", "classes", "reachable", "reasons", "completed", "deferred_to_epoch", "deferred_root"}, "mark") if mark["epoch"] != parsed or mark["completed"] is not True or type(mark["classes"]) is not list or mark["classes"] != sorted(set(mark["classes"])) or any(x not in RETENTION_CLASSES for x in mark["classes"]): _fail("VERIFICATION_FAILED", "mark") deferred = mark["deferred_to_epoch"] if deferred is not None: _u64(deferred) root = mark["deferred_root"] if (deferred is None) != (root is None): _fail("VERIFICATION_FAILED", "deferred root") if root is not None: root = validate_profile_ref(root, profile) pending = store["roots"]["pending"].get(str(deferred), []) consumed = store["roots"]["frozen"].get(str(deferred), {}).get("root_refs", []) if all(typed_key(x) != typed_key(root) for x in pending + consumed): _fail("VERIFICATION_FAILED", "deferred root") expected_mark = _compute_mark(store, frozen, parsed, mark["classes"], deferred, root) if mark != expected_mark: _fail("VERIFICATION_FAILED", "mark recomputation") role = profile.availability_role if store["availability"] and role is None: _fail("VERIFICATION_FAILED", "availability unsupported") available_body_keys: set[str] = set() for key, available in store["availability"].items(): _exact(available, {"envelope_ref", "body_state", "body_record_key", "unavailable_reason"}, "availability") ref = validate_profile_ref(available["envelope_ref"], profile, {role.envelope_id_type}) found = _find_ref(store, ref, profile) if key != typed_key(ref) or not found or found[1]["record_type"] != role.envelope_record_type or available["body_state"] not in ("AVAILABLE", "COLLECTED", "UNAVAILABLE"): _fail("VERIFICATION_FAILED", "availability entry") expected_body_key = role.body_record_type + "|" + key if available["body_state"] == "AVAILABLE": if available["body_record_key"] != expected_body_key or expected_body_key not in store["records"] or available["unavailable_reason"] is not None: _fail("VERIFICATION_FAILED", "available body linkage") body_rec = store["records"][expected_body_key] if body_rec["record_type"] != role.body_record_type or body_rec["key"] != ref or body_rec["fields"].get(role.envelope_ref_field) != ref: _fail("VERIFICATION_FAILED", "body record") if expected_body_key in available_body_keys: _fail("VERIFICATION_FAILED", "duplicate available body") available_body_keys.add(expected_body_key) raw = binary_raw(body_rec["fields"][role.body_field], "bytes") envelope_fields = found[1]["fields"] if envelope_fields[role.length_field] != len(raw) or binary_raw(envelope_fields[role.digest_field], "digest32") != hashlib.sha256(raw).digest(): _fail("VERIFICATION_FAILED", "persisted body digest/length") else: if available["body_record_key"] is not None or expected_body_key in store["records"]: _fail("VERIFICATION_FAILED", "absent body linkage") if available["body_state"] == "COLLECTED" and available["unavailable_reason"] is not None: _fail("VERIFICATION_FAILED", "collected reason") if available["body_state"] == "UNAVAILABLE" and available["unavailable_reason"] is None: _fail("VERIFICATION_FAILED", "unavailable reason") if available["unavailable_reason"] is not None: _nfc(available["unavailable_reason"], "unavailable reason") if role is not None: role_body_keys = {key for key, rec in store["records"].items() if rec["record_type"] == role.body_record_type} if role_body_keys != available_body_keys: _fail("VERIFICATION_FAILED", "availability body reverse closure") _validate_json(_store_image(store)) return store def reconstruct_store(raw: bytes, profile: ProfileSemantics, *, fresh: bool = False) -> dict[str, Any]: image = strict_json_loads(raw, require_lf=True) if type(image) is not dict: _fail("MALFORMED", "store image") if canonical_json(image, lf=True) != raw: _fail("MALFORMED", "non-canonical store bytes") store = copy.deepcopy(image) store["registers"], store["trace"] = {}, [] validate_store(store, profile) # Recompute every derived identity and check ledger preimage evidence. for rec in store["records"].values(): ident = rec["identity"] _exact(ident, {"kind", "domain_id", "symbol", "preimage_hex"}, "identity") if ident["kind"] == "derived": ref = rec["key"] rule = profile.id_rules.get(ref.get("type")) if not rule or rule.kind != "derived": _fail("VERIFICATION_FAILED", "derived rule") selected = {f: rec["fields"][f] for f in rule.fields} expected, pre = _derive(rule, selected) if (expected != ref and not _is_cross_typed_hash_override(store, expected, ref, pre.hex())) or pre.hex() != ident["preimage_hex"]: _fail("VERIFICATION_FAILED", "derived identity") if fresh: # Round-trip through a genuinely new container after complete validation. target = empty_store(profile, store["clock"]) for k in ("records", "allocations", "bindings", "edges", "lifecycle", "snapshots", "override_provenance", "roots", "availability"): target[k] = copy.deepcopy(store[k]) validate_store(target, profile) return target return store class Runtime: def __init__(self, profile: ProfileSemantics, store: dict[str, Any] | None = None): if type(profile) is not ProfileSemantics: _fail("MALFORMED", "profile semantics") # Normalize through the constructor so forged/mutated frozen instances cannot bypass eager role closure. profile = ProfileSemantics(profile.name, profile.fixture_id, dict(profile.id_rules), dict(profile.record_rules), dict(profile.edge_rules), unions=dict(profile.unions), binding_kinds=profile.binding_kinds, state_recipe=_deep_copy_plain(profile.state_recipe), projections=_deep_copy_plain(profile.projections), provenance_types=profile.provenance_types, availability_role=profile.availability_role, pin_role=profile.pin_role, versioned_reference_roles=profile.versioned_reference_roles, provenance_roles=profile.provenance_roles, typed_content_role=profile.typed_content_role) self.profile = profile self.store = empty_store(profile) if store is None else validate_store(copy.deepcopy(store), profile) def _put(self, s: dict[str, Any], args: Any, fault: Any) -> tuple[Any, list[str], list[str]]: _exact(args, {"record_type", "fields", "id", "edges", "out_ref"}, "record.put args") rt = _nfc(args["record_type"], "record type") if rt not in self.profile.record_rules: _fail("UNSUPPORTED_RECORD", rt) rr = self.profile.record_rules[rt] role = self.profile.availability_role if role is not None and rt == role.body_record_type: _fail("TYPE_MISMATCH", "availability owns body records") if type(args["fields"]) is not dict or set(args["fields"]) != set(rr.fields): _fail("MALFORMED", "record fields") fields = {k: _field_validate(args["fields"][k], d, self.profile) for k, d in rr.fields.items()} spec = args["id"] if type(spec) is not dict or "kind" not in spec: _fail("MALFORMED", "id spec") kind = spec["kind"] expected_kind, expected_rule = _expected_identity(rr, fields.get(rr.id_field) if rr.id_field else None, self.profile) if kind != expected_kind: _fail("TYPE_MISMATCH", "record identity kind") preimage = None domain_id = None symbol = None if kind == "opaque": _exact(spec, {"kind", "id_type", "domain_id", "symbol"}, "opaque id") rule = self.profile.id_rules.get(spec["id_type"]) if not rule or rule.kind != "opaque" or rule is not expected_rule: _fail("TYPE_MISMATCH", "opaque ID") domain_id = _nfc(spec["domain_id"], "domain") symbol = _nfc(spec["symbol"], "symbol") key = _opaque_ref(self.profile, rule, domain_id, symbol) elif kind == "derived": _exact(spec, {"kind", "id_type"}, "derived id") rule = self.profile.id_rules.get(spec["id_type"]) if not rule or rule.kind != "derived" or rule is not expected_rule: _fail("TYPE_MISMATCH", "derived ID") selected = {f: fields[f] for f in rule.fields} if set(rule.fields) <= set(fields) else _fail("MALFORMED", "preimage fields") key, preimage = _derive(rule, selected) natural_ref = copy.deepcopy(key); override_provenance = None if fault is not None: _exact(fault, {"kind", "phase", "raw_digest_hex"}, "hash fault") if fault["kind"] != "HASH_RESULT_OVERRIDE" or fault["phase"] != "after-preimage-before-typed-lookup": _fail("MALFORMED", "fault") if s["roots"]["next_epoch"] != 0 or s["roots"]["frozen"]: _fail("CONTEXT_MISMATCH", "hash override is strictly pre-freeze") key["hex"] = _hex(fault["raw_digest_hex"], "override", rule.width).hex() same_typed = [rec for rec in s["records"].values() if rec["key"] == key and _authoritative_record(rec,key,self.profile)] if same_typed and any(rec["identity"]["preimage_hex"] != preimage.hex() for rec in same_typed): _fail("FATAL_INTEGRITY_FAILURE", {"typed_key":typed_key(key)}, fatal=True) sources = [rec for rec in s["records"].values() if type(rec["key"]) is dict and rec["key"].get("$kind") == "ref" and rec["key"]["hex"] == key["hex"] and rec["key"]["type"] != key["type"] and _authoritative_record(rec, rec["key"], self.profile)] if len(sources) != 1: _fail("VERIFICATION_FAILED", "hash override requires one cross-type occupant") source_ref = copy.deepcopy(sources[0]["key"]) override_provenance = {"source_ref": source_ref, "forced_target_ref": copy.deepcopy(key), "raw_digest": bytes_value(bytes.fromhex(key["hex"]), "digest32"), "natural_ref": natural_ref, "natural_preimage_hash": hashlib.sha256(preimage).hexdigest(), "fault_kind": fault["kind"], "fault_phase": fault["phase"], "pre_freeze_epoch": 0, "pre_freeze_token": None} elif kind == "structural": _exact(spec, {"kind", "field"}, "structural id") field_name = _nfc(spec["field"], "structural field") if field_name not in fields: _fail("MALFORMED", "structural field") key = validate_slot(fields[field_name], self.profile) elif kind == "keyed": _exact(spec, {"kind", "key"}, "keyed id") key = _validate_json(spec["key"]); _validate_consumed_refs(key, self.profile) else: _fail("MALFORMED", "ID kind") if rr.id_field is not None and rr.id_field in fields: fields[rr.id_field] = copy.deepcopy(key) rec_key = _record_key(rt, key) existing = s["records"].get(rec_key) if existing is not None: if kind == "derived" and existing["identity"]["preimage_hex"] != preimage.hex(): _fail("FATAL_INTEGRITY_FAILURE", {"typed_key": typed_key(key)}, fatal=True) if existing["fields"] == fields: _fail("ALREADY_EXISTS", {"ref": key}) _fail("DUPLICATE_ID", {"ref": key}) # A semantic typed key has at most one authoritative record. Role-managed # keyed records may coexist with that record and are selected only by type. found = _find_ref(s, key, self.profile) if kind in ("opaque", "derived", "structural") else None if found and _authoritative_record(found[1], key, self.profile): if kind == "derived" and found[1]["identity"]["preimage_hex"] != preimage.hex(): _fail("FATAL_INTEGRITY_FAILURE", {"typed_key": typed_key(key)}, fatal=True) _fail("DUPLICATE_ID", {"ref": key}) order = fields.get(rr.instance_order_field, 0) if rr.instance_order_field else 0 rec = {"record_type": rt, "key": key, "fields": fields, "identity": {"kind": kind, "domain_id": domain_id, "symbol": symbol, "preimage_hex": None if preimage is None else preimage.hex()}, "instance_order": order} # Validate predecessor existence/order and all role-grounding before insertion. for pf in rr.predecessor_fields: vals = fields[pf] if type(fields[pf]) is list else ([] if fields[pf] is None else [fields[pf]]) for pred in vals: got = _find_ref(s, pred, self.profile) if not got: _fail("UNKNOWN_REFERENCE", pf) if got[1]["record_type"] != rt: _fail("TYPE_MISMATCH", pf) if got[1]["instance_order"] >= order: _fail("ORDER_VIOLATION", pf) for role in self.profile.versioned_reference_roles: if rt == role.record_type: _validate_version_predecessor(s, rec, role, self.profile) for role in self.profile.provenance_roles: if rt == role.record_type: _validate_provenance_grounding(s, rec, role, self.profile) if type(args["edges"]) is not list: _fail("MALFORMED", "edges") pending_edges = [] for e in args["edges"]: _exact(e, {"kind", "from", "to"}, "edge spec") if e["kind"] not in self.profile.edge_rules: _fail("TYPE_MISMATCH", "edge kind") def endpoint(x: Any) -> Any: _exact(x, {"source", "value"}, "edge endpoint spec") if x["source"] == "literal": return _validate_json(x["value"]) if x["source"] == "field" and x["value"] in fields: return fields[x["value"]] if x["source"] == "self": return key _fail("MALFORMED", "edge endpoint source") fr, to = endpoint(e["from"]), endpoint(e["to"]) _validate_consumed_refs([fr, to], self.profile) er = self.profile.edge_rules[e["kind"]] if _type_of(fr) not in er.from_types or _type_of(to) not in er.to_types: _fail("TYPE_MISMATCH", "edge endpoint") if fr != key and not _find_ref(s, fr, self.profile) or to != key and not _find_ref(s, to, self.profile): _fail("UNKNOWN_REFERENCE", "edge") pending_edges.append((e["kind"], fr, to, er)) if kind in ("opaque", "derived"): ledger = s["allocations"][key["type"]] raw_key = key["hex"] if raw_key in ledger["used"]: _fail("DUPLICATE_ID", key) if ledger["next_order"] == U64_MAX: _fail("ORDER_VIOLATION", "allocation order overflow") ledger["used"][raw_key] = {"first_order": ledger["next_order"], "preimage_hex": None if preimage is None else preimage.hex(), "domain_id": domain_id, "symbol": symbol} ledger["next_order"] += 1 s["records"][rec_key] = rec if kind == "derived" and override_provenance is not None: provenance_key = _override_key(key) if provenance_key in s["override_provenance"]: _fail("DUPLICATE_ID", "override provenance") s["override_provenance"][provenance_key] = override_provenance for ek, fr, to, er in pending_edges: edge_key = ek + "|" + typed_key(fr) + "|" + typed_key(to) s["edges"][edge_key] = {"kind": ek, "from": fr, "to": to, "retention_class": er.retention_class} if rr.projection: s["lifecycle"] = _lifecycle_projection(s, self.profile) if args["out_ref"] is not None: _nfc(args["out_ref"], "register"); s["registers"][args["out_ref"]] = copy.deepcopy(key) writes = ["records/" + rec_key] writes += ["edges/" + ek + "|" + typed_key(fr) + "|" + typed_key(to) for ek, fr, to, _ in pending_edges] if rr.projection: writes.append("lifecycle/" + typed_key(fields[self.profile.projections[rr.projection]["target_field"]])) if kind in ("opaque", "derived"): writes.append("allocations/" + key["type"] + "/used/" + key["hex"]) if kind == "derived" and override_provenance is not None: writes.append("override_provenance/" + _override_key(key)) if args["out_ref"] is not None: writes.append("registers/" + args["out_ref"]) return {"status": "CREATED", "ref": key}, [], writes def _index(self, s: dict[str, Any], args: Any) -> tuple[Any, list[str], list[str]]: _exact(args, {"binding_kind", "key", "value", "expected_version", "out_version"}, "index args") kind = _nfc(args["binding_kind"]) if kind not in self.profile.binding_kinds: _fail("UNSUPPORTED_BINDING", kind) key, value = _validate_json(args["key"]), _validate_json(args["value"]) bkey = kind + "|" + typed_key(key); old = s["bindings"].get(bkey) expected = args["expected_version"] if expected is not None: _u64(expected) current = 0 if old is None else old["version"] if expected is not None and current != expected: _fail("CONTEXT_MISMATCH", {"expected": expected, "actual": current}) # All embedded refs must exist, except empty/container/scalar fixture values. def refs(v: Any): if type(v) is dict and v.get("$kind") == "ref": yield v elif type(v) is dict: for x in v.values(): yield from refs(x) elif type(v) is list: for x in v: yield from refs(x) for ref in refs([key, value]): validate_profile_ref(ref, self.profile) if not _find_ref(s, ref, self.profile): _fail("UNKNOWN_REFERENCE", ref) if current == U64_MAX: _fail("ORDER_VIOLATION", "binding version overflow") version = current + 1 s["bindings"][bkey] = {"binding_kind": kind, "key": key, "value": value, "version": version} writes = ["bindings/" + bkey] if args["out_version"] is not None: out = _nfc(args["out_version"]); s["registers"][out] = version; writes.append("registers/" + out) return {"status": "SET", "version": version, "value": value}, ["bindings/" + bkey], writes def _availability(self, s: dict[str, Any], args: Any) -> tuple[Any, list[str], list[str]]: _exact(args, {"envelope_ref", "state", "body", "reason"}, "availability args") role = self.profile.availability_role if role is None: _fail("TYPE_MISMATCH", "availability unsupported") ref = validate_profile_ref(args["envelope_ref"], self.profile, {role.envelope_id_type}) got = _find_ref(s, ref, self.profile) if not got: _fail("UNKNOWN", ref) if got[1]["record_type"] != role.envelope_record_type: _fail("TYPE_MISMATCH", "availability envelope role") state = args["state"] if state not in ("AVAILABLE", "COLLECTED", "UNAVAILABLE"): _fail("MALFORMED", "availability state") akey = typed_key(ref); verified = False; body_key = None old = s["availability"].get(akey); writes = ["availability/" + akey] reason = args["reason"] if state == "AVAILABLE": if reason is not None: _fail("MALFORMED", "available reason") raw = binary_raw(args["body"], "bytes") f = got[1]["fields"] digest = binary_raw(f[role.digest_field], "digest32") if digest != hashlib.sha256(raw).digest() or f[role.length_field] != len(raw): _fail("VERIFICATION_FAILED", "body") body_key = role.body_record_type + "|" + akey body_rec = {"record_type": role.body_record_type, "key": ref, "fields": {role.envelope_ref_field: ref, role.body_field: args["body"]}, "identity": {"kind": "keyed", "domain_id": None, "symbol": None, "preimage_hex": None}, "instance_order": 0} existing = s["records"].get(body_key) if existing is not None and existing != body_rec: _fail("DUPLICATE_ID", {"ref": ref}) if existing is None: s["records"][body_key] = body_rec; writes.append("records/" + body_key) verified = True else: if args["body"] is not None: _fail("MALFORMED", "body must be null") if state == "COLLECTED" and reason is not None: _fail("MALFORMED", "collected reason") if state == "UNAVAILABLE": _nfc(reason, "unavailable reason") old_key = old and old.get("body_record_key") if old_key and old_key in s["records"]: del s["records"][old_key]; writes.append("records/" + old_key + "#remove") s["availability"][akey] = {"envelope_ref": ref, "body_state": state, "body_record_key": body_key, "unavailable_reason": reason} return {"status": state, "verified": verified}, ["records/" + got[0], "availability/" + akey], writes def _collect_if_unreachable(self, s: dict[str, Any], args: Any) -> tuple[Any, list[str], list[str]]: _exact(args, {"envelope_ref", "epoch", "expected_reachable"}, "collect args") role=self.profile.availability_role if role is None: _fail("TYPE_MISMATCH", "availability unsupported") ref=validate_profile_ref(args["envelope_ref"],self.profile,{role.envelope_id_type}) epoch=_u64(args["epoch"]) if type(args["expected_reachable"]) is not bool: _fail("MALFORMED", "expected mark membership") found=_find_ref(s,ref,self.profile) if not found or found[1]["record_type"] != role.envelope_record_type: _fail("UNKNOWN_REFERENCE", "envelope") frozen=s["roots"]["frozen"].get(str(epoch)) latest = s["roots"]["next_epoch"] - 1 latest_frozen = s["roots"]["frozen"].get(str(latest)) if latest >= 0 else None if frozen is None or frozen.get("mark") is None or frozen["mark"].get("completed") is not True: _fail("CONTEXT_MISMATCH", "incomplete mark") if epoch != latest or latest_frozen is None or latest_frozen.get("mark") is None or latest_frozen["mark"].get("completed") is not True: _fail("CONTEXT_MISMATCH", "collection requires latest completed epoch") mark=frozen["mark"]; reachable=typed_key(ref) in set(mark["reachable"]) if reachable != args["expected_reachable"]: _fail("CONTEXT_MISMATCH", "mark membership") if reachable: _fail("CONTEXT_MISMATCH", "body reachable") akey=typed_key(ref); available=s["availability"].get(akey) if not available or available["body_state"] != "AVAILABLE" or available["body_record_key"] is None: _fail("CONTEXT_MISMATCH", "body unavailable") body_key=available["body_record_key"] if body_key not in s["records"]: _fail("VERIFICATION_FAILED", "body linkage") del s["records"][body_key] s["availability"][akey]={"envelope_ref":ref,"body_state":"COLLECTED","body_record_key":None,"unavailable_reason":None} return {"status":"COLLECTED","epoch":epoch,"envelope_ref":ref,"reachable":False}, ["roots/frozen/"+str(epoch)+"/mark","records/"+found[0],"records/"+body_key], ["records/"+body_key+"#remove","availability/"+akey] def _snapshot_capture(self, s: dict[str, Any], args: Any, fault: Any) -> tuple[Any, list[str], list[str]]: _exact(args, {"name", "snapshot_record", "binding_kinds", "root_context"}, "snapshot capture") name = _nfc(args["name"]) if name in s["snapshots"]: _fail("DUPLICATE_ID", name) if type(args["binding_kinds"]) is not list or len(args["binding_kinds"]) != len(set(args["binding_kinds"])) or any(type(x) is not str or x not in self.profile.binding_kinds for x in args["binding_kinds"]): _fail("MALFORMED", "binding kinds") before = store_digest(s) output, _, writes = self._put(s, args["snapshot_record"], fault) selected = {k: copy.deepcopy(v) for k, v in s["bindings"].items() if v["binding_kind"] in args["binding_kinds"]} root_context = _validate_json(args["root_context"]) _validate_consumed_refs(root_context, self.profile) payload = {"bindings": selected, "binding_kinds": list(args["binding_kinds"]), "root_context": root_context} s["snapshots"][name] = {"snapshot_ref": output["ref"], **payload, "snapshot_digest": hashlib.sha256(canonical_json(payload)).hexdigest(), "store_digest_before": before} reads = ["bindings/" + k for k in sorted(selected)] + ["store"] return {"status": "CAPTURED", "snapshot_ref": output["ref"], "digest": before}, reads, writes + ["snapshots/" + name] def _snapshot_restore(self, s: dict[str, Any], args: Any) -> tuple[Any, list[str], list[str]]: _exact(args, {"name"}, "snapshot restore") name = _nfc(args["name"]); snap = s["snapshots"].get(name) if not snap: _fail("UNKNOWN", name) payload = {"bindings": snap["bindings"], "binding_kinds": snap["binding_kinds"], "root_context": snap["root_context"]} if hashlib.sha256(canonical_json(payload)).hexdigest() != snap["snapshot_digest"]: _fail("VERIFICATION_FAILED", "snapshot") removed = [] for k in list(s["bindings"]): if s["bindings"][k]["binding_kind"] in snap["binding_kinds"]: del s["bindings"][k]; removed.append("bindings/" + k + "#remove") s["bindings"].update(copy.deepcopy(snap["bindings"])) writes = removed + ["bindings/" + k for k in sorted(snap["bindings"])] return {"status": "RESTORED", "snapshot_ref": snap["snapshot_ref"]}, ["snapshots/" + name], writes def _freeze(self, s: dict[str, Any], args: Any) -> tuple[Any, list[str], list[str]]: _exact(args, {"epoch", "root_refs", "pin_refs", "clock"}, "roots freeze") epoch, clock = _u64(args["epoch"]), _u64(args["clock"]) if epoch != s["roots"]["next_epoch"] or clock < s["clock"]: _fail("CONTEXT_MISMATCH", "epoch/clock") if type(args["root_refs"]) is not list or type(args["pin_refs"]) is not list: _fail("MALFORMED", "roots/pins") roots = [] for r in args["root_refs"]: r = validate_profile_ref(r, self.profile) if not _find_ref(s, r, self.profile): _fail("UNKNOWN_REFERENCE", r) roots.append(r) pins = [] pin_role = self.profile.pin_role if args["pin_refs"] and pin_role is None: _fail("TYPE_MISMATCH", "pin role unsupported") for p in args["pin_refs"]: p = validate_profile_ref(p, self.profile, {pin_role.pin_id_type}); found = _find_ref(s, p, self.profile) if not found: _fail("UNKNOWN_REFERENCE", p) if found[1]["record_type"] != pin_role.pin_record_type: _fail("TYPE_MISMATCH", "pin record role") f = found[1]["fields"] active = f[pin_role.active_field] expires = f[pin_role.expires_field]; target = f[pin_role.target_field] if type(active) is not bool: _fail("MALFORMED", "pin active") expires = _u64(expires, "pin expiry"); target = validate_profile_ref(target, self.profile) if not _find_ref(s, target, self.profile): _fail("UNKNOWN_REFERENCE", "pin target") if active and expires >= clock: pins.append({"pin_ref": p, "target_ref": target, "expires_at": expires}) pending = s["roots"]["pending"].pop(str(epoch), []) roots.extend(pending) unique = {typed_key(x): x for x in roots} if s["roots"]["next_epoch"] == U64_MAX: _fail("ORDER_VIOLATION", "root epoch overflow") root_refs = [unique[k] for k in sorted(unique)]; frozen_pins = sorted(pins, key=lambda p: typed_key(p["pin_ref"])) predecessor = None if epoch == 0 else s["roots"]["frozen"][str(epoch-1)]["freeze_receipt"] receipt = _freeze_receipt(epoch, clock, root_refs, frozen_pins, predecessor) s["roots"]["frozen"][str(epoch)] = {"epoch": epoch, "root_refs": root_refs, "pins": frozen_pins, "frozen_at_clock": clock, "predecessor_receipt": predecessor, "latest_token": epoch, "freeze_receipt": receipt, "mark": None} s["roots"]["next_epoch"] += 1 reads = ["records/" + _find_ref(s, x, self.profile)[0] for x in roots if _find_ref(s, x, self.profile)] + ["roots/pending/" + str(epoch)] writes = ["roots/frozen/" + str(epoch), "roots/next_epoch"] if pending: writes.append("roots/pending/" + str(epoch) + "#remove") return {"status": "FROZEN", "epoch": epoch, "root_count": len(unique), "pin_count": len(pins)}, sorted(set(reads)), writes def _mark(self, s: dict[str, Any], args: Any, fault: Any) -> tuple[Any, list[str], list[str]]: _exact(args, {"epoch", "classes", "out"}, "mark args") epoch = _u64(args["epoch"]); frozen = s["roots"]["frozen"].get(str(epoch)) if not frozen: _fail("UNKNOWN", epoch) if frozen["mark"] is not None: _fail("CONTEXT_MISMATCH", "already marked") if type(args["classes"]) is not list or len(args["classes"]) != len(set(args["classes"])) or any(type(x) is not str or x not in RETENTION_CLASSES for x in args["classes"]): _fail("MALFORMED", "classes") classes = sorted(args["classes"]) deferred = None; deferred_root = None if fault is not None: _exact(fault, {"kind", "phase", "epoch", "root_ref"}, "root fault") if fault["kind"] != "CONCURRENT_ROOT_APPEND" or fault["phase"] != "after-root-copy-before-first-edge": _fail("MALFORMED", "fault") target_epoch = _u64(fault["epoch"]); root = validate_profile_ref(fault["root_ref"], self.profile) if target_epoch != epoch + 1 or target_epoch != s["roots"]["next_epoch"] or not _find_ref(s, root, self.profile): _fail("CONTEXT_MISMATCH", "concurrent root must target exact next epoch") pending=s["roots"]["pending"].setdefault(str(target_epoch), []) if any(typed_key(x)==typed_key(root) for x in pending): _fail("CONTEXT_MISMATCH","duplicate concurrent root") pending.append(root); deferred = target_epoch; deferred_root = root result = _compute_mark(s, frozen, epoch, classes, deferred, deferred_root) frozen["mark"] = result reads = ["roots/frozen/" + str(epoch)] + ["edges/" + k for k in sorted(s["edges"])] writes = ["roots/frozen/" + str(epoch) + "/mark"] if deferred is not None: writes.append("roots/pending/" + str(deferred)) if args["out"] is not None: out = _nfc(args["out"]); s["registers"][out] = copy.deepcopy(result); writes.append("registers/" + out) return result, reads, writes def _reload(self, s: dict[str, Any], args: Any) -> tuple[Any, list[str], list[str], dict[str, Any]]: _exact(args, {"source_register", "destination"}, "reload args") name = _nfc(args["source_register"]) if args["destination"] != "current" or name not in s["registers"]: _fail("MALFORMED", "reload source") raw = binary_raw(s["registers"][name], "bytes") loaded = reconstruct_store(raw, self.profile) return {"status": "RELOADED", "digest": store_digest(loaded)}, ["registers/" + name], ["store"], loaded def apply(self, action: Any) -> dict[str, Any]: """Apply one D001-ACTION-3 action; every outcome is a deterministic trace event.""" seq = len(self.store["trace"]) step = action.get("step_id") if type(action) is dict and type(action.get("step_id")) is str else "invalid-step-" + str(seq) op = action.get("op") if type(action) is dict and type(action.get("op")) is str else "invalid-op" before = store_digest(self.store); candidate = copy.deepcopy(self.store) reads: list[str] = []; writes: list[str] = []; output = None; error = None try: base_keys={"action_version", "step_id", "op", "args", "fault"} if type(action) is not dict or set(action) not in (base_keys,base_keys|{"action_meta"}): _fail("MALFORMED", "action") if "action_meta" in action: meta=action["action_meta"] if type(meta) is not dict or set(meta) not in ({"essential_for"},{"setup_only"}): _fail("MALFORMED", "action meta") if "essential_for" in meta: if type(meta["essential_for"]) is not list or not meta["essential_for"] or any(type(x) is not str or not _nfc(x) for x in meta["essential_for"]): _fail("MALFORMED", "essential action meta") else: _nfc(meta["setup_only"], "setup rationale") if action["action_version"] != "D001-ACTION-3": _fail("MALFORMED", "action version") step, op = _nfc(action["step_id"], "step"), _nfc(action["op"], "opcode") if op not in OPS: _fail("MALFORMED", "opcode") if any(e["step_id"] == step for e in self.store["trace"]): _fail("MALFORMED", "duplicate step") fault = action["fault"] if fault is not None: if type(fault) is not dict or type(fault.get("kind")) is not str or type(fault.get("phase")) is not str: _fail("MALFORMED", "fault") if fault["kind"] not in FAULTS or (fault["kind"], fault["phase"]) not in FAULT_COMPATIBILITY.get(op, set()): _fail("MALFORMED", "fault/op/phase") if op == "record.put": output, reads, writes = self._put(candidate, action["args"], fault) elif op == "index.set": output, reads, writes = self._index(candidate, action["args"]) elif op == "availability.set": output, reads, writes = self._availability(candidate, action["args"]) elif op == "availability.collect-if-unreachable": output, reads, writes = self._collect_if_unreachable(candidate, action["args"]) elif op == "snapshot.capture": output, reads, writes = self._snapshot_capture(candidate, action["args"], fault) elif op == "snapshot.restore": output, reads, writes = self._snapshot_restore(candidate, action["args"]) elif op == "roots.freeze": output, reads, writes = self._freeze(candidate, action["args"]) elif op == "reachability.mark": output, reads, writes = self._mark(candidate, action["args"], fault) else: output, reads, writes, candidate = self._reload(candidate, action["args"]) validate_store(candidate, self.profile) self.store = candidate; status = "COMMITTED" except D001Error as exc: status = "FATAL" if exc.fatal else "NO_COMMIT"; error = exc.as_json(); writes = [] except Exception as exc: status = "NO_COMMIT"; error = D001Error("INTERNAL_ERROR", type(exc).__name__).as_json(); writes = [] after = store_digest(self.store) if status != "COMMITTED" and before != after: raise AssertionError("no-commit digest changed") event = {"seq": seq, "step_id": step, "op": op, "status": status, "reads": reads, "writes": writes, "output": output, "error": error, "before_digest": before, "after_digest": after} self.store["trace"].append(event) return copy.deepcopy(event) def query(self, expr: Any) -> dict[str, Any]: try: value = self._eval(expr) if type(value) is dict and value.get("status") in RESOLVER_ORDER: return value return {"status": "OK", "value": value} except D001Error as exc: if exc.code in RESOLVER_ORDER[:-1]: return {"status": exc.code, "value": None, "denotation_ref": None, "matched_version_ref": None} return {"status": exc.code, "value": None, "error": exc.detail} except Exception as exc: return {"status": "INTERNAL_ERROR", "value": None, "error": type(exc).__name__} def _eval(self, expr: Any) -> Any: if type(expr) is not dict or expr.get("op") not in QUERY_OPS: _fail("MALFORMED", "query") op = expr["op"] if op == "literal": _exact(expr, {"op", "value"}, op); return _validate_json(expr["value"]) if op == "reg": _exact(expr, {"op", "name"}, op); name = _nfc(expr["name"]) if name not in self.store["registers"]: _fail("UNKNOWN", name) return copy.deepcopy(self.store["registers"][name]) if op == "record": _exact(expr, {"op", "type", "key"}, op); rt = _nfc(expr["type"], "record type"); key = self._eval(expr["key"]) if rt not in self.profile.record_rules: _fail("UNSUPPORTED_RECORD", rt) rr = self.profile.record_rules[rt] if rr.id_type is not None: validate_profile_ref(key, self.profile, {rr.id_type}) elif type(key) is dict and key.get("$kind") == "slot": validate_slot(key, self.profile) else: _validate_consumed_refs(key, self.profile) rec = self.store["records"].get(_record_key(rt, key)) if not rec: _fail("UNKNOWN", key) return copy.deepcopy(rec) if op == "field": _exact(expr, {"op", "input", "path"}, op); value = self._eval(expr["input"]) return _pointer(value, expr["path"]) if op == "binding": _exact(expr, {"op", "kind", "key"}, op); kind = _nfc(expr["kind"], "binding kind"); key = self._eval(expr["key"]) if kind not in self.profile.binding_kinds: _fail("UNSUPPORTED_BINDING", kind) _validate_consumed_refs(key, self.profile) b = self.store["bindings"].get(kind + "|" + typed_key(key)) if not b: _fail("UNKNOWN", key) _validate_consumed_refs([b["key"], b["value"]], self.profile) return copy.deepcopy(b) if op == "trace": _exact(expr, {"op", "select", "project", "cardinality"}, op) sel = expr["select"]; _exact(sel, {"step_id", "op", "status"}, "trace select") for token in sel.values(): if token is not None: _nfc(token, "trace selector") found = [e for e in self.store["trace"] if all(sel[k] is None or e[k] == sel[k] for k in sel)] project = expr["project"] if project not in ("event", "output", "error"): _fail("MALFORMED", "trace project") vals = [copy.deepcopy(e if project == "event" else e[project]) for e in found] if expr["cardinality"] == "one": if len(vals) != 1: _fail("CONTEXT_MISMATCH", "trace cardinality") return vals[0] if expr["cardinality"] != "many": _fail("MALFORMED", "trace cardinality") return vals if op == "resolve": _exact(expr, {"op", "request"}, op); return self._resolve(expr["request"]) if op == "edge": _exact(expr, {"op", "kind", "from", "to"}, op); kind = _nfc(expr["kind"], "edge kind"); fr, to = self._eval(expr["from"]), self._eval(expr["to"]) if kind not in self.profile.edge_rules: _fail("TYPE_MISMATCH", "edge kind") _validate_consumed_refs([fr, to], self.profile) er = self.profile.edge_rules[kind] if _type_of(fr) not in er.from_types or _type_of(to) not in er.to_types: _fail("TYPE_MISMATCH", "edge") key = kind + "|" + typed_key(fr) + "|" + typed_key(to) return {"exists": key in self.store["edges"], "kind": kind, "from": fr, "to": to} if op == "equal": _exact(expr, {"op", "left", "right"}, op); a, b = self._eval(expr["left"]), self._eval(expr["right"]) return {"equal": type(a) is type(b) and a == b, "left": a, "right": b} if op == "object-has-key": _exact(expr, {"op", "input", "object_key", "key"}, op); value=self._eval(expr["input"]); object_key=_nfc(expr["object_key"], "object key"); key=_nfc(expr["key"], "object key") if type(value) is not dict or type(value.get(object_key)) is not dict: _fail("TYPE_MISMATCH", "object-has-key input") return {"object_key":object_key,"key":key,"present":key in value[object_key]} if op == "contains": _exact(expr, {"op", "input", "value"}, op); values=self._eval(expr["input"]); value=self._eval(expr["value"]) if type(values) is not list: _fail("TYPE_MISMATCH", "contains input") return {"contains":any(type(x) is type(value) and x==value for x in values),"value":value} if op == "typed-key": _exact(expr, {"op", "value"}, op); value=self._eval(expr["value"]); _validate_consumed_refs(value,self.profile) return typed_key(value) if op == "typed-keys": _exact(expr, {"op", "values"}, op); values=self._eval(expr["values"]) if type(values) is not list: _fail("TYPE_MISMATCH", "typed-keys") _validate_consumed_refs(values,self.profile) return sorted(typed_key(x) for x in values) if op == "list-count": _exact(expr, {"op", "input"}, op); values=self._eval(expr["input"]) if type(values) is not list: _fail("TYPE_MISMATCH","list-count") return {"count":len(values)} if op == "list-select": _exact(expr, {"op", "input", "match"}, op); values=self._eval(expr["input"]); match=self._eval(expr["match"]) if type(values) is not list or type(match) is not dict or not match: _fail("TYPE_MISMATCH", "list-select") found=[x for x in values if type(x) is dict and all(x.get(k)==v for k,v in match.items())] if len(found)!=1: _fail("CONTEXT_MISMATCH", "list-select cardinality") return copy.deepcopy(found[0]) if op == "map-get": _exact(expr, {"op", "input", "key"}, op); value=self._eval(expr["input"]); key=self._eval(expr["key"]) if type(value) is not dict or type(key) is not str: _fail("TYPE_MISMATCH", "map-get input") if key not in value: _fail("UNKNOWN", key) return copy.deepcopy(value[key]) if op == "compare-components": _exact(expr, {"op", "left", "right", "components"}, op); a, b = self._eval(expr["left"]), self._eval(expr["right"]) if type(expr["components"]) is not list: _fail("MALFORMED", "components") out = {} for c in expr["components"]: _exact(c, {"name", "left_path", "right_path"}, "component") name = _nfc(c["name"], "component name") if name in out: _fail("MALFORMED", "duplicate component") x, y = _pointer(a, c["left_path"]), _pointer(b, c["right_path"]) out[name] = {"equal": type(x) is type(y) and x == y, "left": x, "right": y} return {"all_equal": all(x["equal"] for x in out.values()), "components": out} if op == "verify-provenance": _exact(expr, {"op", "claim"}, op); claim = self._eval(expr["claim"]) structured={"claimed_occurrence_ref","actor_ref","commitment_ref","authorization_evidence_ref","causal_inputs"} is_structured=type(claim) is dict and set(claim)==structured ref=validate_profile_ref(claim["claimed_occurrence_ref"] if is_structured else claim,self.profile,set(self.profile.provenance_types)) role=next((x for x in self.profile.provenance_roles if x.occurrence_id_type==ref["type"]),None) if role is None: return {"verified":False,"match_status":"UNSUPPORTED","occurrence_ref":ref,"actor_ref":None,"commitment_ref":None,"evidence_ref":None,"causal_inputs":None} rr=self.profile.record_rules[role.record_type] mapped=(("actor_ref",role.actor_field),("commitment_ref",role.commitment_field),("authorization_evidence_ref",role.authorization_evidence_field)) + (() if role.causal_inputs_field is None else (("causal_inputs",role.causal_inputs_field),)) if is_structured: for claim_field,record_field in mapped: _field_validate(claim[claim_field],rr.fields[record_field],self.profile) if role.causal_inputs_field is None: if type(claim["causal_inputs"]) is not list: _fail("MALFORMED","causal inputs") if claim["causal_inputs"]: return {"verified":False,"match_status":"FIELD_MISMATCH","occurrence_ref":ref,"actor_ref":claim["actor_ref"],"commitment_ref":claim["commitment_ref"],"evidence_ref":claim["authorization_evidence_ref"],"causal_inputs":None} else: seen=set() for causal_ref in claim["causal_inputs"]: clean=validate_profile_ref(causal_ref,self.profile,set(role.causal_types)); key=typed_key(clean) if key in seen: _fail("MALFORMED","duplicate causal input") seen.add(key) try: _require_role_ref(self.store,clean,self.profile,role.causal_types,role.causal_record_types,"causal_inputs") except D001Error: return {"verified":False,"match_status":"FIELD_MISMATCH","occurrence_ref":ref,"actor_ref":claim["actor_ref"],"commitment_ref":claim["commitment_ref"],"evidence_ref":claim["authorization_evidence_ref"],"causal_inputs":claim["causal_inputs"]} found=_find_ref(self.store,ref,self.profile) if not found: return {"verified":False,"match_status":"UNKNOWN_MATCH","occurrence_ref":ref,"actor_ref":claim["actor_ref"] if is_structured else None,"commitment_ref":claim["commitment_ref"] if is_structured else None,"evidence_ref":claim["authorization_evidence_ref"] if is_structured else None,"causal_inputs":claim["causal_inputs"] if is_structured else None} if found[1]["record_type"]!=role.record_type or not _authoritative_record(found[1],ref,self.profile): _fail("TYPE_MISMATCH","provenance record role") fields=found[1]["fields"] fields_match=not is_structured or (all(fields[record_field]==claim[claim_field] for claim_field,record_field in mapped) and (role.causal_inputs_field is not None or claim["causal_inputs"]==[])) grounded=True try: _validate_provenance_grounding(self.store,found[1],role,self.profile) except D001Error: grounded=False verified=fields_match and grounded evidence=fields[role.authorization_evidence_field] return {"verified":verified,"match_status":"MATCH" if verified else "FIELD_MISMATCH","occurrence_ref":ref,"actor_ref":fields[role.actor_field],"commitment_ref":fields[role.commitment_field],"evidence_ref":evidence,"causal_inputs":fields[role.causal_inputs_field] if role.causal_inputs_field is not None else None} if op == "verify-body": _exact(expr, {"op", "envelope_ref", "body"}, op); ref, body = self._eval(expr["envelope_ref"]), self._eval(expr["body"]) role = self.profile.availability_role if role is None: _fail("TYPE_MISMATCH", "availability unsupported") ref = validate_profile_ref(ref, self.profile, {role.envelope_id_type}); raw = binary_raw(body, "bytes") found = _find_ref(self.store, ref, self.profile) if not found: return {"verified": False} if found[1]["record_type"] != role.envelope_record_type: _fail("TYPE_MISMATCH", "availability envelope role") fields = found[1]["fields"] if set(fields) != set(self.profile.record_rules[role.envelope_record_type].fields): _fail("MALFORMED", "availability envelope fields") ok = fields[role.length_field] == len(raw) and binary_raw(fields[role.digest_field], "digest32") == hashlib.sha256(raw).digest() return {"verified": ok} if op == "canonical-value": _exact(expr, {"op", "value", "schema"}, op) value, encoded = canonical_value(expr["value"], expr["schema"], self.profile) canonical_raw = b'{"$kind":"absent"}' if value == {"$kind":"absent"} else canonical_json(value) return {"canonical_json_hex": canonical_raw.hex(), "cve1_hex": encoded.hex()} if op == "canonical-json": _exact(expr, {"op", "value"}, op); value = self._eval(expr["value"]) raw = canonical_json(value) return {"value": value, "canonical_json_hex": raw.hex(), "sha256": hashlib.sha256(raw).hexdigest()} if op == "canonical-json-raw": _exact(expr,{"op","value"},op) def raw_check(v): if v is None or type(v) is bool: return v if type(v) is int: return _u64(v) if type(v) is str: return _nfc(v) if type(v) is list: return [raw_check(x) for x in v] if type(v) is not dict: _fail("MALFORMED","raw JSON") return {_nfc(k):raw_check(x) for k,x in v.items()} value=raw_check(expr["value"]); raw=json.dumps(value,ensure_ascii=False,sort_keys=True,separators=(",", ":")).encode("utf-8") return {"value":value,"canonical_json_hex":raw.hex(),"sha256":hashlib.sha256(raw).hexdigest()} if op == "canonical-json-normalized": _exact(expr,{"op","value"},op); value=self._eval(expr["value"]) def normalize(v): if type(v) is dict and v.get("$kind")=="ref" and set(v)=={"$kind","type","hex"}: return {"$kind":"ref","type":v["type"],"hex":""} if type(v) is dict: return {k:normalize(x) for k,x in v.items()} if type(v) is list: return [normalize(x) for x in v] return v value=normalize(value); raw=json.dumps(value,ensure_ascii=False,sort_keys=True,separators=(",", ":")).encode("utf-8") return {"value":value,"canonical_json_hex":raw.hex(),"sha256":hashlib.sha256(raw).hexdigest()} if op == "identity-mapping": _exact(expr, {"op", "kind"}, op); kind = _nfc(expr["kind"], "identity kind") if kind not in ("derived", "opaque"): _fail("MALFORMED", "identity kind") out = {} for rec in self.store["records"].values(): if rec["identity"]["kind"] != kind or type(rec["key"]) is not dict or rec["key"].get("$kind") != "ref": continue key = typed_key(rec["key"]) out[key] = ({"preimage_hex": rec["identity"]["preimage_hex"], "recomputed_ref": rec["key"]} if kind == "derived" else {"value": rec["key"], "allocation": copy.deepcopy(self.store["allocations"][rec["key"]["type"]]["used"][rec["key"]["hex"]])}) return out if op == "persisted-rebind": _exact(expr, {"op", "record_type", "match_fields"}, op); record_type = _nfc(expr["record_type"], "record type"); match = self._eval(expr["match_fields"]) if record_type not in self.profile.record_rules or type(match) is not dict or not match: _fail("MALFORMED", "persisted rebind") rr = self.profile.record_rules[record_type] if rr.id_type is None or any(k not in rr.fields for k in match): _fail("TYPE_MISMATCH", "persisted rebind") matches = [rec["key"] for rec in self.store["records"].values() if rec["record_type"] == record_type and all(rec["fields"][k] == v for k,v in match.items())] if not matches: _fail("UNKNOWN", "persisted rebind") if len(matches) != 1: _fail("CONTEXT_MISMATCH", "persisted rebind ambiguity") return {"record_type": record_type, "matched_fields": match, "ref": copy.deepcopy(matches[0])} if op == "canonical-id": _exact(expr, {"op", "id_type", "fields", "rule_override"}, op); id_type = _nfc(expr["id_type"], "ID type"); fields = self._eval(expr["fields"]) if expr["rule_override"] is not None and type(expr["rule_override"]) is not dict: _fail("MALFORMED", "rule override") rule = self.profile.id_rules.get(id_type) if not rule or rule.kind != "derived": _fail("TYPE_MISMATCH", "canonical ID") if type(fields) is not dict or set(fields) != set(rule.fields): _fail("MALFORMED", "derived fields") descriptors = [rr.fields for rr in self.profile.record_rules.values() if rr.id_type == id_type and set(rule.fields) <= set(rr.fields)] if descriptors: for name in rule.fields: _field_validate(fields[name], descriptors[0][name], self.profile) else: _validate_consumed_refs(fields, self.profile) ref, pre = _derive(rule, fields, expr["rule_override"]) return {"ref": ref, "preimage_hex": pre.hex()} if op == "typed-content-id": _exact(expr, {"op", "subject_ref", "payload_ref"}, op) role = self.profile.typed_content_role if role is None: _fail("TYPE_MISMATCH", "typed content unsupported") subject = validate_profile_ref(self._eval(expr["subject_ref"]), self.profile, set(self.profile.unions[role.subject_union])) payload = validate_profile_ref(self._eval(expr["payload_ref"]), self.profile, {role.payload_id_type}) canonical = {role.canonical_value_fields[0]: subject, role.canonical_value_fields[1]: payload} raw = canonical_json(canonical) fields = {role.fields[0]: role.constant_type_tag, role.fields[1]: bytes_value(raw)} ref, pre = _derive(self.profile.id_rules[role.state_id_type], fields) result = {"ref": ref, "canonical_bytes": fields[role.fields[1]], "preimage_hex": pre.hex(), "subject_ref": subject, "payload_ref": payload} if tuple(result) != role.result_fields: _fail("MALFORMED", "typed content result schema") return result if op == "trace-effects": _exact(expr, {"op", "event"}, op) event_query=expr["event"] if type(event_query) is not dict or set(event_query)!={"op","select","project","cardinality"} or event_query.get("op")!="trace" or event_query.get("project")!="event" or event_query.get("cardinality")!="one": _fail("MALFORMED", "authenticated trace event query") event = self._eval(event_query); _validate_trace_event(event) reads, writes = event["reads"], event["writes"] def count(section): return sum(path==section or path.startswith(section+"/") for path in writes) # Counts are occurrences in the native ordered trace arrays; duplicates are not uniqued. return {"step_id": event["step_id"], "status": event["status"], "read_count": len(reads), "write_count": len(writes), "record_writes": count("records"), "edge_writes": count("edges"), "binding_writes": count("bindings"), "allocation_writes": count("allocations"), "writes": copy.deepcopy(writes)} if op == "trace-write-excludes": _exact(expr,{"op","after_seq","path"},op) after=_u64(self._eval(expr["after_seq"]),"trace after sequence"); path=_trace_path(expr["path"]) events=[] for event in self.store["trace"]: _validate_trace_event(event) if event["seq"]>after: events.append({"seq":event["seq"],"step_id":event["step_id"],"matching_writes":[x for x in event["writes"] if x==path]}) return {"after_seq":after,"path":path,"excluded":all(not x["matching_writes"] for x in events),"events":events} if op == "serialize-store": _exact(expr, {"op"}, op); raw = serialize_store(self.store) return {"bytes": bytes_value(raw), "sha256": hashlib.sha256(raw).hexdigest()} if op == "reconstruct": _exact(expr, {"op", "bytes", "mode", "rebind"} if "rebind" in expr else {"op", "bytes", "mode"}, op); raw = binary_raw(self._eval(expr["bytes"]), "bytes") if expr["mode"] not in ("validate", "fresh-store"): _fail("MALFORMED", "reconstruct mode") rebuilt = reconstruct_store(raw, self.profile, fresh=expr["mode"] == "fresh-store") _validate_consumed_refs([rebuilt["records"], rebuilt["bindings"], rebuilt["edges"], rebuilt["roots"], rebuilt["availability"]], self.profile) derived, opaque = {}, {} for rec in rebuilt["records"].values(): if type(rec["key"]) is dict and rec["key"].get("$kind") == "ref" and rec["identity"]["kind"] in ("derived","opaque"): target = derived if rec["identity"]["kind"] == "derived" else opaque target[typed_key(rec["key"])] = ({"recomputed_ref": rec["key"], "preimage_hex": rec["identity"]["preimage_hex"]} if target is derived else {"value":rec["key"],"allocation":copy.deepcopy(rebuilt["allocations"][rec["key"]["type"]]["used"][rec["key"]["hex"]])}) rebound = None if "rebind" in expr: spec=expr["rebind"] if type(spec) is not dict or set(spec)!={"record_type","match_fields"}: _fail("MALFORMED","reconstruct rebind") record_type=_nfc(spec["record_type"],"record type"); match=self._eval(spec["match_fields"]) rr=self.profile.record_rules.get(record_type) if rr is None or rr.id_type is None or type(match) is not dict or not match or any(k not in rr.fields for k in match): _fail("MALFORMED","reconstruct rebind") matches=[rec["key"] for rec in rebuilt["records"].values() if rec["record_type"]==record_type and all(rec["fields"][k]==v for k,v in match.items())] if not matches: _fail("UNKNOWN","reconstruct rebind") if len(matches)!=1: _fail("CONTEXT_MISMATCH","reconstruct rebind ambiguity") rebound={"record_type":record_type,"matched_fields":match,"ref":copy.deepcopy(matches[0])} return {"store_digest": store_digest(rebuilt), "derived": derived, "opaque": opaque, "relations": sorted(rebuilt["edges"].values(), key=lambda x: (x["kind"], typed_key(x["from"]), typed_key(x["to"]))), "bindings": copy.deepcopy(rebuilt["bindings"]), "allocations": copy.deepcopy(rebuilt["allocations"]), "lifecycle": copy.deepcopy(rebuilt["lifecycle"]), "snapshots": copy.deepcopy(rebuilt["snapshots"]), "roots": copy.deepcopy(rebuilt["roots"]), "availability": copy.deepcopy(rebuilt["availability"]), "registers": copy.deepcopy(rebuilt["registers"]), "trace": copy.deepcopy(rebuilt["trace"]), "rebound": rebound} if op == "relation-set": _exact(expr, {"op", "kinds"}, op) if type(expr["kinds"]) is not list: _fail("MALFORMED", "relation kinds") kinds = [_nfc(x, "relation kind") for x in expr["kinds"]] if len(kinds) != len(set(kinds)): _fail("MALFORMED", "duplicate relation kind") if any(x not in self.profile.edge_rules for x in kinds): _fail("TYPE_MISMATCH", "relation kind") return sorted([copy.deepcopy(e) for e in self.store["edges"].values() if e["kind"] in kinds], key=lambda x: (x["kind"], typed_key(x["from"]), typed_key(x["to"]))) if op == "record-count": _exact(expr, {"op", "record_type", "key"}, op); record_type=_nfc(expr["record_type"],"record type"); key=self._eval(expr["key"]) if record_type not in self.profile.record_rules: _fail("UNSUPPORTED_RECORD",record_type) _validate_consumed_refs(key,self.profile) return {"record_type":record_type,"key":key,"count":int(_record_key(record_type,key) in self.store["records"])} if op == "records": _exact(expr, {"op", "types"}, op) if type(expr["types"]) is not list: _fail("MALFORMED", "record types") types = [_nfc(x, "record type") for x in expr["types"]] if len(types) != len(set(types)) or any(x not in self.profile.record_rules for x in types): _fail("UNSUPPORTED_RECORD", "record type") return [copy.deepcopy(self.store["records"][k]) for k in sorted(self.store["records"]) if self.store["records"][k]["record_type"] in types] if op == "store-section": _exact(expr, {"op", "section"}, op); section = _nfc(expr["section"], "store section") if section not in ("bindings", "lifecycle", "roots", "availability", "allocations", "snapshots"): _fail("MALFORMED", "store section") return copy.deepcopy(self.store[section]) if op == "version-select": _exact(expr, {"op", "record_type", "match", "as_of"}, op) record_type = _nfc(expr["record_type"], "record type") if record_type not in self.profile.record_rules: _fail("UNSUPPORTED_RECORD", record_type) rr = self.profile.record_rules[record_type] if rr.instance_order_field is None: _fail("TYPE_MISMATCH", "unordered record type") match = self._eval(expr["match"]) if type(match) is not dict: _fail("MALFORMED", "version match") as_of = None if expr["as_of"] is None else _u64(self._eval(expr["as_of"]), "as-of") found = [] for rec in self.store["records"].values(): if rec["record_type"] != record_type: continue if any(field not in rr.fields or rec["fields"][field] != value for field, value in match.items()): continue if as_of is None or rec["instance_order"] <= as_of: found.append(rec) if not found: _fail("UNKNOWN", "version") return copy.deepcopy(max(found, key=lambda rec: rec["instance_order"])) raise AssertionError(op) def _resolve(self, req: Any) -> dict[str, Any]: # Syntax is always first, then type, context, existence, lifecycle, expiry, availability. try: _exact(req, {"mode", "reference", "context", "clock", "expected_types"}, "resolver request") mode = _nfc(req["mode"]); clock = _u64(req["clock"]) if mode not in RESOLVER_MODES: _fail("MALFORMED", "resolver mode") if type(req["expected_types"]) is not list or not req["expected_types"]: _fail("MALFORMED", "expected types") expected_list = [_nfc(x, "expected type") for x in req["expected_types"]] if len(expected_list) != len(set(expected_list)) or any(x not in self.profile.id_rules for x in expected_list): _fail("MALFORMED", "expected types") expected_types = set(expected_list) context = req["context"] bkey = expected_value = None if context is not None: _exact(context, {"binding_kind", "key", "expected_value"}, "resolver context") binding_kind = _nfc(context["binding_kind"], "context binding kind") if binding_kind not in self.profile.binding_kinds: _fail("MALFORMED", "context binding kind") key, expected_value = _validate_json(context["key"]), _validate_json(context["expected_value"]) _validate_consumed_refs([key, expected_value], self.profile) bkey = binding_kind + "|" + typed_key(key) if mode in ("alias-current","alias-as-of"): required={"namespace_ref","label"} | ({"version_ref"} if mode=="alias-as-of" else set()) _exact(req["reference"],required,"version reference") namespace=validate_profile_ref(req["reference"]["namespace_ref"],self.profile) label=_nfc(req["reference"]["label"],"version label") version_ref=None if mode=="alias-current" else validate_profile_ref(req["reference"]["version_ref"],self.profile) roles=[role for role in self.profile.versioned_reference_roles if namespace["type"] in role.namespace_types] if not roles: return _resolution("TYPE_MISMATCH") if version_ref is not None: typed_roles=[role for role in roles if role.id_type==version_ref["type"]] if not typed_roles: return _resolution("TYPE_MISMATCH") roles=typed_roles if not expected_types.intersection({role.id_type for role in roles}): return _resolution("TYPE_MISMATCH") if context is not None and (bkey not in self.store["bindings"] or self.store["bindings"][bkey]["value"] != expected_value): return _resolution("CONTEXT_MISMATCH") candidates=[] for role in roles: for rec in self.store["records"].values(): if rec["record_type"]==role.record_type and rec["fields"][role.namespace_field]==namespace and rec["fields"][role.label_field]==label: candidates.append((role,rec)) if not candidates: return _resolution("UNKNOWN") role=candidates[0][0] if any(x[0]!=role for x in candidates): return _resolution("CONTEXT_MISMATCH") if version_ref is not None: selected=next((rec for _,rec in candidates if rec["key"]==version_ref),None) if selected is None: return _resolution("UNKNOWN") else: predecessor_keys={typed_key(rec["fields"][role.predecessor_field]) for _,rec in candidates if rec["fields"][role.predecessor_field] is not None} heads=[rec for _,rec in candidates if typed_key(rec["key"]) not in predecessor_keys] if len(heads)!=1: return _resolution("UNKNOWN" if not heads else "CONTEXT_MISMATCH") selected=heads[0] matched=selected["key"]; fields=selected["fields"] if matched["type"] not in expected_types: return _resolution("TYPE_MISMATCH") if fields[role.tombstone_field] is True: return _resolution("TOMBSTONED",matched=matched) expires=fields[role.expires_field] if expires is not None and expires < clock: return _resolution("EXPIRED",matched=matched) target=fields[role.target_field] if target is None: return _resolution("UNKNOWN",matched=matched) if _type_of(target) not in role.target_types: return _resolution("TYPE_MISMATCH") if not _find_ref(self.store,target,self.profile): return _resolution("UNKNOWN",matched=matched) return _resolution("RESOLVED",denotation=target,matched=matched) if mode == "locator-current": target = validate_profile_ref(req["reference"], self.profile) if target["type"] not in expected_types: return _resolution("TYPE_MISMATCH") if context is None or context.get("binding_kind") != "endpoint-occupant" or context.get("key") != target: return _resolution("CONTEXT_MISMATCH") locator_ref = validate_profile_ref(context["expected_value"], self.profile, {"LocatorRecordId"}) if bkey not in self.store["bindings"] or self.store["bindings"][bkey]["value"] != locator_ref: return _resolution("CONTEXT_MISMATCH") found = _find_ref(self.store, locator_ref, self.profile) if not found: return _resolution("UNKNOWN") rec = found[1] if rec["record_type"] != "LocatorVersion" or rec["fields"].get("target_ref") != target: return _resolution("CONTEXT_MISMATCH") if rec["fields"].get("tombstone") is True: return _resolution("TOMBSTONED", matched=locator_ref) expires = rec["fields"].get("expires_at") if expires is not None and expires < clock: return _resolution("EXPIRED", matched=locator_ref) if rec["fields"].get("available") is False: return _resolution("UNAVAILABLE", denotation=target, matched=locator_ref) return _resolution("RESOLVED", denotation=target, matched=locator_ref) ref = validate_profile_ref(req["reference"], self.profile) except D001Error: return _resolution("MALFORMED") if ref["type"] not in expected_types: return _resolution("TYPE_MISMATCH") # If raw bytes exist under another concrete type, the requested type is wrong, not unknown. same_raw = [r["key"] for r in self.store["records"].values() if type(r["key"]) is dict and r["key"].get("$kind") == "ref" and r["key"]["hex"] == ref["hex"]] if same_raw and all(x["type"] != ref["type"] for x in same_raw): return _resolution("TYPE_MISMATCH") if context is not None and (bkey not in self.store["bindings"] or self.store["bindings"][bkey]["value"] != expected_value): return _resolution("CONTEXT_MISMATCH") found = _find_ref(self.store, ref, self.profile) if not found: return _resolution("UNKNOWN") rec = found[1]; matched = ref if mode in ("alias-as-of", "locator-current") else None lifecycle = self.store["lifecycle"].get(typed_key(ref)) state_fields = lifecycle if lifecycle is not None else rec["fields"] if state_fields.get("tombstone") is True: return _resolution("TOMBSTONED", matched=matched) expires = state_fields.get("expires_at") if expires is not None and expires < clock: return _resolution("EXPIRED", matched=matched) if mode == "payload-body": av = self.store["availability"].get(typed_key(ref)) if av and av["body_state"] == "COLLECTED": return _resolution("CONTENT_COLLECTED", denotation=ref) if not av or av["body_state"] == "UNAVAILABLE": return _resolution("UNAVAILABLE", denotation=ref) if mode == "locator-current": av = rec["fields"].get("available") if av is False: return _resolution("UNAVAILABLE", denotation=rec["fields"].get("target_ref"), matched=ref) denotation = rec["fields"].get("target_ref", ref) if mode in ("alias-as-of","locator-current") else ref return _resolution("RESOLVED", denotation=denotation, matched=matched, availability="AVAILABLE" if mode == "payload-body" else None) def _resolution(status: str, *, denotation: Any = None, matched: Any = None, availability: Any = None) -> dict[str, Any]: if status == "RESOLVED": return {"status": status, "value": {"denotation_ref": denotation, "availability": availability}, "denotation_ref": denotation, "matched_version_ref": matched} return {"status": status, "value": None, "denotation_ref": denotation, "matched_version_ref": matched} def _pointer(root: Any, path: Any) -> Any: if type(path) is not str or not path.startswith("/"): _fail("MALFORMED", "pointer") cur = root for raw in path[1:].split("/"): i = 0 while i < len(raw): if raw[i] == "~": if i + 1 >= len(raw) or raw[i + 1] not in "01": _fail("MALFORMED", "pointer escape") i += 2 else: i += 1 token = raw.replace("~1", "/").replace("~0", "~") try: if type(cur) is list: if not token.isascii() or not token.isdecimal() or (len(token) > 1 and token[0] == "0"): _fail("MALFORMED", "pointer array index") cur = cur[int(token)] else: cur = cur[token] except (KeyError, IndexError, TypeError): _fail("UNKNOWN", path) return copy.deepcopy(cur) __all__ = [ "D001Error", "IdRule", "RecordRule", "EdgeRule", "AvailabilityRole", "PinRole", "VersionedReferenceRole", "ProvenanceRole", "TypedContentRole", "ProfileSemantics", "Runtime", "STORE_VERSION", "QUERY_VERSION", "OPS", "FAULTS", "QUERY_OPS", "RESOLVER_ORDER", "empty_store", "validate_store", "serialize_store", "reconstruct_store", "store_digest", "canonical_json", "strict_json_loads", "cve1", "cve1_from_json", "validate_ref", "validate_slot", "typed_key", "bytes_value", "binary_raw", ]