#!/usr/bin/env -S uv run --script # /// script # requires-python = ">=3.12" # dependencies = ["pydantic"] # /// """Create or independently execute the hash-bound Todo2-9 composition manifest.""" from __future__ import annotations import hashlib import json import re import subprocess import sys from pathlib import Path ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT)) from pydantic import RootModel # noqa: E402 from scripts.nutricoach_v140_golden_path_composition_models import ( CompositionGroup, CompositionManifest, CompositionMember, ) from scripts.nutricoach_v140_golden_path_composition_spec import GROUPS, GroupSeed from scripts.nutricoach_v140_golden_path_models import CliError EVIDENCE = ROOT / ".omo/evidence/nutricoach-v140-weekly-operations" MANIFEST = EVIDENCE / "task-10-composition-manifest-r3.json" DIGEST = EVIDENCE / "task-10-composition-manifest-r3.sha256" RUNNERS = ( "scripts/nutricoach_v140_golden_path_composition_models.py", "scripts/nutricoach_v140_golden_path_composition_spec.py", "scripts/run_nutricoach_v140_golden_path_composition.py", ) ENV = {"PYTHONDONTWRITEBYTECODE": "1", "PYTHONPATH": "dualcoach/profile", "UV_OFFLINE": "1"} def _sha(path: Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest() def _canonical_command(cwd: str, env: dict[str, str], argv: tuple[str, ...]) -> str: encoded = json.dumps( {"argv": argv, "cwd": cwd, "env": env}, sort_keys=True, separators=(",", ":"), ).encode() return hashlib.sha256(encoded).hexdigest() def _argv(seed: GroupSeed) -> tuple[str, ...]: python = ".venv/bin/python" if seed.cwd == "." else "../../.venv/bin/python" return (python, "-m", "pytest", "-q", "-p", "no:cacheprovider", *seed.test_paths) def _run(group: CompositionGroup, *, collect: bool) -> subprocess.CompletedProcess[str]: argv = list(group.argv) if collect: argv.insert(4, "--collect-only") return subprocess.run( argv, cwd=ROOT / group.cwd, env=group.env, check=False, capture_output=True, text=True, ) def _nodeids(output: str) -> tuple[str, ...]: return tuple(line for line in output.splitlines() if "::" in line and not line.startswith(" ")) def _member_paths(seed: GroupSeed) -> tuple[str, ...]: tests = tuple( str((Path(seed.cwd) / path.split("::", 1)[0]).as_posix()).removeprefix("./") for path in seed.test_paths ) return tuple(sorted(set((*tests, *seed.source_paths)))) def _write() -> None: groups: list[CompositionGroup] = [] for seed in GROUPS: argv = _argv(seed) env = dict(ENV) if seed.cwd == "dualcoach/profile": env["PYTHONPATH"] = "." provisional = CompositionGroup( id=seed.id, capabilities=seed.capabilities, cwd=seed.cwd, env=env, argv=argv, test_paths=seed.test_paths, nodeids_member=CompositionMember(path="pending", sha256="0" * 64), collection_count=0, command_sha256=_canonical_command(seed.cwd, env, argv), members=tuple( CompositionMember(path=path, sha256=_sha(ROOT / path)) for path in _member_paths(seed) ), expected_exit=0, expected_passed=0, isolation_policy=seed.isolation_policy, ) collected = _run(provisional, collect=True) if collected.returncode != 0: raise CliError(f"collection failed: {seed.id}: {collected.stderr}") nodeids = _nodeids(collected.stdout) member_path = EVIDENCE / f"task-10-composition-nodeids-r3-{seed.id}.json" _ = member_path.write_text(json.dumps(nodeids, separators=(",", ":")) + "\n", encoding="utf-8") groups.append(provisional.model_copy(update={ "nodeids_member": CompositionMember( path=member_path.relative_to(ROOT).as_posix(), sha256=_sha(member_path), ), "collection_count": len(nodeids), "expected_passed": len(nodeids), })) manifest = CompositionManifest( schema="nutricoach-v140-task-10-composition-r3", repository=".", runner_members=tuple( CompositionMember(path=path, sha256=_sha(ROOT / path)) for path in RUNNERS ), groups=tuple(groups), ) _ = MANIFEST.write_text( manifest.model_dump_json(by_alias=True, exclude_none=True) + "\n", encoding="utf-8", ) _ = DIGEST.write_text(_sha(MANIFEST) + " " + MANIFEST.name + "\n", encoding="ascii") def _verify() -> None: expected_digest = DIGEST.read_text(encoding="ascii").split()[0] if _sha(MANIFEST) != expected_digest: raise CliError("composition manifest hash mismatch") manifest = CompositionManifest.model_validate_json(MANIFEST.read_text(encoding="utf-8")) for member in manifest.runner_members: if _sha(ROOT / member.path) != member.sha256: raise CliError(f"runner member hash mismatch: {member.path}") for group in manifest.groups: if _canonical_command(group.cwd, group.env, group.argv) != group.command_sha256: raise CliError(f"command hash mismatch: {group.id}") for member in (*group.members, group.nodeids_member): if _sha(ROOT / member.path) != member.sha256: raise CliError(f"group member hash mismatch: {group.id}:{member.path}") expected = tuple(RootModel[list[str]].model_validate_json( (ROOT / group.nodeids_member.path).read_text(encoding="utf-8") ).root) collected = _run(group, collect=True) actual = _nodeids(collected.stdout) if collected.returncode != 0 or actual != expected or len(actual) != group.collection_count: raise CliError(f"collection mismatch: {group.id}") completed = _run(group, collect=False) match = re.search(r"(\d+) passed", completed.stdout) passed = 0 if match is None else int(match.group(1)) if completed.returncode != group.expected_exit or passed != group.expected_passed: raise CliError(f"execution mismatch: {group.id}:{completed.stdout}:{completed.stderr}") print(f"COMPOSITION_GROUP_PASS {group.id} {passed}") print("NUTRICOACH_V140_COMPOSITION_PASS") def main() -> int: try: if sys.argv[1:] == ["--write"]: _write() elif not sys.argv[1:]: _verify() else: raise CliError("usage: run_nutricoach_v140_golden_path_composition.py [--write]") except (CliError, OSError, ValueError) as error: print(f"error: {error}", file=sys.stderr) return 1 return 0 if __name__ == "__main__": raise SystemExit(main())