#!/usr/bin/env python3 """Codex FIX7 N7/N8/P7 atomic authority-seal authoring and verification.""" import argparse import copy import datetime as dt import hashlib import importlib.util import json import os from pathlib import Path

ROOT = Path(file).resolve().parent ENCODER_PATH = ROOT / "authority_seal_encoder.py" FINAL_STATUS = "CODEX_FIX7_N7_N8_P7_AUTHORITY_SEAL_AUTHORED_IMPLEMENTATION_PLANNING_ALLOWED" A1 = "OWNER-FIX7-P0-BIRTH-BLUEPRINT-SEAL-2026-06-11-OPT3" A2 = "D D / owner-operator / nmhuyen@gmail.com" A5 = "OPT3_AUTHORIZE_CODEX_AUTHORITY_SEAL_AND_POST_SEAL_IMPLEMENTATION_PLANNING_ONLY" N6 = "d777e87c73d3b62d36789d9343f346102e98dbf301f2c93f7608470b876b258c" N6_CERT = "055828db8303aaaad0ba22adfff54eecc7d31b1fabc90d021be5503bdf746b96" MEMBERSHIP = "f2bda8effc7be19b54722828126b82d7d2d48bee5e5e5dc0c8f347ce210fe251" N2 = "49c386a9b9666c09786fc4f89bc79776b6046eaee6f4da6d8537d2c753b734d0" N3 = "bb9ca03c7b22db5478e7b22496993bdb1a46a2cd51d6f841955b152cd1ee2a49" N4 = "9b111c980ad28a1d1aabed9cbd49e53a05babc3549c7cf28b30a470f09430c67" N5 = "1144b7fb19350656994492764627b465d256dc14ccbba54c47476e684e1cc66c" PACKET_V3 = "b95df0a5d2f41f80bea0cef8621c1f8bb0f6b49a40175116418494ed4141ca6d" INPUT_PACKET_TREE = "ac6793194510dec0adf8f13878cdf6ad716a2ae346ea95cc861deea5715db0d0" N6_PACKET_TREE = "356a0cee2933cffde603c7d4b32e12c14bc6a7024fa3ff5a5a4a0f19282a8b9b" ALIGNMENT_TREE = "96d00b9e570844c6288b1969f180aeb9598a9227dbebd86727376884803e83c1" ENCODER_SHA = "13344f92cafcaf0d07dcb21700bdb642f38b89351702e08080eacb0e957144b8" CANON_DOC = "knowledge/dev/reports/architecture/t1-fix7-existing-system-refactor-execution-blueprint-2026-06-08/canonicalizer-fix7-canon-v1-ssot.md" REPORT_ROOT = "knowledge/dev/reports/architecture/codex-fix7-n7-n8-p7-authority-seal-reauthor-with-valid-a5-2026-06-11" CHECKPOINT = "knowledge/dev/reports/architecture/checkpoint-codex-fix7-n7-n8-p7-authority-seal-reauthor-with-valid-a5-2026-06-11.md" CURRENT_STATE = "knowledge/current-state/reports/fix7-codex-n7-n8-p7-authority-seal-reauthor-with-valid-a5-2026-06-11.md" REPORT_SET = [ ("knowledge/dev/reports/architecture/codex-fix7-n-number-n6-targeted-recheck-2026-06-11/00-readme-first.md", "1"), ("knowledge/dev/reports/architecture/codex-fix7-n-number-n6-targeted-recheck-2026-06-11/reconstruction-and-command-evidence.md", "1"), ("knowledge/dev/reports/architecture/codex-fix7-n-number-n6-targeted-recheck-2026-06-11/adversarial-probes.md", "1"), ("knowledge/dev/reports/architecture/codex-fix7-n-number-n6-targeted-recheck-2026-06-11/authority-inputs-and-blockers.md", "1"), ("knowledge/dev/reports/architecture/checkpoint-codex-fix7-n-number-n6-targeted-recheck-2026-06-11.md", "1"), ("knowledge/current-state/reports/fix7-codex-n-number-n6-targeted-recheck-2026-06-11.md", "1"), ("knowledge/dev/reports/architecture/fix7-codex-n-number-n6-targeted-repair-report-2026-06-11.md", "2"), ("knowledge/dev/reports/architecture/fix7-codex-n-number-n6-targeted-repair-governance-addendum-2026-06-11.md", "2"), ("knowledge/dev/reports/architecture/fix7-alignment-packet-completeness-repair-report-2026-06-11.md", "2"), ("knowledge/dev/laws/tool-kiem-thu/dev/v0.2-hardening/reports/v02-codex-found-loadbearing-file-completeness-repair-2026-06-11.md", "1"), ("knowledge/dev/reports/architecture/fix7-real-n6-provenance-under-tkt-v02-2026-06-11/real-n6-provenance-certificate.json", "1"), ("knowledge/dev/reports/architecture/fix7-real-n6-provenance-under-tkt-v02-2026-06-11/manifest.json", "2"), ("knowledge/dev/reports/architecture/fix7-real-n6-provenance-under-tkt-v02-2026-06-11/authority_firewall.py", "1"), ("knowledge/dev/reports/architecture/fix7-authority-n-node-tkt-v02-alignment-2026-06-11/README_FOR_OWNER_AND_CODEX.md", "2"), ("knowledge/dev/reports/architecture/fix7-authority-n-node-tkt-v02-alignment-2026-06-11/manifest.json", "2"), ] FORBIDDEN_ACTIONS = { "production_mutation", "pg_mutation", "directus_mutation", "registry_row_mutation", "system_issues_mutation", "implementation_execution", "REAL_RUN", "QT001", "permit", "activation", "repoint", "cutover", } OFFICIAL_FILES = { "n7": "n7-approval-event.json", "n8": "n8-detached-codex-seal.json", "p7": "p7-authoritative-pin.json", }

