libretro/scripts/campaign_rules.py

"""The campaign's fixed rule table: which verdicts settle which profile field.

Each rule is a pure function of verdicts read from core probe records (the
boot oracle of core_probe.py: boots, bios_error, crash, hang, unknown) and
a profile entry. It returns verdicts on profile fields. One with `apply` set
is settled by the table and written into the profile with the record ids as
`evidence:`. Any other goes to the review queue (measurements/findings.jsonl).

Three runs are compared for a file, all on the same content: `none` with no
BIOS at all, `present` with the file alone, and wrong files staged under its
name (zeros of its size, its first half, another region's dump).

  runs                                          field          settled
  none boots                                    hle_fallback   true
  none fails, present boots                     required       true
                                                hle_fallback   false
  zeros refused, present boots                  validation     a content check
  zeros used, present boots                     validation     no content check
  half refused, zeros used                      validation     size
  half used                                     validation     no size check
  opens the staged file under another path      path           that path
  opens the staged file under another name      aliases        name added
  a staged file the content should use, never   -              flagged only
  read
  copies of one file that boot and fail         -              queued

"Fails" is bios_error, crash or hang. A wrong file is refused when the
core reports a BIOS error, or when it boots without the BIOS screens the
right file shows (it fell back to a built-in BIOS). It is used when it
turns the run into a hang or a crash that the run without any file does not
show. Anything else settles nothing. Of validation, the table writes only a
missing `size`. A contradicted content check goes to review: which hash a
core computes is read in its code.

Standard library only.
"""

from __future__ import annotations

from collections.abc import Iterable, Sequence
from dataclasses import dataclass, field

BOOTS = "boots"
BIOS_ERROR = "bios_error"
CRASH = "crash"
HANG = "hang"
UNKNOWN = "unknown"
FAILS = (BIOS_ERROR, CRASH, HANG)
REFUSED = "refused"
USED = "used"
CONTENT_CHECKS = ("md5", "sha1", "sha256", "crc32", "adler32")

# A field an entry leaves out reads as this value.
FIELD_DEFAULTS: dict[str, object] = {"hle_fallback": False, "path": None, "aliases": [],
                                     "validation": []}
SETTLED_FIELDS = ("required", "hle_fallback", "path", "aliases", "validation")


@dataclass(frozen=True)
class Run:
    """The verdict of one variant of a record: what a rule reads."""

    verdict: str
    record_id: str
    label: str
    stages: tuple[str, ...] = ()

    @property
    def shows_bios(self) -> bool:
        return "bios" in self.stages


@dataclass(frozen=True)
class Verdict:
    rule: str
    field: str
    value: object
    statement: str
    evidence: tuple[str, ...]
    apply: bool = False


@dataclass(frozen=True)
class Opened:
    """Where a run found the staged file opened."""

    dest: str
    record_id: str
    extra: dict = field(default_factory=dict)


def variant(record: dict, label: str) -> dict:
    for item in record.get("variants") or ():
        if item.get("label") == label:
            return item
    raise KeyError(f"record {record.get('id')}: no variant {label!r}")


def run_of(record: dict, label: str) -> Run:
    """A variant's verdict as a rule reads it. A record from before the
    oracle has none, and reads as unknown."""
    item = variant(record, label)
    evidence = item.get("verdict_evidence") or {}
    return Run(item.get("verdict", UNKNOWN), record["id"], label, tuple(evidence.get("stages") or ()))


def ids(*runs: Run) -> tuple[str, ...]:
    return tuple(dict.fromkeys(run.record_id for run in runs))


def current(entry: dict, name: str) -> object:
    if name in entry:
        return entry[name]
    return FIELD_DEFAULTS.get(name)


def agrees(entry: dict, verdict: Verdict) -> bool:
    """True when the profile already says what the run showed."""
    if verdict.field == "aliases":
        return set(verdict.value or ()) <= set(entry.get("aliases") or ())
    if verdict.field == "validation":
        return verdict.rule == "validation_agrees"
    return current(entry, verdict.field) == verdict.value


def rule_need(none: Run, present: Run, entry: dict) -> list[Verdict]:
    """No BIOS at all against the file alone: required and hle_fallback."""
    name = entry["name"]
    if none.verdict == BOOTS:
        out = [Verdict("boots_without_file", "hle_fallback", True,
                       f"the core boots the content with no BIOS at all, {name} included",
                       ids(none), True)]
        if entry.get("required") is True:
            out.append(Verdict(
                "boots_without_file", "required", False,
                f"the profile says {name} is required, and the core boots without it. The table "
                f"settles hle_fallback from this run, not required", ids(none)))
        return out
    if none.verdict in FAILS and present.verdict == BOOTS:
        return [
            Verdict("fails_without_file", "required", True,
                    f"with no BIOS the run ends in {none.verdict}, with {name} alone it boots",
                    ids(none, present), True),
            Verdict("fails_without_file", "hle_fallback", False,
                    f"no stand-in for {name}: with no BIOS the run ends in {none.verdict}",
                    ids(none, present), True),
        ]
    return [Verdict("unsettled", "hle_fallback", None,
                    f"with no BIOS the run is {none.verdict} and with {name} alone {present.verdict}: "
                    f"no rule settles a field from that", ids(none, present))]


