#!/usr/bin/env python3 """Compile a deterministic private review packet from Phase-0 evidence. The command verifies the live inventory and receipt, re-runs the exact source collector from the evidence-bound downstream checkout, proves candidate and coverage closure, and emits structural source/live differences plus one review decision for every layout relation and assignment. It does not select layout authority, mint portable identities, or create an initial revision plan. """ from __future__ import annotations import argparse from collections import Counter, defaultdict import copy import hashlib import json from pathlib import Path, PurePosixPath import subprocess import sys import tempfile from typing import Any, Callable, Mapping, Sequence SCRIPT_PATH = Path(__file__).resolve() REPOSITORY_ROOT = SCRIPT_PATH.parents[4] if str(REPOSITORY_ROOT) not in sys.path: sys.path.insert(0, str(REPOSITORY_ROOT)) from deployments.offline.runtime.reconcile import ( # noqa: E402 layout_bundle_phase0_live_inventory as live_inventory, ) from deployments.offline.runtime.reconcile import ( # noqa: E402 layout_bundle_phase0_source_inventory as source_inventory, ) from deployments.offline.runtime.reconcile import ( # noqa: E402 reconcile_order_layout_orderitem as live_target, ) REVIEW_INPUT_VERSION = "nexuscrm.layout-phase0-review-input/v1" SOURCE_COLLECTOR_PATH = PurePosixPath( "deployments/offline/runtime/reconcile/" "layout_bundle_phase0_source_inventory.py" ) LIVE_COLLECTOR_PATH = PurePosixPath( "deployments/offline/runtime/reconcile/" "layout_bundle_phase0_live_inventory.py" ) SOURCE_KEYS = frozenset( { "schemaVersion", "mode", "status", "scope", "sourceInventories", "componentComparisons", "coverage", "blockers", "deferred", "inventoryDigest", } ) INVENTORY_KEYS = frozenset( { "schemaVersion", "mode", "status", "sourceBinding", "layouts", "assignments", "duplicateLayoutLegacyTuples", "duplicateAssignmentRoutes", "findings", "layoutCandidateRelations", "layoutCandidateGaps", "coverage", "deferred", "inventoryDigest", } ) RECEIPT_KEYS = frozenset( { "schemaVersion", "mode", "generatedAt", "target", "sourceInventoryDigest", "inventory", "collectorSha256", "receiptSha256", } ) ATTESTATION_KEYS = frozenset( { "kind", "slot", "runtimeEnvSha256", "deploySummarySha256", "activeSelectorSha256", "releaseMarkerSha256", "releaseMetadataSha256", "liveNginxSha256", "backendPort", "backendService", "candidateBundle", "downstreamCommit", "upstreamCommit", "officialAppPacksCommit", "finalPackSha256", "expectedInstalledPack", "verifyUsernameSha256", } ) EXPECTED_PACK_KEYS = frozenset( { "name", "category", "manifestFormatVersion", "payloadHash", "payloadHashAlgorithm", "runtimeRelatedListRulesSha256", "sourceRevisionPolicy", } ) INSTALLED_PACK_KEYS = frozenset( { "name", "category", "manifestFormatVersion", "status", "sourceKind", "sourceUriSha256", "sourceRevision", "desiredPayloadHash", "appliedPayloadHash", "payloadHashAlgorithm", "runtimeRelatedListRulesSha256", "lastErrorEmpty", } ) LAYOUT_KEYS = frozenset( { "id", "objectApiName", "layoutName", "type", "isDefault", "isDeleted", "config", "configCanonicalSha256", "configDatabaseTextSha256", "layoutVersion", "isCustomized", "source", "generationVersion", "ownerPack", "createdDate", "lastModifiedDate", "legacyLayoutTuple", } ) ASSIGNMENT_KEYS = frozenset( { "id", "objectApiName", "layoutId", "roleId", "profileId", "recordTypeId", "specificity", "derivedSpecificity", "createdDate", "lastModifiedDate", "createdById", "routeSource", "ownerPack", "legacyRouteTuple", "references", } ) RELATION_KEYS = frozenset( { "legacyLayoutTuple", "relation", "sourceComponents", "activeLiveLayoutIds", "deletedLiveLayoutIds", "unknownDeletionStateLiveLayoutIds", } ) GAP_KEYS = frozenset( { "code", "legacyLayoutTuple", "sourceBusinessKeys", "activeLiveLayoutIds", "deletedLiveLayoutIds", "unknownDeletionStateLiveLayoutIds", } ) COVERAGE_KEYS = frozenset( { "layoutRowCount", "activeLayoutRowCount", "deletedLayoutRowCount", "unknownDeletionStateLayoutRowCount", "assignmentRowCount", "selectedFinalPackLayoutCount", "layoutCandidateRelationCount", "layoutCandidateGapCount", "layoutCandidateRelationCounts", "findingCounts", "layoutCandidateGapCounts", } ) PERSISTENCE_ENVELOPE_FIELDS = frozenset( { "id", "createdDate", "lastModifiedDate", "source", "isCustomized", "layoutVersion", "generationVersion", } ) REFERENCE_KEYS = frozenset( {"targetLayout", "profile", "role", "recordType", "createdBy"} ) class ReviewError(RuntimeError): """The private evidence or its exact source binding is invalid.""" def canonical_bytes(value: object) -> bytes: return live_inventory.canonical_document_bytes(value) def sha256_bytes(value: bytes) -> str: return hashlib.sha256(value).hexdigest() def _object(value: Any, label: str) -> dict[str, Any]: if not isinstance(value, dict): raise ReviewError(f"{label} must be a JSON object") return value def _array(value: Any, label: str) -> list[Any]: if not isinstance(value, list): raise ReviewError(f"{label} must be a JSON array") return value def _string(value: Any, label: str) -> str: if type(value) is not str or not value or value != value.strip(): raise ReviewError(f"{label} must be a non-empty canonical string") return value def _nullable_string(value: Any, label: str) -> str | None: if value is None: return None if type(value) is not str: raise ReviewError(f"{label} must be a string or null") return value def _boolean(value: Any, label: str, *, nullable: bool = False) -> bool | None: if value is None and nullable: return None if type(value) is not bool: raise ReviewError(f"{label} must be a JSON boolean") return value def _integer(value: Any, label: str) -> int: if type(value) is not int: raise ReviewError(f"{label} must be a JSON integer") return value def _sha256(value: Any, label: str, *, prefixed: bool = False) -> str: digest = _string(value, label) if prefixed: if not digest.startswith("sha256:"): raise ReviewError(f"{label} must use the sha256: prefix") digest = digest.removeprefix("sha256:") if len(digest) != 64 or any(character not in "0123456789abcdef" for character in digest): raise ReviewError(f"{label} must be a lowercase SHA-256") return digest def _expect_keys(value: Mapping[str, Any], expected: frozenset[str], label: str) -> None: actual = frozenset(value) if actual != expected: raise ReviewError(f"{label} fields do not match its versioned contract") def _parse_document(raw: bytes, label: str) -> dict[str, Any]: try: value = live_inventory.parse_json_document(raw, label) except Exception as exc: raise ReviewError(f"cannot parse {label}") from exc return _object(value, label) def _verify_embedded_digest( document: Mapping[str, Any], field: str, label: str ) -> str: embedded = _string(document.get(field), f"{label}.{field}") if not embedded.startswith("sha256:") or len(embedded) != 71: raise ReviewError(f"{label}.{field} is not a canonical SHA-256") unsigned = dict(document) unsigned.pop(field, None) actual = "sha256:" + sha256_bytes(canonical_bytes(unsigned)) if actual != embedded: raise ReviewError(f"{label}.{field} does not match its contents") return embedded def _verify_source_document(source_document: dict[str, Any]) -> None: _expect_keys(source_document, SOURCE_KEYS, "source inventory") if ( source_document.get("schemaVersion") != source_inventory.SOURCE_INVENTORY_VERSION or source_document.get("mode") != "read_only_source_inventory" or source_document.get("status") != "complete" ): raise ReviewError("recompiled source inventory is not complete") _verify_embedded_digest(source_document, "inventoryDigest", "source inventory") if _array(source_document.get("blockers"), "source inventory.blockers"): raise ReviewError("recompiled source inventory contains blockers") def _read_private_document(path: Path, label: str) -> tuple[dict[str, Any], bytes, str]: try: _, raw, file_sha256 = live_target.read_bound_regular_file(path, label) except Exception as exc: raise ReviewError(f"cannot read {label}") from exc return _parse_document(raw, label), raw, file_sha256 def recompile_exact_source_inventory( downstream_root: Path, downstream_commit: str, official_root: Path, ) -> dict[str, Any]: """Execute the evidence commit's exact source collector from its checkout.""" script = downstream_root.joinpath(*SOURCE_COLLECTOR_PATH.parts) with tempfile.TemporaryDirectory(prefix="layout-phase0-source-review-") as tmp: output = Path(tmp) / "source-inventory.json" result = subprocess.run( [ sys.executable, str(script), "--downstream-root", str(downstream_root), "--downstream-commit", downstream_commit, "--official-root", str(official_root), "--output", str(output), ], cwd=downstream_root, check=False, capture_output=True, text=True, ) if result.returncode != 0 or not output.is_file() or output.is_symlink(): raise ReviewError("exact source inventory recompilation failed") raw = output.read_bytes() return _parse_document(raw, "recompiled source inventory") def _verify_receipt( receipt: dict[str, Any], inventory: Mapping[str, Any], inventory_raw: bytes, ) -> None: _expect_keys(receipt, RECEIPT_KEYS, "receipt") if ( receipt.get("schemaVersion") != live_inventory.LIVE_INVENTORY_RECEIPT_VERSION or receipt.get("mode") != "read_only" ): raise ReviewError("receipt contract is not supported") _string(receipt.get("generatedAt"), "receipt.generatedAt") _verify_embedded_digest(receipt, "receiptSha256", "receipt") _sha256(receipt.get("collectorSha256"), "receipt.collectorSha256") inventory_binding = _object(receipt.get("inventory"), "receipt.inventory") if frozenset(inventory_binding) != frozenset( {"schemaVersion", "inventoryDigest", "fileSha256"} ): raise ReviewError("receipt.inventory fields are invalid") if ( inventory_binding.get("schemaVersion") != inventory.get("schemaVersion") or inventory_binding.get("inventoryDigest") != inventory.get("inventoryDigest") or inventory_binding.get("fileSha256") != sha256_bytes(inventory_raw) ): raise ReviewError("receipt does not bind the exact inventory") _sha256(inventory_binding.get("fileSha256"), "receipt inventory file digest") _sha256( inventory_binding.get("inventoryDigest"), "receipt inventory semantic digest", prefixed=True, ) if receipt.get("sourceInventoryDigest") != _object( inventory.get("sourceBinding"), "inventory.sourceBinding" ).get("inventoryDigest"): raise ReviewError("receipt does not bind the source inventory") def _verify_target_binding( receipt: Mapping[str, Any], source_document: Mapping[str, Any] ) -> None: target = _object(receipt.get("target"), "receipt.target") if frozenset(target) != frozenset( {"environment", "databaseTargetSha256", "attestation", "installedPack"} ) or target.get("environment") != "staging": raise ReviewError("receipt target is not the staging inventory contract") _sha256(target.get("databaseTargetSha256"), "receipt target database digest") scope = _object(source_document.get("scope"), "source inventory.scope") downstream = _object(scope.get("downstream"), "source scope.downstream") upstream = _object(scope.get("upstream"), "source scope.upstream") official = _object( scope.get("officialAppPacks"), "source scope.officialAppPacks" ) pack_build = _object(scope.get("packBuild"), "source scope.packBuild") attestation = _object(target.get("attestation"), "receipt target.attestation") attestation_keys = frozenset(attestation) allowed_attestation_keys = (ATTESTATION_KEYS, ATTESTATION_KEYS | {"composedCheckouts"}) if attestation_keys not in allowed_attestation_keys: raise ReviewError("receipt target attestation fields are invalid") if attestation.get("kind") != "trusted-active-runtime-v1": raise ReviewError("receipt target is not a trusted active runtime") for field in ( "runtimeEnvSha256", "deploySummarySha256", "activeSelectorSha256", "liveNginxSha256", "verifyUsernameSha256", ): _sha256(attestation.get(field), f"receipt attestation {field}") for field in ("releaseMarkerSha256", "releaseMetadataSha256"): value = attestation.get(field) if value is not None: _sha256(value, f"receipt attestation {field}") _string(attestation.get("slot"), "receipt attestation.slot") _string(attestation.get("backendPort"), "receipt attestation.backendPort") _nullable_string( attestation.get("backendService"), "receipt attestation.backendService" ) candidate_bundle = attestation.get("candidateBundle") if candidate_bundle is not None: candidate_bundle = _object( candidate_bundle, "receipt attestation.candidateBundle" ) if frozenset(candidate_bundle) != frozenset( {"archiveSha256", "manifestSha256"} ): raise ReviewError("receipt candidate bundle fields are invalid") for field in ("archiveSha256", "manifestSha256"): _sha256( candidate_bundle.get(field), f"receipt candidate bundle {field}" ) if "composedCheckouts" in attestation: raise ReviewError( "receipt cannot bind both a candidate bundle and composed checkouts" ) else: composed = _object( attestation.get("composedCheckouts"), "receipt attestation.composedCheckouts", ) if frozenset(composed) != frozenset( { "algorithm", "downstreamMaterializationSha256", "runtimeMaterializationSha256", } ): raise ReviewError("receipt composed checkout fields are invalid") _string(composed.get("algorithm"), "receipt composed checkout algorithm") for field in ( "downstreamMaterializationSha256", "runtimeMaterializationSha256", ): _sha256(composed.get(field), f"receipt composed checkout {field}") expected_commits = ( (attestation.get("downstreamCommit"), downstream.get("commit")), (attestation.get("upstreamCommit"), upstream.get("commit")), (attestation.get("officialAppPacksCommit"), official.get("commit")), (attestation.get("finalPackSha256"), pack_build.get("finalPackTreeSha256")), ) if any(actual != expected for actual, expected in expected_commits): raise ReviewError("live target attestation disagrees with source scope") expected_pack = _object( attestation.get("expectedInstalledPack"), "receipt target expectedInstalledPack", ) _expect_keys(expected_pack, EXPECTED_PACK_KEYS, "expected installed pack") installed_pack = _object( target.get("installedPack"), "receipt target.installedPack" ) _expect_keys(installed_pack, INSTALLED_PACK_KEYS, "installed pack") if ( expected_pack.get("sourceRevisionPolicy") != "exact-current-downstream" or installed_pack.get("status") != "active" or installed_pack.get("sourceKind") != "local_path" or installed_pack.get("lastErrorEmpty") is not True ): raise ReviewError("installed pack is not exact active local-path evidence") for value, label in ( ( expected_pack.get("runtimeRelatedListRulesSha256"), "expected pack related-list rules digest", ), (installed_pack.get("sourceUriSha256"), "installed pack source URI digest"), ( installed_pack.get("runtimeRelatedListRulesSha256"), "installed pack related-list rules digest", ), ): _sha256(value, label) checks = ( (installed_pack.get("name"), expected_pack.get("name")), (installed_pack.get("category"), expected_pack.get("category")), ( installed_pack.get("manifestFormatVersion"), expected_pack.get("manifestFormatVersion"), ), (installed_pack.get("desiredPayloadHash"), expected_pack.get("payloadHash")), (installed_pack.get("appliedPayloadHash"), expected_pack.get("payloadHash")), ( installed_pack.get("payloadHashAlgorithm"), expected_pack.get("payloadHashAlgorithm"), ), ( installed_pack.get("runtimeRelatedListRulesSha256"), expected_pack.get("runtimeRelatedListRulesSha256"), ), (installed_pack.get("sourceRevision"), downstream.get("commit")), ) if any(actual != expected for actual, expected in checks): raise ReviewError("installed pack disagrees with the attested source") def _counter(values: Sequence[Mapping[str, Any]], field: str) -> dict[str, int]: return dict(sorted(Counter(str(item[field]) for item in values).items())) def _duplicate_assignment_routes( assignments: Sequence[Mapping[str, Any]], ) -> list[dict[str, Any]]: groups: dict[bytes, list[str]] = defaultdict(list) tuples: dict[bytes, list[Any]] = {} for assignment in assignments: route = _array(assignment.get("legacyRouteTuple"), "assignment route tuple") key = canonical_bytes(route) groups[key].append(_string(assignment.get("id"), "assignment.id")) tuples[key] = route result = [] for key in sorted(groups): if len(groups[key]) > 1: result.append( { "legacyRouteTuple": tuples[key], "assignmentIds": sorted(groups[key]), } ) return result def _validate_reference( value: Any, *, assignment_id: str, kind: str, source_id: str | None, object_api_name: str, ) -> tuple[dict[str, Any] | None, list[dict[str, Any]]]: label = f"assignment {assignment_id} {kind} reference" if source_id is None: if value is not None: raise ReviewError(f"{label} exists without its source dimension") return None, [] reference = _object(value, label) object_bound = kind in {"target", "record_type"} expected_keys = {"state", "idExact"} if object_bound: expected_keys.update({"objectApiName", "objectMatches"}) if frozenset(reference) != frozenset(expected_keys): raise ReviewError(f"{label} fields are invalid") state = _string(reference.get("state"), f"{label}.state") allowed_states = {"active", "deleted", "missing"} if kind in {"profile", "record_type"}: allowed_states.add("inactive") if state not in allowed_states: raise ReviewError(f"{label}.state is invalid") exact = reference.get("idExact") if state == "missing": if exact is not None: raise ReviewError(f"{label}.idExact must be null when missing") else: _boolean(exact, f"{label}.idExact") findings: list[dict[str, Any]] = [] if state == "missing": findings.append( {"code": f"assignment_{kind}_missing", "assignmentId": assignment_id} ) elif state == "deleted": findings.append( {"code": f"assignment_{kind}_deleted", "assignmentId": assignment_id} ) if exact is False: findings.append( { "code": f"assignment_{kind}_id_case_mismatch", "assignmentId": assignment_id, } ) if object_bound: referenced_object = _nullable_string( reference.get("objectApiName"), f"{label}.objectApiName" ) object_matches = _boolean( reference.get("objectMatches"), f"{label}.objectMatches" ) expected_match = ( referenced_object is not None and referenced_object.casefold() == object_api_name.casefold() ) if object_matches != expected_match: raise ReviewError(f"{label}.objectMatches is not derived from its facts") if referenced_object is not None and not object_matches: findings.append( { "code": f"assignment_{kind}_object_mismatch", "assignmentId": assignment_id, } ) if kind == "profile" and state == "inactive": findings.append( {"code": "assignment_profile_inactive", "assignmentId": assignment_id} ) if kind == "record_type" and state == "inactive": findings.append( { "code": "assignment_record_type_inactive", "assignmentId": assignment_id, } ) return reference, findings def _validate_assignment( assignment: Mapping[str, Any], ) -> list[dict[str, Any]]: assignment_id = _string(assignment.get("id"), "assignment.id") object_api_name = _string( assignment.get("objectApiName"), f"assignment {assignment_id}.objectApiName" ) layout_id = _string( assignment.get("layoutId"), f"assignment {assignment_id}.layoutId" ) role_id = _nullable_string( assignment.get("roleId"), f"assignment {assignment_id}.roleId" ) profile_id = _nullable_string( assignment.get("profileId"), f"assignment {assignment_id}.profileId" ) record_type_id = _nullable_string( assignment.get("recordTypeId"), f"assignment {assignment_id}.recordTypeId" ) created_by_id = _nullable_string( assignment.get("createdById"), f"assignment {assignment_id}.createdById" ) _nullable_string( assignment.get("routeSource"), f"assignment {assignment_id}.routeSource" ) _nullable_string( assignment.get("ownerPack"), f"assignment {assignment_id}.ownerPack" ) for field in ("createdDate", "lastModifiedDate"): value = assignment.get(field) if value is not None: _string(value, f"assignment {assignment_id}.{field}") specificity = _integer( assignment.get("specificity"), f"assignment {assignment_id}.specificity" ) derived_specificity = sum( value is not None for value in (role_id, profile_id, record_type_id) ) if assignment.get("derivedSpecificity") != derived_specificity: raise ReviewError( f"assignment {assignment_id}.derivedSpecificity is not derived from its route" ) expected_route = [object_api_name.strip().casefold(), role_id, profile_id, record_type_id] if assignment.get("legacyRouteTuple") != expected_route: raise ReviewError(f"assignment {assignment_id} legacy route tuple is invalid") findings: list[dict[str, Any]] = [] for dimension, value in ( ("role", role_id), ("profile", profile_id), ("record_type", record_type_id), ): if value == "": findings.append( { "code": "assignment_empty_dimension", "assignmentId": assignment_id, "dimension": dimension, } ) if specificity != derived_specificity: findings.append( { "code": "assignment_specificity_mismatch", "assignmentId": assignment_id, "actual": specificity, "expected": derived_specificity, } ) references = _object( assignment.get("references"), f"assignment {assignment_id}.references" ) if frozenset(references) != REFERENCE_KEYS: raise ReviewError(f"assignment {assignment_id} reference fields are invalid") reference_specs = ( ("target", "targetLayout", layout_id), ("profile", "profile", profile_id), ("role", "role", role_id), ("record_type", "recordType", record_type_id), ("created_by", "createdBy", created_by_id), ) for kind, field, source_id in reference_specs: _, reference_findings = _validate_reference( references.get(field), assignment_id=assignment_id, kind=kind, source_id=source_id, object_api_name=object_api_name, ) findings.extend(reference_findings) return findings def _validate_layout(layout: Mapping[str, Any]) -> None: layout_id = _string(layout.get("id"), "layout.id") object_api_name = _string( layout.get("objectApiName"), f"layout {layout_id}.objectApiName" ) layout_name = _string( layout.get("layoutName"), f"layout {layout_id}.layoutName" ) _nullable_string(layout.get("type"), f"layout {layout_id}.type") _boolean(layout.get("isDefault"), f"layout {layout_id}.isDefault") _boolean( layout.get("isDeleted"), f"layout {layout_id}.isDeleted", nullable=True ) config = _object(layout.get("config"), f"layout {layout_id}.config") _sha256( layout.get("configCanonicalSha256"), f"layout {layout_id}.configCanonicalSha256", ) if layout.get("configCanonicalSha256") != sha256_bytes(canonical_bytes(config)): raise ReviewError(f"layout {layout_id} canonical config digest is invalid") _sha256( layout.get("configDatabaseTextSha256"), f"layout {layout_id}.configDatabaseTextSha256", ) _integer(layout.get("layoutVersion"), f"layout {layout_id}.layoutVersion") _boolean(layout.get("isCustomized"), f"layout {layout_id}.isCustomized") _nullable_string(layout.get("source"), f"layout {layout_id}.source") generation = layout.get("generationVersion") if generation is not None: _integer(generation, f"layout {layout_id}.generationVersion") _nullable_string(layout.get("ownerPack"), f"layout {layout_id}.ownerPack") for field in ("createdDate", "lastModifiedDate"): value = layout.get(field) if value is not None: _string(value, f"layout {layout_id}.{field}") if layout.get("legacyLayoutTuple") != [ object_api_name.strip().casefold(), layout_name.strip().casefold(), ]: raise ReviewError(f"layout {layout_id} legacy tuple is invalid") def _expected_findings( layouts: Sequence[Mapping[str, Any]], assignments: Sequence[Mapping[str, Any]], ) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: duplicate_layouts, duplicate_findings = ( live_inventory.compile_duplicate_layout_tuples(layouts) ) findings = list(duplicate_findings) for assignment in assignments: findings.extend(_validate_assignment(assignment)) duplicate_routes = _duplicate_assignment_routes(assignments) findings.extend( {"code": "duplicate_assignment_route_tuple", **entry} for entry in duplicate_routes ) findings.sort(key=canonical_bytes) return duplicate_layouts, findings def _verify_assignment_target_closure( layouts: Sequence[Mapping[str, Any]], assignments: Sequence[Mapping[str, Any]], ) -> None: layouts_by_folded_id: dict[str, list[Mapping[str, Any]]] = defaultdict(list) for layout in layouts: layouts_by_folded_id[str(layout["id"]).casefold()].append(layout) for assignment in assignments: assignment_id = str(assignment["id"]) layout_id = str(assignment["layoutId"]) matches = layouts_by_folded_id.get(layout_id.casefold(), []) if len(matches) > 1: raise ReviewError( f"assignment {assignment_id} target layout identity is ambiguous" ) reference = _object( _object( assignment.get("references"), f"assignment {assignment_id}.references", ).get("targetLayout"), f"assignment {assignment_id} target layout reference", ) if not matches: expected = { "state": "missing", "idExact": None, "objectApiName": None, "objectMatches": False, } else: layout = matches[0] target_object = str(layout["objectApiName"]) expected = { "state": "deleted" if layout.get("isDeleted") is True else "active", "idExact": layout_id == layout.get("id"), "objectApiName": target_object, "objectMatches": target_object.casefold() == str(assignment["objectApiName"]).casefold(), } if reference != expected: raise ReviewError( f"assignment {assignment_id} target layout reference is not closed over layouts" ) def _verify_inventory_closure( inventory: dict[str, Any], source_document: dict[str, Any] ) -> None: _expect_keys(inventory, INVENTORY_KEYS, "inventory") if ( inventory.get("schemaVersion") != live_inventory.LIVE_INVENTORY_VERSION or inventory.get("mode") != "read_only_consistent_snapshot" or inventory.get("status") != "complete" ): raise ReviewError("inventory contract is not supported or incomplete") _verify_embedded_digest(inventory, "inventoryDigest", "inventory") source_binding = _object( inventory.get("sourceBinding"), "inventory.sourceBinding" ) if frozenset(source_binding) != frozenset({"inventoryDigest", "scope"}): raise ReviewError("inventory source binding fields are invalid") if ( source_binding.get("inventoryDigest") != source_document.get("inventoryDigest") or source_binding.get("scope") != source_document.get("scope") ): raise ReviewError("inventory does not match recompiled source authority") layouts = [ _object(item, f"inventory.layouts[{index}]") for index, item in enumerate(_array(inventory.get("layouts"), "layouts")) ] assignments = [ _object(item, f"inventory.assignments[{index}]") for index, item in enumerate( _array(inventory.get("assignments"), "assignments") ) ] relations = [ _object(item, f"inventory.layoutCandidateRelations[{index}]") for index, item in enumerate( _array( inventory.get("layoutCandidateRelations"), "layoutCandidateRelations", ) ) ] gaps = [ _object(item, f"inventory.layoutCandidateGaps[{index}]") for index, item in enumerate( _array(inventory.get("layoutCandidateGaps"), "layoutCandidateGaps") ) ] for layout in layouts: _expect_keys(layout, LAYOUT_KEYS, "layout row") _validate_layout(layout) for assignment in assignments: _expect_keys(assignment, ASSIGNMENT_KEYS, "assignment row") for relation in relations: _expect_keys(relation, RELATION_KEYS, "layout candidate relation") for gap in gaps: _expect_keys(gap, GAP_KEYS, "layout candidate gap") layout_ids = [_string(item.get("id"), "layout.id") for item in layouts] assignment_ids = [ _string(item.get("id"), "assignment.id") for item in assignments ] if len(layout_ids) != len(set(layout_ids)) or len(assignment_ids) != len( set(assignment_ids) ): raise ReviewError("inventory contains duplicate physical row identities") if layout_ids != sorted(layout_ids) or assignment_ids != sorted(assignment_ids): raise ReviewError("inventory physical rows are not in canonical identity order") _verify_assignment_target_closure(layouts, assignments) selected = source_inventory.final_page_layout_candidates(source_document) expected_relations, expected_gaps = ( live_inventory.compile_layout_candidate_relations(selected, layouts) ) if relations != expected_relations or gaps != expected_gaps: raise ReviewError("layout candidate relation closure does not verify") expected_duplicates, expected_findings = _expected_findings( layouts, assignments ) if inventory.get("duplicateLayoutLegacyTuples") != expected_duplicates: raise ReviewError("duplicate layout tuple closure does not verify") if inventory.get("duplicateAssignmentRoutes") != _duplicate_assignment_routes( assignments ): raise ReviewError("duplicate assignment route closure does not verify") findings = [ _object(item, f"inventory.findings[{index}]") for index, item in enumerate(_array(inventory.get("findings"), "findings")) ] if findings != expected_findings: raise ReviewError("inventory finding closure does not verify") coverage = _object(inventory.get("coverage"), "inventory.coverage") _expect_keys(coverage, COVERAGE_KEYS, "inventory coverage") expected_coverage = { "layoutRowCount": len(layouts), "activeLayoutRowCount": sum(item.get("isDeleted") is False for item in layouts), "deletedLayoutRowCount": sum(item.get("isDeleted") is True for item in layouts), "unknownDeletionStateLayoutRowCount": sum( item.get("isDeleted") is None for item in layouts ), "assignmentRowCount": len(assignments), "selectedFinalPackLayoutCount": len(selected), "layoutCandidateRelationCount": len(relations), "layoutCandidateGapCount": len(gaps), "layoutCandidateRelationCounts": _counter(relations, "relation"), "findingCounts": _counter(findings, "code"), "layoutCandidateGapCounts": _counter(gaps, "code"), } if coverage != expected_coverage: raise ReviewError("inventory coverage does not close over its evidence") def verify_evidence( *, inventory_path: Path, receipt_path: Path, downstream_root: Path, official_root: Path, source_recompiler: Callable[[Path, str, Path], dict[str, Any]] = ( recompile_exact_source_inventory ), ) -> tuple[dict[str, Any], dict[str, Any], dict[str, Any], dict[str, str]]: inventory, inventory_raw, inventory_file_sha256 = _read_private_document( inventory_path, "live inventory" ) receipt, _, receipt_file_sha256 = _read_private_document( receipt_path, "live inventory receipt" ) _verify_receipt(receipt, inventory, inventory_raw) source_binding = _object( inventory.get("sourceBinding"), "inventory.sourceBinding" ) scope = _object(source_binding.get("scope"), "inventory source scope") downstream = _object(scope.get("downstream"), "inventory downstream scope") downstream_commit = source_inventory.require_commit( downstream.get("commit"), "inventory downstream commit" ) try: source_document = source_recompiler( downstream_root, downstream_commit, official_root ) except Exception as exc: if isinstance(exc, ReviewError): raise raise ReviewError("source inventory recompilation failed") from exc _verify_source_document(source_document) if source_document.get("inventoryDigest") != source_binding.get( "inventoryDigest" ): raise ReviewError("recompiled source inventory digest changed") _verify_inventory_closure(inventory, source_document) _verify_target_binding(receipt, source_document) live_collector = source_inventory.read_exact_tracked_file( downstream_root, downstream_commit, LIVE_COLLECTOR_PATH, "evidence-bound live inventory collector", ) if receipt.get("collectorSha256") != sha256_bytes(live_collector): raise ReviewError("receipt collector digest does not match its bound source") return ( inventory, receipt, source_document, { "inventoryFileSha256": inventory_file_sha256, "receiptFileSha256": receipt_file_sha256, }, ) def _pointer(path: Sequence[str | int]) -> str: if not path: return "" return "/" + "/".join( str(part).replace("~", "~0").replace("/", "~1") for part in path ) def _side(*, present: bool, value: Any = None) -> dict[str, Any]: result = {"present": present} if present: result["value"] = value return result def _difference( *, path: tuple[str | int, ...], kind: str, source_present: bool, live_present: bool, source_value: Any = None, live_value: Any = None, ) -> dict[str, Any]: return { "pointer": _pointer(path), "kind": kind, "source": _side(present=source_present, value=source_value), "live": _side(present=live_present, value=live_value), } def structural_diff( source: Any, live: Any, path: tuple[str | int, ...] = () ) -> list[dict[str, Any]]: """Return deterministic RFC-6901-addressed source-to-live differences.""" if type(source) is not type(live): return [ _difference( path=path, kind="replace", source_present=True, live_present=True, source_value=source, live_value=live, ) ] if isinstance(source, dict): result: list[dict[str, Any]] = [] for key in sorted(set(source) | set(live)): child = (*path, key) if key not in source: result.append( _difference( path=child, kind="add", source_present=False, live_present=True, live_value=live[key], ) ) elif key not in live: result.append( _difference( path=child, kind="remove", source_present=True, live_present=False, source_value=source[key], ) ) else: result.extend(structural_diff(source[key], live[key], child)) return result if isinstance(source, list): result = [] shared = min(len(source), len(live)) for index in range(shared): result.extend(structural_diff(source[index], live[index], (*path, index))) for index in range(shared, len(source)): result.append( _difference( path=(*path, index), kind="remove", source_present=True, live_present=False, source_value=source[index], ) ) for index in range(shared, len(live)): result.append( _difference( path=(*path, index), kind="add", source_present=False, live_present=True, live_value=live[index], ) ) return result if source != live: return [ _difference( path=path, kind="replace", source_present=True, live_present=True, source_value=source, live_value=live, ) ] return [] def _content_projection(config: Mapping[str, Any], *, is_default: bool) -> dict[str, Any]: projected = copy.deepcopy(dict(config)) for field in PERSISTENCE_ENVELOPE_FIELDS: projected.pop(field, None) projected["isDefault"] = is_default return projected def _content_decision(relation: str) -> str: if relation == "one_to_one_candidate": return "select_source_live_or_reviewed_content" if relation == "source_only": return "materialize_source_or_exclude_slot" if relation == "live_only": return "retain_live_content_or_retire_residue" if relation == "deleted_live_only": return "resolve_tombstone_only_lineage" if relation == "live_deletion_state_unknown": return "resolve_deletion_state" return "resolve_legacy_tuple_cardinality" def _load_selected_payloads( source_document: Mapping[str, Any], downstream_root: Path ) -> dict[str, dict[str, Any]]: scope = _object(source_document.get("scope"), "source inventory.scope") downstream_commit = _string( _object(scope.get("downstream"), "source scope.downstream").get("commit"), "source downstream commit", ) final_path = source_inventory.safe_relative_path( _object(scope.get("packBuild"), "source scope.packBuild").get( "finalPackPath" ), "source final-pack path", ) selected: dict[str, dict[str, Any]] = {} for component in source_inventory.final_page_layout_candidates(dict(source_document)): business_key = _string(component.get("businessKey"), "source businessKey") if business_key in selected: raise ReviewError("selected source contains a duplicate business key") component_path = source_inventory.safe_relative_path( component.get("path"), f"source {business_key} path" ) relative = PurePosixPath(final_path, component_path) raw = source_inventory.read_exact_tracked_file( downstream_root, downstream_commit, relative, f"selected source payload {business_key}", ) if sha256_bytes(raw) != component.get("payloadRawSha256"): raise ReviewError("selected source payload hash changed") selected[business_key] = _parse_document( raw, f"selected source payload {business_key}" ) return selected def compile_review_packet( *, inventory: Mapping[str, Any], receipt: Mapping[str, Any], source_document: Mapping[str, Any], source_payloads: Mapping[str, Mapping[str, Any]], evidence_files: Mapping[str, str], ) -> dict[str, Any]: layouts = { _string(item.get("id"), "layout.id"): item for item in _array(inventory.get("layouts"), "inventory.layouts") } assignments = [ _object(item, "inventory assignment") for item in _array(inventory.get("assignments"), "inventory.assignments") ] relation_items: list[dict[str, Any]] = [] comparison_count = 0 content_difference_count = 0 for relation_ordinal, raw_relation in enumerate(_array( inventory.get("layoutCandidateRelations"), "layout candidate relations" )): relation = _object(raw_relation, "layout candidate relation") source_components = [ _object(item, "source component") for item in _array(relation.get("sourceComponents"), "source components") ] live_ids = [ *list(relation.get("activeLiveLayoutIds", [])), *list(relation.get("deletedLiveLayoutIds", [])), *list(relation.get("unknownDeletionStateLiveLayoutIds", [])), ] comparisons: list[dict[str, Any]] = [] for component in source_components: business_key = _string(component.get("businessKey"), "source businessKey") source_payload = _object( source_payloads.get(business_key), f"source payload {business_key}" ) source_default = component.get("isDefault") is True source_content = _content_projection( source_payload, is_default=source_default ) for layout_id in live_ids: live_layout = _object(layouts.get(layout_id), f"live layout {layout_id}") live_config = _object(live_layout.get("config"), "live layout config") live_content = _content_projection( live_config, is_default=live_layout.get("isDefault") is True ) differences = structural_diff(source_content, live_content) comparison_count += 1 if differences: content_difference_count += 1 comparisons.append( { "sourceBusinessKey": business_key, "liveLayoutId": layout_id, "liveDeletionState": live_layout.get("isDeleted"), "sourcePayloadSha256": "sha256:" + sha256_bytes(canonical_bytes(source_payload)), "liveConfigSha256": "sha256:" + sha256_bytes(canonical_bytes(live_config)), "sourceContentSha256": "sha256:" + sha256_bytes(canonical_bytes(source_content)), "liveContentSha256": "sha256:" + sha256_bytes(canonical_bytes(live_content)), "differences": differences, } ) live_facts = [] for layout_id in live_ids: layout = _object(layouts.get(layout_id), f"live layout {layout_id}") live_facts.append( { "layoutId": layout_id, "deletionState": layout.get("isDeleted"), "source": layout.get("source"), "isCustomized": layout.get("isCustomized"), "ownerPack": layout.get("ownerPack"), "isDefault": layout.get("isDefault"), "configCanonicalSha256": layout.get("configCanonicalSha256"), } ) relation_name = _string(relation.get("relation"), "relation") required_decisions: list[dict[str, Any]] = [ { "decision": "content_disposition_required", "choice": _content_decision(relation_name), }, { "decision": "registry_identity_required", "evidenceReference": { "layoutRelationOrdinal": relation_ordinal, }, }, ] required_decisions.extend( { "decision": "tombstone_disposition_required", "evidenceReference": { "layoutRelationOrdinal": relation_ordinal, "legacyLayoutId": layout_id, }, } for layout_id in relation.get("deletedLiveLayoutIds", []) ) for layout_id in relation.get("unknownDeletionStateLiveLayoutIds", []): required_decisions.append( { "decision": "deletion_state_required", "evidenceReference": { "layoutRelationOrdinal": relation_ordinal, "legacyLayoutId": layout_id, }, } ) relation_items.append( { "itemKind": "layout_relation", "itemOrdinal": relation_ordinal, "status": "review_required", "legacyLayoutTuple": relation.get("legacyLayoutTuple"), "relation": relation_name, "sourceComponents": copy.deepcopy(source_components), "liveLayouts": live_facts, "structuralComparisons": comparisons, "requiredDecisions": required_decisions, } ) assignment_items = [] for assignment_ordinal, assignment in enumerate(assignments): required_decisions = [ { "decision": "route_disposition_required", "evidenceReference": { "assignmentOrdinal": assignment_ordinal, "legacyAssignmentId": assignment.get("id"), }, }, { "decision": "registry_identity_required", "identityKind": "target_layout_bundle_slot", }, ] if assignment.get("recordTypeId") is not None: required_decisions.append( { "decision": "registry_identity_required", "identityKind": "catalog_record_type_key", } ) if assignment.get("profileId") is not None: required_decisions.append( { "decision": "registry_identity_required", "identityKind": "security_profile_key", } ) if assignment.get("roleId") is not None: required_decisions.append( { "decision": "registry_identity_required", "identityKind": "security_role_key", } ) assignment_items.append( { "itemKind": "layout_assignment", "itemOrdinal": assignment_ordinal, "status": "review_required", "assignmentId": assignment.get("id"), "legacyRouteTuple": assignment.get("legacyRouteTuple"), "objectApiName": assignment.get("objectApiName"), "layoutId": assignment.get("layoutId"), "routeSource": assignment.get("routeSource"), "ownerPack": assignment.get("ownerPack"), "references": copy.deepcopy(assignment.get("references")), "requiredDecisions": required_decisions, } ) findings = _array(inventory.get("findings"), "inventory findings") gaps = _array(inventory.get("layoutCandidateGaps"), "layout candidate gaps") blockers = [ {"source": "inventory_finding", "finding": copy.deepcopy(item)} for item in findings ] blockers.extend( {"source": "layout_candidate_gap", "gap": copy.deepcopy(item)} for item in gaps ) source_binding = _object(inventory.get("sourceBinding"), "source binding") packet: dict[str, Any] = { "schemaVersion": REVIEW_INPUT_VERSION, "mode": "private_review_input", "status": "review_required", "evidence": { "inventoryFileSha256": evidence_files["inventoryFileSha256"], "inventoryDigest": inventory.get("inventoryDigest"), "receiptFileSha256": evidence_files["receiptFileSha256"], "receiptSha256": receipt.get("receiptSha256"), "sourceInventoryDigest": source_binding.get("inventoryDigest"), "collectorSha256": receipt.get("collectorSha256"), "compilerSha256": sha256_bytes(SCRIPT_PATH.read_bytes()), "scope": copy.deepcopy(source_binding.get("scope")), "target": copy.deepcopy(receipt.get("target")), }, "layoutRelations": relation_items, "assignments": assignment_items, "blockers": blockers, "coverage": { "layoutRelationReviewItemCount": len(relation_items), "assignmentReviewItemCount": len(assignment_items), "structuralComparisonCount": comparison_count, "contentDifferenceComparisonCount": content_difference_count, "inventoryFindingCount": len(findings), "layoutCandidateGapCount": len(gaps), "unresolvedDecisionCount": sum( len(item["requiredDecisions"]) for item in [*relation_items, *assignment_items] ), }, "claims": { "authoritySelected": False, "portableIdentityMinted": False, "revisionPlanned": False, "servingBehaviorChanged": False, }, } packet["reviewInputDigest"] = "sha256:" + sha256_bytes(canonical_bytes(packet)) return packet def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--source-root", required=True, type=Path) parser.add_argument("--official-root", required=True, type=Path) parser.add_argument("--inventory", required=True, type=Path) parser.add_argument("--receipt", required=True, type=Path) parser.add_argument("--output", required=True, type=Path) return parser def _require_output_outside_checkouts( output: Path, checkout_roots: Sequence[Path] ) -> Path: resolved = output.expanduser().resolve() for checkout in checkout_roots: root = checkout.expanduser().resolve() try: resolved.relative_to(root) except ValueError: continue raise ReviewError("review output must stay outside every source checkout") if resolved.exists() or resolved.is_symlink(): raise ReviewError("review output already exists") return resolved def main(argv: Sequence[str] | None = None) -> int: args = build_parser().parse_args(argv) try: downstream_root = args.source_root.expanduser().resolve() official_root = args.official_root.expanduser().resolve() inventory, receipt, source_document, evidence_files = verify_evidence( inventory_path=args.inventory.expanduser().resolve(), receipt_path=args.receipt.expanduser().resolve(), downstream_root=downstream_root, official_root=official_root, ) source_payloads = _load_selected_payloads(source_document, downstream_root) packet = compile_review_packet( inventory=inventory, receipt=receipt, source_document=source_document, source_payloads=source_payloads, evidence_files=evidence_files, ) output = _require_output_outside_checkouts( args.output, (REPOSITORY_ROOT, downstream_root, official_root), ) source_inventory.write_exclusive_json(output, packet, REPOSITORY_ROOT) print(json.dumps({"reviewInputDigest": packet["reviewInputDigest"]})) return 0 except Exception as exc: # noqa: BLE001 - private command boundary print(f"phase0 review: {exc}", file=sys.stderr) return 2 if __name__ == "__main__": raise SystemExit(main())