blind_bundle_integrity.py

"""Source-derived, reduced-schema blind-review controls; no API or filesystem writes.

Public adaptation of Decision v3. See README.md for the intentionally different
wire format. Shared candidates are for authorized judges, NOT public Git hosting.
A trusted manifest detects changes; it does not authenticate execution or people.
"""
from __future__ import annotations

import copy
import hashlib
import json
import math
import re
import secrets
from pathlib import Path
from typing import Any, Mapping, Sequence

VERSION = "portfolio-blind-v1"
NA = "not_applicable"
CASE_FIELDS = {"case_id", "dependency_group_id", "category", "track", "messages",
               "reference_answer", "rubric"}
SCORE_FIELDS = {"professional_correctness", "completeness", "clarity", "evidence_faithfulness"}
REVIEW_FIELDS = {"case_id", "candidate_id", "bundle_sha256", "protocol_sha256",
                 "reviewer_id", "reviewer_type", "scores", "critical_error",
                 "preferred", "safety", "failure_tags"}


def canonical_sha256(value: Any) -> str:
    """Hash canonical JSON, not private source-file bytes; reject non-JSON numbers."""
    payload = json.dumps(value, sort_keys=True, ensure_ascii=False,
                         separators=(",", ":"), allow_nan=False)
    return hashlib.sha256(payload.encode("utf-8")).hexdigest()


def _text(value: Any) -> str:
    if not isinstance(value, str) or not value.strip():
        raise ValueError("Expected nonempty text")
    return value


def _hash(value: Any) -> str:
    if not isinstance(value, str) or re.fullmatch(r"[0-9a-f]{64}", value) is None:
        raise ValueError("Expected lowercase SHA256")
    return value


def _score(value: Any) -> float:
    if type(value) not in (int, float) or not math.isfinite(value) or not 0 <= value <= 5:
        raise ValueError("Scores must be finite numbers in [0, 5], not booleans")
    return float(value)


def _index(rows: Sequence[Mapping[str, Any]], fields: tuple[str, ...]) -> dict:
    result = {}
    for row in rows:
        if not isinstance(row, Mapping):
            raise ValueError("Expected a record object")
        key = tuple(_text(row.get(field)) for field in fields)
        if key in result:
            raise ValueError("Duplicate record identity")
        result[key] = row
    return result


def check_separate_directories(shared: Path, private: Path) -> None:
    """Resolve symlinks and reject equal/nested directories; create nothing.

    Callers own permissions, race-free filesystem access and no-overwrite writes.
    """
    shared, private = shared.resolve(), private.resolve()
    if shared == private or shared.is_relative_to(private) or private.is_relative_to(shared):
        raise ValueError("Judge and owner directories must be disjoint")


