runtime/src/decisionnode_edge/systemone.py

"""POST /v1/systemone: the request and answer format public decision-model benchmarks use, read onto the runtime's own.

    translate(body)         the format's pictures onto the runtime's, before the parse:
                            - `assets`: {reference: {"type": "image", "format": "png" | "jpg" | "jpeg", "data_base64"}}.
                              Every place a reference appears (as a whole string value or inside a longer string, in the
                              state, a question's instructions or its criteria; never in an object key) is the picture
                              "in its place". The model reads every picture of a request before the state, labelled
                              `Picture <n> (<id>):`, so each reference is replaced in the text by `Picture <n>` and the
                              asset becomes picture n. A picture is read once however many times it is referenced, and
                              every question sees every picture.
                            - chat content parts, `{"type": "image_url", "image_url": {"url": "data:..."}}` (or
                              `"image_url": "data:..."`), anywhere in the state, the instructions or the criteria: the
                              same, the part replaced by `Picture <n>`; an array made only of such parts and
                              `{"type": "text", "text"}` parts is read as its text, the parts joined by a line break.
                            - `images` as a list of data URLs.
                            - `box` questions: in the picture the instructions name (`image_for_box`) when the request
                              holds several of the format's pictures; answered in fractions (`normalised_boxes`, after
                              the checks). A box with `max` keeps the runtime's list answer.
                            The legend of a score whose levels held a reference echoes the levels as sent.
    box_answer(answer)      a one-box answer in the format's shape: {"type": "box", "box": {"top", "bottom", "left",
                            "right"}, "confidence"}, the edges as fractions of the picture's height and width, and
                            `confidence` the probability that the box is right where the model folder has a box
                            calibration (BoxCalibration), else `present`.
    mark_capacity(...)      the words benchmark harnesses look for in an error ("maximum context length", "options per
                            choice", "a score takes 2 to 10 levels"), so that a limit of the model is recorded as one.

Choice and score probabilities on this route are normalised first and rounded to lists that sum to exactly 1.00
(answers.round_probabilities with exact_sum). Headers the format defines for pictures and long inputs are accepted and
never required. A multipart body (`part` assets) is not taken: an asset with `part` is a 400 that says to send
`data_base64`.
"""

from __future__ import annotations

import base64
import binascii
import hashlib
import json
import math
import re
from dataclasses import dataclass, field
from typing import Any, Iterable

from .answers import round2
from .request import IMAGE_MEDIA_TYPES, MAX_LEVELS, MAX_OPTIONS, RequestError

ASSET_FORMATS = {"png": "image/png", "jpg": "image/jpeg", "jpeg": "image/jpeg"}  # the format's Asset.format
MAX_REFERENCE_BYTES = 1024  # one reference, UTF-8
MAX_REFERENCES_BYTES = 65536  # every reference of a request together
MAX_NESTING = 128  # a body nested deeper is left as it is (the parse refuses it)
PICTURE = "Picture {n}"  # what the text reads in a reference's place: the model's own label of picture n
BOX_CALIBRATION_FILE = "box_calibration.json"  # in the model folder, optional
BOX_CALIBRATION_FORMAT = "dn-box-calibration-1"
BOX_CALIBRATION_FEATURES = ["logit_present", "logit_confidence"]
_WHITESPACE = re.compile(r"[\t\n\x0b\x0c\r ]")
_BACKTICKED = re.compile(r"`([^`\n]{1,512})`")
_PATH_PART = re.compile(r"([^.\[\]]+)|\[(\d+)\]")


@dataclass
class Translation:
    """What translate() did to a body, for the parse and the answer."""

    images: list[dict] = field(default_factory=list)  # the image objects it added, in picture order
    legends: dict[str, list] = field(default_factory=dict)  # score question id -> its levels as sent
    boxes: list[str] = field(default_factory=list)  # the box questions, answered in the format's shape
    references: int = 0  # the references replaced in the text


