iop/scripts/agent_benchmark/rubric.py

"""Strict versioned benchmark worksheet contract.

Automatic web gates establish scoring eligibility; they are intentionally not
represented in this 100-point worksheet and can never contribute points.
"""

from __future__ import annotations

import json
import os
import stat
from dataclasses import dataclass
from pathlib import Path
from types import MappingProxyType
from typing import Any, Mapping

from scripts.agent_benchmark.manifest import (
    ONE_SHOT_RUBRIC_VERSION,
    RUBRIC_VERSION,
)


MAX_WORKSHEET_BYTES = 64 * 1024
MAX_EVIDENCE_CHARS = 4096
RUBRIC_CATEGORIES = (
    ("task_fidelity", 25),
    ("visual_hierarchy", 25),
    ("responsive_composition", 20),
    ("typography_readability", 15),
    ("polish_consistency", 15),
)
ONE_SHOT_RUBRIC_CATEGORIES = (
    ("requirements_fidelity", 25),
    ("visual_completeness", 25),
    ("responsive_accessibility", 15),
    ("image_detail_usage", 10),
    ("behavior_stability", 10),
    ("code_quality", 10),
    ("self_verification", 5),
)
RUBRIC_CATEGORIES_BY_VERSION: Mapping[str, tuple[tuple[str, int], ...]] = (
    MappingProxyType(
        {
            RUBRIC_VERSION: RUBRIC_CATEGORIES,
            ONE_SHOT_RUBRIC_VERSION: ONE_SHOT_RUBRIC_CATEGORIES,
        }
    )
)


class RubricError(Exception):
    """A worksheet is missing, malformed, non-canonical, or out of bounds."""


def rubric_categories(version: str) -> tuple[tuple[str, int], ...]:
    """Return the immutable ordered category table for a known rubric version."""
    if not isinstance(version, str):
        raise RubricError("rubric version is invalid")
    try:
        return RUBRIC_CATEGORIES_BY_VERSION[version]
    except KeyError as exc:
        raise RubricError("rubric version is invalid") from exc


@dataclass(frozen=True)
class CategoryScore:
    id: str
    max_score: int
    score: int
    evidence: str


@dataclass(frozen=True)
class Worksheet:
    rubric_version: str
    categories: tuple[CategoryScore, ...]
    total: int

    def as_dict(self) -> dict[str, Any]:
        return {
            "rubric_version": self.rubric_version,
            "categories": [
                {
                    "id": item.id,
                    "max_score": item.max_score,
                    "score": item.score,
                    "evidence": item.evidence,
                }
                for item in self.categories
            ],
            "total": self.total,
        }


def canonical_worksheet_bytes(worksheet: Worksheet) -> bytes:
    if not isinstance(worksheet, Worksheet):
        raise RubricError("worksheet object is invalid")
    return (
        json.dumps(
            worksheet.as_dict(),
            sort_keys=True,
            separators=(",", ":"),
            ensure_ascii=True,
        ).encode("ascii")
        + b"\n"
    )


def validate_worksheet(
    value: Any, expected_version: str | None = None
) -> Worksheet:
    if not isinstance(value, dict) or set(value) != {
        "rubric_version", "categories", "total",
    }:
        raise RubricError("worksheet schema is invalid")
    rubric_version = value["rubric_version"]
    expected_categories = rubric_categories(rubric_version)
    if expected_version is not None:
        rubric_categories(expected_version)
    if expected_version is not None and rubric_version != expected_version:
        raise RubricError("worksheet rubric version is invalid")
    raw_categories = value["categories"]
    if not isinstance(raw_categories, list) or len(raw_categories) != len(
        expected_categories
    ):
        raise RubricError("worksheet categories are invalid")

    categories: list[CategoryScore] = []
    for raw, (expected_id, expected_max) in zip(
        raw_categories, expected_categories
    ):
        if not isinstance(raw, dict) or set(raw) != {
            "id", "max_score", "score", "evidence",
        }:
            raise RubricError("worksheet category schema is invalid")
        score = raw["score"]
        evidence = raw["evidence"]
        if (
            raw["id"] != expected_id
            or raw["max_score"] != expected_max
            or isinstance(score, bool)
            or not isinstance(score, int)
            or not 0 <= score <= expected_max
            or not isinstance(evidence, str)
            or not evidence.strip()
            or len(evidence) > MAX_EVIDENCE_CHARS
            or any(ord(char) < 0x20 and char not in "\n\t" for char in evidence)
        ):
            raise RubricError("worksheet category is invalid")
        categories.append(
            CategoryScore(expected_id, expected_max, score, evidence)
        )

    total = value["total"]
    expected_total = sum(item.score for item in categories)
    if (
        isinstance(total, bool)
        or not isinstance(total, int)
        or total != expected_total
        or not 0 <= total <= 100
    ):
        raise RubricError("worksheet total is invalid")
    return Worksheet(rubric_version, tuple(categories), total)


def load_worksheet(
    path: str | Path, expected_version: str | None = None
) -> Worksheet:
    target = Path(path)
    flags = os.O_RDONLY | os.O_CLOEXEC | os.O_NONBLOCK
    if hasattr(os, "O_NOFOLLOW"):
        flags |= os.O_NOFOLLOW
    try:
        fd = os.open(target, flags)
    except OSError as exc:
        raise RubricError("worksheet is unavailable") from exc
    try:
        info = os.fstat(fd)
        if not stat.S_ISREG(info.st_mode) or info.st_size > MAX_WORKSHEET_BYTES:
            raise RubricError("worksheet must be a bounded regular file")
        raw = bytearray()
        while len(raw) < info.st_size:
            chunk = os.read(fd, info.st_size - len(raw))
            if not chunk:
                raise RubricError("worksheet changed while reading")
            raw.extend(chunk)
        if os.read(fd, 1):
            raise RubricError("worksheet changed while reading")
    except OSError as exc:
        raise RubricError("worksheet is unavailable") from exc
    finally:
        os.close(fd)
    try:
        value = json.loads(bytes(raw).decode("utf-8"))
    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise RubricError("worksheet JSON is invalid") from exc
    return validate_worksheet(value, expected_version=expected_version)