protocol.py

"""Shared study protocol: booklet assignment, trial order and response validation.

Used unchanged by the Hugging Face backend (hf_backend_app.py), the local server
(serve.py), the response simulator and the analysis, so every path enforces the same
rules. Depends on the standard library only.
"""

import copy
import hashlib
import json
import os
import random
import re
import threading
import uuid
from datetime import datetime, timedelta, timezone
from pathlib import Path

SCHEMA_VERSION = 1
MAX_PAYLOAD_BYTES = 1024 * 1024
MAX_FEEDBACK_CHARS = 2000
# Crockford base32 without I, L, O, U, grouped as XXXX-XXXX (40 random bits).
PARTICIPANT_PATTERN = re.compile(r"^[0-9A-HJKMNP-TV-Z]{4}-[0-9A-HJKMNP-TV-Z]{4}$")
CHOICES = {"a", "b", "tie"}
SKIP_REASONS = {"video_error"}
# A session that saved within this window still holds its booklet.
ACTIVE_WINDOW = timedelta(hours=2)
# Codes like TEST-0001 are for piloting the live deployment: they never count toward
# booklet balancing and analyze.py ignores them.
TEST_PREFIX = "TEST-"


# ---------------------------------------------------------------------------
# Generic helpers


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}.{os.getpid()}.{uuid.uuid4().hex}.tmp")
    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 utc_now():
    return datetime.now(timezone.utc)