def build_bundle(cases: Sequence[Mapping[str, Any]], outputs: Sequence[Mapping[str, Any]],
                 *, left_model: str, right_model: str, run_config_sha256: str,
                 protocol_sha256: str) -> tuple[list[dict], list[dict], dict]:
    """Return (judge_candidates, owner_mapping, owner_manifest), all in memory.

    Input outputs must contain exactly the selected two models for every case.
    The caller must explicitly select a comparison before invoking this function.
    """
    left_model, right_model = _text(left_model), _text(right_model)
    if left_model == right_model:
        raise ValueError("Comparison requires different model identities")
    run_hash, protocol_hash = _hash(run_config_sha256), _hash(protocol_sha256)
    case_map = _index(cases, ("case_id",))
    if not case_map:
        raise ValueError("Empty evaluation is not evidence")
    output_map = _index(outputs, ("case_id", "model"))
    expected = {(cid[0], model) for cid in case_map for model in (left_model, right_model)}
    if set(output_map) != expected:
        raise ValueError("Output coverage must exactly match the selected comparison")
    shared, mapping = [], []
    rng = secrets.SystemRandom()
    for (cid,), case in sorted(case_map.items()):
        if set(case) != CASE_FIELDS:
            raise ValueError("Unexpected/missing case fields for the reduced public schema")
        for field in CASE_FIELDS - {"messages"}:
            _text(case[field])
        category, track = case["category"], case["track"]
        if category not in {"domain", "calculation", "evidence", "safety", "general"}:
            raise ValueError("Unsupported category")
        expected_track = ("evidence_control" if category == "evidence" else
                          "general_retention" if category == "general" else "no_rag")
        if track != expected_track:
            raise ValueError("Category and track disagree")
        messages = case["messages"]
        if not isinstance(messages, list) or not messages:
            raise ValueError("Expected prompt messages")
        for message in messages:
            if not isinstance(message, Mapping) or set(message) != {"role", "content"}:
                raise ValueError("Text-only message schema required")
            if message["role"] not in {"system", "user", "assistant", "tool"}:
                raise ValueError("Unsupported message role")
            _text(message["content"])
        pair = []
        for model in (left_model, right_model):
            output = output_map[cid, model]
            if output.get("smoke") is not False:
                raise ValueError("Explicit non-smoke output required")
            if output.get("status") not in {"completed", "truncated"}:
                raise ValueError("Failed/missing output blocks review")
            answer = _text(output.get("answer"))
            finish = "length" if output["status"] == "truncated" else "stop"
            if output.get("finish_reason") != finish:
                raise ValueError("Status and finish reason disagree")
            if output.get("run_config_sha256") != run_hash:
                raise ValueError("Unmatched generation configuration")
            if output.get("input_sha256") != canonical_sha256(messages):
                raise ValueError("Input identity mismatch")
            if output.get("answer_sha256") != canonical_sha256(answer):
                raise ValueError("Answer identity mismatch")
            pair.append(output)
        rng.shuffle(pair)
        for label, output in zip(("A", "B"), pair):
            candidate = copy.deepcopy(dict(case))
            candidate.update(candidate_id=label, answer=output["answer"],
                             generation_outcome={"status": output["status"],
                                                 "finish_reason": output["finish_reason"]})
            shared.append(candidate)
            mapping.append({"case_id": cid, "candidate_id": label, "model": output["model"],
                            "comparison_role": "left" if output["model"] == left_model else "right",
                            "candidate_sha256": canonical_sha256(candidate)})
    manifest = {"schema_version": VERSION, "case_count": len(case_map),
                "candidate_count": len(shared), "case_ids": sorted(cid[0] for cid in case_map),
                "left_model": left_model, "right_model": right_model,
                "run_config_sha256": run_hash, "protocol_sha256": protocol_hash,
                "candidates_sha256": canonical_sha256(shared),
                "mapping_sha256": canonical_sha256(mapping)}
    return shared, mapping, manifest


