src/turn_detection/provenance.py

"""Hash-bound freeze manifests for one-shot official-test evaluation."""

from __future__ import annotations

import hashlib
import hmac
import json
import math
from collections.abc import Mapping
from pathlib import Path
from typing import Any

FREEZE_BINDING_FORMAT_VERSION = 1
FREEZE_BINDING_ALGORITHM = "sha256"
FREEZE_BINDING_CANONICALIZATION = "json-sorted-compact-v1"
_FREEZE_BINDING_FIELD = "policy_binding"


def sha256_file(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for block in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(block)
    return digest.hexdigest()


def file_evidence(path: Path, root: Path) -> dict[str, int | str]:
    if path.is_symlink() or not path.is_file():
        raise ValueError(f"frozen file must be a regular file: {path}")
    resolved = path.resolve()
    try:
        relative = resolved.relative_to(root.resolve())
    except ValueError as exc:
        raise ValueError(f"frozen file escapes project root: {path}") from exc
    return {
        "path": relative.as_posix(),
        "bytes": resolved.stat().st_size,
        "sha256": sha256_file(resolved),
    }


def freeze_policy_sha256(manifest: Mapping[str, Any]) -> str:
    """Hash the complete freeze policy, excluding only its own binding field.

    Compact, key-sorted JSON gives freeze generation and verification one stable
    byte representation.  Binding every other top-level field prevents policy
    pointers (checkpoint/config), thresholds, controller settings, or dataset
    identity from changing independently after the freeze is written.
    """

    payload = {key: value for key, value in manifest.items() if key != _FREEZE_BINDING_FIELD}
    serialized = json.dumps(
        payload,
        sort_keys=True,
        separators=(",", ":"),
        ensure_ascii=False,
        allow_nan=False,
    ).encode("utf-8")
    return hashlib.sha256(serialized).hexdigest()


def build_freeze_policy_binding(manifest: Mapping[str, Any]) -> dict[str, int | str]:
    """Create the self-describing canonical binding stored in a freeze manifest."""

    return {
        "format_version": FREEZE_BINDING_FORMAT_VERSION,
        "algorithm": FREEZE_BINDING_ALGORITHM,
        "canonicalization": FREEZE_BINDING_CANONICALIZATION,
        "sha256": freeze_policy_sha256(manifest),
    }


def _is_sha256(value: object) -> bool:
    return (
        isinstance(value, str)
        and len(value) == 64
        and all(character in "0123456789abcdef" for character in value)
    )


def _verify_policy_binding(manifest: Mapping[str, Any]) -> None:
    binding = manifest.get(_FREEZE_BINDING_FIELD)
    expected_keys = {"format_version", "algorithm", "canonicalization", "sha256"}
    if not isinstance(binding, dict) or set(binding) != expected_keys:
        raise ValueError("freeze manifest has no valid policy binding")
    if binding.get("format_version") != FREEZE_BINDING_FORMAT_VERSION:
        raise ValueError("unsupported freeze policy-binding format")
    if binding.get("algorithm") != FREEZE_BINDING_ALGORITHM:
        raise ValueError("unsupported freeze policy-binding algorithm")
    if binding.get("canonicalization") != FREEZE_BINDING_CANONICALIZATION:
        raise ValueError("unsupported freeze policy canonicalization")
    claimed = binding.get("sha256")
    if not _is_sha256(claimed):
        raise ValueError("freeze policy binding has an invalid SHA-256")
    try:
        actual = freeze_policy_sha256(manifest)
    except (TypeError, ValueError) as exc:
        raise ValueError("freeze policy is not canonical JSON") from exc
    if not hmac.compare_digest(claimed, actual):
        raise ValueError("freeze policy binding mismatch")


def _verify_file_evidence(
    evidence: object,
    root: Path,
    *,
    seen: set[str],
) -> tuple[str, dict[str, Any]]:
    if not isinstance(evidence, dict) or not isinstance(evidence.get("path"), str):
        raise ValueError("invalid frozen file evidence")
    relative = evidence["path"]
    if relative in seen:
        raise ValueError(f"duplicate frozen file: {relative}")
    if (
        isinstance(evidence.get("bytes"), bool)
        or not isinstance(evidence.get("bytes"), int)
        or evidence["bytes"] < 0
        or not _is_sha256(evidence.get("sha256"))
    ):
        raise ValueError(f"invalid frozen file evidence: {relative}")
    unresolved = root / relative
    if unresolved.is_symlink() or not unresolved.is_file():
        raise ValueError(f"frozen file is missing: {relative}")
    path = unresolved.resolve()
    try:
        canonical_relative = path.relative_to(root).as_posix()
    except ValueError as exc:
        raise ValueError(f"frozen path escapes project root: {relative}") from exc
    if relative != canonical_relative:
        raise ValueError(f"frozen path is not canonical: {relative}")
    if evidence["bytes"] != path.stat().st_size or not hmac.compare_digest(
        evidence["sha256"], sha256_file(path)
    ):
        raise ValueError(f"frozen file changed: {relative}")
    seen.add(relative)
    return relative, evidence


def _verify_bound_pointer(
    manifest: Mapping[str, Any],
    name: str,
    evidence_by_path: Mapping[str, dict[str, Any]],
) -> dict[str, Any]:
    pointer = manifest.get(name)
    if not isinstance(pointer, dict) or not isinstance(pointer.get("path"), str):
        raise ValueError(f"freeze manifest has no valid {name} pointer")
    bound = evidence_by_path.get(pointer["path"])
    if bound is None or pointer != bound:
        raise ValueError(f"frozen {name} pointer is not identical to hash-bound file evidence")
    return pointer


def verify_freeze_manifest(
    manifest_path: Path,
    root: Path,
    *,
    checkpoint_path: Path | None = None,
    threshold: float | None = None,
) -> dict[str, Any]:
    try:
        manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        raise ValueError(f"invalid freeze manifest: {manifest_path}") from exc
    if not isinstance(manifest, dict) or manifest.get("status") != "frozen_for_official_test":
        raise ValueError("manifest is not frozen_for_official_test")
    if manifest.get("format_version") != 1:
        raise ValueError("unsupported freeze manifest format")
    _verify_policy_binding(manifest)
    files = manifest.get("files")
    if not isinstance(files, list) or not files:
        raise ValueError("freeze manifest has no bound files")
    root = root.resolve()
    seen: set[str] = set()
    evidence_by_path: dict[str, dict[str, Any]] = {}
    for evidence in files:
        relative, verified = _verify_file_evidence(evidence, root, seen=seen)
        evidence_by_path[relative] = verified

    checkpoint_pointer = _verify_bound_pointer(manifest, "checkpoint", evidence_by_path)
    _verify_bound_pointer(manifest, "training_config", evidence_by_path)

    frozen_threshold = manifest.get("threshold")
    if (
        isinstance(frozen_threshold, bool)
        or not isinstance(frozen_threshold, int | float)
        or not math.isfinite(float(frozen_threshold))
        or not 0.0 <= float(frozen_threshold) <= 1.0
    ):
        raise ValueError("freeze manifest has no valid threshold")
    controller = manifest.get("controller")
    controller_threshold = (
        controller.get("endpoint_threshold") if isinstance(controller, dict) else None
    )
    if (
        isinstance(controller_threshold, bool)
        or not isinstance(controller_threshold, int | float)
        or not math.isclose(
            float(frozen_threshold),
            float(controller_threshold),
            rel_tol=0.0,
            abs_tol=1e-9,
        )
    ):
        raise ValueError("frozen controller endpoint threshold differs from frozen threshold")
    if checkpoint_path is not None:
        try:
            checkpoint_relative = checkpoint_path.resolve().relative_to(root).as_posix()
        except ValueError as exc:
            raise ValueError("evaluated checkpoint escapes project root") from exc
        if checkpoint_relative != checkpoint_pointer["path"]:
            raise ValueError("evaluated checkpoint differs from frozen checkpoint")
    if threshold is not None:
        invalid_threshold = isinstance(threshold, bool) or not isinstance(threshold, int | float)
        if invalid_threshold or not math.isclose(
            float(frozen_threshold),
            float(threshold),
            rel_tol=0.0,
            abs_tol=1e-9,
        ):
            raise ValueError("evaluated threshold differs from frozen threshold")
    official = manifest.get("official_test")
    if not isinstance(official, dict):
        raise ValueError("freeze manifest has no official-test identity")
    if official.get("dataset_id") != "pipecat-ai/smart-turn-data-v3.2-test":
        raise ValueError("freeze manifest targets the wrong official dataset")
    if official.get("revision") != "0500378e8ed6d38e37b016e24d261e8e6c6a6859":
        raise ValueError("freeze manifest targets the wrong official revision")
    return manifest