# ---------------------------------------------------------------------------
# Errors
# ---------------------------------------------------------------------------


def field_error(kind: str, loc: Iterable[Any], msg: str, value: Any, **extra: Any) -> dict:
    return {"type": kind, "loc": list(loc), "msg": msg, "input": value, **extra}


def _invalid(entry: dict) -> RequestError:
    return RequestError(f"{'.'.join(str(part) for part in entry['loc'])}: {entry['msg']}", 422, errors=[entry])


def _asset_error(ref: str, problem: str) -> RequestError:
    shown = ref if len(ref) <= 80 else ref[:77] + "..."
    return RequestError(f"Asset {shown!r}: {problem}.")


def _too_large(message: str) -> RequestError:
    return RequestError(message, 413, "request_too_large_error")


# ---------------------------------------------------------------------------
# Pictures
# ---------------------------------------------------------------------------


def _sniff(text: str) -> str | None:
    """The type of a base64 picture from its first bytes (PNG, JPEG, WebP), or None."""
    head = text[:64]
    try:
        blob = base64.b64decode(head[: len(head) - len(head) % 4])
    except (binascii.Error, ValueError):
        return None
    if blob.startswith(b"\x89PNG\r\n\x1a\n"):
        return "image/png"
    if blob.startswith(b"\xff\xd8\xff"):
        return "image/jpeg"
    if blob[:4] == b"RIFF" and blob[8:12] == b"WEBP":
        return "image/webp"
    return None


def _clean_base64(text: str) -> str:
    """Base64 text as the decoder reads it: a `data:` prefix and any ASCII whitespace dropped, missing padding added."""
    if text[:5].lower() == "data:":
        text = text.partition(",")[2]
    if _WHITESPACE.search(text):
        text = _WHITESPACE.sub("", text)
    if len(text) % 4 in (2, 3) and "=" not in text:
        text += "=" * (4 - len(text) % 4)
    return text


def _asset_image(ref: str, asset: Any) -> dict:
    """The runtime's image object (without its id) for one asset. Raises RequestError."""
    if asset.get("type") is not None and asset.get("type") != "image":
        raise _invalid(field_error("literal_error", ("body", "assets", ref, "type"), "Input should be 'image'", asset.get("type"), ctx={"expected": "'image'"}))
    if asset.get("part") is not None:
        raise _asset_error(ref, "multipart image parts are not taken by this server; send the image base64-encoded in data_base64 in a JSON body")
    data = asset.get("data_base64")
    if data is None:
        raise _asset_error(ref, "data_base64 is required")
    if not isinstance(data, str):
        raise _invalid(field_error("string_type", ("body", "assets", ref, "data_base64"), "Input should be a valid string", data))
    fmt = asset.get("format")
    declared = None
    if fmt is not None:
        if not isinstance(fmt, str) or fmt not in ASSET_FORMATS:
            raise _asset_error(ref, "format must be png, jpg or jpeg")
        declared = ASSET_FORMATS[fmt]
    data = _clean_base64(data)
    if not data:
        raise _asset_error(ref, "data_base64 is empty")
    media = _sniff(data) or declared  # the format is read from the bytes
    if media is None:
        raise _asset_error(ref, "the data is not a PNG or JPEG image")
    return {"media_type": media, "data": data}


def _part_image(part: dict) -> dict:
    """The runtime's image object (without its id) for an `image_url` content part. Raises RequestError."""
    url = part.get("image_url")
    if isinstance(url, dict):
        url = url.get("url")
    if not isinstance(url, str) or not url:
        raise RequestError("An image_url part needs a url: a data:image/...;base64 URL.")
    scheme = url[:8].lower()
    if scheme.startswith("data:"):
        head, comma, data = url.partition(",")
        fields = head[5:].split(";")
        if not comma or len(fields) < 2 or fields[-1].strip().lower() != "base64":
            raise RequestError("An image_url data URL must be data:<media type>;base64,<data>.")
        data = _clean_base64(data)
        declared = fields[0].strip().lower()
        declared = "image/jpeg" if declared == "image/jpg" else declared
        media = _sniff(data) or (declared if declared in IMAGE_MEDIA_TYPES else None)
        if media is None:
            raise RequestError("An image_url part must hold a PNG, JPEG or WebP image.")
        return {"media_type": media, "data": data}
    if scheme == "https://":
        raise RequestError("Image URLs are not fetched on this deployment: send the image as a base64 data URL.")
    raise RequestError("An image_url part must be a base64 data URL.")


