#!/usr/bin/env python3 """FIX7 P0 implementation-planning packet validator (fail-closed).

Reads the design JSON files in this directory and enforces the planning gates. Exit 0 => every planning gate passes AND implementation execution + production mutation remain blocked (paper-only state intact). Exit 1 => a fail-closed condition fired (gate violated / overclaim / fabrication). Exit 2 => a required file is missing or unreadable.

This validator authorizes nothing. It only proves the planning packet is internally consistent and that nothing here claims execution-readiness. """ import json import os import sys

HERE = os.path.dirname(os.path.abspath(file))

Official Codex seal chain (consumed read-only from the verified seal packet

tree 3890cd34..a234). A mismatch in seal-consumption.json must fail closed.

EXPECTED_N7 = "efb0c5747ae5f56c0e8b5d99c35438a3d6231253570e6ab4d2658ab9e1853d32" EXPECTED_N8 = "daa70c39a91a64696fb1a24ca2d3f620d04049583c9047689902ebf26117e1a1" EXPECTED_P7 = "9ddb27c35a06ca11ee616e3b0399c705d0d97f5b8f284c12045c396f7c034550" EXPECTED_OWNER_DECISION = ( "OPT3_AUTHORIZE_CODEX_AUTHORITY_SEAL_AND_POST_SEAL_IMPLEMENTATION_PLANNING_ONLY" ) FORBIDDEN_SURFACES = ["PG", "Directus", "registry-row", "system_issues", "production"]

def _fail(reasons, code): reasons.append(code)

---- gate functions: each returns a list of fail-codes (empty == pass) ----

def check_seal(seal): r = [] if not seal.get("p7_present", False): _fail(r, "MISSING_P7") if seal.get("p7_digest") != EXPECTED_P7: _fail(r, "P7_DIGEST_MISMATCH") if seal.get("n7_digest") != EXPECTED_N7: _fail(r, "N7_DIGEST_MISMATCH") if seal.get("n8_digest") != EXPECTED_N8: _fail(r, "N8_DIGEST_MISMATCH") if seal.get("implementation_execution_allowed_now") is not False: _fail(r, "IMPLEMENTATION_EXECUTION_CLAIMED_NOW") if seal.get("production_mutation_allowed_now") is not False: _fail(r, "PRODUCTION_MUTATION_ALLOWED_NOW") if seal.get("owner_decision") != EXPECTED_OWNER_DECISION: _fail(r, "OWNER_DECISION_NOT_OPT3") if seal.get("implementation_authorized_by_p7_alone") is not False: _fail(r, "P7_ALONE_AUTHORIZES_IMPL") return r

def check_mutation_inventory(inv): r = [] surfaces = inv.get("surfaces") if not isinstance(surfaces, list) or not surfaces: _fail(r, "MUTATION_INVENTORY_EMPTY") return r required_keys = { "surface", "classification", "mutation_class", "proposed_mutation", "reason", "prerequisite", "dryrun_method", "rollback_method", "evidence_required", "allowed_now", } inventoried = " ".join(s.get("surface", "") for s in surfaces).lower() for s in surfaces: missing = required_keys - set(s.keys()) if missing: _fail(r, "SURFACE_FIELD_MISSING:" + s.get("surface", "?")) if s.get("allowed_now") is not False: _fail(r, "MUTATION_ALLOWED_NOW:" + s.get("surface", "?")) if not str(s.get("rollback_method", "")).strip(): _fail(r, "SURFACE_NO_ROLLBACK:" + s.get("surface", "?")) if not str(s.get("dryrun_method", "")).strip(): _fail(r, "SURFACE_NO_DRYRUN:" + s.get("surface", "?")) # registry/index mutations must be explicitly classified if ("registry" in s.get("surface", "").lower() or "index" in s.get("surface", "").lower()): if not str(s.get("mutation_class", "")).strip(): _fail(r, "REGISTRY_MUTATION_NOT_CLASSIFIED:" + s.get("surface", "?")) for fs in FORBIDDEN_SURFACES: if fs.lower() not in inventoried: _fail(r, "SURFACE_NOT_INVENTORIED:" + fs) if inv.get("any_surface_allowed_now") is not False: _fail(r, "ANY_SURFACE_ALLOWED_NOW") return r

def check_precondition(pc): r = [] items = pc.get("items") if not isinstance(items, list) or not items: _fail(r, "PRECONDITION_EMPTY") return r any_required_not_pass = False for it in items: if it.get("required"): status = it.get("status") if status not in ("PASS", "FAIL", "UNKNOWN"): _fail(r, "PRECONDITION_BAD_STATUS:" + str(it.get("id"))) if not str(it.get("evidence_path", "")).strip(): _fail(r, "MISSING_PRECONDITION_EVIDENCE:" + str(it.get("id"))) if status != "PASS": any_required_not_pass = True # a non-PASS required item must carry a blocker id if not str(it.get("blocker_id", "")).strip(): _fail(r, "PRECONDITION_NO_BLOCKER:" + str(it.get("id"))) # execution_ready must be False whenever any required item is not PASS if any_required_not_pass and pc.get("execution_ready") is not False: _fail(r, "EXECUTION_READY_OVERCLAIM") if pc.get("execution_ready") is True: # this packet is planning-only; execution-ready must never be asserted here _fail(r, "EXECUTION_READY_CLAIMED_IN_PLANNING") return r

def check_dryrun(dr): r = [] if dr.get("production_target") is not False: _fail(r, "DRYRUN_TARGETS_PRODUCTION") if dr.get("staging_only") is not True: _fail(r, "DRYRUN_NOT_STAGING_ONLY") steps = dr.get("steps") if not isinstance(steps, list) or not steps: _fail(r, "DRYRUN_NO_STEPS") return r for st in steps: if st.get("production") is not False: _fail(r, "DRYRUN_STEP_PRODUCTION:" + str(st.get("id"))) ref = str(st.get("rollback_ref", "")).strip() if not ref: _fail(r, "DRYRUN_CMD_WITHOUT_ROLLBACK:" + str(st.get("id"))) if dr.get("production_mutation_in_dryrun") is not False: _fail(r, "DRYRUN_PRODUCTION_MUTATION") return r

def check_rollback(rb): r = [] status = rb.get("rollback_proof_status") if status == "PROVEN": _fail(r, "ROLLBACK_PROOF_FABRICATED") elif status != "NOT_YET_PROVEN": _fail(r, "ROLLBACK_PROOF_STATUS_INVALID") if not rb.get("snapshot_requirements"): _fail(r, "ROLLBACK_NO_SNAPSHOT_REQ") rbs = rb.get("rollbacks") if not isinstance(rbs, list) or not rbs: _fail(r, "ROLLBACK_NO_ENTRIES") return r for e in rbs: if not str(e.get("restore_path", "")).strip(): _fail(r, "ROLLBACK_ENTRY_NO_RESTORE:" + str(e.get("id"))) if not str(e.get("verification_after_rollback", "")).strip(): _fail(r, "ROLLBACK_ENTRY_NO_VERIFY:" + str(e.get("id"))) return r

def check_owner(ow): r = [] if ow.get("default_decision") != "HOLD": _fail(r, "DEFAULT_NOT_HOLD") if ow.get("execution_authorization_status") != "NOT_AUTHORIZED": _fail(r, "EXECUTION_AUTHORIZED_IN_TEMPLATE") sig = str(ow.get("owner_signature", "")).strip().upper() selected = ow.get("selected_option") # paper-only: owner has not chosen; a non-null selection without a real # signature is a fabricated approval and must fail closed. if selected not in (None, "", "HOLD"): if sig in ("", "UNSIGNED"): _fail(r, "OWNER_APPROVAL_FABRICATED") return r

---- driver ----

FILES = { "seal": "seal-consumption.json", "inv": "mutation-inventory.json", "pc": "precondition-checklist.json", "dr": "dryrun-design.json", "rb": "rollback-recovery-design.json", "ow": "owner-execution-decision-template.json", }

CHECKS = [ ("seal-consumption", "seal", check_seal), ("mutation-inventory", "inv", check_mutation_inventory), ("precondition-checklist", "pc", check_precondition), ("dryrun-design", "dr", check_dryrun), ("rollback-recovery", "rb", check_rollback), ("owner-decision-template", "ow", check_owner), ]

def load_all(base=HERE): data = {} for key, fn in FILES.items(): path = os.path.join(base, fn) if not os.path.exists(path): print("MISSING_FILE:" + fn) sys.exit(2) try: with open(path, "r", encoding="utf-8") as fh: data[key] = json.load(fh) except Exception as exc: # noqa: BLE001 print("UNREADABLE_FILE:" + fn + ":" + str(exc)) sys.exit(2) return data

def run(data): all_fail = [] for label, key, fn in CHECKS: fails = fn(data[key]) status = "PASS" if not fails else "FAIL" print("[%s] %s%s" % (status, label, "" if not fails else " -> " + "; ".join(fails))) all_fail.extend(fails) return all_fail

def main(): data = load_all() fails = run(data) if fails: print("PLANNING_VALIDATOR_RESULT: FAIL (%d fail-closed conditions)" % len(fails)) sys.exit(1) print("PLANNING_VALIDATOR_RESULT: PASS (all planning gates; execution still blocked)") sys.exit(0)

if name == "main": main()