1# Author: Simon-Pierre Boucher — simon-pierre.boucher@uqo.ca2"""Protocol freeze: results/freeze.json with the SHA-256 of every role prompt, every config file, the3schema module, the code of agents/ and debate/, and the model ids of the cost-matched pairs.4Any later change is detectable (`av freeze --check`) and must be reported as a protocol deviation."""5from __future__ import annotations67import hashlib8import json9import platform10import subprocess11from datetime import datetime, timezone12from pathlib import Path13from typing import Any1415from .. import __version__16from ..agents.roles import OPTIONAL_ROLE_IDS, ROLES, placebo_roles, role_prompt_hashes17from ..config import CONFIGS, RESULTS, ROOT, load_config1819SRC = ROOT / "src" / "agentic_valuation"20FREEZE_PATH = RESULTS / "freeze.json"212223def _sha_file(p: Path) -> str:24 return hashlib.sha256(p.read_bytes()).hexdigest()252627def _sha_text(s: str) -> str:28 return hashlib.sha256(s.encode("utf-8")).hexdigest()293031def _code_hashes() -> dict[str, str]:32 out: dict[str, str] = {}33 for sub in ("agents", "debate"):34 for p in sorted((SRC / sub).glob("*.py")):35 out[f"{sub}/{p.name}"] = _sha_file(p)36 out["schemas.py"] = _sha_file(SRC / "schemas.py")37 for p in sorted((SRC / "providers").glob("*.py")):38 out[f"providers/{p.name}"] = _sha_file(p)39 return out404142def _git_commit() -> str | None:43 try:44 r = subprocess.run(["git", "-C", str(ROOT), "rev-parse", "HEAD"], capture_output=True, text=True, timeout=5)45 return r.stdout.strip() or None46 except Exception:47 return None484950def build_freeze() -> dict[str, Any]:51 models = load_config("models")52 prompts = {rid: _sha_text(r.system_prompt()) for rid, r in ROLES.items()}53 prompts |= {r.id: _sha_text(r.system_prompt()) for r in placebo_roles(12)}54 doc: dict[str, Any] = {55 "frozen_at": datetime.now(timezone.utc).isoformat(),56 "package_version": __version__,57 "git_commit": _git_commit(),58 "python": platform.python_version(),59 "role_prompt_sha256": prompts,60 "role_prompt_hash16": role_prompt_hashes(),61 "optional_roles": list(OPTIONAL_ROLE_IDS),62 "configs_sha256": {p.name: _sha_file(p) for p in sorted(CONFIGS.glob("*.yaml"))},63 "code_sha256": _code_hashes(),64 "model_ids": {"cost_matched_pairs": models["cost_matched_pairs"], "default_tier": models.get("default_tier"), "request_defaults": models.get("request_defaults")},65 "treatments": load_config("treatments"),66 }67 doc["freeze_digest"] = _sha_text(json.dumps({k: v for k, v in doc.items() if k not in ("frozen_at", "git_commit", "python")}, sort_keys=True, default=str))68 return doc697071def write_freeze(path: Path | None = None) -> Path:72 path = path or FREEZE_PATH73 path.parent.mkdir(parents=True, exist_ok=True)74 path.write_text(json.dumps(build_freeze(), indent=1, ensure_ascii=False, default=str))75 return path767778def check_freeze(path: Path | None = None) -> list[str]:79 """Return the list of items whose hash differs from the frozen copy (empty = protocol unchanged)."""80 path = path or FREEZE_PATH81 if not path.exists():82 return [f"no freeze file at {path}"]83 old = json.loads(path.read_text())84 new = build_freeze()85 diffs: list[str] = []86 for key in ("role_prompt_sha256", "configs_sha256", "code_sha256"):87 for k, v in new.get(key, {}).items():88 if old.get(key, {}).get(k) != v:89 diffs.append(f"{key}:{k}")90 for k in old.get(key, {}):91 if k not in new.get(key, {}):92 diffs.append(f"{key}:{k} (removed)")93 if old.get("model_ids") != new.get("model_ids"):94 diffs.append("model_ids")95 return diffs96