def _is_image_part(value: Any) -> bool:
    return isinstance(value, dict) and value.get("type") == "image_url" and "image_url" in value


def _is_text_part(value: Any) -> bool:
    return isinstance(value, dict) and value.get("type") == "text" and isinstance(value.get("text"), str) and set(value) <= {"type", "text"}


# ---------------------------------------------------------------------------
# The walk
# ---------------------------------------------------------------------------


class _Walk:
    """Replaces references and image parts in the free-form parts of a request, numbering the pictures as it meets them."""

    def __init__(self, refs: Iterable[str], first: int):
        self.refs = list(refs)
        self.first = first  # the picture number of the first picture this walk adds, less one (the body's own images)
        self.order: list[tuple[str, Any]] = []  # ("ref", reference) or ("part", image object), in picture order
        self.by_ref: dict[str, int] = {}
        self.by_url: dict[str, int] = {}
        self.replaced = 0
        ordered = sorted(self.refs, key=len, reverse=True)  # at one position the longest reference wins
        self.pattern = re.compile("|".join(re.escape(ref) for ref in ordered)) if ordered else None
        self.whole = set(self.refs)

    def _number_of_ref(self, ref: str) -> int:
        if ref not in self.by_ref:
            self.order.append(("ref", ref))
            self.by_ref[ref] = self.first + len(self.order)
        return self.by_ref[ref]

    def _number_of_part(self, part: dict) -> int:
        image = _part_image(part)
        key = image.get("data") or ""
        if key not in self.by_url:
            self.order.append(("part", image))
            self.by_url[key] = self.first + len(self.order)
        return self.by_url[key]

    def text(self, value: str) -> str:
        if self.pattern is None:
            return value
        if value in self.whole:  # a whole value first
            self.replaced += 1
            return PICTURE.format(n=self._number_of_ref(value))
        if not any(ref in value for ref in self.refs):
            return value

        def one(match: "re.Match[str]") -> str:
            self.replaced += 1
            return PICTURE.format(n=self._number_of_ref(match.group(0)))

        return self.pattern.sub(one, value)

    def value(self, value: Any) -> Any:
        """`value` with every reference and image part replaced (a new object where anything changed)."""
        if isinstance(value, str):
            return self.text(value)
        if _is_image_part(value):
            self.replaced += 1
            return PICTURE.format(n=self._number_of_part(value))
        if isinstance(value, list):
            if value and any(_is_image_part(item) for item in value) and all(_is_image_part(item) or _is_text_part(item) for item in value):
                # a content-part array: its text, the pictures in their places
                return "\n".join(self.value(item) if _is_image_part(item) else self.text(item["text"]) for item in value)
            out = [self.value(item) for item in value]
            return out if any(a is not b for a, b in zip(out, value)) else value
        if isinstance(value, dict):
            out = {key: self.value(item) for key, item in value.items()}  # keys stay text
            return out if any(out[key] is not value[key] for key in value) else value
        return value


def _question(walk: _Walk, question: Any) -> Any:
    """A question with its instructions and criteria walked (its type and every other field as sent)."""
    if not isinstance(question, dict):
        return question
    out = dict(question)
    if "instructions" in out:
        out["instructions"] = walk.value(out["instructions"])
    criteria = out.get("criteria")
    if isinstance(criteria, dict):
        out["criteria"] = {key: walk.value(item) for key, item in criteria.items()}  # an option's name is a key: never replaced
    elif isinstance(criteria, list):
        out["criteria"] = [walk.value(item) for item in criteria]
    return out