class GateReject(Exception): def init(self, code, detail=""): super().init(f"{code}: {detail}") self.code = code

def load_encoder(): if hashlib.sha256(ENCODER_PATH.read_bytes()).hexdigest() != ENCODER_SHA: raise GateReject("ENCODER_HASH_MISMATCH") spec = importlib.util.spec_from_file_location("fix7_encoder", ENCODER_PATH) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module

def base_config(timestamp_utc, timestamp_local): return { "a1": A1, "a2": A2, "a3": timestamp_utc, "a3_asia_ho_chi_minh": timestamp_local, "a5": A5, "n6": N6, "n6_certificate": N6_CERT, "n6_status": "RATIFIED_ENGINEERING_VERIFIED_CANDIDATE", "n6_is_official_before_seal": False, "n7_envelope_current_blocker": "AUTHORITY_INPUTS_CLOSED_BY_OWNER_OPT3", "report_set": copy.deepcopy(REPORT_SET), "expected_report_set": copy.deepcopy(REPORT_SET), "claimed_report_digest": None, "existing_authority_ids": [], "report_id": f"{REPORT_ROOT}/00-readme-first.md@1", "checkpoint_id": f"{CHECKPOINT}@1", "requested_actions": [], "n7_extra": [], "n8_extra": [], "p7_drop": [], "probe_fixture": False, }

def validate_config(c): for field in ("a1", "a2", "a3", "a5"): if not c.get(field): raise GateReject(f"MISSING_{field.upper()}") probe_a5 = A5 if c["a5"] != A5 and not (c.get("probe_fixture") and c["a5"] == probe_a5): raise GateReject("OWNER_DECISION_NOT_OPT3") if c["n6"] != N6: raise GateReject("STALE_N6_DIGEST") if c["n6_certificate"] != N6_CERT: raise GateReject("N6_CERTIFICATE_MISSING_OR_MISMATCH") if c["n6_status"] != "RATIFIED_ENGINEERING_VERIFIED_CANDIDATE": raise GateReject("N6_STATUS_INVALID") if c["n6_is_official_before_seal"]: raise GateReject("N6_FALSELY_OFFICIAL_BEFORE_SEAL") if c["n7_envelope_current_blocker"] == "SEAL_REAL_N6_NOT_AVAILABLE": raise GateReject("STALE_N7_ENVELOPE_ACTIVE_BLOCKER") attempted = sorted(set(c["requested_actions"]) & FORBIDDEN_ACTIONS) if attempted: raise GateReject("FORBIDDEN_MUTATION_OR_EXECUTION", ",".join(attempted)) if c["report_set"] != c["expected_report_set"]: raise GateReject("N8_REPORT_SET_MISSING_OR_DRIFTED") if c["report_id"] in c["existing_authority_ids"] or c["checkpoint_id"] in c["existing_authority_ids"]: raise GateReject("P7_AUTHORITY_ID_COLLISION")

