protocol.py

"""Shared manifest ordering and response validation for the human evaluation."""

import hashlib
import json
import os
import random
import re
import uuid
from pathlib import Path


DIMENSIONS = {
    "edit_a_applied",
    "edit_b_applied",
    "composition_success",
    "source_preservation",
    "absence_unintended_changes",
    "visual_realism",
    "overall",
}
MAX_REQUEST_BYTES = 2 * 1024 * 1024
RATER_ID_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_-]{2,63}$")


def read_json(path):
    with Path(path).open("r", encoding="utf-8") as handle:
        return json.load(handle)


def write_json_atomic(value, path):
    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)
    temporary = path.with_name(f".{path.name}.tmp.{os.getpid()}.{uuid.uuid4().hex}")
    temporary.write_text(
        json.dumps(value, indent=2, sort_keys=True) + "\n", encoding="utf-8"
    )
    temporary.replace(path)


def stable_seed(*parts):
    payload = ":".join(str(part) for part in parts)
    return int.from_bytes(hashlib.sha256(payload.encode("utf-8")).digest()[:8], "big")


def validate_rater_id(rater_id):
    if not isinstance(rater_id, str) or not RATER_ID_PATTERN.fullmatch(rater_id):
        raise ValueError(
            "Participant code must be 3-64 characters using letters, numbers, "
            "underscores, or hyphens."
        )


def public_manifest(study_root, mode, rater_id):
    if mode != "main":
        raise ValueError("mode must be main")
    validate_rater_id(rater_id)
    manifest = read_json(Path(study_root) / f"{mode}_public.json")
    assignments = manifest.pop("assignment_orders", {})
    by_id = {item["blind_id"]: item for item in manifest["items"]}
    order = assignments.get(rater_id)
    if order is None:
        order = list(by_id)
        random.Random(stable_seed(manifest["study_id"], mode, rater_id)).shuffle(order)
    manifest["items"] = [by_id[item_id] for item_id in order]
    manifest["rater_id"] = rater_id
    return manifest


def validate_submission(payload, manifest):
    if payload.get("schema_version") != manifest["schema_version"]:
        raise ValueError("Response schema version does not match the study")
    if payload.get("study_id") != manifest["study_id"]:
        raise ValueError("Response study id does not match")
    if payload.get("set") != manifest["set"]:
        raise ValueError("Response set does not match")
    validate_rater_id(payload.get("rater_id"))
    if payload.get("rater_id") != manifest["rater_id"]:
        raise ValueError("Response rater code does not match")
    tutorial_version = payload.get("tutorial_version", 0)
    if (
        isinstance(tutorial_version, bool)
        or not isinstance(tutorial_version, int)
        or not 0 <= tutorial_version <= 100
    ):
        raise ValueError("tutorial_version must be an integer between 0 and 100")
    allowed_ids = {item["blind_id"] for item in manifest["items"]}
    seen = set()
    responses = payload.get("responses")
    if not isinstance(responses, list):
        raise ValueError("responses must be a list")
    for response in responses:
        blind_id = response.get("blind_id")
        if blind_id not in allowed_ids or blind_id in seen:
            raise ValueError(f"Invalid or repeated blinded item id: {blind_id!r}")
        seen.add(blind_id)
        ratings = response.get("ratings")
        unclear = response.get("unclear")
        if not isinstance(ratings, dict) or set(ratings) != DIMENSIONS:
            raise ValueError(f"Item {blind_id} has an invalid rating key set")
        if not isinstance(unclear, dict) or set(unclear) != {"edit_a", "edit_b"}:
            raise ValueError(f"Item {blind_id} has invalid unclear flags")
        if not all(isinstance(value, bool) for value in unclear.values()):
            raise ValueError(f"Item {blind_id} unclear flags must be booleans")
        comment = response.get("comment", "")
        if not isinstance(comment, str) or len(comment) > 1000:
            raise ValueError(f"Item {blind_id} has an invalid optional comment")
        for dimension, value in ratings.items():
            may_be_null = (
                dimension == "edit_a_applied" and unclear["edit_a"]
            ) or (dimension == "edit_b_applied" and unclear["edit_b"])
            if value is None and may_be_null:
                continue
            if (
                isinstance(value, bool)
                or not isinstance(value, int)
                or not 1 <= value <= 5
            ):
                raise ValueError(f"Item {blind_id} has invalid {dimension} rating")
        elapsed = response.get("elapsed_seconds")
        if elapsed is not None and (
            isinstance(elapsed, bool)
            or not isinstance(elapsed, (int, float))
            or elapsed < 0
        ):
            raise ValueError(f"Item {blind_id} has invalid elapsed_seconds")
    if payload.get("complete") and seen != allowed_ids:
        raise ValueError("A completed response must contain every assigned item")