"""Semantic checks for the NutriCoach v1.4.0 candidate manifest.""" from __future__ import annotations import sys from dataclasses import dataclass from pathlib import Path from scripts.verify_nutricoach_v140_candidate_core import ( CandidateContractError, JsonValue, derive_candidate_digest, inventory_digest, load_json, path_beneath, require_digest, require_int, require_list, require_object, require_string, sha256_file, verify_hash_entries, ) from scripts.verify_nutricoach_v140_candidate_wheels import member_sha256, verify_delta from scripts.verify_nutricoach_v140_candidate_physical import verify_physical_inventory @dataclass(frozen=True, slots=True) class Qualification: candidate_digest: str hermes_wheel_sha256: str profile_wheel_sha256: str source_tree_digest: str source_member_count: int evidence_member_count: int def _strings(value: JsonValue | None, label: str) -> tuple[str, ...]: return tuple(require_string(item, label) for item in require_list(value, label)) def _verify_interpreter(manifest: dict[str, JsonValue]) -> str: interpreter = require_object(manifest.get("interpreter"), "interpreter") expected_path = require_string(interpreter.get("resolved_path"), "interpreter.resolved_path") actual_path = str(Path(sys.executable).resolve()) if actual_path != expected_path: raise CandidateContractError("interpreter identity path mismatch") digest = require_digest(interpreter.get("sha256"), "interpreter.sha256") if sha256_file(Path(actual_path)) != digest: raise CandidateContractError("interpreter identity byte mismatch") if require_string(interpreter.get("version"), "interpreter.version") != sys.version: raise CandidateContractError("interpreter identity version mismatch") if interpreter.get("mode") != "sealed-candidate": raise CandidateContractError("interpreter mode is not sealed-candidate") return digest def _verify_semantic_receipts(root: Path, manifest: dict[str, JsonValue], candidate: str) -> None: receipts = require_object(manifest.get("receipts"), "receipts") source = require_object(load_json(path_beneath(root, receipts.get("source_golden_path"), "source_golden_path")), "source receipt") installed = require_object(load_json(path_beneath(root, receipts.get("installed_golden_path"), "installed_golden_path")), "installed receipt") for label, receipt in (("source", source), ("installed", installed)): if receipt.get("candidate_digest") != candidate: raise CandidateContractError(f"{label} Golden Path candidate binding is invalid") if receipt.get("status") != "PASS": raise CandidateContractError(f"{label} Golden Path status is not PASS") counts = require_object(receipt.get("counts"), f"{label}.counts") expected: dict[str, JsonValue] = { "submitted": 4, "late_submitted": 1, "missed": 2, "completed_days": 5, "logical_topic59_cards": 7, "monday_owner_cards": 1, "privacy_leaks": 0, "automatic_customer_coaching_calls": 0, } for name, value in expected.items(): if counts.get(name) != value: raise CandidateContractError(f"{label} Golden Path count mismatch: {name}") if counts.get("adherence_percent") != 71.43: raise CandidateContractError(f"{label} Golden Path adherence mismatch") source_normalized = require_digest(source.get("normalized_semantic_sha256"), "source normalized receipt") installed_normalized = require_digest(installed.get("normalized_semantic_sha256"), "installed normalized receipt") if source_normalized != installed_normalized: raise CandidateContractError("source and installed Golden Path semantics differ") migration = require_object(load_json(path_beneath(root, receipts.get("migration"), "migration")), "migration receipt") rollback = require_object(load_json(path_beneath(root, receipts.get("rollback"), "rollback")), "rollback receipt") default_off = require_object(load_json(path_beneath(root, receipts.get("default_off"), "default_off")), "default-OFF receipt") if any(receipt.get("candidate_digest") != candidate for receipt in (migration, rollback, default_off)): raise CandidateContractError("operational receipt candidate binding is invalid") if migration.get("status") != "PASS" or migration.get("no_backfill") is not True or migration.get("append_only_initialization") is not True or migration.get("live_paths_touched") != 0: raise CandidateContractError("migration receipt is invalid") if rollback.get("status") != "PASS" or rollback.get("baseline_bytes_restored") is not True or rollback.get("live_paths_touched") != 0: raise CandidateContractError("rollback receipt is invalid") if default_off.get("status") != "PASS" or default_off.get("baseline_byte_parity") is not True: raise CandidateContractError("default-OFF baseline parity is invalid") for field in ("v140_imports", "v140_artifacts", "v140_calls"): if require_int(default_off.get(field), f"default_off.{field}") != 0: raise CandidateContractError(f"default-OFF {field} is not zero") def _verify_source_bindings( root: Path, manifest: dict[str, JsonValue], wheels: dict[str, Path], ) -> None: allowlists = require_object(manifest.get("changed_member_allowlists"), "changed_member_allowlists") seen: dict[str, set[str]] = {"hermes": set(), "profile": set()} for index, raw in enumerate(require_list(manifest.get("wheel_source_bindings"), "wheel_source_bindings")): entry = require_object(raw, f"wheel_source_bindings[{index}]") kind = require_string(entry.get("wheel"), "source binding wheel") if kind not in wheels: raise CandidateContractError("wheel source binding kind is invalid") member = require_string(entry.get("member"), "source binding member") source = path_beneath(root, entry.get("path"), "source binding path") if member in seen[kind] or sha256_file(source) != member_sha256(wheels[kind], member): raise CandidateContractError(f"wheel source byte mismatch: {member}") seen[kind].add(member) for kind in ("hermes", "profile"): expected = {member for member in _strings(allowlists.get(kind), f"{kind} allowlist") if ".dist-info/" not in member} if seen[kind] != expected: raise CandidateContractError(f"{kind} wheel source binding inventory mismatch") def _verify_installed(root: Path, manifest: dict[str, JsonValue], wheels: dict[str, Path]) -> None: for index, raw in enumerate(require_list(manifest.get("installed_members"), "installed_members")): entry = require_object(raw, f"installed_members[{index}]") kind = require_string(entry.get("wheel"), "installed wheel kind") if kind not in wheels: raise CandidateContractError("installed member wheel kind is invalid") installed = path_beneath(root, entry.get("path"), "installed member path") member = require_string(entry.get("member"), "installed wheel member") expected = member_sha256(wheels[kind], member) if sha256_file(installed) != expected: raise CandidateContractError(f"installed loaded byte mismatch: {member}") def _verify_delta_receipts(root: Path, manifest: dict[str, JsonValue]) -> None: receipts = require_object(manifest.get("delta_receipts"), "delta_receipts") static = require_object(load_json(path_beneath(root, receipts.get("static"), "static delta")), "static delta") loc = require_object(load_json(path_beneath(root, receipts.get("loc"), "LOC delta")), "LOC delta") gateway = require_object(load_json(path_beneath(root, receipts.get("gateway"), "gateway delta")), "gateway delta") if ( static.get("status") != "PASS" or static.get("new_diagnostics") != 0 or static.get("changed_line_diagnostics") != 0 or static.get("new_suppressions") != 0 ): raise CandidateContractError("static baseline delta is not clean") if ( loc.get("status") != "PASS" or loc.get("new_module_over_250") != 0 or loc.get("legacy_pure_loc_increases") != 0 ): raise CandidateContractError("LOC baseline delta is not clean") if ( gateway.get("status") != "PASS" or gateway.get("nodeid_count") != 7751 or gateway.get("candidate_only_failures") != [] or gateway.get("unclassified_failures") != [] ): raise CandidateContractError("gateway baseline delta is not clean") def verify_candidate(root: Path, manifest_path: Path) -> Qualification: """Independently verify every identity and behavioral receipt.""" manifest = require_object(load_json(manifest_path), "manifest") if manifest.get("schema") != "nutricoach-v140-candidate-manifest-v1": raise CandidateContractError("candidate manifest schema is invalid") physical_inventory = require_list(manifest.get("physical_inventory"), "physical_inventory") _ = verify_physical_inventory( root, physical_inventory, require_digest(manifest.get("physical_seal_digest"), "physical_seal_digest") ) _verify_delta_receipts(root, manifest) source_entries = verify_hash_entries(root, manifest.get("source_inventory"), "source_inventory") evidence_entries = verify_hash_entries(root, manifest.get("evidence_inventory"), "evidence_inventory") _ = verify_hash_entries(root, manifest.get("qualification_evidence"), "qualification_evidence") config_entries = verify_hash_entries(root, manifest.get("config_inventory"), "config_inventory") policy_entries = verify_hash_entries(root, manifest.get("policy_inventory"), "policy_inventory") identities = require_object(manifest.get("identities"), "identities") source_digest = inventory_digest(source_entries) evidence_digest = inventory_digest(evidence_entries) config_digest = inventory_digest(config_entries) policy_digest = inventory_digest(policy_entries) for name, actual in (("source_tree_digest", source_digest), ("evidence_digest", evidence_digest), ("config_digest", config_digest), ("policy_digest", policy_digest)): if require_digest(identities.get(name), f"identities.{name}") != actual: raise CandidateContractError(f"{name} mismatch") baseline = require_object(manifest.get("baseline_wheels"), "baseline_wheels") builds = require_list(manifest.get("builds"), "builds") if len(builds) != 2: raise CandidateContractError("exactly two builds are required") build_paths: list[dict[str, Path]] = [] for index, raw in enumerate(builds): build = require_object(raw, f"builds[{index}]") build_paths.append({kind: path_beneath(root, build.get(kind), f"builds[{index}].{kind}") for kind in ("hermes", "profile")}) wheel_digests: dict[str, str] = {} allowlists = require_object(manifest.get("changed_member_allowlists"), "changed_member_allowlists") for kind in ("hermes", "profile"): first_digest = sha256_file(build_paths[0][kind]) if sha256_file(build_paths[1][kind]) != first_digest: raise CandidateContractError(f"{kind} wheel builds are not reproducible") if require_digest(identities.get(f"{kind}_wheel_sha256"), f"identities.{kind}_wheel_sha256") != first_digest: raise CandidateContractError(f"{kind} wheel identity mismatch") baseline_path = path_beneath(root, baseline.get(kind), f"baseline_wheels.{kind}") _ = verify_delta(baseline_path, build_paths[0][kind], _strings(allowlists.get(kind), f"{kind} allowlist")) _ = verify_delta(baseline_path, build_paths[1][kind], _strings(allowlists.get(kind), f"{kind} allowlist")) wheel_digests[kind] = first_digest _verify_source_bindings(root, manifest, build_paths[0]) interpreter_digest = _verify_interpreter(manifest) derivation = require_object(manifest.get("derivation_inputs"), "derivation_inputs") expected_derivation: dict[str, JsonValue] = { "hermes_wheel_sha256": wheel_digests["hermes"], "profile_wheel_sha256": wheel_digests["profile"], "source_tree_digest": source_digest, "config_digest": config_digest, "policy_digest": policy_digest, "evidence_digest": evidence_digest, "interpreter_sha256": interpreter_digest, } if derivation != expected_derivation: raise CandidateContractError("full canonical product derivation inputs differ") candidate = derive_candidate_digest(derivation, claimed=require_digest(identities.get("candidate_digest"), "candidate_digest")) _verify_installed(root, manifest, build_paths[0]) _verify_semantic_receipts(root, manifest, candidate) return Qualification(candidate, wheel_digests["hermes"], wheel_digests["profile"], source_digest, len(source_entries), len(evidence_entries))