def build_chain(c): validate_config(c) enc = load_encoder() report_digest = enc.report_documents_digest(c["report_set"]) if c.get("claimed_report_digest") not in (None, report_digest): raise GateReject("N8_REPORT_SET_DIGEST_MISMATCH") n7_pairs = [ ("schema_version", enc.SCHEMA), ("node_id", "N7"), ("membership_sha256", MEMBERSHIP), ("canonicalizer_sha256", N2), ("marker_fence_registry_sha256", N3), ("superseded_boundary_sha256", N4), ("guard_set_sha256", N5), ("active_corpus_sha256", N6), ("approval_event_id", c["a1"]), ("approver_identity", c["a2"]), ("approval_event_timestamp", c["a3"]), ("owner_blueprint_decision", c["a5"]), ("approval_scope", "BLUEPRINT_SEAL_ONLY_NO_IMPLEMENTATION"), ] + c["n7_extra"] n7_prov = {x: "ENGINEERING_VERIFIED_CANDIDATE" for x in ( "active_corpus_sha256", "membership_sha256", "canonicalizer_sha256", "marker_fence_registry_sha256", "superseded_boundary_sha256", "guard_set_sha256")} n7_prov.update({"approval_event_id": "CODEX_AUTHORED", "approver_identity": "AUTHORITY_INPUT", "approval_event_timestamp": "CODEX_AUTHORED", "owner_blueprint_decision": "AUTHORITY_INPUT"}) n7 = enc.encode_real_n7(n7_pairs, n7_prov, real_n6_available=True) n8_pairs = [ ("schema_version", enc.SCHEMA), ("node_id", "N8"), ("canonicalizer_sha256", N2), ("guard_set_sha256", N5), ("active_corpus_sha256", N6), ("envelope_manifest_sha256", n7), ("sealed_by", "CODEX"), ("sealed_at", c["a3"]), ("parent_checkpoint", c["checkpoint_id"]), ("report_documents_digest", report_digest), ("seal_scope", "BLUEPRINT_SEAL_ONLY_NO_IMPLEMENTATION"), ] + c["n8_extra"] n8_prov = {x: "CODEX_AUTHORED" for x in ("sealed_by", "sealed_at", "parent_checkpoint", "report_documents_digest")} n8 = enc.encode_real_n8(n8_pairs, n8_prov, real_upstream=True) p7_pairs = [ ("schema_version", enc.SCHEMA), ("node_id", "P7"), ("pinned_canonicalizer_document_id", CANON_DOC), ("pinned_canonicalizer_revision", "3"), ("pinned_canonicalizer_utf8_bytes", "38756"), ("pinned_canonicalizer_sha256", N2), ("pinned_packet_v3_tree_sha256", PACKET_V3), ("codex_report_document", c["report_id"]), ("codex_checkpoint_document", c["checkpoint_id"]), ("envelope_manifest_sha256", n7), ("detached_seal_sha256", n8), ("approval_event_id", c["a1"]), ("pin_scope", "CANDIDATE_TO_AUTHORITATIVE_PIN_BLUEPRINT_ONLY"), ] p7_pairs = [pair for pair in p7_pairs if pair[0] not in c["p7_drop"]] p7_prov = {x: "CODEX_AUTHORED" for x in ("codex_report_document", "codex_checkpoint_document", "approval_event_id")} p7 = enc.encode_real_p7(p7_pairs, p7_prov, real_upstream=True) return {"n7": n7, "n8": n8, "p7": p7, "report_digest": report_digest, "n7_pairs": n7_pairs, "n8_pairs": n8_pairs, "p7_pairs": p7_pairs}

def write_json(name, data): (ROOT / name).write_text(json.dumps(data, indent=2, ensure_ascii=False) + "\n")