def validate_reviews(candidates: Sequence[Mapping[str, Any]], mapping: Sequence[Mapping[str, Any]],
                     reviews: Sequence[Mapping[str, Any]], manifest: Mapping[str, Any],
                     *, reviewer_id: str, reviewer_type: str,
                     protocol_sha256: str) -> list[dict]:
    """Validate one frozen comparison/assessment and return paired numeric records.

    Keep the manifest and mapping owner-side. Reviewer labels are declared, not
    authenticated. No grading, deduplication of assessments, or approval is inferred.
    """
    reviewer_id, reviewer_type = _text(reviewer_id), _text(reviewer_type)
    if reviewer_type not in {"expert", "codex", "external_llm"}:
        raise ValueError("Unsupported reviewer type")
    if manifest.get("schema_version") != VERSION:
        raise ValueError("Unsupported bundle version")
    _hash(manifest.get("run_config_sha256"))
    if manifest.get("protocol_sha256") != _hash(protocol_sha256):
        raise ValueError("Unexpected protocol identity")
    bundle_hash = canonical_sha256(candidates)
    if bundle_hash != manifest.get("candidates_sha256"):
        raise ValueError("Frozen candidates changed")
    if canonical_sha256(mapping) != manifest.get("mapping_sha256"):
        raise ValueError("Frozen mapping changed")
    indexes = [_index(rows, ("case_id", "candidate_id")) for rows in (candidates, mapping, reviews)]
    cm, mm, rm = indexes
    if not cm or set(cm) != set(mm) or set(cm) != set(rm):
        raise ValueError("Review, mapping and candidate coverage must match exactly")
    if any(type(manifest.get(field)) is not int for field in ("case_count", "candidate_count")):
        raise ValueError("Manifest counts must be integers")
    case_ids = sorted({cid for cid, _ in cm})
    if (manifest.get("case_ids") != case_ids or manifest.get("case_count") != len(case_ids)
            or manifest.get("candidate_count") != len(cm) or len(cm) != 2 * len(case_ids)):
        raise ValueError("Manifest coverage mismatch")
    left, right = _text(manifest.get("left_model")), _text(manifest.get("right_model"))
    if left == right:
        raise ValueError("Same model on both sides")
    pairs = {}
    for key in sorted(cm):
        candidate, binding, review = cm[key], mm[key], rm[key]
        if key[1] not in {"A", "B"} or binding.get("candidate_sha256") != canonical_sha256(candidate):
            raise ValueError("Mapping must bind the entire candidate")
        role = binding.get("comparison_role")
        if role not in {"left", "right"} or binding.get("model") != (left if role == "left" else right):
            raise ValueError("Inconsistent comparison role/model")
        if set(review) != REVIEW_FIELDS:
            raise ValueError("Unexpected/missing review fields")
        if (review["reviewer_id"], review["reviewer_type"]) != (reviewer_id, reviewer_type):
            raise ValueError("Mixed or unexpected reviewer identity")
        if review["bundle_sha256"] != bundle_hash or review["protocol_sha256"] != protocol_sha256:
            raise ValueError("Review belongs to another bundle or protocol")
        scores = review["scores"]
        if not isinstance(scores, Mapping) or set(scores) != SCORE_FIELDS:
            raise ValueError("Incomplete score schema")
        for name in SCORE_FIELDS - {"evidence_faithfulness"}:
            _score(scores[name])
        faithfulness = scores["evidence_faithfulness"]
        if candidate["track"] == "evidence_control":
            _score(faithfulness)
        elif faithfulness != NA:
            raise ValueError("Evidence faithfulness must be not_applicable on this track")
        for field in ("preferred", "critical_error"):
            if type(review[field]) is not bool:
                raise ValueError("Flags must be booleans, not coerced strings/numbers")
        allowed_safety = {"pass", "fail"} if candidate["category"] == "safety" else {NA}
        if review["safety"] not in allowed_safety:
            raise ValueError("Safety applicability mismatch")
        tags = review["failure_tags"]
        if not isinstance(tags, list) or not tags:
            raise ValueError("Failure tags required")
        for tag in tags:
            _text(tag)
        if len(set(tags)) != len(tags) or ("none" in tags and len(tags) != 1):
            raise ValueError("Inconsistent failure tags")
        if "none" in tags and (review["critical_error"] or review["safety"] == "fail"):
            raise ValueError("Failure cannot be tagged none")
        meta = {field: candidate[field] for field in
                ("case_id", "dependency_group_id", "category", "track")}
        pair = pairs.setdefault(key[0], {**meta, "left_model": left, "right_model": right})
        if any(pair[field] != value for field, value in meta.items()) or role in pair:
            raise ValueError("Pair metadata or role mismatch")
        pair[role] = {**dict(scores), "TaskCore2": (scores["professional_correctness"] +
                                                   scores["completeness"]) / 2,
                      "critical_error": review["critical_error"], "preferred": review["preferred"],
                      "safety": review["safety"],
                      "truncated": candidate["generation_outcome"]["status"] == "truncated"}
    for pair in pairs.values():
        if "left" not in pair or "right" not in pair:
            raise ValueError("Exactly one left/right candidate required")
        if pair["left"]["preferred"] and pair["right"]["preferred"]:
            raise ValueError("Both candidates cannot be preferred")
    return [pairs[cid] for cid in sorted(pairs)]