diff --git a/future/d007_checker_v3/producer.py b/future/d007_checker_v3/producer.py new file mode 100644 index 0000000000..fef4b1556d --- /dev/null +++ b/future/d007_checker_v3/producer.py @@ -0,0 +1,542 @@ +import hashlib +import json +import unicodedata + +U = 18446744073709551615 +FRESH = ("execution_id", "activation_id", "validation_id", "invocation_nonce", "freshness_ref") +GLOB = ("BUDGET_LEDGER", "COMMITMENT_IDENTITY", "DELIVERY", "EXECUTION_RESOURCE_ACQUISITION", "EXTERNAL_OCCURRENCE_SETTLEMENT", "INGRESS", "REVOCATION") +MODES = ("NO_GENERIC_MERGE", "STRUCTURAL_THREE_WAY", "TYPED_SERVICE_RELATION") +LAW = ("WHOLE_OBSERVATION_CLOSURE", "CLOSED_IDENTITY_AND_SCOPE", "D003_AUTHORITY_FIRST", "BASE_READ_POLICY_FRESHNESS", "LINEAGE_AND_MATERIALIZED_RELATION", "CONFLICT_SOUNDNESS", "GLOBAL_FACT_IMMUTABILITY", "EXACT_ATOMIC_PUBLICATION", "FAULT_RELOAD_IDEMPOTENCY", "PROMOTION_SEPARATION", "CANONICAL_DETERMINISM") + + +def fail(): + return int("invalid") + + +def pairs(rows): + out = {} + for key, value in rows: + if key in out: + fail() + out[key] = value + return out + + +def tree(value): + if isinstance(value, float): + fail() + if isinstance(value, int) and not isinstance(value, bool) and not 0 <= value <= U: + fail() + if isinstance(value, str): + value.encode("utf-8", "strict") + if unicodedata.normalize("NFC", value) != value: + fail() + elif isinstance(value, dict): + for key, item in value.items(): + if not isinstance(key, str): + fail() + tree(key) + tree(item) + elif isinstance(value, list): + for item in value: + tree(item) + elif value is not None and not isinstance(value, (int, bool)): + fail() + + +def canon(value): + tree(value) + return (json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False) + "\n").encode() + + +def decode(raw): + value = json.loads(raw.decode("utf-8", "strict"), object_pairs_hook=pairs, parse_float=fail, parse_constant=fail) + if canon(value) != raw: + fail() + return value + + +def hash_bytes(raw): + return hashlib.sha256(raw).hexdigest() + + +def hash_value(value): + return hash_bytes(canon(value)) + + +def copy_value(value): + return json.loads(json.dumps(value)) + + +def closed(value, keys): + if not isinstance(value, dict) or set(value) != set(keys): + fail() + + +def word(value): + if not isinstance(value, str) or not value: + fail() + tree(value) + return value + + +def number(value): + if not isinstance(value, int) or isinstance(value, bool) or not 0 <= value <= U: + fail() + return value + + +def hexsum(value): + word(value) + if len(value) != 64 or any(ch not in "0123456789abcdef" for ch in value): + fail() + return value + + +def unique(rows, key): + out = {} + for row in rows: + value = word(row[key]) + if value in out: + fail() + out[value] = row + return out + + +def artifact(kind, raw): + return {"byte_length": len(raw), "kind": kind, "sha256": hash_bytes(raw)} + + +def triple(value): + closed(value, ("generation", "root", "world_id")) + number(value["generation"]) + word(value["root"]) + word(value["world_id"]) + + +def handoff_shape(value): + keys = ("activation_id", "actor_id", "audience", "commitment_id", "current_activation_id", "current_actor_id", "current_commitment_id", "current_execution_id", "current_generation", "current_grant_ref", "current_selection_id", "current_world_id", "execution_id", "freshness_ref", "generation", "invocation_nonce", "presentation_ref", "right", "selected_grant_id", "selection_id", "target_candidate_world_id", "validation_id", "verdict", "world_id") + closed(value, keys) + for key, item in value.items(): + if key in ("generation", "current_generation"): + number(item) + else: + word(item) + if value["verdict"] not in ("ALLOW", "DENY"): + fail() + + +def policy_shape(value): + closed(value, ("generation", "policy_id", "root")) + number(value["generation"]) + word(value["policy_id"]) + word(value["root"]) + + +def invariant_shape(value): + closed(value, ("digest", "invariant_id", "version")) + hexsum(value["digest"]) + word(value["invariant_id"]) + number(value["version"]) + + +def predicate_shape(value): + closed(value, ("digest", "predicate_id", "read_keys", "version")) + hexsum(value["digest"]) + word(value["predicate_id"]) + number(value["version"]) + if not isinstance(value["read_keys"], list) or not value["read_keys"] or len(value["read_keys"]) != len(set(value["read_keys"])): + fail() + for item in value["read_keys"]: + word(item) + + +def revision_shape(value): + closed(value, ("actor_id", "attempt_id", "causal_revision_ids", "codec_id", "commitment_id", "operation_id", "payload_id", "predecessor_revision_ids", "revision_id", "schema_id", "tombstone", "typed_state_slot_ref")) + for key in ("actor_id", "attempt_id", "codec_id", "commitment_id", "operation_id", "payload_id", "revision_id", "schema_id", "typed_state_slot_ref"): + word(value[key]) + if value["tombstone"] is not False or not isinstance(value["predecessor_revision_ids"], list) or len(value["predecessor_revision_ids"]) not in (0, 1) or not isinstance(value["causal_revision_ids"], list) or len(value["causal_revision_ids"]) != len(set(value["causal_revision_ids"])): + fail() + for item in value["predecessor_revision_ids"] + value["causal_revision_ids"]: + word(item) + + +def intent_shape(value): + closed(value, ("base", "bound_handoff", "bound_invariants", "bound_policy", "bound_predicates", "declared_common_base_world_id", "expires_at", "materialized_result", "mode_id", "nonce", "operation_id", "output_bindings", "owner_scoped_slots", "proposal_id", "proposed_revisions", "relation_id", "retention_epoch", "source_world_ids", "target")) + for key in ("declared_common_base_world_id", "mode_id", "nonce", "operation_id", "proposal_id", "relation_id"): + word(value[key]) + if value["mode_id"] not in MODES: + fail() + number(value["retention_epoch"]) + number(value["expires_at"]) + if value["retention_epoch"] >= value["expires_at"]: + fail() + for key in ("base", "materialized_result", "target"): + triple(value[key]) + handoff_shape(value["bound_handoff"]) + policy_shape(value["bound_policy"]) + if not isinstance(value["source_world_ids"], list) or len(value["source_world_ids"]) != 2 or len(set(value["source_world_ids"])) != 2: + fail() + scopes = [] + for row in value["owner_scoped_slots"]: + closed(row, ("actor_id", "schema_id", "typed_state_slot_ref")) + for item in row.values(): + word(item) + scopes.append((row["actor_id"], row["typed_state_slot_ref"])) + if not scopes or len(scopes) != len(set(scopes)): + fail() + for row in value["bound_invariants"]: + invariant_shape(row) + for row in value["bound_predicates"]: + predicate_shape(row) + unique(value["bound_invariants"], "invariant_id") + unique(value["bound_predicates"], "predicate_id") + revs = unique(value["proposed_revisions"], "revision_id") + for row in value["proposed_revisions"]: + revision_shape(row) + outputs = [] + schemas = {(row["actor_id"], row["typed_state_slot_ref"]): row["schema_id"] for row in value["owner_scoped_slots"]} + for row in value["output_bindings"]: + closed(row, ("actor_id", "revision_id", "typed_state_slot_ref")) + for item in row.values(): + word(item) + key = (row["actor_id"], row["typed_state_slot_ref"]) + outputs.append(key) + hit = revs.get(row["revision_id"]) + if hit is not None and (hit["actor_id"], hit["typed_state_slot_ref"], hit["schema_id"]) != (key[0], key[1], schemas[key]): + fail() + if len(outputs) != len(set(outputs)) or set(outputs) != set(scopes): + fail() + if set(revs) != {row["revision_id"] for row in value["output_bindings"] if row["revision_id"] in revs}: + fail() + + +def stable_hash(value): + made = copy_value(value) + for key in FRESH: + made["bound_handoff"][key] = "" + return hash_value(made) + + +def result_row(value): + return {"candidate_world_id": value["materialized_result"]["world_id"], "operation_id": value["operation_id"], "proposal_id": value["proposal_id"], "status": "COMMITTED"} + + +def retention_row(value): + return {"expires_at": value["expires_at"], "operation_id": value["operation_id"], "outcome": result_row(value), "proposal": copy_value(value), "proposal_sha256": stable_hash(value), "retention_epoch": value["retention_epoch"]} + + +def ancestry_closed(revisions): + for start in revisions: + frontier = [(start, set())] + found_genesis = False + while frontier: + here, path = frontier.pop() + if here in path: + fail() + row = revisions.get(here) + if row is None: + fail() + parents = row["predecessor_revision_ids"] + if not parents: + found_genesis = True + else: + next_path = path.copy() + next_path.add(here) + for parent in parents: + frontier.append((parent, next_path.copy())) + if not found_genesis: + fail() + return True + + +def state_shape(value): + closed(value, ("authority_context", "canonical", "globals", "invariants", "policies", "predicates", "promotion_records", "proposal_tombstones", "retained_proposals", "revision_occurrences", "service_records", "slot_bindings", "worlds")) + worlds = unique(value["worlds"], "world_id") + for row in value["worlds"]: + closed(row, ("generation", "kind", "parent_world_ids", "root", "world_id")) + number(row["generation"]); word(row["root"]); word(row["world_id"]) + if row["kind"] not in ("CANONICAL", "CANDIDATE", "BRANCH", "BASE") or not isinstance(row["parent_world_ids"], list) or len(row["parent_world_ids"]) != len(set(row["parent_world_ids"])): + fail() + if not row["parent_world_ids"] and row["generation"] != 0: + fail() + for parent in row["parent_world_ids"]: + if parent not in worlds or row["generation"] <= worlds[parent]["generation"]: + fail() + triple(value["canonical"]) + cw = worlds.get(value["canonical"]["world_id"]) + if cw is None or (cw["root"], cw["generation"], cw["kind"]) != (value["canonical"]["root"], value["canonical"]["generation"], "CANONICAL"): + fail() + revs = unique(value["revision_occurrences"], "revision_id") + ops = set(); attempts = set() + for row in value["revision_occurrences"]: + revision_shape(row) + if row["operation_id"] in ops or row["attempt_id"] in attempts: + fail() + ops.add(row["operation_id"]); attempts.add(row["attempt_id"]) + for row in value["revision_occurrences"]: + for parent in row["predecessor_revision_ids"]: + old = revs.get(parent) + if old is None or (old["actor_id"], old["typed_state_slot_ref"]) != (row["actor_id"], row["typed_state_slot_ref"]): + fail() + if any(item not in revs for item in row["causal_revision_ids"]): + fail() + ancestry_closed(revs) + keys = set() + for row in value["slot_bindings"]: + closed(row, ("actor_id", "revision_id", "typed_state_slot_ref", "world_id")) + key = (row["world_id"], row["actor_id"], row["typed_state_slot_ref"]) + hit = revs.get(row["revision_id"]) + if key in keys or row["world_id"] not in worlds or hit is None or (hit["actor_id"], hit["typed_state_slot_ref"]) != (row["actor_id"], row["typed_state_slot_ref"]): + fail() + keys.add(key) + closed(value["globals"], GLOB) + for kind in GLOB: + ids = [] + for row in value["globals"][kind]: + closed(row, ("fact_id", "payload_sha256")); ids.append(word(row["fact_id"])); hexsum(row["payload_sha256"]) + if len(ids) != len(set(ids)): + fail() + for row in value["policies"]: policy_shape(row) + for row in value["invariants"]: invariant_shape(row) + for row in value["predicates"]: predicate_shape(row) + unique(value["policies"], "policy_id"); unique(value["invariants"], "invariant_id"); unique(value["predicates"], "predicate_id") + authority = value["authority_context"] + closed(authority, ("activation_id", "actor_id", "audience", "commitment_id", "execution_id", "generation", "grant_ref", "presentation_ref", "right", "selection_id", "target_candidate_world_id", "world_id")) + for key, item in authority.items(): number(item) if key == "generation" else word(item) + if authority["target_candidate_world_id"] not in worlds or worlds[authority["target_candidate_world_id"]]["kind"] != "CANDIDATE": fail() + promotion_ids = [] + for row in value["promotion_records"]: + closed(row, ("canonical_world_id", "promotion_id", "root")); promotion_ids.append(word(row["promotion_id"])); hit = worlds.get(word(row["canonical_world_id"])); word(row["root"]) + if hit is None or hit["kind"] != "CANONICAL" or hit["root"] != row["root"]: fail() + if len(promotion_ids) != len(set(promotion_ids)): fail() + service_ids = [] + for row in value["service_records"]: + closed(row, ("base_revision_id", "invariant_digests", "left_revision_id", "output_revision_id", "policy_root", "predicate_digests", "relation_id", "right_revision_id", "service_record_id")); service_ids.append(word(row["service_record_id"])) + for key in ("base_revision_id", "left_revision_id", "output_revision_id", "policy_root", "relation_id", "right_revision_id"): word(row[key]) + for key in ("invariant_digests", "predicate_digests"): + if not row[key] or len(row[key]) != len(set(row[key])): fail() + for item in row[key]: hexsum(item) + if len(service_ids) != len(set(service_ids)): fail() + retained_ops = [] + for row in value["retained_proposals"]: + closed(row, ("expires_at", "operation_id", "outcome", "proposal", "proposal_sha256", "retention_epoch")); intent = row["proposal"]; intent_shape(intent) + if row != retention_row(intent): fail() + expected_revs = {item["revision_id"]: item for item in intent["proposed_revisions"]}; actual_revs = [item for item in value["revision_occurrences"] if item["operation_id"] == intent["operation_id"]] + if len(actual_revs) != len(expected_revs) or any(item != expected_revs.get(item["revision_id"]) for item in actual_revs): fail() + target = intent["materialized_result"] + if worlds.get(target["world_id"]) != {"generation": target["generation"], "kind": "CANDIDATE", "parent_world_ids": list(intent["source_world_ids"]), "root": target["root"], "world_id": target["world_id"]}: fail() + wanted = [{"actor_id": item["actor_id"], "revision_id": item["revision_id"], "typed_state_slot_ref": item["typed_state_slot_ref"], "world_id": target["world_id"]} for item in intent["output_bindings"]] + actual = [item for item in value["slot_bindings"] if item["world_id"] == target["world_id"]] + if sorted(wanted, key=canon) != sorted(actual, key=canon): fail() + retained_ops.append(intent["operation_id"]) + tomb_ops = [] + for row in value["proposal_tombstones"]: + closed(row, ("operation_id", "prior_outcome", "prior_proposal", "prior_proposal_sha256", "retired_at")); intent = row["prior_proposal"]; intent_shape(intent); number(row["retired_at"]) + if row["operation_id"] != intent["operation_id"] or row["prior_outcome"] != result_row(intent) or row["prior_proposal_sha256"] != stable_hash(intent) or row["retired_at"] < intent["expires_at"]: fail() + tomb_ops.append(row["operation_id"]) + if len(retained_ops) != len(set(retained_ops)) or len(tomb_ops) != len(set(tomb_ops)) or set(retained_ops) & set(tomb_ops): fail() + ids = [row["proposal"]["proposal_id"] for row in value["retained_proposals"]] + [row["prior_proposal"]["proposal_id"] for row in value["proposal_tombstones"]] + if len(ids) != len(set(ids)): fail() + return value + + +def world_bases(state, left, right): + worlds = {row["world_id"]: row for row in state["worlds"]} + reaches = [] + for start in (left, right): + found = set(); todo = [start] + while todo: + here = todo.pop() + if here not in found: + found.add(here); todo = todo + worlds[here]["parent_world_ids"] + reaches.append(found) + common = reaches[0] & reaches[1]; best = [] + for item in common: + dominated = False + for other in common: + if item != other: + found = set(); todo = [other] + while todo: + here = todo.pop() + if here not in found: found.add(here); todo = todo + worlds[here]["parent_world_ids"] + if item in found: dominated = True + if not dominated: best.append(item) + return worlds, sorted(best) + + +def bindings(state, world): + return {(row["actor_id"], row["typed_state_slot_ref"]): row["revision_id"] for row in state["slot_bindings"] if row["world_id"] == world} + + +def authority_ok(state, intent, handoff): + current = state["authority_context"] + direct = ("actor_id", "audience", "commitment_id", "generation", "presentation_ref", "right", "selection_id", "target_candidate_world_id", "world_id") + joins = {"current_activation_id": "activation_id", "current_actor_id": "actor_id", "current_commitment_id": "commitment_id", "current_execution_id": "execution_id", "current_generation": "generation", "current_grant_ref": "grant_ref", "current_selection_id": "selection_id", "current_world_id": "world_id", "selected_grant_id": "grant_ref"} + return handoff["verdict"] == "ALLOW" and all(handoff[key] == current[key] for key in direct) and all(handoff[key] == current[value] for key, value in joins.items()) and handoff["target_candidate_world_id"] == intent["target"]["world_id"] and all(handoff[key] == intent["bound_handoff"][key] for key in handoff if key not in FRESH) + + +def context_ok(state, intent): + return intent["bound_policy"] in state["policies"] and sorted(intent["bound_invariants"], key=canon) == sorted(state["invariants"], key=canon) and sorted(intent["bound_predicates"], key=canon) == sorted(state["predicates"], key=canon) + + +def lineage_ok(state, intent): + old = {row["revision_id"]: row for row in state["revision_occurrences"]}; new = {row["revision_id"]: row for row in intent["proposed_revisions"]} + if set(old) & set(new) or len(new) != len(intent["proposed_revisions"]): return False + if intent["operation_id"] in {row["operation_id"] for row in state["revision_occurrences"] + state["retained_proposals"] + state["proposal_tombstones"]}: return False + attempts = {row["attempt_id"] for row in state["revision_occurrences"]}; fresh_attempts = [row["attempt_id"] for row in intent["proposed_revisions"]] + if len(fresh_attempts) != len(set(fresh_attempts)) or attempts & set(fresh_attempts): return False + joined = old.copy(); joined.update(new) + scopes = {(row["actor_id"], row["typed_state_slot_ref"]): row["schema_id"] for row in intent["owner_scoped_slots"]} + for row in intent["proposed_revisions"]: + if row["operation_id"] != intent["operation_id"] or row["commitment_id"] != intent["bound_handoff"]["commitment_id"] or len(row["predecessor_revision_ids"]) != 1: return False + parent = joined.get(row["predecessor_revision_ids"][0]) + if parent is None or (parent["actor_id"], parent["typed_state_slot_ref"]) != (row["actor_id"], row["typed_state_slot_ref"]) or any(item not in joined for item in row["causal_revision_ids"]): return False + seen = set(); here = row + while here["predecessor_revision_ids"]: + if here["revision_id"] in seen: return False + seen.add(here["revision_id"]); here = joined.get(here["predecessor_revision_ids"][0]) + if here is None: return False + if here["revision_id"] not in old: return False + for row in intent["output_bindings"]: + hit = joined.get(row["revision_id"]); key = (row["actor_id"], row["typed_state_slot_ref"]) + if hit is None or (hit["actor_id"], hit["typed_state_slot_ref"], hit["schema_id"], hit["commitment_id"]) != (key[0], key[1], scopes[key], intent["bound_handoff"]["commitment_id"]): return False + return True + + +def relation_ok(state, intent): + if intent["mode_id"] == "NO_GENERIC_MERGE": return False + out = {(row["actor_id"], row["typed_state_slot_ref"]): row["revision_id"] for row in intent["output_bindings"]} + if intent["mode_id"] == "STRUCTURAL_THREE_WAY": + base, left, right = [bindings(state, world) for world in (intent["base"]["world_id"], intent["source_world_ids"][0], intent["source_world_ids"][1])] + for scope in intent["owner_scoped_slots"]: + key = (scope["actor_id"], scope["typed_state_slot_ref"]); b = base.get(key); l = left.get(key); r = right.get(key) + allowed = l if l == r or r == b else r if l == b else None + if allowed is None or out.get(key) != allowed: return False + return True + if len(state["service_records"]) != 1 or len(intent["owner_scoped_slots"]) != 1: return False + scope = intent["owner_scoped_slots"][0]; key = (scope["actor_id"], scope["typed_state_slot_ref"]); base = bindings(state, intent["base"]["world_id"]).get(key); left = bindings(state, intent["source_world_ids"][0]).get(key); right = bindings(state, intent["source_world_ids"][1]).get(key); output = out.get(key); hit = {row["revision_id"]: row for row in intent["proposed_revisions"]}.get(output); service = state["service_records"][0] + wanted = {"base_revision_id": base, "invariant_digests": [row["digest"] for row in intent["bound_invariants"]], "left_revision_id": left, "output_revision_id": output, "policy_root": intent["bound_policy"]["root"], "predicate_digests": [row["digest"] for row in intent["bound_predicates"]], "relation_id": intent["relation_id"], "right_revision_id": right, "service_record_id": service["service_record_id"]} + return service == wanted and hit is not None and hit["predecessor_revision_ids"] == [left] and right in hit["causal_revision_ids"] + + +def publication(intent): + target = intent["materialized_result"] + return {"candidate_world": {"generation": target["generation"], "kind": "CANDIDATE", "parent_world_ids": list(intent["source_world_ids"]), "root": target["root"], "world_id": target["world_id"]}, "canonical_update": None, "global_operations": [], "promotion_operations": [], "retained_proposal_insert": retention_row(intent), "revision_occurrence_inserts": copy_value(intent["proposed_revisions"]), "slot_binding_inserts": [{"actor_id": row["actor_id"], "revision_id": row["revision_id"], "typed_state_slot_ref": row["typed_state_slot_ref"], "world_id": target["world_id"]} for row in intent["output_bindings"]]} + + +def status_relation(state, intent, handoff, now, attempted): + if not authority_ok(state, intent, handoff): return "DENY_AUTHORITY", None + stored = [row for row in state["retained_proposals"] + state["proposal_tombstones"] if row["operation_id"] == intent["operation_id"]] + if stored: + row = stored[0]; digest = row["proposal_sha256"] if "proposal_sha256" in row else row["prior_proposal_sha256"] + if digest != stable_hash(intent): return "BINDING_MISMATCH", None + return ("RECONCILE_EXPIRED" if "prior_outcome" in row or now >= row.get("expires_at", 0) else "REPLAY"), intent["materialized_result"]["world_id"] + if now < intent["retention_epoch"] or now >= intent["expires_at"]: return "OUTSIDE_RETENTION", None + worlds, best = world_bases(state, intent["source_world_ids"][0], intent["source_world_ids"][1]); target = worlds.get(intent["target"]["world_id"]) + if target is None or (target["root"], target["generation"], target["kind"]) != (intent["target"]["root"], intent["target"]["generation"], "CANDIDATE") or any(worlds.get(item, {}).get("kind") != "BRANCH" for item in intent["source_world_ids"]): return "STALE_BASE", None + if len(best) != 1: return "AMBIGUOUS_BASE", None + if best[0] != intent["declared_common_base_world_id"] or intent["base"] != {key: worlds[best[0]][key] for key in ("generation", "root", "world_id")}: return "STALE_BASE", None + delta_keys = ("candidate_world", "canonical_update", "global_operations", "promotion_operations", "retained_proposal_insert", "revision_occurrence_inserts", "slot_binding_inserts") + if not isinstance(attempted, dict) or set(attempted) != set(delta_keys): return "INVALID_DELTA", None + if attempted["global_operations"] or attempted["canonical_update"] is not None or attempted["promotion_operations"]: return "GLOBAL_FACT_VIOLATION", None + if not context_ok(state, intent): return "STALE_CONTEXT", None + if not lineage_ok(state, intent): return "INVALID_LINEAGE", None + if intent["mode_id"] == "NO_GENERIC_MERGE": return "UNSUPPORTED", None + if not relation_ok(state, intent): return ("CONFLICT" if intent["mode_id"] == "STRUCTURAL_THREE_WAY" else "INVALID_SERVICE_RELATION"), None + target = intent["materialized_result"] + if target["world_id"] in worlds or target["root"] in {row["root"] for row in worlds.values()} or target["generation"] != max(worlds[item]["generation"] for item in intent["source_world_ids"]) + 1 or attempted != publication(intent): return "INVALID_DELTA", None + return "COMMITTED", target["world_id"] + + +def publish(before, delta): + after = copy_value(before) + after["worlds"].append(copy_value(delta["candidate_world"])); after["slot_bindings"] = after["slot_bindings"] + copy_value(delta["slot_binding_inserts"]); after["revision_occurrences"] = after["revision_occurrences"] + copy_value(delta["revision_occurrence_inserts"]); after["retained_proposals"].append(copy_value(delta["retained_proposal_insert"])); state_shape(after) + return after + + +def observe(inputs): + closed(inputs, ("initial_state", "phases")); state_shape(inputs["initial_state"]); current = copy_value(inputs["initial_state"]); rows = []; rights = {}; seen = {key: set() for key in FRESH}; epoch = 0 + for phase in inputs["phases"]: + before = copy_value(current); kind = phase["kind"] + if kind == "RELOAD": + closed(phase, ("kind",)); current = decode(canon(current)); epoch += 1 + for right in rights.values(): right["epoch"] = epoch + rows.append({"classification": None, "delta": None, "durable_after": copy_value(current), "durable_before": before, "kind": kind, "reload_bytes_sha256": hash_value(before), "response_visible": False}) + else: + closed(phase, ("attempted_delta", "fault", "handoff", "kind", "logical_now", "proposal") if kind == "SUBMIT" else ("handoff", "kind", "logical_now", "proposal")); intent = phase["proposal"]; handoff = phase["handoff"]; intent_shape(intent); handoff_shape(handoff); number(phase["logical_now"]) + for key in FRESH: + if handoff[key] in seen[key]: fail() + seen[key].add(handoff[key]) + op = intent["operation_id"]; authority = [handoff[key] for key in ("actor_id", "audience", "commitment_id", "current_activation_id", "current_execution_id", "current_grant_ref", "current_selection_id", "current_world_id", "generation", "presentation_ref", "right", "selection_id", "target_candidate_world_id", "world_id")]; token = {"authority": authority, "epoch": None, "proposal_sha256": stable_hash(intent), "state_sha256": hash_value(before), "target": [intent["target"][key] for key in ("world_id", "root", "generation")]} + if kind == "RETRY": + right = rights.get(op) + if right is None or right["epoch"] != epoch or right["epoch"] is None or any(right[key] != token[key] for key in ("authority", "proposal_sha256", "target")) or right["state_sha256"] != hash_value(before): fail() + attempted = phase["attempted_delta"] if kind == "SUBMIT" else publication(intent); status, candidate = status_relation(before, intent, handoff, phase["logical_now"], attempted); fault = phase.get("fault") + if kind == "SUBMIT" and fault in ("BEFORE_COMMIT", "AFTER_COMMIT_BEFORE_RESPONSE"): + if op in rights: fail() + rights[op] = token + commits = status == "COMMITTED" and not (kind == "SUBMIT" and fault == "BEFORE_COMMIT"); delta = publication(intent) if commits else None + if delta is not None: current = publish(before, delta) + if kind == "SUBMIT" and fault == "AFTER_COMMIT_BEFORE_RESPONSE": rights[op]["state_sha256"] = hash_value(current) + if kind == "RETRY": rights.pop(op) + rows.append({"classification": {"candidate_world_id": candidate, "operation_id": op, "proposal_id": intent["proposal_id"], "status": status}, "delta": delta, "durable_after": copy_value(current), "durable_before": before, "kind": kind, "reload_bytes_sha256": None, "response_visible": not (kind == "SUBMIT" and fault in ("BEFORE_COMMIT", "AFTER_COMMIT_BEFORE_RESPONSE"))}) + return {"final_state": current, "input_sha256": hash_value(inputs), "phases": rows} + + +def relation_laws(inputs, observation): + if observation["input_sha256"] != hash_value(inputs) or observation["final_state"] != (observation["phases"][-1]["durable_after"] if observation["phases"] else inputs["initial_state"]): fail() + state_shape(inputs["initial_state"]); ancestry_closed(unique(inputs["initial_state"]["revision_occurrences"], "revision_id")) + state_shape(observation["final_state"]); ancestry_closed(unique(observation["final_state"]["revision_occurrences"], "revision_id")); prior = inputs["initial_state"] + for phase, row in zip(inputs["phases"], observation["phases"]): + if row["durable_before"] != prior or row["kind"] != phase["kind"]: fail() + state_shape(row["durable_before"]); ancestry_closed(unique(row["durable_before"]["revision_occurrences"], "revision_id")) + state_shape(row["durable_after"]); ancestry_closed(unique(row["durable_after"]["revision_occurrences"], "revision_id")) + if row["durable_before"]["globals"] != row["durable_after"]["globals"] or row["durable_before"]["canonical"] != row["durable_after"]["canonical"] or row["durable_before"]["promotion_records"] != row["durable_after"]["promotion_records"]: fail() + if row["delta"] is None: + if row["durable_before"] != row["durable_after"]: fail() + else: + intent = phase["proposal"] + if row["delta"] != publication(intent) or publish(row["durable_before"], row["delta"]) != row["durable_after"]: fail() + prior = row["durable_after"] + if len(observation["phases"]) != len(inputs["phases"]): fail() + return {key: True for key in LAW} + + +def manifest_pin(value): + closed(value, ("algebra", "boundary", "budget", "canonicalization", "cases", "decision", "evidence", "projections", "python_grammar", "resolution", "schema", "stage")) + if value["schema"] != "D007-MANIFEST-3" or value["stage"] != "PREREGISTERED_V3" or value["python_grammar"] != "3.13" or value["algebra"] != {"id": "WORLD_PROPOSAL_ALGEBRA", "representation_neutral": True}: fail() + budget = value["budget"] + if budget != {"cell_count": 30, "engineer_hours_cap": 56, "family_count": 10, "projection_count": 3, "review_gate_count": 4, "review_gates": ["EXECUTOR_INDEPENDENT_REVIEW", "CHECKER_INDEPENDENT_REVIEW", "EVALUATOR_REVIEW", "PRE_FREEZE_SIGNOFF"], "risk_tier": "FULL", "stop_date": "2026-08-30", "subcase_count": 30}: fail() + if [row["id"] for row in value["cases"]] != ["F01", "F02", "F03", "F04", "F05", "F06", "F07", "F08", "F09", "F10"] or [len(row["subcases"]) for row in value["cases"]] != [4, 3, 4, 4, 2, 4, 2, 4, 2, 1]: fail() + paths = value["evidence"]["producer_sources"] + if paths != {"CHECKER": "future/d007_checker_v3/producer.py", "EXECUTOR": "future/d007_executor_v3/producer.py"} or value["evidence"]["producer_interface"] != "produce(packet_bytes, manifest_bytes, identity_bytes)": fail() + edges = [["M", "PF"], ["P", "PF"], ["VS", "PF"], ["IX", "PF"], ["IC", "PF"], ["IR", "PF"], ["PF", "F"], ["M", "F"], ["P", "F"], ["VS", "F"], ["IX", "F"], ["IC", "F"], ["IR", "F"], ["M", "O1"], ["IX", "O1"], ["M", "O2"], ["IC", "O2"], ["F", "V"], ["IR", "V"], ["O1", "V"], ["O2", "V"], ["V", "E"]] + if value["evidence"]["dag"] != {"edges": edges}: fail() + + +def identity_pin(value, manifest_bytes, packet_bytes): + closed(value, ("clean", "closure_sha256", "dependencies", "locks", "role", "root", "schema", "sources")) + if value["schema"] != "D007-IDENTITY-3" or value["role"] != "CHECKER" or value["root"] != "future/d007_checker_v3" or value["dependencies"] != []: fail() + source = value["sources"] + locks = value["locks"] + if len(source) != 1 or source[0]["path"] != "future/d007_checker_v3/producer.py" or locks != [{"byte_length": len(packet_bytes), "path": "decisions/D-007-PACKET.md", "sha256": hash_bytes(packet_bytes)}, {"byte_length": len(manifest_bytes), "path": "decisions/d007-world-merge-manifest-v1.json", "sha256": hash_bytes(manifest_bytes)}]: fail() + for row in source + locks: + closed(row, ("byte_length", "path", "sha256")); number(row["byte_length"]); word(row["path"]); hexsum(row["sha256"]) + if value["closure_sha256"] != hash_value(sorted(source + locks, key=lambda row: row["path"])): fail() + clean = value["clean"] + closed(clean, ("head_commit", "index_tree", "modified_paths", "tracked", "untracked_paths", "worktree_tree")) + for key in ("head_commit", "index_tree", "worktree_tree"): + if len(clean[key]) != 40 or any(ch not in "0123456789abcdef" for ch in clean[key]): fail() + if clean["index_tree"] != clean["worktree_tree"] or clean["modified_paths"] or clean["untracked_paths"] or clean["tracked"] != sorted(source + locks, key=lambda row: row["path"]): fail() + + +def produce(packet_bytes, manifest_bytes, identity_bytes): + if len(packet_bytes) != 17133 or hash_bytes(packet_bytes) != "1430c77dbc3df0dadd3a752b1a5bb97e5c358ee5ad126f538f09ab98ce833790": fail() + if len(manifest_bytes) != 276765 or hash_bytes(manifest_bytes) != "33663d1141bbabcc170763d9751a12c4c8d0eef4c80e8ce4c43dacf42983491f": fail() + manifest = decode(manifest_bytes); identity = decode(identity_bytes); manifest_pin(manifest); identity_pin(identity, manifest_bytes, packet_bytes) + cells = []; assertions = [] + for family in manifest["cases"]: + for subcase in family["subcases"]: + observation = observe(subcase["inputs"]); laws = relation_laws(subcase["inputs"], observation); cells.append({"case_id": family["id"], "observation": observation, "subcase_id": subcase["id"]}); assertions.append({"case_id": family["id"], "laws": laws, "pass": all(laws.values()), "subcase_id": subcase["id"]}) + return canon({"assertions": assertions, "cells": cells, "claims": list(LAW), "manifest": artifact("MANIFEST", manifest_bytes), "producer_identity": artifact("CHECKER_IDENTITY", identity_bytes), "producer_role": "CHECKER", "schema": "D007-O-3"}) diff --git a/future/d007_executor_v3/producer.py b/future/d007_executor_v3/producer.py new file mode 100644 index 0000000000..70d244dbfb --- /dev/null +++ b/future/d007_executor_v3/producer.py @@ -0,0 +1,412 @@ +import hashlib +import json +import unicodedata + +PACKET_LEN = 17133 +PACKET_SHA = "1430c77dbc3df0dadd3a752b1a5bb97e5c358ee5ad126f538f09ab98ce833790" +MANIFEST_LEN = 276765 +MANIFEST_SHA = "33663d1141bbabcc170763d9751a12c4c8d0eef4c80e8ce4c43dacf42983491f" +SOURCE_PATH = "future/d007_executor_v3/producer.py" +LOCK_PATHS = ("decisions/D-007-PACKET.md", "decisions/d007-world-merge-manifest-v1.json") +FRESH = ("execution_id", "activation_id", "validation_id", "invocation_nonce", "freshness_ref") +LAW_NAMES = ("WHOLE_OBSERVATION_CLOSURE", "CLOSED_IDENTITY_AND_SCOPE", "D003_AUTHORITY_FIRST", "BASE_READ_POLICY_FRESHNESS", "LINEAGE_AND_MATERIALIZED_RELATION", "CONFLICT_SOUNDNESS", "GLOBAL_FACT_IMMUTABILITY", "EXACT_ATOMIC_PUBLICATION", "FAULT_RELOAD_IDEMPOTENCY", "PROMOTION_SEPARATION", "CANONICAL_DETERMINISM") + + +def reject_number(text): + assert False + + +def unique_object(items): + made = {} + for key, value in items: + assert key not in made + made[key] = value + return made + + +def inspect_tree(value): + if isinstance(value, dict): + for key, item in value.items(): + assert isinstance(key, str) and key and unicodedata.normalize("NFC", key) == key + for char in key: + assert not 55296 <= int(char.encode("unicode_escape").decode("ascii").replace("\\u", ""), 16) <= 57343 if char.startswith("\\u") else True + inspect_tree(item) + elif isinstance(value, list): + for item in value: + inspect_tree(item) + elif isinstance(value, str): + assert unicodedata.normalize("NFC", value) == value + for char in value: + assert not 55296 <= int(char.encode("unicode_escape").decode("ascii").replace("\\u", ""), 16) <= 57343 if char.startswith("\\u") else True + elif isinstance(value, bool): + assert value is True or value is False + elif isinstance(value, int): + assert 0 <= value <= 18446744073709551615 + else: + assert value is None + + +def encode_value(value): + return (json.dumps(value, ensure_ascii=False, separators=(",", ":"), sort_keys=True) + "\n").encode("utf-8") + + +def read_value(raw): + assert isinstance(raw, bytes) and raw.endswith(b"\n") + value = json.loads(raw.decode("utf-8"), object_pairs_hook=unique_object, parse_float=reject_number, parse_constant=reject_number) + inspect_tree(value) + assert encode_value(value) == raw + return value + + +def copy_value(value): + return json.loads(json.dumps(value, ensure_ascii=False)) + + +def hex_of(raw): + return hashlib.sha256(raw).hexdigest() + + +def value_hex(value): + return hex_of(encode_value(value)) + + +def artifact(kind, raw): + return {"byte_length": len(raw), "kind": kind, "sha256": hex_of(raw)} + + +def exact_keys(value, keys): + return isinstance(value, dict) and set(value.keys()) == set(keys) + + +def text_value(value): + return isinstance(value, str) and len(value) > 0 and unicodedata.normalize("NFC", value) == value + + +def digest_value(value): + return isinstance(value, str) and len(value) == 64 and all(char in "0123456789abcdef" for char in value) + + +def stable_form(item): + changed = copy_value(item) + for field in FRESH: + changed["bound_handoff"][field] = "" + return changed + + +def stable_hex(item): + return value_hex(stable_form(item)) + + +def committed_result(item): + return {"candidate_world_id": item["materialized_result"]["world_id"], "operation_id": item["operation_id"], "proposal_id": item["proposal_id"], "status": "COMMITTED"} + + +def retained_item(item): + return {"expires_at": item["expires_at"], "operation_id": item["operation_id"], "outcome": committed_result(item), "proposal": copy_value(item), "proposal_sha256": stable_hex(item), "retention_epoch": item["retention_epoch"]} + + +def publication(item): + result = item["materialized_result"] + return {"candidate_world": {"generation": result["generation"], "kind": "CANDIDATE", "parent_world_ids": list(item["source_world_ids"]), "root": result["root"], "world_id": result["world_id"]}, "canonical_update": None, "global_operations": [], "promotion_operations": [], "retained_proposal_insert": retained_item(item), "revision_occurrence_inserts": copy_value(item["proposed_revisions"]), "slot_binding_inserts": [{"actor_id": row["actor_id"], "revision_id": row["revision_id"], "typed_state_slot_ref": row["typed_state_slot_ref"], "world_id": result["world_id"]} for row in item["output_bindings"]]} + + +def world_table(state): + return {row["world_id"]: row for row in state["worlds"]} + + +def slot_table(state, world): + return {(row["actor_id"], row["typed_state_slot_ref"]): row["revision_id"] for row in state["slot_bindings"] if row["world_id"] == world} + + +def reachable(worlds, start): + found = set() + pending = [start] + while pending: + current = pending.pop() + if current not in found: + found.add(current) + pending += worlds[current]["parent_world_ids"] + return found + + +def maximal_shared(state, left, right): + worlds = world_table(state) + shared = reachable(worlds, left) & reachable(worlds, right) + answer = [] + for candidate in shared: + shadowed = False + for other in shared: + if candidate != other and candidate in reachable(worlds, other): + shadowed = True + if not shadowed: + answer.append(candidate) + return sorted(answer) + + +def authority_ok(state, item, handoff): + current = state["authority_context"] + fixed = ("actor_id", "audience", "commitment_id", "generation", "presentation_ref", "right", "selection_id", "target_candidate_world_id", "world_id") + if handoff["verdict"] != "ALLOW" or any(handoff[key] != item["bound_handoff"][key] for key in handoff if key not in FRESH): + return False + if any(handoff[key] != current[key] for key in fixed): + return False + joins = (("current_activation_id", "activation_id"), ("current_actor_id", "actor_id"), ("current_commitment_id", "commitment_id"), ("current_execution_id", "execution_id"), ("current_generation", "generation"), ("current_selection_id", "selection_id"), ("current_world_id", "world_id")) + return all(handoff[a] == current[b] for a, b in joins) and handoff["current_grant_ref"] == current["grant_ref"] and handoff["selected_grant_id"] == current["grant_ref"] and handoff["target_candidate_world_id"] == item["target"]["world_id"] + + +def context_ok(state, item): + policy = None + for row in state["policies"]: + if row["policy_id"] == item["bound_policy"]["policy_id"]: + policy = row + return policy == item["bound_policy"] and sorted(state["invariants"], key=lambda row: row["invariant_id"]) == sorted(item["bound_invariants"], key=lambda row: row["invariant_id"]) and sorted(state["predicates"], key=lambda row: row["predicate_id"]) == sorted(item["bound_predicates"], key=lambda row: row["predicate_id"]) + + +def ancestry_ok(state, item): + old = {row["revision_id"]: row for row in state["revision_occurrences"]} + new = {row["revision_id"]: row for row in item["proposed_revisions"]} + if len(new) != len(item["proposed_revisions"]) or set(old.keys()) & set(new.keys()): + return False + old_ops = {row["operation_id"] for row in state["revision_occurrences"]} + old_attempts = {row["attempt_id"] for row in state["revision_occurrences"]} + attempts = [row["attempt_id"] for row in item["proposed_revisions"]] + if item["operation_id"] in old_ops or len(attempts) != len(set(attempts)) or any(value in old_attempts for value in attempts): + return False + if item["operation_id"] in {row["operation_id"] for row in state["retained_proposals"] + state["proposal_tombstones"]}: + return False + joined = old.copy() + joined.update(new) + for row in item["proposed_revisions"]: + if row["operation_id"] != item["operation_id"] or len(row["predecessor_revision_ids"]) != 1 or row["commitment_id"] != item["bound_handoff"]["commitment_id"]: + return False + parent = joined.get(row["predecessor_revision_ids"][0]) + if parent is None or (parent["actor_id"], parent["typed_state_slot_ref"]) != (row["actor_id"], row["typed_state_slot_ref"]) or any(value not in joined for value in row["causal_revision_ids"]): + return False + seen = set() + cursor = row["revision_id"] + while joined[cursor]["predecessor_revision_ids"]: + if cursor in seen: + return False + seen.add(cursor) + cursor = joined[cursor]["predecessor_revision_ids"][0] + if cursor not in old: + return False + scopes = {(row["actor_id"], row["typed_state_slot_ref"]): row["schema_id"] for row in item["owner_scoped_slots"]} + for binding in item["output_bindings"]: + row = joined.get(binding["revision_id"]) + key = (binding["actor_id"], binding["typed_state_slot_ref"]) + if row is None or (row["actor_id"], row["typed_state_slot_ref"], row["schema_id"], row["commitment_id"]) != (key[0], key[1], scopes[key], item["bound_handoff"]["commitment_id"]): + return False + return True + + +def structural_ok(state, item): + base = slot_table(state, item["base"]["world_id"]) + left = slot_table(state, item["source_world_ids"][0]) + right = slot_table(state, item["source_world_ids"][1]) + output = {(row["actor_id"], row["typed_state_slot_ref"]): row["revision_id"] for row in item["output_bindings"]} + for scope in item["owner_scoped_slots"]: + key = (scope["actor_id"], scope["typed_state_slot_ref"]) + b = base.get(key) + l = left.get(key) + r = right.get(key) + if l == r: + wanted = l + elif l == b: + wanted = r + elif r == b: + wanted = l + else: + return False + if output.get(key) != wanted: + return False + return True + + +def service_ok(state, item): + if len(state["service_records"]) != 1 or len(item["owner_scoped_slots"]) != 1: + return False + scope = item["owner_scoped_slots"][0] + key = (scope["actor_id"], scope["typed_state_slot_ref"]) + base = slot_table(state, item["base"]["world_id"]).get(key) + left = slot_table(state, item["source_world_ids"][0]).get(key) + right = slot_table(state, item["source_world_ids"][1]).get(key) + output = {(row["actor_id"], row["typed_state_slot_ref"]): row["revision_id"] for row in item["output_bindings"]}.get(key) + record = state["service_records"][0] + expected = {"base_revision_id": base, "invariant_digests": [row["digest"] for row in item["bound_invariants"]], "left_revision_id": left, "output_revision_id": output, "policy_root": item["bound_policy"]["root"], "predicate_digests": [row["digest"] for row in item["bound_predicates"]], "relation_id": item["relation_id"], "right_revision_id": right, "service_record_id": record["service_record_id"]} + selected = None + for row in item["proposed_revisions"]: + if row["revision_id"] == output: + selected = row + return record == expected and selected is not None and selected["predecessor_revision_ids"] == [left] and right in selected["causal_revision_ids"] and (selected["actor_id"], selected["typed_state_slot_ref"], selected["schema_id"]) == (scope["actor_id"], scope["typed_state_slot_ref"], scope["schema_id"]) + + +def decide_status(state, item, handoff, now, attempted): + if not authority_ok(state, item, handoff): + return "DENY_AUTHORITY", None + binding = stable_hex(item) + matches = [row for row in state["retained_proposals"] + state["proposal_tombstones"] if row["operation_id"] == item["operation_id"]] + if matches: + row = matches[0] + stored = row.get("proposal_sha256", row.get("prior_proposal_sha256")) + if stored != binding: + return "BINDING_MISMATCH", None + if "prior_outcome" in row or now >= row["expires_at"]: + return "RECONCILE_EXPIRED", item["materialized_result"]["world_id"] + return "REPLAY", item["materialized_result"]["world_id"] + if now < item["retention_epoch"] or now >= item["expires_at"]: + return "OUTSIDE_RETENTION", None + worlds = world_table(state) + target = worlds.get(item["target"]["world_id"]) + if target is None or (target["root"], target["generation"], target["kind"]) != (item["target"]["root"], item["target"]["generation"], "CANDIDATE") or any(worlds.get(value, {}).get("kind") != "BRANCH" for value in item["source_world_ids"]): + return "STALE_BASE", None + bases = maximal_shared(state, item["source_world_ids"][0], item["source_world_ids"][1]) + if len(bases) != 1: + return "AMBIGUOUS_BASE", None + chosen = worlds[bases[0]] + if bases[0] != item["declared_common_base_world_id"] or item["base"] != {"generation": chosen["generation"], "root": chosen["root"], "world_id": chosen["world_id"]}: + return "STALE_BASE", None + if not exact_keys(attempted, ("candidate_world", "canonical_update", "global_operations", "promotion_operations", "retained_proposal_insert", "revision_occurrence_inserts", "slot_binding_inserts")): + return "INVALID_DELTA", None + if attempted["global_operations"] or attempted["canonical_update"] is not None or attempted["promotion_operations"]: + return "GLOBAL_FACT_VIOLATION", None + if not context_ok(state, item): + return "STALE_CONTEXT", None + if not ancestry_ok(state, item): + return "INVALID_LINEAGE", None + if item["mode_id"] == "NO_GENERIC_MERGE": + return "UNSUPPORTED", None + if item["mode_id"] == "STRUCTURAL_THREE_WAY" and not structural_ok(state, item): + return "CONFLICT", None + if item["mode_id"] == "TYPED_SERVICE_RELATION" and not service_ok(state, item): + return "INVALID_SERVICE_RELATION", None + result = item["materialized_result"] + sources = [worlds[value] for value in item["source_world_ids"]] + if result["world_id"] in worlds or result["root"] in {row["root"] for row in state["worlds"]} or result["generation"] != max(row["generation"] for row in sources) + 1 or attempted != publication(item): + return "INVALID_DELTA", None + return "COMMITTED", result["world_id"] + + +def install(state, delta): + changed = copy_value(state) + changed["worlds"].append(copy_value(delta["candidate_world"])) + changed["revision_occurrences"] += copy_value(delta["revision_occurrence_inserts"]) + changed["slot_bindings"] += copy_value(delta["slot_binding_inserts"]) + changed["retained_proposals"].append(copy_value(delta["retained_proposal_insert"])) + return changed + + +def recovery_key(item, handoff, state): + fields = ("actor_id", "audience", "commitment_id", "current_activation_id", "current_execution_id", "current_grant_ref", "current_selection_id", "current_world_id", "generation", "presentation_ref", "right", "selection_id", "target_candidate_world_id", "world_id") + return {"authority": [handoff[key] for key in fields], "proposal_sha256": stable_hex(item), "state_sha256": value_hex(state), "target": [item["target"][key] for key in ("world_id", "root", "generation")]} + + +def execute_input(raw_input): + state = copy_value(raw_input["initial_state"]) + trace = [] + rights = {} + seen = {field: set() for field in FRESH} + epoch = 0 + for phase in raw_input["phases"]: + before = copy_value(state) + if phase["kind"] == "RELOAD": + state = read_value(encode_value(state)) + epoch += 1 + for right in rights.values(): + right["epoch"] = epoch + trace.append({"classification": None, "delta": None, "durable_after": copy_value(state), "durable_before": before, "kind": "RELOAD", "reload_bytes_sha256": value_hex(before), "response_visible": False}) + continue + item = phase["proposal"] + handoff = phase["handoff"] + operation = item["operation_id"] + for field in FRESH: + assert handoff[field] not in seen[field] + seen[field].add(handoff[field]) + attempted = phase["attempted_delta"] if phase["kind"] == "SUBMIT" else publication(item) + key = recovery_key(item, handoff, state) + if phase["kind"] == "RETRY": + right = rights.get(operation) + assert right is not None and right["epoch"] == epoch + assert right["authority"] == key["authority"] and right["proposal_sha256"] == key["proposal_sha256"] and right["target"] == key["target"] and right["state_sha256"] == value_hex(state) + status, candidate = decide_status(state, item, handoff, phase["logical_now"], attempted) + classification = {"candidate_world_id": candidate, "operation_id": operation, "proposal_id": item["proposal_id"], "status": status} + fault = phase.get("fault") + if phase["kind"] == "SUBMIT" and fault in ("BEFORE_COMMIT", "AFTER_COMMIT_BEFORE_RESPONSE"): + assert operation not in rights + key["epoch"] = None + rights[operation] = key + delta = publication(item) if status == "COMMITTED" else None + if phase["kind"] == "SUBMIT" and fault == "BEFORE_COMMIT": + delta = None + if delta is not None: + state = install(state, delta) + if phase["kind"] == "SUBMIT" and fault == "AFTER_COMMIT_BEFORE_RESPONSE": + rights[operation]["state_sha256"] = value_hex(state) + visible = not (phase["kind"] == "SUBMIT" and fault in ("BEFORE_COMMIT", "AFTER_COMMIT_BEFORE_RESPONSE")) + trace.append({"classification": classification, "delta": copy_value(delta), "durable_after": copy_value(state), "durable_before": before, "kind": phase["kind"], "reload_bytes_sha256": None, "response_visible": visible}) + if phase["kind"] == "RETRY": + rights.pop(operation) + return {"final_state": state, "input_sha256": value_hex(raw_input), "phases": trace} + + +def trace_laws(raw_input, observation): + phase_pairs = list(zip(raw_input["phases"], observation["phases"])) + whole = len(phase_pairs) == len(raw_input["phases"]) and observation["input_sha256"] == value_hex(raw_input) and (not observation["phases"] or observation["final_state"] == observation["phases"][-1]["durable_after"]) + closed = whole and all(exact_keys(row, ("classification", "delta", "durable_after", "durable_before", "kind", "reload_bytes_sha256", "response_visible")) for row in observation["phases"]) + authority = all(phase["kind"] == "RELOAD" or row["classification"]["status"] == decide_status(row["durable_before"], phase["proposal"], phase["handoff"], phase["logical_now"], phase["attempted_delta"] if phase["kind"] == "SUBMIT" else publication(phase["proposal"]))[0] for phase, row in phase_pairs) + freshness = all(phase["kind"] == "RELOAD" or context_ok(row["durable_before"], phase["proposal"]) or row["classification"]["status"] in ("DENY_AUTHORITY", "BINDING_MISMATCH", "REPLAY", "RECONCILE_EXPIRED", "STALE_BASE", "AMBIGUOUS_BASE", "GLOBAL_FACT_VIOLATION", "STALE_CONTEXT") for phase, row in phase_pairs) + lineage = all(phase["kind"] == "RELOAD" or ancestry_ok(row["durable_before"], phase["proposal"]) or row["classification"]["status"] in ("DENY_AUTHORITY", "BINDING_MISMATCH", "REPLAY", "RECONCILE_EXPIRED", "STALE_BASE", "AMBIGUOUS_BASE", "GLOBAL_FACT_VIOLATION", "STALE_CONTEXT", "INVALID_LINEAGE") for phase, row in phase_pairs) + conflict = all(phase["kind"] == "RELOAD" or phase["proposal"]["mode_id"] != "STRUCTURAL_THREE_WAY" or structural_ok(row["durable_before"], phase["proposal"]) or row["classification"]["status"] == "CONFLICT" for phase, row in phase_pairs) + globals_fixed = all(row["durable_before"]["globals"] == row["durable_after"]["globals"] for row in observation["phases"]) + atomic = all(phase["kind"] == "RELOAD" or row["delta"] is None and row["durable_before"] == row["durable_after"] or row["delta"] == publication(phase["proposal"]) and row["durable_after"] == install(row["durable_before"], row["delta"]) for phase, row in phase_pairs) + fault = whole and all(phase["kind"] != "RELOAD" or row["durable_before"] == row["durable_after"] and row["reload_bytes_sha256"] == value_hex(row["durable_before"]) for phase, row in phase_pairs) + separation = all(row["durable_before"]["canonical"] == row["durable_after"]["canonical"] and row["durable_before"]["promotion_records"] == row["durable_after"]["promotion_records"] for row in observation["phases"]) + deterministic = read_value(encode_value(raw_input)) == raw_input and observation["input_sha256"] == value_hex(raw_input) + return dict(zip(LAW_NAMES, (whole, closed, authority, freshness, lineage, conflict, globals_fixed, atomic, fault, separation, deterministic))) + + +def validate_manifest_header(manifest): + assert exact_keys(manifest, ("algebra", "boundary", "budget", "canonicalization", "cases", "decision", "evidence", "projections", "python_grammar", "resolution", "schema", "stage")) + assert manifest["schema"] == "D007-MANIFEST-3" and manifest["stage"] == "PREREGISTERED_V3" and manifest["python_grammar"] == "3.13" + assert manifest["decision"] == {"frozen": False, "id": "D-007", "status": "OPEN", "v1": "INVALIDATED_PRE_FREEZE_PRE_EVIDENCE"} + assert manifest["algebra"] == {"id": "WORLD_PROPOSAL_ALGEBRA", "representation_neutral": True} + assert manifest["budget"]["cell_count"] == 30 and manifest["budget"]["family_count"] == 10 and manifest["budget"]["subcase_count"] == 30 and manifest["budget"]["projection_count"] == 3 and manifest["budget"]["review_gate_count"] == 4 + assert manifest["evidence"]["producer_interface"] == "produce(packet_bytes, manifest_bytes, identity_bytes)" and manifest["evidence"]["producer_sources"] == {"CHECKER": "future/d007_checker_v3/producer.py", "EXECUTOR": SOURCE_PATH} + assert [row["id"] for row in manifest["projections"]["options"]] == ["NO_GENERIC_MERGE", "STRUCTURAL_THREE_WAY", "TYPED_SERVICE_RELATION"] and all(row["rank"] is None for row in manifest["projections"]["options"]) + assert len(manifest["cases"]) == 10 and sum(len(row["subcases"]) for row in manifest["cases"]) == 30 + + +def validate_identity_input(identity, packet_bytes, manifest_bytes): + assert exact_keys(identity, ("clean", "closure_sha256", "dependencies", "locks", "role", "root", "schema", "sources")) + assert identity["schema"] == "D007-IDENTITY-3" and identity["role"] == "EXECUTOR" and identity["root"] == "future/d007_executor_v3" and identity["dependencies"] == [] + # The recursive outer identity validator authenticates this source record against source bytes. + assert isinstance(identity["sources"], list) and len(identity["sources"]) == 1 + source = identity["sources"][0] + assert exact_keys(source, ("byte_length", "path", "sha256")) and source["path"] == SOURCE_PATH and isinstance(source["byte_length"], int) and not isinstance(source["byte_length"], bool) and source["byte_length"] > 0 and digest_value(source["sha256"]) + wanted_locks = [{"byte_length": len(packet_bytes), "path": LOCK_PATHS[0], "sha256": hex_of(packet_bytes)}, {"byte_length": len(manifest_bytes), "path": LOCK_PATHS[1], "sha256": hex_of(manifest_bytes)}] + assert identity["locks"] == wanted_locks + clean = identity["clean"] + assert exact_keys(clean, ("head_commit", "index_tree", "modified_paths", "tracked", "untracked_paths", "worktree_tree")) and clean["modified_paths"] == [] and clean["untracked_paths"] == [] and clean["index_tree"] == clean["worktree_tree"] + assert all(isinstance(clean[key], str) and len(clean[key]) == 40 and all(char in "0123456789abcdef" for char in clean[key]) for key in ("head_commit", "index_tree", "worktree_tree")) + inventory = sorted([source] + wanted_locks, key=lambda row: row["path"]) + assert clean["tracked"] == inventory and identity["closure_sha256"] == value_hex(inventory) + + +def produce(packet_bytes, manifest_bytes, identity_bytes): + assert isinstance(packet_bytes, bytes) and len(packet_bytes) == PACKET_LEN and hex_of(packet_bytes) == PACKET_SHA + assert isinstance(manifest_bytes, bytes) and len(manifest_bytes) == MANIFEST_LEN and hex_of(manifest_bytes) == MANIFEST_SHA + manifest = read_value(manifest_bytes) + identity = read_value(identity_bytes) + validate_manifest_header(manifest) + validate_identity_input(identity, packet_bytes, manifest_bytes) + cells = [] + assertions = [] + for family in manifest["cases"]: + for subcase in family["subcases"]: + observation = execute_input(subcase["inputs"]) + laws = trace_laws(subcase["inputs"], observation) + assert all(laws.values()) + cells.append({"case_id": family["id"], "observation": observation, "subcase_id": subcase["id"]}) + assertions.append({"case_id": family["id"], "laws": laws, "pass": True, "subcase_id": subcase["id"]}) + output = {"assertions": assertions, "cells": cells, "claims": list(LAW_NAMES), "manifest": artifact("MANIFEST", manifest_bytes), "producer_identity": artifact("EXECUTOR_IDENTITY", identity_bytes), "producer_role": "EXECUTOR", "schema": "D007-O-3"} + return encode_value(output)