def write_official_artifacts(c, chain): common = {"final_status": FINAL_STATUS, "authoring_timestamp_utc": c["a3"], "authoring_timestamp_asia_ho_chi_minh": c["a3_asia_ho_chi_minh"], "production_mutation": False, "implementation_execution": False} n7 = {**common, "node": "N7", "authority": "OFFICIAL_CODEX_AUTHORED_APPROVAL_EVENT", "approval_event_id": c["a1"], "approver_identity": c["a2"], "approval_event_timestamp": c["a3"], "owner_blueprint_decision": c["a5"], "n_number_table_status": "RATIFIED_FOR_BINDING_USE", "membership_sha256": MEMBERSHIP, "canonicalizer_sha256": N2, "marker_fence_registry_sha256": N3, "superseded_boundary_sha256": N4, "guard_set_sha256": N5, "active_corpus_sha256": N6, "n6_certificate_binding_sha256": N6_CERT, "input_roster_packet_tree_sha256": INPUT_PACKET_TREE, "depends_on": ["N1", "N2", "N3", "N4", "N5", "N6"], "must_not_depend_on": ["N8", "P7"], "envelope_manifest_sha256": chain["n7"]} n8 = {**common, "node": "N8", "authority": "OFFICIAL_CODEX_DETACHED_SEAL", "sealed_by": "CODEX", "sealed_at": c["a3"], "parent_checkpoint": c["checkpoint_id"], "n7_digest": chain["n7"], "n6_digest": N6, "report_documents": [{"document_id": p, "revision": int(r)} for p, r in REPORT_SET], "report_documents_digest": chain["report_digest"], "depends_on": ["N2", "N5", "N6", "N7"], "must_not_depend_on": ["P7"], "detached_seal_payload": chain["n8_pairs"], "detached_seal_sha256": chain["n8"]} p7 = {**common, "node": "P7", "authority": "OFFICIAL_CODEX_AUTHORED_AUTHORITATIVE_PIN", "canonicalizer_revision": 3, "canonicalizer_utf8_bytes": 38756, "canonicalizer_sha256": N2, "packet_v3_tree_sha256": PACKET_V3, "n7_digest": chain["n7"], "n8_digest": chain["n8"], "n6_digest": N6, "codex_report_document": c["report_id"], "codex_checkpoint_document": c["checkpoint_id"], "depends_on": ["N2", "N7", "N8"], "must_not_depend_on": ["P7"], "implementation_authorized_by_p7_alone": False, "implementation_boundary": "IMPLEMENTATION_EXECUTION_REMAINS_BLOCKED; POST_SEAL_IMPLEMENTATION_PLANNING_ONLY_ALLOWED", "authority_seal_pin_sha256": chain["p7"]} write_json(OFFICIAL_FILES["n7"], n7) write_json(OFFICIAL_FILES["n8"], n8) write_json(OFFICIAL_FILES["p7"], p7) for key, data, title in (("n7", n7, "N7 Approval Event"), ("n8", n8, "N8 Detached Codex Seal"), ("p7", p7, "P7 Authoritative Pin")): digest_key = {"n7": "envelope_manifest_sha256", "n8": "detached_seal_sha256", "p7": "authority_seal_pin_sha256"}[key] (ROOT / OFFICIAL_FILES[key].replace(".json", ".md")).write_text( f"# {title}\n\n- Status: {FINAL_STATUS}\n- Authored UTC: {c['a3']}\n" f"- Authored Asia/Ho_Chi_Minh: {c['a3_asia_ho_chi_minh']}\n- Digest: {data[digest_key]}\n" "- Production mutation: NO\n- Implementation execution: NO\n" + ("- P7 alone does not authorize implementation execution.\n" if key == "p7" else ""))

def verify_existing(): enc = load_encoder() n7 = json.loads((ROOT / OFFICIAL_FILES["n7"]).read_text()) n8 = json.loads((ROOT / OFFICIAL_FILES["n8"]).read_text()) p7 = json.loads((ROOT / OFFICIAL_FILES["p7"]).read_text()) c = base_config(n7["approval_event_timestamp"], n7["authoring_timestamp_asia_ho_chi_minh"]) chain = build_chain(c) expected = {"n7": n7["envelope_manifest_sha256"], "n8": n8["detached_seal_sha256"], "p7": p7["authority_seal_pin_sha256"]} actual = {k: chain[k] for k in ("n7", "n8", "p7")} if actual != expected or enc.has_cycle(enc.EDGES): raise GateReject("EXISTING_CHAIN_VERIFY_FAILED", f"{actual} != {expected}") print(f"CHAIN_VERIFY: PASS N7={actual['n7']} N8={actual['n8']} P7={actual['p7']}")

def main(): parser = argparse.ArgumentParser() parser.add_argument("--author", action="store_true") parser.add_argument("--verify", action="store_true") args = parser.parse_args() if args.author: utc = dt.datetime.now(dt.timezone.utc).replace(microsecond=0) local = utc.astimezone(dt.timezone(dt.timedelta(hours=7))) c = base_config(utc.isoformat().replace("+00:00", "Z"), local.isoformat()) chain = build_chain(c) write_official_artifacts(c, chain) print(f"AUTHORED: N7={chain['n7']} N8={chain['n8']} P7={chain['p7']} REPORT_SET={chain['report_digest']}") if args.verify: verify_existing() if not args.author and not args.verify: parser.error("use --author or --verify")

if name == "main": main()