# ---------------------------------------------------------------------------
# Box questions
# ---------------------------------------------------------------------------


def _resolve(state: Any, path: str) -> Any:
    """The value a backticked path names in the state (`photo`, `frames[2].frame`, `faces.happy`), or None."""
    value = state
    for name, index in _PATH_PART.findall(path.strip()):
        if name:
            if not isinstance(value, dict) or name not in value:
                return None
            value = value[name]
        else:
            if not isinstance(value, list) or int(index) >= len(value):
                return None
            value = value[int(index)]
    return value


def _first_ref(value: Any, refs: set[str]) -> str | None:
    """The first reference held whole by a string inside `value` (depth first, in order)."""
    if isinstance(value, str):
        return value if value in refs else None
    items = value.values() if isinstance(value, dict) else value if isinstance(value, list) else ()
    for item in items:
        found = _first_ref(item, refs)
        if found is not None:
            return found
    return None


def image_for_box(instructions: Any, state: Any, refs: set[str], numbers: dict[str, int], in_state: list[int], own: set[int], count: int) -> int:
    """The picture number a box question looks in when the request holds several pictures (the path in backticks names
    which one to search): the picture a backticked path of the instructions names in the state (a reference held there,
    whole); else the first picture of the state; else the first picture the instructions do not hold themselves (a
    picture in a box's instructions shows the target, it is not the picture to search); else picture 1."""
    text = instructions if isinstance(instructions, str) else "" if instructions is None else str(instructions)
    for path in _BACKTICKED.findall(text):
        found = _first_ref(_resolve(state, path), refs)
        if found is not None and found in numbers:
            return numbers[found]
    if in_state:
        return in_state[0]
    for n in range(1, count + 1):
        if n not in own:
            return n
    return 1


def question_refs(question: dict, refs: Iterable[str]) -> set[str]:
    """The references a question's own instructions and criteria hold (whole or inside a string)."""
    found: set[str] = set()
    wanted = list(refs)

    def visit(value: Any) -> None:
        if isinstance(value, str):
            found.update(ref for ref in wanted if ref in value)
        elif isinstance(value, dict):
            for item in value.values():
                visit(item)
        elif isinstance(value, list):
            for item in value:
                visit(item)

    visit(question.get("instructions"))
    visit(question.get("criteria"))
    return found


# ---------------------------------------------------------------------------
# The request
# ---------------------------------------------------------------------------


def _within_nesting(value: Any, limit: int = MAX_NESTING) -> bool:
    """At most `limit` containers inside each other (walked without recursion)."""
    stack = [(value, 0)]
    while stack:
        item, depth = stack.pop()
        if isinstance(item, (dict, list)):
            if depth + 1 > limit:
                return False
            stack.extend((child, depth + 1) for child in (item.values() if isinstance(item, dict) else item))
    return True


def _check_assets(assets: Any) -> dict:
    if not isinstance(assets, dict):
        raise _invalid(field_error("dict_type", ("body", "assets"), "Input should be a valid dictionary", assets))
    if not assets:
        raise _invalid(field_error("too_short", ("body", "assets"), "Dictionary should have at least 1 item after validation, not 0", assets, ctx={"field_type": "Dictionary", "min_length": 1, "actual_length": 0}))
    total = 0
    for ref, asset in assets.items():
        if not isinstance(asset, dict):
            raise _invalid(field_error("model_attributes_type", ("body", "assets", ref), "Input should be a valid dictionary or object to extract fields from", asset))
        if not ref.strip():
            raise RequestError("An asset reference cannot be empty or only whitespace.")
        size = len(ref.encode("utf-8", "surrogatepass"))
        if size > MAX_REFERENCE_BYTES:
            raise _too_large(f"An asset reference is at most {MAX_REFERENCE_BYTES} bytes of UTF-8.")
        total += size
    if total > MAX_REFERENCES_BYTES:
        raise _too_large(f"The asset references of a request are at most {MAX_REFERENCES_BYTES} bytes of UTF-8 together.")
    return assets