def iso(moment):
    return moment.astimezone(timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z")


def parse_iso(value):
    return datetime.fromisoformat(str(value).replace("Z", "+00:00"))


def deep_merge(base, override):
    merged = copy.deepcopy(base)
    for key, value in override.items():
        if isinstance(value, dict) and isinstance(merged.get(key), dict):
            merged[key] = deep_merge(merged[key], value)
        else:
            merged[key] = copy.deepcopy(value)
    return merged


def load_config(path, override=None):
    config = read_json(path)
    if override:
        config = deep_merge(config, read_json(override))
    return config


def normalize_participant_id(value):
    if not isinstance(value, str):
        raise ValueError("Participant code must be text")
    code = value.strip().upper().replace(" ", "")
    if len(code) == 8 and "-" not in code:
        code = f"{code[:4]}-{code[4:]}"
    if not PARTICIPANT_PATTERN.fullmatch(code):
        raise ValueError("Participant codes look like 7K2M-QX4P")
    return code


# ---------------------------------------------------------------------------
# Trial order


def session_order(manifest, booklet_id, participant_id):
    """Deterministic per-participant order of one booklet's trials.

    Main trials are shuffled with a participant-specific seed; consecutive trials
    never share a prompt when that can be avoided. The first attention check is the
    third pair, so a session abandoned after two pairs can still be screened; the
    other checks are spread evenly over the rest.
    """
    booklet = booklet_by_id(manifest, booklet_id)
    trials = manifest["trials"]
    rng = random.Random(stable_seed(manifest["study_id"], "order", participant_id))

    main = list(booklet["trials"])
    rng.shuffle(main)
    for index in range(1, len(main)):
        if trials[main[index]]["prompt"] != trials[main[index - 1]]["prompt"]:
            continue
        for swap in range(index + 1, len(main)):
            if trials[main[swap]]["prompt"] != trials[main[index - 1]]["prompt"]:
                main[index], main[swap] = main[swap], main[index]
                break

    checks = list(booklet["checks"])
    rng.shuffle(checks)
    total = len(main) + len(checks)
    first = min(2, total - 1)
    positions = [first] if checks else []
    for index in range(1, len(checks)):
        target = first + round(index * (total - first) / len(checks)) + rng.randint(-1, 1)
        positions.append(max(positions[-1] + 2, min(target, total - 1)))

    order = []
    main_iter = iter(main)
    check_iter = iter(checks)
    position_set = set(positions)
    for slot in range(total):
        order.append(next(check_iter) if slot in position_set else next(main_iter, None))
    order = [trial for trial in order if trial is not None]
    order.extend(check_iter)
    return order


def booklet_by_id(manifest, booklet_id):
    for booklet in manifest["booklets"]:
        if booklet["id"] == booklet_id:
            return booklet
    raise ValueError(f"Unknown booklet {booklet_id!r}")


# ---------------------------------------------------------------------------
# Booklet assignment


class Assigner:
    """Hands each new participant the least-filled booklet, in booklet order on ties.

    The load of a booklet counts completed sessions, sessions saved within the
    active window, and the answered fraction of abandoned sessions, so coverage of
    every (prompt, comparison) cell stays balanced even when volunteers drop out.
    Retired booklets (manifest "retired_booklets") go to nobody new. Someone rating
    another set gets the next booklet they have not done, by the same rule, so the
    booklets are used strictly in order and every pair is shown once before any pair
    again. State lives in one JSON file next to the responses.
    """

    def __init__(self, manifest, state_path):
        self.manifest = manifest
        self.state_path = Path(state_path)
        self.lock = threading.Lock()
        self.state = {"participants": {}}
        if self.state_path.exists():
            self.state = read_json(self.state_path)

    def _load(self, now):
        load = {booklet["id"]: 0.0 for booklet in self.manifest["booklets"]}
        for participant, record in self.state["participants"].items():
            if record["booklet"] not in load or participant.startswith(TEST_PREFIX):
                continue
            if record.get("complete"):
                weight = 1.0
            elif now - parse_iso(record["updated_at"]) < ACTIVE_WINDOW:
                weight = 1.0
            else:
                weight = record.get("answered", 0) / max(1, record.get("total", 1))
            load[record["booklet"]] += weight
        return load

    def _person_booklets(self, root):
        return {
            record["booklet"]
            for record in self.state["participants"].values()
            if record.get("root") == root
        }

    def assign(self, participant_id, linked_to=None, now=None):
        now = now or utc_now()
        with self.lock:
            record = self.state["participants"].get(participant_id)
            if record is not None:
                return record["booklet"]
            root = participant_id
            if linked_to and linked_to in self.state["participants"]:
                root = self.state["participants"][linked_to].get("root", linked_to)
            seen = self._person_booklets(root)
            load = self._load(now)
            retired = set(self.manifest.get("retired_booklets", []))
            order = [booklet["id"] for booklet in self.manifest["booklets"] if booklet["id"] not in retired]
            candidates = [booklet for booklet in order if booklet not in seen] or order
            booklet = min(candidates, key=lambda key: (load[key], order.index(key)))
            self.state["participants"][participant_id] = {
                "booklet": booklet,
                "root": root,
                "linked_to": linked_to,
                "assigned_at": iso(now),
                "updated_at": iso(now),
                "answered": 0,
                "total": len(booklet_by_id(self.manifest, booklet)["trials"])
                + len(booklet_by_id(self.manifest, booklet)["checks"]),
                "complete": False,
            }
            write_json_atomic(self.state, self.state_path)
            return booklet

    def booklet_of(self, participant_id):
        record = self.state["participants"].get(participant_id)
        return None if record is None else record["booklet"]

    def record_progress(self, participant_id, answered, complete, now=None):
        now = now or utc_now()
        with self.lock:
            record = self.state["participants"][participant_id]
            record.update(answered=answered, complete=bool(complete), updated_at=iso(now))
            write_json_atomic(self.state, self.state_path)


# ---------------------------------------------------------------------------
# Response validation


def _is_count(value):
    return isinstance(value, int) and not isinstance(value, bool) and value >= 0


def validate_payload(payload, manifest, booklet_id):
    """Raise ValueError unless payload is a well-formed session for booklet_id."""
    if not isinstance(payload, dict):
        raise ValueError("Payload must be an object")
    expected = {
        "schema_version": SCHEMA_VERSION,
        "study_id": manifest["study_id"],
        "booklet_id": booklet_id,
    }
    for key, value in expected.items():
        if payload.get(key) != value:
            raise ValueError(f"{key} does not match this study build")
    # Builds made with build_study.py --extend keep every earlier trial, so pages
    # loaded before an extension can keep saving.
    if payload.get("build_id") not in [manifest["build_id"], *manifest.get("compatible_builds", [])]:
        raise ValueError("build_id does not match this study build")
    normalize_participant_id(payload.get("participant_id"))
    linked_to = payload.get("linked_to")
    if linked_to is not None:
        normalize_participant_id(linked_to)
    for key in ("started_at", "updated_at"):
        parse_iso(payload.get(key))
    if not isinstance(payload.get("complete"), bool):
        raise ValueError("complete must be true or false")

    profile = payload.get("profile", {})
    allowed_profile = {option["id"] for option in manifest["profile"]["options"]} | {None}
    if not isinstance(profile, dict) or profile.get("experience") not in allowed_profile:
        raise ValueError("Invalid profile answer")
    client = payload.get("client", {})
    if not isinstance(client, dict) or len(json.dumps(client)) > 1000:
        raise ValueError("Invalid client information")
    feedback = payload.get("feedback", "")
    if not isinstance(feedback, str) or len(feedback) > MAX_FEEDBACK_CHARS:
        raise ValueError("Feedback is too long")

    booklet = booklet_by_id(manifest, booklet_id)
    allowed = set(booklet["trials"]) | set(booklet["checks"])
    criteria = {criterion["id"] for criterion in manifest["criteria"]}
    choices_allowed = CHOICES if manifest["allow_tie"] else CHOICES - {"tie"}
    trials = payload.get("trials")
    if not isinstance(trials, list):
        raise ValueError("trials must be a list")
    seen = set()
    for entry in trials:
        trial_id = entry.get("trial_id") if isinstance(entry, dict) else None
        if trial_id not in allowed or trial_id in seen:
            raise ValueError(f"Invalid or repeated trial {trial_id!r}")
        seen.add(trial_id)
        trial = manifest["trials"][trial_id]
        if entry.get("a") != trial["a"] or entry.get("b") != trial["b"]:
            raise ValueError(f"Trial {trial_id} videos do not match the study")
        skipped = entry.get("skipped")
        choices = entry.get("choices")
        if skipped is not None:
            if skipped not in SKIP_REASONS or choices != {}:
                raise ValueError(f"Trial {trial_id} has an invalid skip")
        elif not isinstance(choices, dict) or set(choices) != criteria or not set(
            choices.values()
        ) <= choices_allowed:
            raise ValueError(f"Trial {trial_id} must answer every question")
        for key in ("position", "plays", "watch_ms", "elapsed_ms"):
            if not _is_count(entry.get(key)):
                raise ValueError(f"Trial {trial_id} has an invalid {key}")
        parse_iso(entry.get("answered_at"))
    if payload["complete"] and seen != allowed:
        raise ValueError("A complete session must contain every trial")


# ---------------------------------------------------------------------------
# API operations shared by the hosted backend and the local server


class StudyService:
    def __init__(self, manifest, response_root):
        self.manifest = manifest
        root = Path(response_root) / manifest["study_id"]
        self.sessions_dir = root / "sessions"
        self.assigner = Assigner(manifest, root / "assignments.json")

    def start_session(self, participant_id, linked_to=None):
        participant_id = normalize_participant_id(participant_id)
        linked_to = normalize_participant_id(linked_to) if linked_to else None
        booklet_id = self.assigner.assign(participant_id, linked_to)
        saved_path = self.sessions_dir / f"{participant_id}.json"
        saved = read_json(saved_path) if saved_path.exists() else None
        return {
            "participant_id": participant_id,
            "booklet_id": booklet_id,
            "build_id": self.manifest["build_id"],
            "order": session_order(self.manifest, booklet_id, participant_id),
            "saved": saved,
        }

    def save_session(self, payload):
        if isinstance(payload, str):
            payload = json.loads(payload)
        if len(json.dumps(payload).encode("utf-8")) > MAX_PAYLOAD_BYTES:
            raise ValueError("Payload is too large")
        participant_id = normalize_participant_id(payload.get("participant_id"))
        if payload["participant_id"] != participant_id:
            raise ValueError("Participant code is not normalized")
        booklet_id = self.assigner.booklet_of(participant_id)
        if booklet_id is None:
            raise ValueError("Unknown participant; start a session first")
        validate_payload(payload, self.manifest, booklet_id)
        stored = dict(payload, received_at=iso(utc_now()))
        write_json_atomic(stored, self.sessions_dir / f"{participant_id}.json")
        self.assigner.record_progress(participant_id, len(payload["trials"]), payload["complete"])
        return {"saved": True, "answered": len(payload["trials"]), "complete": payload["complete"]}