def file_use(wrong: Run, none: Run, present: Run) -> str | None:
    """Whether the core refused a wrong file or ran it, or None."""
    if present.verdict != BOOTS:
        return None
    if wrong.verdict == BIOS_ERROR:
        return REFUSED
    if wrong.verdict == BOOTS:
        if wrong.shows_bios:
            return USED
        return REFUSED if present.shows_bios else None
    if wrong.verdict in (HANG, CRASH) and none.verdict in (BOOTS, BIOS_ERROR):
        return USED
    return None


def declared_checks(entry: dict) -> tuple[set[str], bool]:
    """The checks the entry's validation names, and whether it is one list
    (a per-build dict is never edited by the table)."""
    value = entry.get("validation") or []
    if isinstance(value, dict):
        return {str(v) for items in value.values() for v in items or ()}, False
    return {str(v) for v in value}, True


def rule_validation(zeros: Run | None, half: Run | None, none: Run, present: Run,
                    entry: dict) -> Verdict | None:
    """Same-size and short wrong files against the right one: validation."""
    name = entry["name"]
    content = {REFUSED: True, USED: False}.get(file_use(zeros, none, present)) if zeros else None
    half_use = file_use(half, none, present) if half else None
    size: bool | None = None
    if half_use == USED:
        size, content = False, False if content is None else content
    elif half_use == REFUSED and content is False:
        size = True
    if content is None and size is None:
        return None
    declared, editable = declared_checks(entry)
    evidence = ids(*(run for run in (none, present, zeros, half) if run))
    said = sorted(declared)
    shown = []
    problems = []
    if content is not None:
        shown.append("checks the content" if content else "does not check the content")
        if content != bool(declared & set(CONTENT_CHECKS)):
            problems.append("a same-size file of zeros is " + ("refused" if content else "run")
                            + f", the profile declares validation {said}")
    missing_size = size is True and "size" not in declared
    if size is not None:
        shown.append("checks the size" if size else "does not check the size")
        if size is False and "size" in declared:
            problems.append(f"the first half of {name} is run, the profile declares a size check")
    statement = f"the core {' and '.join(shown)} of {name}"
    if problems:
        return Verdict("validation_contradicted", "validation", said,
                       f"{statement}: {'; '.join(problems)}", evidence)
    if missing_size:
        if not editable:
            return Verdict("validation_contradicted", "validation", said,
                           f"{statement}: the per-build validation has no size check", evidence)
        value = [*(entry.get("validation") or []), "size"]
        return Verdict("validation_size_added", "validation", value,
                       f"{statement}: half of the file is refused while zeros of its size are run",
                       evidence, True)
    return Verdict("validation_agrees", "validation", entry.get("validation") or [], statement,
                   evidence, editable)


def rule_region(dump: Run, present: Run, entry: dict, game_region: str | None) -> str:
    """What the core did with another region's dump under this file's name."""
    return (f"{dump.label} staged as {entry['name']} with a {game_region or 'region-less'} game: "
            f"{dump.verdict}, {entry['name']} itself: {present.verdict}")


def rule_path(entry: dict, declared: str, opened: Opened) -> Verdict | None:
    """The core opened the file at another place under the system directory."""
    if opened.dest == declared:
        return None
    return Verdict("opened_under_other_path", "path", opened.dest,
                   f"the core opened {entry['name']} as {opened.dest}, the profile declares "
                   f"{declared}", (opened.record_id,), True)


def rule_alias(entry: dict, opened: Opened) -> Verdict | None:
    """The core opened the file under a name the entry does not declare."""
    name = opened.dest.rsplit("/", 1)[-1]
    known = [entry["name"], *(entry.get("aliases") or ())]
    if name in known:
        return None
    return Verdict("opened_other_name", "aliases", [*(entry.get("aliases") or ()), name],
                   f"the core opened {entry['name']} under the name {name}",
                   (opened.record_id,), True)


def rule_never_opened(entry: dict, record_ids: Sequence[str], content: str) -> Verdict:
    return Verdict("never_opened", "opened", False,
                   f"{entry['name']} was staged and the core never opened it while running "
                   f"{content}", tuple(record_ids))


def rule_variants(runs: Iterable[Run], entry: dict) -> Verdict | None:
    """Copies of one file whose verdicts differ: a pack choice to review."""
    runs = list(runs)
    if len({run.verdict for run in runs}) < 2:
        return None
    shown = ", ".join(f"{run.label} {run.verdict}" for run in runs)
    return Verdict("variant_changes_verdict", "variants", sorted(run.label for run in runs),
                   f"the copies of {entry['name']} do not boot alike: {shown}", ids(*runs))


def merge_evidence(existing: dict | None, verdict: Verdict,
                   keep: object = lambda rid: True) -> dict:
    """The entry's evidence: map with the verdict's record ids added under its
    field. An id `keep` refuses is dropped from that field: a record from
    before the oracle no longer proves anything."""
    out = {key: list(value) for key, value in (existing or {}).items()}
    ids_ = [rid for rid in out.get(verdict.field, []) if keep(rid)]
    for rid in verdict.evidence:
        if rid not in ids_:
            ids_.append(rid)
    out[verdict.field] = ids_
    return {key: out[key] for key in sorted(out) if out[key]}