#!/usr/bin/env -S uv run --script # /// script # requires-python = ">=3.12,<3.14" # dependencies = ["pydantic>=2,<3"] # /// # ─── How to run ─── # uv run scripts/seal_nutricoach_v150_candidate.py --help # ────────────────── """Materialize the immutable NutriCoach v1.5 combined candidate seal.""" from __future__ import annotations import shutil import sys from dataclasses import dataclass from pathlib import Path from typing import override sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from scripts.nutricoach_v150_candidate_cli import CliNamespace, candidate_parser from scripts.nutricoach_v150_r71b_task10_evidence import ( ValidatedTask10Evidence, validate_fixed_task10_evidence_sources, write_candidate_task10_evidence, ) from scripts.nutricoach_v150_candidate_source_inventory import ( CANDIDATE_REQUIRED_SOURCE_PATHS, ) from scripts.verify_nutricoach_v150_candidate_inputs import ( R71B_MAINTENANCE_OVERLAY_PATHS, ) from scripts.verify_nutricoach_v140_candidate_core import ( JsonValue, canonical, inventory_digest, load_json, require_list, require_object, require_string, sha256_bytes, sha256_file, ) @dataclass(frozen=True, slots=True) class CandidateSealError(RuntimeError): reason: str @override def __str__(self) -> str: return self.reason def _inventory(root: Path, paths: list[Path]) -> list[dict[str, JsonValue]]: return [ {"path": path.relative_to(root).as_posix(), "sha256": sha256_file(path)} for path in sorted(paths) ] def _copy(source: Path, destination: Path) -> Path: _ = destination.parent.mkdir(parents=True, exist_ok=True) _ = shutil.copy2(source, destination) return destination def _v17_input(v17: Path, direct_name: str, sealed_name: str) -> Path: direct = v17 / direct_name if direct.is_file(): return direct return v17 / "inputs" / sealed_name def r71b_evidence_inputs(args: CliNamespace) -> dict[str, ValidatedTask10Evidence]: """Validate and map only the three fixed Task10 sources into candidate inputs.""" r70_error, observer, health_recovery = validate_fixed_task10_evidence_sources( args.r70_error_evidence, args.observer_evidence, args.fixed_collector_evidence, ) return { "inputs/r70-canonical-authority-drift.json": r70_error, "inputs/r71b-observer-final.json": observer, "inputs/r71b-fixed-collector.json": health_recovery, } def copy_r71b_evidence(args: CliNamespace, successor: Path) -> list[Path]: """Produce canonical candidate copies with exact Task10 source provenance.""" return _write_r71b_evidence(r71b_evidence_inputs(args), successor) def _write_r71b_evidence( evidence_inputs: dict[str, ValidatedTask10Evidence], successor: Path, ) -> list[Path]: copies: list[Path] = [] for relative, evidence in evidence_inputs.items(): destination = successor / relative write_candidate_task10_evidence(destination, evidence) copies.append(destination) return copies def candidate_source_paths( base_manifest: dict[str, JsonValue], v17: Path ) -> list[Path]: source_root = Path(__file__).resolve().parents[1] paths: set[Path] = set() for raw in require_list( base_manifest.get("source_inventory"), "base source inventory" ): entry = require_object(raw, "base source inventory") relative = Path(require_string(entry.get("path"), "base source path")) paths.add(Path(*relative.parts[2:])) patch_manifest = _v17_input(v17, "patch-tree.sha256", "v17-patch-tree.sha256") for line in patch_manifest.read_text(encoding="utf-8").splitlines(): _, absolute = line.split(" ", maxsplit=1) parts = Path(absolute).parts patch_index = parts.index("patch-tree") paths.add(Path(*parts[patch_index + 1 :])) paths.update(Path(relative) for relative in CANDIDATE_REQUIRED_SOURCE_PATHS) missing = sorted(str(path) for path in paths if not (source_root / path).is_file()) if missing: raise FileNotFoundError(f"candidate source members missing: {missing}") paths.update(Path(relative) for relative in R71B_MAINTENANCE_OVERLAY_PATHS) return sorted(paths) def main() -> int: args = candidate_parser().parse_args(namespace=CliNamespace()) source_root = Path(__file__).resolve().parents[1] successor = args.successor.resolve() if successor.exists() or successor.is_symlink(): raise CandidateSealError("successor_exists") evidence_inputs = r71b_evidence_inputs(args) _ = successor.mkdir(parents=True) base_manifest = require_object(load_json(args.base), "base manifest") source_files = [ _copy(source_root / relative, successor / "snapshot/source" / relative) for relative in candidate_source_paths(base_manifest, args.v17_evidence) ] input_sources = { "inputs/base-manifest.json": args.base, "inputs/v17-evidence-files.sha256": _v17_input( args.v17_evidence, "evidence-files.sha256", "v17-evidence-files.sha256", ), "inputs/v17-exact-source-set.sha256": _v17_input( args.v17_evidence, "exact-current-source-set.final.sha256", "v17-exact-source-set.sha256", ), "inputs/v17-patch-tree.sha256": _v17_input( args.v17_evidence, "patch-tree.sha256", "v17-patch-tree.sha256", ), "inputs/v17-sandbox.diff": _v17_input( args.v17_evidence, "sandbox.diff", "v17-sandbox.diff", ), "inputs/successor-overlay.sha256": args.successor_overlay, } input_files = [ _copy(source, successor / relative) for relative, source in input_sources.items() ] # These Task10 reports are copied while the candidate is sealed, so # preseal consumes candidate-bound evidence rather than injected files. r71b_evidence_files = _write_r71b_evidence(evidence_inputs, successor) input_files.extend(r71b_evidence_files) wheel_sources = { "artifacts/build-1/hermes_agent-0.17.0-py3-none-any.whl": args.build_1_hermes, "artifacts/build-1/physique_checkin_cli-0.1.0-py3-none-any.whl": args.build_1_profile, "artifacts/build-2/hermes_agent-0.17.0-py3-none-any.whl": args.build_2_hermes, "artifacts/build-2/physique_checkin_cli-0.1.0-py3-none-any.whl": args.build_2_profile, } wheel_files = [ _copy(source, successor / relative) for relative, source in wheel_sources.items() ] receipt_files = [ _copy(path, successor / "receipts" / path.relative_to(args.receipts_root)) for path in sorted(args.receipts_root.rglob("*")) if path.is_file() ] source_inventory = _inventory(successor, source_files) input_inventory = _inventory(successor, input_files) wheel_inventory = _inventory(successor, wheel_files) evidence_inventory = _inventory(successor, receipt_files) hermes_digest = sha256_file(wheel_files[0]) profile_digest = sha256_file(wheel_files[1]) interpreter = Path(sys.executable).resolve() derivation_inputs: dict[str, JsonValue] = { "base_candidate_digest": require_string( require_object(base_manifest["identities"], "base identities").get( "candidate_digest" ), "base candidate digest", ), "base_manifest_sha256": sha256_file(args.base), "evidence_digest": inventory_digest(evidence_inventory), "hermes_wheel_sha256": hermes_digest, "interpreter_sha256": sha256_file(interpreter), "profile_wheel_sha256": profile_digest, "r71b_evidence_digest": inventory_digest( _inventory(successor, r71b_evidence_files) ), "source_tree_digest": inventory_digest(source_inventory), "successor_overlay_sha256": sha256_file(args.successor_overlay), "v17_evidence_manifest_sha256": sha256_file( _v17_input( args.v17_evidence, "evidence-files.sha256", "v17-evidence-files.sha256", ) ), "v17_patch_tree_sha256": sha256_file( _v17_input( args.v17_evidence, "patch-tree.sha256", "v17-patch-tree.sha256", ) ), "v17_sandbox_diff_sha256": sha256_file( _v17_input( args.v17_evidence, "sandbox.diff", "v17-sandbox.diff", ) ), "v17_source_set_sha256": sha256_file( _v17_input( args.v17_evidence, "exact-current-source-set.final.sha256", "v17-exact-source-set.sha256", ) ), } candidate_digest = sha256_bytes(canonical(derivation_inputs)) source_values: list[JsonValue] = list(source_inventory) input_values: list[JsonValue] = list(input_inventory) wheel_values: list[JsonValue] = list(wheel_inventory) evidence_values: list[JsonValue] = list(evidence_inventory) manifest: dict[str, JsonValue] = { "schema": "nutricoach-v150-combined-candidate-manifest-v1", "status": "QUALIFIED_PENDING_LIVE_AUTHORIZATION", "candidate_digest": candidate_digest, "capabilities": { "nutricoach_channel_inbox_v1": { "compiled": True, "configured": False, "authorized": False, }, "nutricoach_multi_customer_v1": { "compiled": True, "configured": False, "authorized": False, "authorized_capacity": 0, "post_migration_capacity": 5, }, }, "derivation_inputs": derivation_inputs, "source_inventory": source_values, "input_inventory": input_values, "wheel_inventory": wheel_values, "evidence_inventory": evidence_values, "builds": [ { "hermes": wheel_inventory[0]["path"], "profile": wheel_inventory[1]["path"], }, { "hermes": wheel_inventory[2]["path"], "profile": wheel_inventory[3]["path"], }, ], "interpreter": { "path": str(interpreter), "sha256": sha256_file(interpreter), "version": sys.version, }, } _ = (successor / "manifest.json").write_bytes(canonical(manifest) + b"\n") qualification: dict[str, JsonValue] = { "candidate_digest": candidate_digest, "status": "QUALIFIED_PENDING_LIVE_AUTHORIZATION", } _ = (successor / "qualification.json").write_bytes(canonical(qualification) + b"\n") for path in successor.rglob("*"): if path.is_file() and not path.is_symlink(): if path.suffix in {".pyc", ".pyo"} or "__pycache__" in path.parts: raise CandidateSealError("candidate_bytecode") path.chmod(0o444) for path in sorted( (entry for entry in successor.rglob("*") if entry.is_dir()), reverse=True, ): path.chmod(0o555) successor.chmod(0o555) print(candidate_digest) return 0 if __name__ == "__main__": raise SystemExit(main())