def _may_hold_parts(value: Any) -> bool:
    """Cheap: whether a value may hold an image_url part (any dict with that type somewhere), walked without recursion."""
    stack = [value]
    while stack:
        item = stack.pop()
        if isinstance(item, dict):
            if item.get("type") == "image_url":
                return True
            stack.extend(child for child in item.values() if isinstance(child, (dict, list)))
        elif isinstance(item, list):
            stack.extend(child for child in item if isinstance(child, (dict, list)))
    return False


def translate(body: Any, *, max_images: int) -> tuple[Any, Translation]:
    """A /v1/systemone body with the format's pictures on the runtime's (module docstring), and what was done. A body
    that is not an object with a question map is returned as it is (the parse refuses it). Raises RequestError for an
    asset or an image part the runtime cannot take."""
    done = Translation()
    if not isinstance(body, dict) or not isinstance(body.get("questions"), dict):
        return body, done
    assets = body.get("assets")
    has_assets = assets is not None
    questions = body["questions"]
    has_parts = _may_hold_parts(body.get("state")) or any(_may_hold_parts(q.get("instructions")) or _may_hold_parts(q.get("criteria")) for q in questions.values() if isinstance(q, dict))
    boxes = [qid for qid, q in questions.items() if isinstance(q, dict) and q.get("type") == "box"]
    url_images = isinstance(body.get("images"), list) and any(isinstance(image, str) for image in body["images"])
    if not has_assets and not has_parts and not boxes and not url_images:
        if "assets" in body:  # `"assets": null` equals no assets
            body = {name: value for name, value in body.items() if name != "assets"}
        return body, done
    if not _within_nesting(body):
        return body, done  # the parse refuses it as it refuses any body that deep
    out = {name: value for name, value in body.items() if name != "assets"}
    refs: list[str] = list(_check_assets(assets)) if has_assets else []
    if len(refs) > max_images:
        raise RequestError(f"Too many images: {len(refs)}, the limit is {max_images}.")
    own_images = body.get("images") if isinstance(body.get("images"), list) else []
    if url_images:
        # `images` as a list of data URLs: each an image object, in the list's order
        if len(own_images) > max_images:
            raise RequestError(f"Too many images: {len(own_images)}, the limit is {max_images}.")
        listed = []
        for i, image in enumerate(own_images):
            if isinstance(image, str):
                image = {"id": "image" if len(own_images) == 1 else f"image_{i + 1}", **_part_image({"image_url": image})}
            listed.append(image)
        own_images = listed
        out["images"] = listed
    walk = _Walk(refs, len(own_images))
    state = body.get("state")
    out["state"] = walk.value(state)
    in_state = sorted(set(walk.by_ref.values()) | set(walk.by_url.values()))
    walked: dict[str, Any] = {}
    for qid, question in questions.items():
        walked[qid] = _question(walk, question)
        if isinstance(question, dict) and question.get("type") == "score" and walked[qid].get("criteria") != question.get("criteria"):
            done.legends[str(qid)] = question.get("criteria")  # the legend echoes the levels as sent
    unused = [ref for ref in refs if ref not in walk.by_ref]
    if unused:
        raise _asset_error(unused[0], "it is not referenced in the state, the instructions or the criteria")
    added = len(walk.order)
    total = len(own_images) + added
    if total > max_images:
        raise RequestError(f"Too many images: {total}, the limit is {max_images}.")
    taken = {image.get("id") for image in own_images if isinstance(image, dict)}
    images = []
    for i, (kind, item) in enumerate(walk.order):
        n = len(own_images) + i + 1
        image_id = "image" if total == 1 else f"image_{n}"
        while image_id in taken:
            image_id += "_"
        taken.add(image_id)
        image = _asset_image(item, assets[item]) if kind == "ref" else dict(item)
        images.append({"id": image_id, **image})
    if images:
        out["images"] = list(own_images) + images
    done.images = images
    done.references = walk.replaced
    ids = [image.get("id") for image in out.get("images") or [] if isinstance(image, dict)]
    numbers = dict(walk.by_ref)
    for qid in boxes:
        # the picture is chosen only for the format's pictures (assets, image parts, data URLs), which no `image` field
        # can name; with image objects alone a box without `image` among several stays the 422 of the runtime's parse
        if walked[qid].get("image") is None and len(ids) > 1 and (added or url_images):
            question = dict(walked[qid])
            own = {numbers[ref] for ref in question_refs(questions[qid], refs) if ref in numbers}
            n = image_for_box(questions[qid].get("instructions"), state, set(refs), numbers, in_state, own, len(ids))
            question["image"] = ids[n - 1]
            walked[qid] = question
        done.boxes.append(str(qid))
    out["questions"] = walked
    return out, done


def normalised_boxes(specs: Iterable[Any]) -> None:
    """The box questions of a /v1/systemone request answered in fractions of the picture (the format's edges), set on the
    parsed specs (places.PlaceSpec) after their checks. A box with `max` keeps the units the caller sent."""
    for spec in specs:
        if spec.shape == "box" and spec.max is None:
            spec.units = "normalized"


# ---------------------------------------------------------------------------
# The answer
# ---------------------------------------------------------------------------


class BoxCalibrationError(ValueError):
    """A box calibration file the runtime does not take (the server does not start on it)."""


@dataclass(frozen=True)
class BoxCalibration:
    """The `confidence` of a box on /v1/systemone: the probability that the box is right (an intersection over union of
    0.5 or more with the true box), read from the one-box answer's own two numbers, `present` and `confidence`, as the
    answer writes them (2 decimals). A logistic map, fitted per model and kept beside it as box_calibration.json:

        {"format": "dn-box-calibration-1", "features": ["logit_present", "logit_confidence"], "clip": 0.005,
         "intercept": a, "coef": [b, c], "model": ..., "fit": {...}}

    p = sigmoid(a + b logit(present) + c logit(confidence)), each input clipped to [clip, 1 - clip] first, the result
    rounded to 2 decimals. Monotone: no coefficient below 0, so neither number can lower the probability by rising."""

    intercept: float
    coef: tuple[float, float]
    clip: float
    sha256: str
    model: str = ""

    @classmethod
    def load(cls, path: str, sha256: str = "") -> "BoxCalibration":
        """Read and check a calibration file; `sha256`, when given, must be the file's."""
        try:
            with open(path, "rb") as f:
                data = f.read()
        except OSError as error:
            raise BoxCalibrationError(f"box calibration {path}: {error}") from None
        digest = hashlib.sha256(data).hexdigest()
        if sha256 and sha256.lower() != digest:
            raise BoxCalibrationError(f"box calibration {path}: sha256 {digest}, expected {sha256}")
        try:
            spec = json.loads(data)
        except ValueError as error:
            raise BoxCalibrationError(f"box calibration {path}: not JSON ({error})") from None
        if not isinstance(spec, dict) or spec.get("format") != BOX_CALIBRATION_FORMAT:
            raise BoxCalibrationError(f"box calibration {path}: the format is not {BOX_CALIBRATION_FORMAT}")
        if spec.get("features") != BOX_CALIBRATION_FEATURES:
            raise BoxCalibrationError(f"box calibration {path}: features must be {BOX_CALIBRATION_FEATURES}")
        coef, intercept, clip = spec.get("coef"), spec.get("intercept"), spec.get("clip")
        numbers = [intercept, clip] + (list(coef) if isinstance(coef, list) else [])
        if not isinstance(coef, list) or len(coef) != 2 or any(isinstance(x, bool) or not isinstance(x, (int, float)) or x != x or abs(x) == float("inf") for x in numbers):
            raise BoxCalibrationError(f"box calibration {path}: intercept, coef (two numbers) and clip must be finite numbers")
        if any(c < 0 for c in coef):
            raise BoxCalibrationError(f"box calibration {path}: a coefficient below 0 is not monotone")
        if not 0 < clip < 0.5:
            raise BoxCalibrationError(f"box calibration {path}: clip must be above 0 and below 0.5")
        return cls(float(intercept), (float(coef[0]), float(coef[1])), float(clip), digest, str(spec.get("model") or ""))

    def probability(self, present: float, confidence: float) -> float:
        """The unrounded probability that the box is right."""

        def logit(p: float) -> float:
            p = min(1.0 - self.clip, max(self.clip, float(p)))
            return math.log(p / (1.0 - p))

        z = self.intercept + self.coef[0] * logit(present) + self.coef[1] * logit(confidence)
        return 1.0 / (1.0 + math.exp(-z)) if z >= 0 else math.exp(z) / (1.0 + math.exp(z))

    def __call__(self, present: float, confidence: float) -> float:
        return round2(self.probability(present, confidence))


def box_answer(answer: dict, calibration: BoxCalibration | None = None) -> dict:
    """A one-box answer (normalized units) in the format's shape. `confidence`: the calibrated probability that the box
    is right where there is a calibration, else `present`."""
    return {
        "type": "box",
        "box": {"top": answer["y1"], "bottom": answer["y2"], "left": answer["x1"], "right": answer["x2"]},
        "confidence": calibration(answer["present"], answer["confidence"]) if calibration is not None else answer["present"],
    }


def box_answers(body: dict, calibration: BoxCalibration | None = None) -> dict:
    """The response body with every one-box answer in the format's shape (a list answer, `max`, stays the runtime's)."""
    answers = body.get("answers")
    if isinstance(answers, dict):
        for qid, answer in answers.items():
            if isinstance(answer, dict) and answer.get("type") == "box" and "x1" in answer:
                answers[qid] = box_answer(answer, calibration)
    return body


# ---------------------------------------------------------------------------
# Errors with the capacity words
# ---------------------------------------------------------------------------

CONTEXT_MARKER = "maximum context length"
OPTIONS_MARKER = "options per choice"
LEVELS_MARKER = "a score takes 2 to 10 levels"
TOO_MANY_LEVELS = f"Too many score levels. Must have at most {MAX_LEVELS} levels."


def mark_capacity(error: RequestError) -> RequestError:
    """A refusal of /v1/systemone for a limit of the model, in the words and bodies benchmark harnesses read:

        over the length limit   {"detail": {"error_type": "max_tokens_exceeded", "message": "The request does not fit
                                the model's maximum context length (<the count and the limit>)."}}
        more than 255 options   {"detail": "Too many choices. Must have at most 255 choices (255 options per choice)."}
        more than 10 levels     {"detail": "Too many score levels. Must have at most 10 levels (a score takes 2 to 10
                                levels; one level is answered without the model)."}

    Every other refusal is returned as it is."""
    message = str(error)
    if error.status != 400:
        return error
    if error.error_type == "max_tokens_exceeded":
        text = f"The request does not fit the model's {CONTEXT_MARKER} ({message})."
        return RequestError(text, 400, error.error_type, detail={"error_type": error.error_type, "message": text})
    if message.startswith("Too many choices"):
        text = f"Too many choices. Must have at most {MAX_OPTIONS} choices ({MAX_OPTIONS} {OPTIONS_MARKER})."
        return RequestError(text, 400, error.error_type, detail=text)
    if message == TOO_MANY_LEVELS:
        text = f"Too many score levels. Must have at most {MAX_LEVELS} levels ({LEVELS_MARKER}; one level is answered without the model)."
        return RequestError(text, 400, error.error_type, detail=text)
    return error